Restart scheduler on error

If the scheduler receives an error, it will never restart again since
`bgw_restart_time` is set to `BGW_NEVER_RESTART`, which will prevent
all jobs from executing.

This commit adds the GUC `timescaledb.bgw_scheduler_restart_time` that
can be set to the restart time for the scheduler. It defaults
to 60 seconds, which is the default restart interval for background
workers defined by PostgreSQL.

It also adds `timescaledb.debug_bgw_scheduler_exit_status` to be able
to shutdown the scheduler with a non-zero exit status, which allows the
restart functionality to be tested.

It also ensures that `backend_type` is explicitly set up rather than
copied from `application_name` and add some more information to
`application_name`. It also updates the tests to use `backend_type`
where applicable.

To avoid exhausting slots when the launcher restarts, it will kill all
existing schedulers and start new ones. Since background worker slots
are easily exhausted in the `bgw_launcher` test, we do not run it
repeatedly in the flakes workflow.

.github/gh_matrix_builder.py

@@ -53,6 +53,13 @@
     "memoize",
 }
 
+# Tests that we do not run as part of a Flake tests
+flaky_exclude_tests = {
+    # Not executed as a flake test since it easily exhausts available
+    # background worker slots.
+    "bgw_launcher",
+}
+
 
 # helper functions to generate matrix entries
 # the release and apache config inherit from the
@@ -309,11 +316,13 @@ def macos_config(overrides):
             sys.exit(1)
 
     if tests:
