Track logrep apply workers' last start times to avoid useless waits.

Enforce wal_retrieve_retry_interval on a per-subscription basis,
rather than globally, and arrange to skip that delay in case of
an intentional worker exit.  This probably makes little difference
in the field, where apply workers wouldn't be restarted often;
but it has a significant impact on the runtime of our logical
replication regression tests (even though those tests use
artificially-small wal_retrieve_retry_interval settings already).

Nathan Bossart, with mostly-cosmetic editorialization by me

Discussion: https://postgr.es/m/20221122004119.GA132961@nathanxps13

1 files changed, 20 insertions, 0 deletions

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a0084c7ef69..cfb2ab62481 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -174,6 +174,7 @@
 #include "postmaster/walwriter.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
+#include "replication/logicallauncher.h"
 #include "replication/logicalproto.h"
 #include "replication/logicalrelation.h"
 #include "replication/logicalworker.h"
@@ -3811,6 +3812,15 @@ apply_worker_exit(void)
 		return;
 	}
 
+	/*
+	 * Reset the last-start time for this apply worker so that the launcher
+	 * will restart it without waiting for wal_retrieve_retry_interval if the
+	 * subscription is still active, and so that we won't leak that hash table
+	 * entry if it isn't.
+	 */
+	if (!am_tablesync_worker())
+		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
+
 	proc_exit(0);
 }
 
@@ -3851,6 +3861,9 @@ maybe_reread_subscription(void)
 				(errmsg("%s for subscription \"%s\" will stop because the subscription was removed",
 						get_worker_name(), MySubscription->name)));
 
+		/* Ensure we remove no-longer-useful entry for worker's start time */
+		if (!am_tablesync_worker() && !am_parallel_apply_worker())
+			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 		proc_exit(0);
 	}
 
@@ -4421,6 +4434,9 @@ InitializeApplyWorker(void)
 				(errmsg("%s for subscription %u will not start because the subscription was removed during startup",
 						get_worker_name(), MyLogicalRepWorker->subid)));
 
+		/* Ensure we remove no-longer-useful entry for worker's start time */
+		if (!am_tablesync_worker() && !am_parallel_apply_worker())
+			ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 		proc_exit(0);
 	}
 
@@ -4678,6 +4694,10 @@ DisableSubscriptionAndExit(void)
 	DisableSubscription(MySubscription->oid);
 	CommitTransactionCommand();
 
+	/* Ensure we remove no-longer-useful entry for worker's start time */
+	if (!am_tablesync_worker() && !am_parallel_apply_worker())
+		ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
+
 	/* Notify the subscription has been disabled and exit */
 	ereport(LOG,
 			errmsg("subscription \"%s\" has been disabled because of an error",