workshop: add plan option "notify_progress"

Author: MaxKellermann

debian/changelog

@@ -1,6 +1,6 @@
 cm4all-workshop (7.3) unstable; urgency=low
 
-  * 
+  * workshop: add plan option "notify_progress"
 
  --   
 

doc/index.rst

@@ -315,6 +315,10 @@ The following options are available:
 
   The ``control_channel`` option is allowed, but not ``allow_spawn``.
 
+* :samp:`notify_progress`: send a PostgreSQL notify
+  ``job_progress:PLAN_NAME`` after the ``progress`` column of a job
+  was updated.
+
 * :samp:`control_channel`: see `Control Channel`_.
 
 .. _allow_spawn:

src/workshop/Job.cxx

@@ -6,9 +6,9 @@
 #include "Queue.hxx"
 
 bool
-WorkshopJob::SetProgress(unsigned progress, const char *timeout) noexcept
+WorkshopJob::SetProgress(unsigned progress, const char *timeout, bool notify) noexcept
 {
-	return queue.SetJobProgress(*this, progress, timeout);
+	return queue.SetJobProgress(*this, progress, timeout, notify);
 }
 
 void

src/workshop/Job.hxx

@@ -39,7 +39,7 @@ struct WorkshopJob {
 	 * string that is understood by PostgreSQL)
 	 * @return true on success
 	 */
-	bool SetProgress(unsigned progress, const char *timeout) noexcept;
+	bool SetProgress(unsigned progress, const char *timeout, bool notify) noexcept;
 
 	/**
 	 * Add more environment variables to the record in the "jobs"

src/workshop/Operator.cxx

@@ -409,7 +409,7 @@ WorkshopOperator::OnProgress(unsigned progress) noexcept
 		   LogBridge::Flush() call */
 		return;
 
-	job.SetProgress(progress, plan->timeout.c_str());
+	job.SetProgress(progress, plan->timeout.c_str(), plan->notify_progress);
 
 	/* refresh the timeout */
 	ScheduleTimeout();

src/workshop/Plan.hxx

@@ -66,6 +66,8 @@ struct Plan {
 
 	bool translate = false;
 
+	bool notify_progress = false;
+
 	Plan() = default;
 
 	Plan(Plan &&) = default;

src/workshop/PlanLoader.cxx

@@ -71,6 +71,8 @@ PlanLoader::ParseLine(FileLineParser &line)
 		seen_exec_option = true;
 	} else if (StringIsEqual(key, "translate")) {
 		plan.translate = true;
+	} else if (StringIsEqual(key, "notify_progress")) {
+		plan.notify_progress = true;
 	} else if (StringIsEqual(key, "control_channel")) {
 		/* previously, the "yes"/"no" parameter was mandatory, but
 		   that's deprecated since 2.0.36 */

src/workshop/Queue.cxx

@@ -33,6 +33,7 @@ WorkshopQueue::WorkshopQueue(const Logger &parent_logger,
 	 db(event_loop, std::move(_db_config), *this),
 	 check_notify_event(event_loop, BIND_THIS_METHOD(CheckNotify)),
 	 timer_event(event_loop, BIND_THIS_METHOD(OnTimer)),
+	 progress_notify_timer(event_loop, BIND_THIS_METHOD(OnProgressNotifyTimer)),
 	 handler(_handler)
 {
 }
@@ -48,6 +49,19 @@ WorkshopQueue::OnTimer() noexcept
 	Run();
 }
 
+void
+WorkshopQueue::OnProgressNotifyTimer() noexcept
+{
+	try {
+		for (std::string_view plan_name : progress_notify_plans)
+			db.Execute(fmt::format("NOTIFY \"job_progress:{}\"", plan_name).c_str());
+	} catch (...) {
+		db.CheckError(std::current_exception());
+	}
+
+	progress_notify_plans.clear();
+}
+
 bool
 WorkshopQueue::GetNextScheduled(int *span_r)
 {
@@ -345,14 +359,19 @@ WorkshopQueue::CheckRateLimit(const char *plan_name,
 
 bool
 WorkshopQueue::SetJobProgress(const WorkshopJob &job, unsigned progress,
-			      const char *timeout) noexcept
+			      const char *timeout, bool notify) noexcept
 {
 	assert(&job.queue == this);
 
 	logger(5, "job ", job.id, " progress=", progress);
 
 	ScheduleCheckNotify();
 
+	if (notify)
+		if (auto [it, inserted] = progress_notify_plans.emplace(job.plan_name);
+		    inserted && !progress_notify_timer.IsPending())
+			progress_notify_timer.Schedule(std::chrono::milliseconds{250});
+
 	try {
 		pg_set_job_progress(db, job.id.c_str(), progress, timeout);
 		return true;
@@ -497,6 +516,7 @@ WorkshopQueue::OnDisconnect() noexcept
 {
 	logger(4, "disconnected from database");
 
+	progress_notify_timer.Cancel();
 	timer_event.Cancel();
 	check_notify_event.Cancel();
 }

src/workshop/Queue.hxx

@@ -9,6 +9,7 @@
 #include "pg/AsyncConnection.hxx"
 #include "io/Logger.hxx"
 
+#include <set>
 #include <string>
 #include <chrono>
 
@@ -72,6 +73,19 @@ class WorkshopQueue final : private Pg::AsyncConnectionHandler {
 	 */
 	FineTimerEvent timer_event;
 
+	/**
+	 * This timer delays the submission of progress notifies by
+	 * some time to allow merging several notifies into one.
+	 */
+	FineTimerEvent progress_notify_timer;
+
+	/**
+	 * The list of plan names that have seen progress updates, to
+	 * be handled by #progress_notify_timer /
+	 * OnProgressNotifyTimer().
+	 */
+	std::set<std::string, std::less<>> progress_notify_plans;
+
 	std::string plans_include, plans_exclude, plans_lowprio;
 	std::chrono::steady_clock::time_point next_expire_check =
 		std::chrono::steady_clock::time_point::min();
@@ -147,10 +161,11 @@ public:
 					    unsigned max_count) noexcept;
 
 	/**
+	 * @param notify send a PostgreSQL NOTIFY?
 	 * @return true on success
 	 */
 	bool SetJobProgress(const WorkshopJob &job, unsigned progress,
-			    const char *timeout) noexcept;
+			    const char *timeout, bool notify) noexcept;
 
 	void SetJobEnv(const WorkshopJob &job, const char *more_env);
 
@@ -197,6 +212,8 @@ private:
 		timer_event.Schedule(d);
 	}
 
+	void OnProgressNotifyTimer() noexcept;
+
 	/**
 	 * Schedule a queue run.  It will occur "very soon" (in a few
 	 * milliseconds).