+        flake_tests = [t for t in tests if t not in flaky_exclude_tests]
+        installcheck_args = f'TESTS="{" ".join(flake_tests * 20)}"'
         m["include"].append(
             build_debug_config(
                 {
                     "coverage": False,
-                    "installcheck_args": f'TESTS="{" ".join(list(tests) * 20)}"',
+                    "installcheck_args": installcheck_args,
                     "name": "Flaky Check Debug",
                     "pg": PG16_LATEST,
                     "pginstallcheck": False,
@@ -324,7 +333,7 @@ def macos_config(overrides):
             build_debug_config(
                 {
                     "coverage": False,
-                    "installcheck_args": f'TESTS="{" ".join(list(tests) * 20)}"',
+                    "installcheck_args": installcheck_args,
                     "name": "Flaky Check Debug",
                     "pg": PG17_LATEST,
                     "pginstallcheck": False,

.unreleased/pr_7527

@@ -0,0 +1 @@
+Fixes: #7527 Restart scheduler on error

src/bgw/scheduler.c

@@ -795,9 +795,11 @@ ts_bgw_scheduler_process(int32 run_for_interval_ms,
 	 * exit. */
 	if (ts_guc_restoring || IsBinaryUpgrade)
 	{
-		elog(LOG,
-			 "scheduler for database %u exiting since the database is restoring or upgrading",
-			 MyDatabaseId);
+		ereport(LOG,
+				errmsg("scheduler for database %u exiting with exit status %d",
+					   MyDatabaseId,
+					   ts_debug_bgw_scheduler_exit_status),
+				errdetail("the database is restoring or upgrading"));
 		terminate_all_jobs_and_release_workers();
 		goto scheduler_exit;
 	}
@@ -866,7 +868,10 @@ ts_bgw_scheduler_process(int32 run_for_interval_ms,
 		MemoryContextReset(scratch_mctx);
 	}
 
-	elog(DEBUG1, "database scheduler for database %u exiting", MyDatabaseId);
+	elog(DEBUG1,
+		 "scheduler for database %u exiting with exit status %d",
+		 MyDatabaseId,
+		 ts_debug_bgw_scheduler_exit_status);
 
 #ifdef TS_DEBUG
 	if (ts_shutdown_bgw)
@@ -879,6 +884,7 @@ ts_bgw_scheduler_process(int32 run_for_interval_ms,
 	wait_for_all_jobs_to_shutdown();
 	check_for_stopped_and_timed_out_jobs();
 	scheduled_jobs = NIL;
+	proc_exit(ts_debug_bgw_scheduler_exit_status);
 }
 
 static void

src/compat/compat.h

@@ -655,6 +655,12 @@ RelationGetSmgr(Relation rel)
 	GenerationContextCreate(parent, name, blockSize)
 #endif
 
+#if PG16_GE
+#define pgstat_get_local_beentry_by_index_compat(idx) pgstat_get_local_beentry_by_index(idx)
+#else
+#define pgstat_get_local_beentry_by_index_compat(idx) pgstat_fetch_stat_local_beentry(idx)
+#endif
+
 /*
  * PG16 adds a new parameter to DefineIndex, total_parts, that takes
  * in the total number of direct and indirect partitions of the relation.

src/guc.c

@@ -183,6 +183,20 @@ bool ts_guc_debug_require_batch_sorted_merge = false;
 
 bool ts_guc_debug_allow_cagg_with_deprecated_funcs = false;
 
+/*
+ * Exit code for the scheduler.
+ *
+ * Normally it exits with a zero which means that it will not restart. If an
+ * error is raised, it exits with error code 1, which will trigger a
+ * restart.
+ *
+ * This variable exists to be able to trigger a restart for a normal exit,
+ * which is useful when debugging.
+ *
+ * See backend/postmaster/bgworker.c
+ */
+int ts_debug_bgw_scheduler_exit_status = 0;
+
 #ifdef TS_DEBUG
 bool ts_shutdown_bgw = false;
 char *ts_current_timestamp_mock = NULL;
@@ -1067,6 +1081,19 @@ _guc_init(void)
 							 /* assign_hook= */ NULL,
 							 /* show_hook= */ NULL);
 
+	DefineCustomIntVariable(/* name= */ MAKE_EXTOPTION("debug_bgw_scheduler_exit_status"),
+							/* short_desc= */ "exit status to use when shutting down the scheduler",
+							/* long_desc= */ "this is for debugging purposes",
+							/* valueAddr= */ &ts_debug_bgw_scheduler_exit_status,
+							/* bootValue= */ 0,
+							/* minValue= */ 0,
+							/* maxValue= */ 255,
+							/* context= */ PGC_SIGHUP,
+							/* flags= */ 0,
+							/* check_hook= */ NULL,
+							/* assign_hook= */ NULL,
+							/* show_hook= */ NULL);
+
 	DefineCustomStringVariable(/* name= */ MAKE_EXTOPTION("current_timestamp_mock"),
 							   /* short_desc= */ "set the current timestamp",
 							   /* long_desc= */ "this is for debugging purposes",

src/guc.h

@@ -92,6 +92,14 @@ extern TSDLLEXPORT bool ts_guc_auto_sparse_indexes;
 extern TSDLLEXPORT bool ts_guc_enable_columnarscan;
 extern TSDLLEXPORT int ts_guc_bgw_log_level;
 
+/*
+ * Exit code to use when scheduler exits.
+ *
+ * Mostly used for debugging, but defined also for non-debug builds since that
+ * simplifies the code (and also simplifies debugging non-debug builds).
+ */
+extern TSDLLEXPORT int ts_debug_bgw_scheduler_exit_status;
+
 #ifdef TS_DEBUG
 extern bool ts_shutdown_bgw;
 extern char *ts_current_timestamp_mock;

src/loader/bgw_launcher.c

@@ -84,6 +84,8 @@ typedef enum SchedulerState
 
 static volatile sig_atomic_t got_SIGHUP = false;
 
+int ts_guc_bgw_scheduler_restart_time_sec = BGW_DEFAULT_RESTART_INTERVAL;
+
 static void
 launcher_sighup(SIGNAL_ARGS)
 {
@@ -238,13 +240,27 @@ terminate_background_worker(BackgroundWorkerHandle *handle)
 }
 
 extern void
-ts_bgw_cluster_launcher_register(void)
+ts_bgw_cluster_launcher_init(void)
 {
 	BackgroundWorker worker;
 
+	DefineCustomIntVariable(/* name= */ MAKE_EXTOPTION("bgw_scheduler_restart_time"),
+							/* short_desc= */ "Restart time for scheduler in seconds",
+							/* long_desc= */
+							"The number of seconds until the scheduler restart on failure.",
+							/* valueAddr= */ &ts_guc_bgw_scheduler_restart_time_sec,
+							/* bootValue= */ BGW_DEFAULT_RESTART_INTERVAL,
+							/* minValue= */ 1,
+							/* maxValue= */ 3600,
+							/* context= */ PGC_SIGHUP,
+							/* flags= */ GUC_UNIT_S,
+							/* check_hook= */ NULL,
+							/* assign_hook= */ NULL,
+							/* show_hook= */ NULL);
+
 	memset(&worker, 0, sizeof(worker));
 	/* set up worker settings for our main worker */
-	snprintf(worker.bgw_name, BGW_MAXLEN, "TimescaleDB Background Worker Launcher");
+	snprintf(worker.bgw_name, BGW_MAXLEN, TS_BGW_TYPE_LAUNCHER);
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_restart_time = BGW_LAUNCHER_RESTART_TIME_S;
 
@@ -274,9 +290,10 @@ register_entrypoint_for_db(Oid db_id, VirtualTransactionId vxid, BackgroundWorke
 	BackgroundWorker worker;
 
 	memset(&worker, 0, sizeof(worker));
-	snprintf(worker.bgw_name, BGW_MAXLEN, "TimescaleDB Background Worker Scheduler");
+	snprintf(worker.bgw_type, BGW_MAXLEN, TS_BGW_TYPE_SCHEDULER);
+	snprintf(worker.bgw_name, BGW_MAXLEN, "%s for database %d", TS_BGW_TYPE_SCHEDULER, db_id);
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
-	worker.bgw_restart_time = BGW_NEVER_RESTART;
+	worker.bgw_restart_time = ts_guc_bgw_scheduler_restart_time_sec,
 	worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	snprintf(worker.bgw_library_name, BGW_MAXLEN, EXTENSION_NAME);
 	snprintf(worker.bgw_function_name, BGW_MAXLEN, BGW_ENTRYPOINT_FUNCNAME);
@@ -332,15 +349,89 @@ db_hash_entry_create_if_not_exists(HTAB *db_htab, Oid db_oid)
 	return db_he;
 }
 
+/*
+ * Result from signalling a backend.
+ *
+ * Error codes are non-zero, and success is zero.
+ */
+enum SignalBackendResult
+{
+	SIGNAL_BACKEND_SUCCESS = 0,
+	SIGNAL_BACKEND_ERROR,
+	SIGNAL_BACKEND_NOPERMISSION,
+	SIGNAL_BACKEND_NOSUPERUSER,
+};
+
+/*
+ * Terminate a background worker.
+ *
+ * This is copied from pg_signal_backend() in
+ * src/backend/storage/ipc/signalfuncs.c but tweaked to not require a database
+ * connection since the launcher does not have one.
+ */
+static enum SignalBackendResult
+ts_signal_backend(int pid, int sig)
+{
+	PGPROC *proc = BackendPidGetProc(pid);
+
+	if (unlikely(proc == NULL))
+	{
+		ereport(WARNING, (errmsg("PID %d is not a PostgreSQL backend process", pid)));
+		return SIGNAL_BACKEND_ERROR;
+	}
+
+	if (unlikely(kill(pid, sig)))
+	{
+		/* Again, just a warning to allow loops */
+		ereport(WARNING, (errmsg("could not send signal to process %d: %m", pid)));
+		return SIGNAL_BACKEND_ERROR;
+	}
+
+	return SIGNAL_BACKEND_SUCCESS;
+}
+
+/*
+ * Terminate backends by backend type.
+ *
+ * We iterate through all backends and mark those that match the given backend
+ * type as terminated.
+ *
+ * Note that there is potentially a delay between marking backends as
+ * terminated and their actual termination, so the backends have to be able to
+ * run even if there are multiple instances accessing the same data.
+ *
+ * Parts of this code is taken from pg_stat_get_activity() in
+ * src/backend/utils/adt/pgstatfuncs.c.
+ */
+static void
+terminate_backends_by_backend_type(const char *backend_type)
+{
+	Assert(backend_type);
+
+	const int num_backends = pgstat_fetch_stat_numbackends();
+	for (int curr_backend = 1; curr_backend <= num_backends; ++curr_backend)
+	{
+		const LocalPgBackendStatus *local_beentry =
+			pgstat_get_local_beentry_by_index_compat(curr_backend);
+		const PgBackendStatus *beentry = &local_beentry->backendStatus;
+		const char *bgw_type = GetBackgroundWorkerTypeByPid(beentry->st_procpid);
+		if (bgw_type && strcmp(backend_type, bgw_type) == 0)
+		{
+			int error = ts_signal_backend(beentry->st_procpid, SIGTERM);
+			if (error)
+				elog(LOG, "failed to terminate backend with pid %d", beentry->st_procpid);
+		}
+	}
+}
+
 /*
  * Model this on autovacuum.c -> get_database_list.
  *
- * Note that we are not doing
- * all the things around memory context that they do, because the hashtable
- * we're using to store db entries is automatically created in its own memory
- * context (a child of TopMemoryContext) This can get called at two different
- * times 1) when the cluster launcher starts and is looking for dbs and 2) if
- * it restarts due to a postmaster signal.
+ * Note that we are not doing all the things around memory context that they
+ * do, because the hashtable we're using to store db entries is automatically
+ * created in its own memory context (a child of TopMemoryContext) This can
+ * get called at two different times 1) when the cluster launcher starts and
+ * is looking for dbs and 2) if it restarts due to a postmaster signal.
  */
 static void
 populate_database_htab(HTAB *db_htab)
@@ -757,6 +848,16 @@ ts_bgw_cluster_launcher_main(PG_FUNCTION_ARGS)
 	db_htab = init_database_htab();
 	*htab_storage = db_htab;
 
+	/*
+	 * If the launcher was restarted and discovers old schedulers, these has
+	 * to be terminated to avoid exhausting the worker slots.
+	 *
+	 * We cannot easily pick up the old schedulers since we do not have access
+	 * to the slots array in PostgreSQL, so instead we scan for something that
+	 * looks like schedulers for databases, and kill them. New ones will then
+	 * be spawned below.
+	 */
+	terminate_backends_by_backend_type(TS_BGW_TYPE_SCHEDULER);
 	populate_database_htab(db_htab);
 
 	while (true)

src/loader/bgw_launcher.h

@@ -8,7 +8,12 @@
 #include <postgres.h>
 #include <fmgr.h>
 
-extern void ts_bgw_cluster_launcher_register(void);
+#define TS_BGW_TYPE_LAUNCHER "TimescaleDB Background Worker Launcher"
+#define TS_BGW_TYPE_SCHEDULER "TimescaleDB Background Worker Scheduler"
+
+extern int ts_guc_bgw_scheduler_restart_time_sec;
+
+extern void ts_bgw_cluster_launcher_init(void);
 
 /*called by postmaster at launcher bgw startup*/
 TSDLLEXPORT extern Datum ts_bgw_cluster_launcher_main(PG_FUNCTION_ARGS);

src/loader/loader.c

@@ -591,7 +591,7 @@ _PG_init(void)
 	timescaledb_shmem_request_hook();
 #endif
 
-	ts_bgw_cluster_launcher_register();
+	ts_bgw_cluster_launcher_init();
 	ts_bgw_counter_setup_gucs();
 	ts_bgw_interface_register_api_version();