Subject: Synchronizing slots from primary to standby

From: Peter Eisentraut <[email-masked]>
Date: Sun, 31 Oct 2021 11:08:18 +0100

I want to reactivate $subject.  I took Petr Jelinek's patch from [0], 
rebased it, added a bit of testing.  It basically works, but as 
mentioned in [0], there are various issues to work out.

The idea is that the standby runs a background worker to periodically 
fetch replication slot information from the primary.  On failover, a 
logical subscriber would then ideally find up-to-date replication slots 
on the new publisher and can just continue normally.

The previous thread didn't have a lot of discussion, but I have gathered 
from off-line conversations that there is a wider agreement on this 
approach.  So the next steps would be to make it more robust and 
configurable and documented.  As I said, I added a small test case to 
show that it works at all, but I think a lot more tests should be added. 
  I have also found that this breaks some seemingly unrelated tests in 
the recovery test suite.  I have disabled these here.  I'm not sure if 
the patch actually breaks anything or if these are just differences in 
timing or implementation dependencies.  This patch adds a LIST_SLOTS 
replication command, but I think this could be replaced with just a 
SELECT FROM pg_replication_slots query now.  (This patch is originally 
older than when you could run SELECT queries over the replication protocol.)

So, again, this isn't anywhere near ready, but there is already a lot 
here to gather feedback about how it works, how it should work, how to 
configure it, and how it fits into an overall replication and HA 
architecture.

[0]: 
https://www.postgresql.org/message-id/flat/3095349b-44d4-bf11-1b33-7eefb585d578%402ndquadrant.com
From 3b3c3fa1d1e92d6b39ab0c869cb9398bf7791d48 Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <[email-masked]>
Date: Sun, 31 Oct 2021 10:49:29 +0100
Subject: [PATCH v1] Synchronize logical replication slots from primary to
 standby

---
 src/backend/commands/subscriptioncmds.c       |   4 +-
 src/backend/postmaster/bgworker.c             |   3 +
 .../libpqwalreceiver/libpqwalreceiver.c       |  73 ++++
 src/backend/replication/logical/Makefile      |   1 +
 src/backend/replication/logical/launcher.c    | 199 +++++++----
 src/backend/replication/logical/slotsync.c    | 311 ++++++++++++++++++
 src/backend/replication/logical/tablesync.c   |  13 +-
 src/backend/replication/repl_gram.y           |  13 +-
 src/backend/replication/repl_scanner.l        |   1 +
 src/backend/replication/slotfuncs.c           |   2 +-
 src/backend/replication/walsender.c           | 169 +++++++++-
 src/backend/utils/activity/wait_event.c       |   3 +
 src/include/commands/subscriptioncmds.h       |   3 +
 src/include/nodes/nodes.h                     |   1 +
 src/include/nodes/replnodes.h                 |   8 +
 src/include/replication/logicalworker.h       |   1 +
 src/include/replication/slot.h                |   4 +-
 src/include/replication/walreceiver.h         |  16 +
 src/include/replication/worker_internal.h     |   8 +-
 src/include/utils/wait_event.h                |   1 +
 src/test/recovery/t/007_sync_rep.pl           |   3 +-
 .../t/010_logical_decoding_timelines.pl       |   3 +-
 src/test/recovery/t/030_slot_sync.pl          |  51 +++
 23 files changed, 819 insertions(+), 72 deletions(-)
 create mode 100644 src/backend/replication/logical/slotsync.c
 create mode 100644 src/test/recovery/t/030_slot_sync.pl

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index c47ba26369..2bab813440 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -743,7 +743,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
 
 				RemoveSubscriptionRel(sub->oid, relid);
 
-				logicalrep_worker_stop(sub->oid, relid);
+				logicalrep_worker_stop(MyDatabaseId, sub->oid, relid);
 
 				/*
 				 * For READY state, we would have already dropped the
@@ -1244,7 +1244,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	{
 		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
 
-		logicalrep_worker_stop(w->subid, w->relid);
+		logicalrep_worker_stop(w->dbid, w->subid, w->relid);
 	}
 	list_free(subworkers);
 
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index c05f500639..818b8a35e9 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -128,6 +128,9 @@ static const struct
 	},
 	{
 		"ApplyWorkerMain", ApplyWorkerMain
+	},
+	{
+		"ReplSlotSyncMain", ReplSlotSyncMain
 	}
 };
 
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 5c6e56a5b2..5d2871eb08 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -58,6 +58,7 @@ static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
 									char **sender_host, int *sender_port);
 static char *libpqrcv_identify_system(WalReceiverConn *conn,
 									  TimeLineID *primary_tli);
+static List *libpqrcv_list_slots(WalReceiverConn *conn);
 static int	libpqrcv_server_version(WalReceiverConn *conn);
 static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
 											 TimeLineID tli, char **filename,
@@ -89,6 +90,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
 	libpqrcv_get_conninfo,
 	libpqrcv_get_senderinfo,
 	libpqrcv_identify_system,
+	libpqrcv_list_slots,
 	libpqrcv_server_version,
 	libpqrcv_readtimelinehistoryfile,
 	libpqrcv_startstreaming,
@@ -385,6 +387,77 @@ libpqrcv_server_version(WalReceiverConn *conn)
 	return PQserverVersion(conn->streamConn);
 }
 
+/*
+ * Get list of slots from primary.
+ */
+static List *
+libpqrcv_list_slots(WalReceiverConn *conn)
+{
+	PGresult   *res;
+	int			i;
+	List	   *slots = NIL;
+	int			ntuples;
+	WalRecvReplicationSlotData *slot_data;
+
+	res = libpqrcv_PQexec(conn->streamConn, "LIST_SLOTS");
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("could not receive list of slots the primary server: %s",
+						pchomp(PQerrorMessage(conn->streamConn)))));
+	}
+	if (PQnfields(res) < 10)
+	{
+		int			nfields = PQnfields(res);
+
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("invalid response from primary server"),
+				 errdetail("Could not get list of slots: got %d fields, expected %d or more fields.",
+						   nfields, 10)));
+	}
+
+	ntuples = PQntuples(res);
+	for (i = 0; i < ntuples; i++)
+	{
+		char   *slot_type;
+
+		slot_data = palloc0(sizeof(WalRecvReplicationSlotData));
+		namestrcpy(&slot_data->name, PQgetvalue(res, i, 0));
+		if (!PQgetisnull(res, i, 1))
+			namestrcpy(&slot_data->plugin, PQgetvalue(res, i, 1));
+		slot_type = PQgetvalue(res, i, 2);
+		if (!PQgetisnull(res, i, 3))
+			slot_data->database = atooid(PQgetvalue(res, i, 3));
+		if (strcmp(slot_type, "physical") == 0)
+		{
+			if (OidIsValid(slot_data->database))
+				elog(ERROR, "unexpected physical replication slot with database set");
+		}
+		if (pg_strtoint32(PQgetvalue(res, i, 5)) == 1)
+			slot_data->persistency = RS_TEMPORARY;
+		else
+			slot_data->persistency = RS_PERSISTENT;
+		if (!PQgetisnull(res, i, 6))
+			slot_data->xmin = atooid(PQgetvalue(res, i, 6));
+		if (!PQgetisnull(res, i, 7))
+			slot_data->catalog_xmin = atooid(PQgetvalue(res, i, 7));
+		if (!PQgetisnull(res, i, 8))
+			slot_data->restart_lsn = pg_strtouint64(PQgetvalue(res, i, 8),
+													NULL, 10);
+		if (!PQgetisnull(res, i, 9))
+			slot_data->confirmed_flush = pg_strtouint64(PQgetvalue(res, i, 9),
+														NULL, 10);
+
+		slots = lappend(slots, slot_data);
+	}
+
+	PQclear(res);
+
+	return slots;
+}
+
 /*
  * Start streaming WAL data from given streaming options.
  *
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb71..bc3f23b5a2 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -24,6 +24,7 @@ OBJS = \
 	proto.o \
 	relation.o \
 	reorderbuffer.o \
+	slotsync.o \
 	snapbuild.o \
 	tablesync.o \
 	worker.o
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 3fb4caa803..4919f74ef5 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -22,6 +22,7 @@
 #include "access/htup_details.h"
 #include "access/tableam.h"
 #include "access/xact.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
 #include "funcapi.h"
@@ -212,7 +213,7 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
  * subscription id and relid.
  */
 LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running)
 {
 	int			i;
 	LogicalRepWorker *res = NULL;
@@ -224,8 +225,8 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->in_use && w->subid == subid && w->relid == relid &&
-			(!only_running || w->proc))
+		if (w->in_use && w->dbid == dbid && w->subid == subid &&
+			w->relid == relid && (!only_running || w->proc))
 		{
 			res = w;
 			break;
@@ -275,9 +276,13 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	int			nsyncworkers;
 	TimestampTz now;
 
-	ereport(DEBUG1,
-			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
-							 subname)));
+	if (OidIsValid(subid))
+		ereport(DEBUG1,
+				(errmsg_internal("starting logical replication worker for subscription \"%s\"",
+								 subname)));
+	else
+		ereport(DEBUG1,
+				(errmsg("starting replication slot synchronization worker")));
 
 	/* Report this after the initial starting message for consistency. */
 	if (max_replication_slots == 0)
@@ -314,7 +319,9 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	 * reason we do this is because if some worker failed to start up and its
 	 * parent has crashed while waiting, the in_use state was never cleared.
 	 */
-	if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
+	if (worker == NULL ||
+		(OidIsValid(relid) &&
+		 nsyncworkers >= max_sync_workers_per_subscription))
 	{
 		bool		did_cleanup = false;
 
@@ -348,7 +355,7 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	 * silently as we might get here because of an otherwise harmless race
 	 * condition.
 	 */
-	if (nsyncworkers >= max_sync_workers_per_subscription)
+	if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return;
@@ -395,15 +402,22 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	memset(&bgw, 0, sizeof(bgw));
 	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
 		BGWORKER_BACKEND_DATABASE_CONNECTION;
-	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+	bgw.bgw_start_time = BgWorkerStart_ConsistentState;
 	snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
-	snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+	if (OidIsValid(subid))
+		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+	else
+		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ReplSlotSyncMain");
 	if (OidIsValid(relid))
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
 				 "logical replication worker for subscription %u sync %u", subid, relid);
-	else
+	else if (OidIsValid(subid))
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
 				 "logical replication worker for subscription %u", subid);
+	else
+		snprintf(bgw.bgw_name, BGW_MAXLEN,
+				 "replication slot synchronization worker");
+
 	snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
 
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
@@ -434,14 +448,14 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
  * it detaches from the slot.
  */
 void
-logicalrep_worker_stop(Oid subid, Oid relid)
+logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid)
 {
 	LogicalRepWorker *worker;
 	uint16		generation;
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-	worker = logicalrep_worker_find(subid, relid, false);
+	worker = logicalrep_worker_find(dbid, subid, relid, false);
 
 	/* No worker, nothing to do. */
 	if (!worker)
@@ -531,13 +545,13 @@ logicalrep_worker_stop(Oid subid, Oid relid)
  * Wake up (using latch) any logical replication worker for specified sub/rel.
  */
 void
-logicalrep_worker_wakeup(Oid subid, Oid relid)
+logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid)
 {
 	LogicalRepWorker *worker;
 
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-	worker = logicalrep_worker_find(subid, relid, true);
+	worker = logicalrep_worker_find(dbid, subid, relid, true);
 
 	if (worker)
 		logicalrep_worker_wakeup_ptr(worker);
@@ -714,7 +728,7 @@ ApplyLauncherRegister(void)
 	memset(&bgw, 0, sizeof(bgw));
 	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
 		BGWORKER_BACKEND_DATABASE_CONNECTION;
-	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+	bgw.bgw_start_time = BgWorkerStart_ConsistentState;
 	snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
 	snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
 	snprintf(bgw.bgw_name, BGW_MAXLEN,
@@ -795,6 +809,113 @@ ApplyLauncherWakeup(void)
 		kill(LogicalRepCtx->launcher_pid, SIGUSR1);
 }
 
+static void
+ApplyLauncherStartSlotSync(TimestampTz *last_start_time, long *wait_time)
+{
+	WalReceiverConn *wrconn;
+	TimestampTz now;
+	char	   *err;
+	List	   *slots;
+	ListCell   *lc;
+	MemoryContext tmpctx;
+	MemoryContext oldctx;
+
+	wrconn = walrcv_connect(PrimaryConnInfo, false,
+							"Logical Replication Launcher", &err);
+	if (!wrconn)
+		ereport(ERROR,
+				(errmsg("could not connect to the primary server: %s", err)));
+
+	/* Use temporary context for the slot list and worker info. */
+	tmpctx = AllocSetContextCreate(TopMemoryContext,
+									"Logical Replication Launcher slot sync ctx",
+									ALLOCSET_DEFAULT_SIZES);
+	oldctx = MemoryContextSwitchTo(tmpctx);
+
+	slots = walrcv_list_slots(wrconn);
+
+	now = GetCurrentTimestamp();
+
+	foreach (lc, slots)
+	{
+		WalRecvReplicationSlotData *slot_data = lfirst(lc);
+		LogicalRepWorker *w;
+
+		if (!OidIsValid(slot_data->database))
+			continue;
+
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+		w = logicalrep_worker_find(slot_data->database, InvalidOid,
+								   InvalidOid, false);
+		LWLockRelease(LogicalRepWorkerLock);
+
+		if (w == NULL)
+		{
+			*last_start_time = now;
+			*wait_time = wal_retrieve_retry_interval;
+
+			logicalrep_worker_launch(slot_data->database, InvalidOid, NULL,
+									 BOOTSTRAP_SUPERUSERID, InvalidOid);
+		}
+	}
+
+	/* Switch back to original memory context. */
+	MemoryContextSwitchTo(oldctx);
+	/* Clean the temporary memory. */
+	MemoryContextDelete(tmpctx);
+
+	walrcv_disconnect(wrconn);
+}
+
+static void
+ApplyLauncherStartSubs(TimestampTz *last_start_time, long *wait_time)
+{
+	TimestampTz now;
+	List	   *sublist;
+	ListCell   *lc;
+	MemoryContext subctx;
+	MemoryContext oldctx;
+
+	now = GetCurrentTimestamp();
+
+	/* Use temporary context for the database list and worker info. */
+	subctx = AllocSetContextCreate(TopMemoryContext,
+								   "Logical Replication Launcher sublist",
+								   ALLOCSET_DEFAULT_SIZES);
+	oldctx = MemoryContextSwitchTo(subctx);
+
+	/* search for subscriptions to start or stop. */
+	sublist = get_subscription_list();
+
+	/* Start the missing workers for enabled subscriptions. */
+	foreach(lc, sublist)
+	{
+		Subscription *sub = (Subscription *) lfirst(lc);
+		LogicalRepWorker *w;
+
+		if (!sub->enabled)
+			continue;
+
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+		w = logicalrep_worker_find(sub->dbid, sub->oid, InvalidOid, false);
+		LWLockRelease(LogicalRepWorkerLock);
+
+		if (w == NULL)
+		{
+			*last_start_time = now;
+			*wait_time = wal_retrieve_retry_interval;
+
+			logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
+									 sub->owner, InvalidOid);
+		}
+	}
+
+	/* Switch back to original memory context. */
+	MemoryContextSwitchTo(oldctx);
+	/* Clean the temporary memory. */
+	MemoryContextDelete(subctx);
+}
+
 /*
  * Main loop for the apply launcher process.
  */
@@ -822,14 +943,12 @@ ApplyLauncherMain(Datum main_arg)
 	 */
 	BackgroundWorkerInitializeConnection(NULL, NULL, 0);
 
+	load_file("libpqwalreceiver", false);
+
 	/* Enter main loop */
 	for (;;)
 	{
 		int			rc;
-		List	   *sublist;
-		ListCell   *lc;
-		MemoryContext subctx;
-		MemoryContext oldctx;
 		TimestampTz now;
 		long		wait_time = DEFAULT_NAPTIME_PER_CYCLE;
 
@@ -841,42 +960,10 @@ ApplyLauncherMain(Datum main_arg)
 		if (TimestampDifferenceExceeds(last_start_time, now,
 									   wal_retrieve_retry_interval))
 		{
-			/* Use temporary context for the database list and worker info. */
-			subctx = AllocSetContextCreate(TopMemoryContext,
-										   "Logical Replication Launcher sublist",
-										   ALLOCSET_DEFAULT_SIZES);
-			oldctx = MemoryContextSwitchTo(subctx);
-
-			/* search for subscriptions to start or stop. */
-			sublist = get_subscription_list();
-
-			/* Start the missing workers for enabled subscriptions. */
-			foreach(lc, sublist)
-			{
-				Subscription *sub = (Subscription *) lfirst(lc);
-				LogicalRepWorker *w;
-
-				if (!sub->enabled)
-					continue;
-
-				LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-				w = logicalrep_worker_find(sub->oid, InvalidOid, false);
-				LWLockRelease(LogicalRepWorkerLock);
-
-				if (w == NULL)
-				{
-					last_start_time = now;
-					wait_time = wal_retrieve_retry_interval;
-
-					logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
-											 sub->owner, InvalidOid);
-				}
-			}
-
-			/* Switch back to original memory context. */
-			MemoryContextSwitchTo(oldctx);
-			/* Clean the temporary memory. */
-			MemoryContextDelete(subctx);
+			if (!RecoveryInProgress())
+				ApplyLauncherStartSubs(&last_start_time, &wait_time);
+			else
+				ApplyLauncherStartSlotSync(&last_start_time, &wait_time);
 		}
 		else
 		{
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
new file mode 100644
index 0000000000..9b1e56f6d8
--- /dev/null
+++ b/src/backend/replication/logical/slotsync.c
@@ -0,0 +1,311 @@
+/*-------------------------------------------------------------------------
+ * slotsync.c
+ *	   PostgreSQL worker for synchronizing slots to a standby from primary
+ *
+ * Copyright (c) 2016-2018, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/slotsync.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "commands/dbcommands.h"
+#include "pgstat.h"
+#include "postmaster/bgworker.h"
+#include "replication/logicalworker.h"
+#include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "storage/procarray.h"
+#include "utils/builtins.h"
+#include "utils/pg_lsn.h"
+
+
+/*
+ * Wait for remote slot to pass localy reserved position.
+ */
+static void
+wait_for_primary_slot_catchup(WalReceiverConn *wrconn, char *slot_name,
+							  XLogRecPtr min_lsn)
+{
+	WalRcvExecResult *res;
+	TupleTableSlot *slot;
+	Oid				slotRow[1] = {LSNOID};
+	StringInfoData	cmd;
+	bool			isnull;
+	XLogRecPtr		restart_lsn;
+
+	for (;;)
+	{
+		int	rc;
+
+		CHECK_FOR_INTERRUPTS();
+
+		initStringInfo(&cmd);
+		appendStringInfo(&cmd,
+						 "SELECT restart_lsn"
+						 "  FROM pg_catalog.pg_replication_slots"
+						 " WHERE slot_name = %s",
+						 quote_literal_cstr(slot_name));
+		res = walrcv_exec(wrconn, cmd.data, 1, slotRow);
+
+		if (res->status != WALRCV_OK_TUPLES)
+			ereport(ERROR,
+					(errmsg("could not fetch slot info for slot \"%s\" from primary: %s",
+							slot_name, res->err)));
+
+		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+		if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+			ereport(ERROR,
+					(errmsg("slot \"%s\" disapeared from provider",
+							slot_name)));
+
+		restart_lsn = DatumGetLSN(slot_getattr(slot, 1, &isnull));
+		Assert(!isnull);
+
+		ExecClearTuple(slot);
+		walrcv_clear_result(res);
+
+		if (restart_lsn >= min_lsn)
+			break;
+
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   wal_retrieve_retry_interval,
+					   WAIT_EVENT_REPL_SLOT_SYNC_MAIN);
+
+		ResetLatch(MyLatch);
+
+		/* emergency bailout if postmaster has died */
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+	}
+}
+
+/*
+ * Synchronize single slot to given position.
+ *
+ * This optionally creates new slot if there is no existing one.
+ */
+static void
+synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database,
+					 char *plugin_name, XLogRecPtr target_lsn)
+{
+	int			i;
+	bool		found = false;
+	XLogRecPtr	endlsn;
+
+	/* Search for the named slot and mark it active if we find it. */
+	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+	for (i = 0; i < max_replication_slots; i++)
+	{
+		ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+		if (!s->in_use)
+			continue;
+
+		if (strcmp(NameStr(s->data.name), slot_name) == 0)
+		{
+			found = true;
+			break;
+		}
+	}
+	LWLockRelease(ReplicationSlotControlLock);
+
+	StartTransactionCommand();
+
+	/* Already existing slot, acquire */
+	if (found)
+	{
+		ReplicationSlotAcquire(slot_name, true);
+
+		if (target_lsn < MyReplicationSlot->data.confirmed_flush)
+		{
+			elog(DEBUG1,
+				 "not synchronizing slot %s; synchronization would move it backward",
+				 slot_name);
+
+			ReplicationSlotRelease();
+			CommitTransactionCommand();
+			return;
+		}
+	}
+	/* Otherwise create the slot first. */
+	else
+	{
+		TransactionId xmin_horizon = InvalidTransactionId;
+		ReplicationSlot	   *slot;
+
+		ReplicationSlotCreate(slot_name, true, RS_EPHEMERAL, false);
+		slot = MyReplicationSlot;
+
+		SpinLockAcquire(&slot->mutex);
+		slot->data.database = get_database_oid(database, false);
+		namestrcpy(&slot->data.plugin, plugin_name);
+		SpinLockRelease(&slot->mutex);
+
+		ReplicationSlotReserveWal();
+
+		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+		xmin_horizon = GetOldestSafeDecodingTransactionId(true);
+		slot->effective_catalog_xmin = xmin_horizon;
+		slot->data.catalog_xmin = xmin_horizon;
+		ReplicationSlotsComputeRequiredXmin(true);
+		LWLockRelease(ProcArrayLock);
+
+		if (target_lsn < MyReplicationSlot->data.restart_lsn)
+		{
+			elog(LOG, "waiting for remote slot %s lsn (%X/%X) to pass local slot lsn (%X/%X)",
+				 slot_name,
+				 (uint32) (target_lsn >> 32),
+				 (uint32) (target_lsn),
+				 (uint32) (MyReplicationSlot->data.restart_lsn >> 32),
+				 (uint32) (MyReplicationSlot->data.restart_lsn));
+
+			wait_for_primary_slot_catchup(wrconn, slot_name,
+										  MyReplicationSlot->data.restart_lsn);
+		}
+
+		ReplicationSlotPersist();
+	}
+
+	endlsn = pg_logical_replication_slot_advance(target_lsn);
+
+	elog(DEBUG3, "synchronized slot %s to lsn (%X/%X)",
+		 slot_name, (uint32) (endlsn >> 32), (uint32) (endlsn));
+
+	ReplicationSlotRelease();
+	CommitTransactionCommand();
+}
+
+static void
+synchronize_slots(void)
+{
+	WalRcvExecResult *res;
+	WalReceiverConn *wrconn = NULL;
+	TupleTableSlot *slot;
+	Oid				slotRow[3] = {TEXTOID, TEXTOID, LSNOID};
+	StringInfoData	s;
+	char		   *database;
+	char		   *err;
+	MemoryContext	oldctx = CurrentMemoryContext;
+
+	if (!WalRcv)
+		return;
+
+	/* syscache access needs a transaction env. */
+	StartTransactionCommand();
+	/* make dbname live outside TX context */
+	MemoryContextSwitchTo(oldctx);
+
+	database = get_database_name(MyDatabaseId);
+	initStringInfo(&s);
+	appendStringInfo(&s, "%s dbname=%s", PrimaryConnInfo, database);
+	wrconn = walrcv_connect(s.data, true, "slot_sync", &err);
+
+	if (wrconn == NULL)
+		ereport(ERROR,
+				(errmsg("could not connect to the primary server: %s", err)));
+
+	resetStringInfo(&s);
+	/* TODO filter slot names? */
+	appendStringInfo(&s,
+					 "SELECT slot_name, plugin, confirmed_flush_lsn"
+					 "  FROM pg_catalog.pg_replication_slots"
+					 " WHERE database = %s",
+					 quote_literal_cstr(database));
+	res = walrcv_exec(wrconn, s.data, 3, slotRow);
+	pfree(s.data);
+
+	if (res->status != WALRCV_OK_TUPLES)
+		ereport(ERROR,
+				(errmsg("could not fetch slot info from primary: %s",
+						res->err)));
+
+	CommitTransactionCommand();
+	/* CommitTransactionCommand switches to TopMemoryContext */
+	MemoryContextSwitchTo(oldctx);
+
+	slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+	while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+	{
+		char	   *slot_name;
+		char	   *plugin_name;
+		XLogRecPtr	confirmed_flush_lsn;
+		bool		isnull;
+
+		slot_name = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+		Assert(!isnull);
+
+		plugin_name = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+		Assert(!isnull);
+
+		confirmed_flush_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull));
+		Assert(!isnull);
+
+		synchronize_one_slot(wrconn, slot_name, database, plugin_name,
+							 confirmed_flush_lsn);
+
+		ExecClearTuple(slot);
+	}
+
+	walrcv_clear_result(res);
+	pfree(database);
+
+	walrcv_disconnect(wrconn);
+}
+
+/*
+ * The main loop of our worker process.
+ */
+void
+ReplSlotSyncMain(Datum main_arg)
+{
+	int			worker_slot = DatumGetInt32(main_arg);
+
+	/* Attach to slot */
+	logicalrep_worker_attach(worker_slot);
+
+	/* Establish signal handlers. */
+	BackgroundWorkerUnblockSignals();
+
+	/* Load the libpq-specific functions */
+	load_file("libpqwalreceiver", false);
+
+	/* Connect to our database. */
+	BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
+											  MyLogicalRepWorker->userid,
+											  0);
+
+	StartTransactionCommand();
+	ereport(LOG,
+			(errmsg("replication slot synchronization worker for database \"%s\" has started",
+					get_database_name(MyLogicalRepWorker->dbid))));
+	CommitTransactionCommand();
+
+	/* Main wait loop. */
+	for (;;)
+	{
+		int		rc;
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!RecoveryInProgress())
+			return;
+
+		synchronize_slots();
+
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   wal_retrieve_retry_interval,
+					   WAIT_EVENT_REPL_SLOT_SYNC_MAIN);
+
+		ResetLatch(MyLatch);
+
+		/* emergency bailout if postmaster has died */
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+	}
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a43c..0e0593f716 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -100,6 +100,7 @@
 #include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_type.h"
 #include "commands/copy.h"
+#include "commands/subscriptioncmds.h"
 #include "miscadmin.h"
 #include "parser/parse_relation.h"
 #include "pgstat.h"
@@ -151,7 +152,8 @@ finish_sync_worker(void)
 	CommitTransactionCommand();
 
 	/* Find the main apply worker and signal it. */
-	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
+	logicalrep_worker_wakeup(MyLogicalRepWorker->dbid,
+							 MyLogicalRepWorker->subid, InvalidOid);
 
 	/* Stop gracefully */
 	proc_exit(0);
@@ -191,7 +193,8 @@ wait_for_relation_state_change(Oid relid, char expected_state)
 
 		/* Check if the sync worker is still running and bail if not. */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-		worker = logicalrep_worker_find(MyLogicalRepWorker->subid, relid,
+		worker = logicalrep_worker_find(MyLogicalRepWorker->dbid,
+										MyLogicalRepWorker->subid, relid,
 										false);
 		LWLockRelease(LogicalRepWorkerLock);
 		if (!worker)
@@ -238,7 +241,8 @@ wait_for_worker_state_change(char expected_state)
 		 * waiting.
 		 */
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-		worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+		worker = logicalrep_worker_find(MyLogicalRepWorker->dbid,
+										MyLogicalRepWorker->subid,
 										InvalidOid, false);
 		if (worker && worker->proc)
 			logicalrep_worker_wakeup_ptr(worker);
@@ -484,7 +488,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 			 */
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
-			syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+			syncworker = logicalrep_worker_find(MyLogicalRepWorker->dbid,
+												MyLogicalRepWorker->subid,
 												rstate->relid, false);
 
 			if (syncworker)
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index dcb1108579..ab4869f312 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -91,11 +91,12 @@ static SQLCmd *make_sqlcmd(void);
 %token K_USE_SNAPSHOT
 %token K_MANIFEST
 %token K_MANIFEST_CHECKSUMS
+%token K_LIST_SLOTS
 
 %type <node>	command
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
-				read_replication_slot timeline_history show sql_cmd
+				read_replication_slot timeline_history show sql_cmd list_slots
 %type <list>	base_backup_legacy_opt_list generic_option_list
 %type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
@@ -129,6 +130,7 @@ command:
 			| read_replication_slot
 			| timeline_history
 			| show
+			| list_slots
 			| sql_cmd
 			;
 
@@ -141,6 +143,15 @@ identify_system:
 					$$ = (Node *) makeNode(IdentifySystemCmd);
 				}
 			;
+/*
+ * LIST_SLOTS
+ */
+list_slots:
+			K_LIST_SLOTS
+				{
+					$$ = (Node *) makeNode(ListSlotsCmd);
+				}
+			;
 
 /*
  * READ_REPLICATION_SLOT %s
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index 1b599c255e..9ee638355d 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -85,6 +85,7 @@ identifier		{ident_start}{ident_cont}*
 BASE_BACKUP			{ return K_BASE_BACKUP; }
 FAST			{ return K_FAST; }
 IDENTIFY_SYSTEM		{ return K_IDENTIFY_SYSTEM; }
+LIST_SLOTS		{ return K_LIST_SLOTS; }
 READ_REPLICATION_SLOT	{ return K_READ_REPLICATION_SLOT; }
 SHOW		{ return K_SHOW; }
 LABEL			{ return K_LABEL; }
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 17df99c2ac..0cf1c85d52 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -484,7 +484,7 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto)
  * WAL and removal of old catalog tuples.  As decoding is done in fast_forward
  * mode, no changes are generated anyway.
  */
-static XLogRecPtr
+XLogRecPtr
 pg_logical_replication_slot_advance(XLogRecPtr moveto)
 {
 	LogicalDecodingContext *ctx;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index d9ab6d6de2..f09f4c13ec 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -458,6 +458,167 @@ IdentifySystem(void)
 	end_tup_output(tstate);
 }
 
+/*
+ * Handle the LIST_SLOTS command.
+ */
+static void
+ListSlots(void)
+{
+	DestReceiver *dest;
+	TupOutputState *tstate;
+	TupleDesc	tupdesc;
+	int			slotno;
+
+	dest = CreateDestReceiver(DestRemoteSimple);
+
+	/* need a tuple descriptor representing four columns */
+	tupdesc = CreateTemplateTupleDesc(10);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "slot_name",
+							  TEXTOID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "plugin",
+							  TEXTOID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 3, "slot_type",
+							  TEXTOID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 4, "datoid",
+							  INT8OID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 5, "database",
+							  TEXTOID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 6, "temporary",
+							  INT4OID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 7, "xmin",
+							  INT8OID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 8, "catalog_xmin",
+							  INT8OID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 9, "restart_lsn",
+							  TEXTOID, -1, 0);
+	TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 10, "confirmed_flush",
+							  TEXTOID, -1, 0);
+
+	/* prepare for projection of tuples */
+	tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
+
+	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+	for (slotno = 0; slotno < max_replication_slots; slotno++)
+	{
+		ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[slotno];
+		char		restart_lsn_str[MAXFNAMELEN];
+		char		confirmed_flush_lsn_str[MAXFNAMELEN];
+		Datum		values[10];
+		bool		nulls[10];
+
+		ReplicationSlotPersistency persistency;
+		TransactionId xmin;
+		TransactionId catalog_xmin;
+		XLogRecPtr	restart_lsn;
+		XLogRecPtr	confirmed_flush_lsn;
+		Oid			datoid;
+		NameData	slot_name;
+		NameData	plugin;
+		int			i;
+		int64		tmpbigint;
+
+		if (!slot->in_use)
+			continue;
+
+		SpinLockAcquire(&slot->mutex);
+
+		xmin = slot->data.xmin;
+		catalog_xmin = slot->data.catalog_xmin;
+		datoid = slot->data.database;
+		restart_lsn = slot->data.restart_lsn;
+		confirmed_flush_lsn = slot->data.confirmed_flush;
+		namestrcpy(&slot_name, NameStr(slot->data.name));
+		namestrcpy(&plugin, NameStr(slot->data.plugin));
+		persistency = slot->data.persistency;
+
+		SpinLockRelease(&slot->mutex);
+
+		memset(nulls, 0, sizeof(nulls));
+
+		i = 0;
+		values[i++] = CStringGetTextDatum(NameStr(slot_name));
+
+		if (datoid == InvalidOid)
+			nulls[i++] = true;
+		else
+			values[i++] = CStringGetTextDatum(NameStr(plugin));
+
+		if (datoid == InvalidOid)
+			values[i++] = CStringGetTextDatum("physical");
+		else
+			values[i++] = CStringGetTextDatum("logical");
+
+		if (datoid == InvalidOid)
+			nulls[i++] = true;
+		else
+		{
+			tmpbigint = datoid;
+			values[i++] = Int64GetDatum(tmpbigint);
+		}
+
+		if (datoid == InvalidOid)
+			nulls[i++] = true;
+		else
+		{
+			MemoryContext cur = CurrentMemoryContext;
+
+			/* syscache access needs a transaction env. */
+			StartTransactionCommand();
+			/* make dbname live outside TX context */
+			MemoryContextSwitchTo(cur);
+			values[i++] = CStringGetTextDatum(get_database_name(datoid));
+			CommitTransactionCommand();
+			/* CommitTransactionCommand switches to TopMemoryContext */
+			MemoryContextSwitchTo(cur);
+		}
+
+		values[i++] = Int32GetDatum(persistency == RS_TEMPORARY ? 1 : 0);
+
+		if (xmin != InvalidTransactionId)
+		{
+			tmpbigint = xmin;
+			values[i++] = Int64GetDatum(tmpbigint);
+		}
+		else
+			nulls[i++] = true;
+
+		if (catalog_xmin != InvalidTransactionId)
+		{
+			tmpbigint = catalog_xmin;
+			values[i++] = Int64GetDatum(tmpbigint);
+		}
+		else
+			nulls[i++] = true;
+
+		if (restart_lsn != InvalidXLogRecPtr)
+		{
+			snprintf(restart_lsn_str, sizeof(restart_lsn_str), "%X/%X",
+					 (uint32) (restart_lsn >> 32),
+					 (uint32) restart_lsn);
+			values[i++] = CStringGetTextDatum(restart_lsn_str);
+		}
+		else
+			nulls[i++] = true;
+
+		if (confirmed_flush_lsn != InvalidXLogRecPtr)
+		{
+			snprintf(confirmed_flush_lsn_str, sizeof(confirmed_flush_lsn_str),
+					 "%X/%X",
+					 (uint32) (confirmed_flush_lsn >> 32),
+					 (uint32) confirmed_flush_lsn);
+			values[i++] = CStringGetTextDatum(confirmed_flush_lsn_str);
+		}
+		else
+			nulls[i++] = true;
+
+		/* send it to dest */
+		do_tup_output(tstate, values, nulls);
+	}
+	LWLockRelease(ReplicationSlotControlLock);
+
+	end_tup_output(tstate);
+}
+
 /* Handle READ_REPLICATION_SLOT command */
 static void
 ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
@@ -556,7 +717,6 @@ ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
 	end_tup_output(tstate);
 }
 
-
 /*
  * Handle TIMELINE_HISTORY command.
  */
@@ -1746,6 +1906,13 @@ exec_replication_command(const char *cmd_string)
 			EndReplicationCommand(cmdtag);
 			break;
 
+		case T_ListSlotsCmd:
+			cmdtag = "LIST_SLOTS";
+			set_ps_display(cmdtag);
+			ListSlots();
+			EndReplicationCommand(cmdtag);
+			break;
+
 		case T_StartReplicationCmd:
 			{
 				StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 4a5b7502f5..937be34b3d 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -230,6 +230,9 @@ pgstat_get_wait_activity(WaitEventActivity w)
 		case WAIT_EVENT_LOGICAL_LAUNCHER_MAIN:
 			event_name = "LogicalLauncherMain";
 			break;
+		case WAIT_EVENT_REPL_SLOT_SYNC_MAIN:
+			event_name = "ReplSlotSyncMain";
+			break;
 		case WAIT_EVENT_PGSTAT_MAIN:
 			event_name = "PgStatMain";
 			break;
diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index aec7e478ab..1cc19e0c99 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -17,6 +17,7 @@
 
 #include "catalog/objectaddress.h"
 #include "parser/parse_node.h"
+#include "replication/walreceiver.h"
 
 extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 										bool isTopLevel);
@@ -26,4 +27,6 @@ extern void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel);
 extern ObjectAddress AlterSubscriptionOwner(const char *name, Oid newOwnerId);
 extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId);
 
+extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
+
 #endif							/* SUBSCRIPTIONCMDS_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 7c657c1241..920a510c4c 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -501,6 +501,7 @@ typedef enum NodeTag
 	T_StartReplicationCmd,
 	T_TimeLineHistoryCmd,
 	T_SQLCmd,
+	T_ListSlotsCmd,
 
 	/*
 	 * TAGS FOR RANDOM OTHER STUFF
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index a746fafc12..06aad4fc6e 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -33,6 +33,14 @@ typedef struct IdentifySystemCmd
 	NodeTag		type;
 } IdentifySystemCmd;
 
+/* ----------------------
+ *		LIST_SLOTS command
+ * ----------------------
+ */
+typedef struct ListSlotsCmd
+{
+	NodeTag		type;
+} ListSlotsCmd;
 
 /* ----------------------
  *		BASE_BACKUP command
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 2ad61a001a..902789f815 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -13,6 +13,7 @@
 #define LOGICALWORKER_H
 
 extern void ApplyWorkerMain(Datum main_arg);
+extern void ReplSlotSyncMain(Datum main_arg);
 
 extern bool IsLogicalWorker(void);
 
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 53d773ccff..a29e517707 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -216,7 +216,6 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid);
 extern bool InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
 extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
 extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, int szslot);
-extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
 
 extern void StartupReplicationSlots(void);
 extern void CheckPointReplicationSlots(void);
@@ -224,4 +223,7 @@ extern void CheckPointReplicationSlots(void);
 extern void CheckSlotRequirements(void);
 extern void CheckSlotPermissions(void);
 
+extern XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto);
+
+
 #endif							/* SLOT_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 0b607ed777..ff8b755c67 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -18,6 +18,7 @@
 #include "pgtime.h"
 #include "port/atomics.h"
 #include "replication/logicalproto.h"
+#include "replication/slot.h"
 #include "replication/walsender.h"
 #include "storage/condition_variable.h"
 #include "storage/latch.h"
@@ -187,6 +188,13 @@ typedef struct
 	}			proto;
 } WalRcvStreamOptions;
 
+/*
+ * Slot information receiver from remote.
+ *
+ * Currently this is same as ReplicationSlotPersistentData
+ */
+#define WalRecvReplicationSlotData ReplicationSlotPersistentData
+
 struct WalReceiverConn;
 typedef struct WalReceiverConn WalReceiverConn;
 
@@ -274,6 +282,11 @@ typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn,
 typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
 											TimeLineID *primary_tli);
 
+/*
+ * TODO
+ */
+typedef List *(*walrcv_list_slots_fn) (WalReceiverConn *conn);
+
 /*
  * walrcv_server_version_fn
  *
@@ -387,6 +400,7 @@ typedef struct WalReceiverFunctionsType
 	walrcv_get_conninfo_fn walrcv_get_conninfo;
 	walrcv_get_senderinfo_fn walrcv_get_senderinfo;
 	walrcv_identify_system_fn walrcv_identify_system;
+	walrcv_list_slots_fn walrcv_list_slots;
 	walrcv_server_version_fn walrcv_server_version;
 	walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;
 	walrcv_startstreaming_fn walrcv_startstreaming;
@@ -411,6 +425,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 	WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
 #define walrcv_identify_system(conn, primary_tli) \
 	WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
+#define walrcv_list_slots(conn) \
+	WalReceiverFunctions->walrcv_list_slots(conn)
 #define walrcv_server_version(conn) \
 	WalReceiverFunctions->walrcv_server_version(conn)
 #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index c00be2a2b6..e7226fcb6e 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -57,7 +57,7 @@ typedef struct LogicalRepWorker
 	 * exits.  Under this, separate buffiles would be created for each
 	 * transaction which will be deleted after the transaction is finished.
 	 */
-	FileSet    *stream_fileset;
+	struct FileSet *stream_fileset;
 
 	/* Stats. */
 	XLogRecPtr	last_lsn;
@@ -80,13 +80,13 @@ extern LogicalRepWorker *MyLogicalRepWorker;
 extern bool in_remote_transaction;
 
 extern void logicalrep_worker_attach(int slot);
-extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
+extern LogicalRepWorker *logicalrep_worker_find(Oid dbid, Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
 extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid);
-extern void logicalrep_worker_stop(Oid subid, Oid relid);
-extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
+extern void logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid);
+extern void logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
 extern int	logicalrep_sync_worker_count(Oid subid);
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index c22142365f..ba7ca743f5 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -42,6 +42,7 @@ typedef enum
 	WAIT_EVENT_CHECKPOINTER_MAIN,
 	WAIT_EVENT_LOGICAL_APPLY_MAIN,
 	WAIT_EVENT_LOGICAL_LAUNCHER_MAIN,
+	WAIT_EVENT_REPL_SLOT_SYNC_MAIN,
 	WAIT_EVENT_PGSTAT_MAIN,
 	WAIT_EVENT_RECOVERY_WAL_STREAM,
 	WAIT_EVENT_SYSLOGGER_MAIN,
diff --git a/src/test/recovery/t/007_sync_rep.pl b/src/test/recovery/t/007_sync_rep.pl
index 0d0e60b772..69504d84ee 100644
--- a/src/test/recovery/t/007_sync_rep.pl
+++ b/src/test/recovery/t/007_sync_rep.pl
@@ -6,7 +6,8 @@
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 11;
+#use Test::More tests => 11;
+use Test::More skip_all => 'FIXME';
 
 # Query checking sync_priority and sync_state of each standby
 my $check_sql =
diff --git a/src/test/recovery/t/010_logical_decoding_timelines.pl b/src/test/recovery/t/010_logical_decoding_timelines.pl
index 68d94ac91c..d45066c1f2 100644
--- a/src/test/recovery/t/010_logical_decoding_timelines.pl
+++ b/src/test/recovery/t/010_logical_decoding_timelines.pl
@@ -26,7 +26,8 @@
 
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 13;
+#use Test::More tests => 13;
+use Test::More skip_all => 'FIXME';
 use File::Copy;
 use IPC::Run ();
 use Scalar::Util qw(blessed);
diff --git a/src/test/recovery/t/030_slot_sync.pl b/src/test/recovery/t/030_slot_sync.pl
new file mode 100644
index 0000000000..109daaa9fe
--- /dev/null
+++ b/src/test/recovery/t/030_slot_sync.pl
@@ -0,0 +1,51 @@
+
+# Copyright (c) 2021, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More tests => 2;
+
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+$node_primary->init(allows_streaming => 'logical');
+$node_primary->start;
+
+$node_primary->backup('backup');
+
+my $node_phys_standby = PostgreSQL::Test::Cluster->new('phys_standby');
+$node_phys_standby->init_from_backup($node_primary, 'backup', has_streaming => 1);
+$node_phys_standby->start;
+
+$node_primary->safe_psql('postgres', "CREATE TABLE t1 (a int PRIMARY KEY)");
+$node_primary->safe_psql('postgres', "INSERT INTO t1 VALUES (1), (2), (3)");
+
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+$node_subscriber->safe_psql('postgres', "CREATE TABLE t1 (a int PRIMARY KEY)");
+
+$node_primary->safe_psql('postgres', "CREATE PUBLICATION pub1 FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub1 CONNECTION '" . ($node_primary->connstr . ' dbname=postgres') . "' PUBLICATION pub1");
+
+# Wait for initial sync of all subscriptions
+my $synced_query =
+  "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+my $result = $node_primary->safe_psql('postgres',
+	"SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_type = 'logical'");
+
+is($result, qq(sub1|pgoutput|postgres), 'logical slot on primary');
+
+# FIXME: standby needs restart to pick up new slots
+$node_phys_standby->restart;
+sleep 3;
+
+$result = $node_phys_standby->safe_psql('postgres',
+	"SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_type = 'logical'");
+
+is($result, qq(sub1|pgoutput|postgres), 'logical slot on standby');

base-commit: e6c60719e6c6ee9bd396f430879e1de9079bf74c
-- 
2.33.1

---

From: Masahiko Sawada <[email-masked]>
Date: Wed, 24 Nov 2021 15:11:51 +0900

> mentioned in [0], there are various issues to work out.

Thank you for working on this feature!

> timing or implementation dependencies.

I haven’t looked at the patch deeply but regarding 007_sync_rep.pl,
the tests seem to fail since the tests rely on the order of the wal
sender array on the shared memory. Since a background worker for
synchronizing replication slots periodically connects to the walsender
on the primary and disconnects, it breaks the assumption of the order.
Regarding 010_logical_decoding_timelines.pl, I guess that the patch
breaks the test because the background worker for synchronizing slots
on the replica periodically advances the replica's slot. I think we
need to have a way to disable the slot synchronization or to specify
the slot name to sync with the primary. I'm not sure we already
discussed this topic but I think we need it at least for testing
purposes.

Regards,

--
Masahiko Sawada
EDB:  https://www.enterprisedb.com/

---

From: Dimitri Fontaine <[email-masked]>
Date: Wed, 24 Nov 2021 17:25:47 +0100

Hi all,

> mentioned in [0], there are various issues to work out.

Thanks for working on that topic, I believe it's an important part of
Postgres' HA story.

> The idea is that the standby runs a background worker to periodically fetch
> replication slot information from the primary.  On failover, a logical
> subscriber would then ideally find up-to-date replication slots on the new
> publisher and can just continue normally.

Is there a case to be made about doing the same thing for physical
replication slots too?

That's what pg_auto_failover [1] does by default: it creates replication
slots on every node for every other node, in a way that a standby
Postgres instance now maintains a replication slot for the primary. This
ensures that after a promotion, the standby knows to retain any and all
WAL segments that the primary might need when rejoining, at pg_rewind
time.

> configurable and documented.

I suppose part of the configuration would then include taking care of
physical slots. Some people might want to turn that off and use the
Postgres 13+ ability to use the remote primary restore_command to fetch
missing WAL files, instead. Well, people who have setup an archiving
system, anyway.

> As I said, I added a small test case to 
> show that it works at all, but I think a lot more tests should be added.   I
> have also found that this breaks some seemingly unrelated tests in the
> recovery test suite.  I have disabled these here.  I'm not sure if the patch
> actually breaks anything or if these are just differences in timing or
> implementation dependencies.  This patch adds a LIST_SLOTS replication
> command, but I think this could be replaced with just a SELECT FROM
> pg_replication_slots query now.  (This patch is originally older than when
> you could run SELECT queries over the replication protocol.)

Given the admitted state of the patch, I didn't focus on tests. I could
successfully apply the patch on-top of current master's branch, and
cleanly compile and `make check`.

Then I also updated pg_auto_failover to support Postgres 15devel [2] so
that I could then `make NODES=3 cluster` there and play with the new
replication command:

  $ psql -d "port=5501 replication=1" -c "LIST_SLOTS;"
  psql:/Users/dim/.psqlrc:24: ERROR:  XX000: cannot execute SQL commands in WAL sender for physical replication
  LOCATION:  exec_replication_command, walsender.c:1830
  ...

I'm not too sure about this idea of running SQL in a replication
protocol connection that you're mentioning, but I suppose that's just me
needing to brush up on the topic.

> So, again, this isn't anywhere near ready, but there is already a lot here
> to gather feedback about how it works, how it should work, how to configure
> it, and how it fits into an overall replication and HA architecture.

Maybe the first question about configuration would be about selecting
which slots a standby should maintain from the primary. Is it all of the
slots that exists on both the nodes, or a sublist of that?

Is it possible to have a slot with the same name on a primary and a
standby node, in a way that the standby's slot would be a completely
separate entity from the primary's slot? If yes (I just don't know at
the moment), well then, should we continue to allow that?

Other aspects of the configuration might include a list of databases in
which to make the new background worker active, and the polling delay,
etc.

Also, do we want to even consider having the slot management on a
primary node depend on the ability to sync the advancing on one or more
standby nodes? I'm not sure to see that one as a good idea, but maybe we
want to kill it publically very early then ;-)

Regards,
-- 
dim

Author of “The Art of PostgreSQL”, see https://theartofpostgresql.com

[1]: https://github.com/citusdata/pg_auto_failover
[2]: https://github.com/citusdata/pg_auto_failover/pull/838

---

From: Bharath Rupireddy <[email-masked]>
Date: Sun, 28 Nov 2021 12:22:45 +0530

> https://www.postgresql.org/message-id/flat/3095349b-44d4-bf11-1b33-7eefb585d578%402ndquadrant.com

Thanks for working on this patch. This feature will be useful as it
avoids manual intervention during the failover.

Here are some thoughts:
1) Instead of a new LIST_SLOT command, can't we use
READ_REPLICATION_SLOT (slight modifications needs to be done to make
it support logical replication slots and to get more information from
the subscriber).

2) How frequently the new bg worker is going to sync the slot info?
How can it ensure that the latest information exists say when the
subscriber is down/crashed before it picks up the latest slot
information?

3) Instead of the subscriber pulling the slot info, why can't the
publisher (via the walsender or a new bg worker maybe?) push the
latest slot info? I'm not sure we want to add more functionality to
the walsender, if yes, isn't it going to be much simpler?

4) IIUC, the proposal works only for logical replication slots but do
you also see the need for supporting some kind of synchronization of
physical replication slots as well? IMO, we need a better and
consistent way for both types of replication slots. If the walsender
can somehow push the slot info from the primary (for physical
replication slots)/publisher (for logical replication slots) to the
standby/subscribers, this will be a more consistent and simplistic
design. However, I'm not sure if this design is doable at all.

Regards,
Bharath Rupireddy.

---

From: SATYANARAYANA NARLAPURAM <[email-masked]>
Date: Sun, 28 Nov 2021 12:17:46 -0800

> the walsender, if yes, isn't it going to be much simpler?

Standby pulling the information or at least making a first attempt to
connect to the  primary is a better design as primary doesn't need to spend
its cycles repeatedly connecting to an unreachable standby. In fact,
primary wouldn't even need to know the followers, for example followers /
log shipping standbys

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 29 Nov 2021 09:39:59 +0530

On Mon, Nov 29, 2021 at 1:48 AM SATYANARAYANA NARLAPURAM
<[email-masked]> wrote:
> Standby pulling the information or at least making a first attempt to connect to the  primary is a better design as primary doesn't need to spend its cycles repeatedly connecting to an unreachable standby. In fact, primary wouldn't even need to know the followers, for example followers / log shipping standbys

My idea was to let the existing walsender from the primary/publisher
to send the slot info (both logical and physical replication slots) to
the standby/subscriber, probably by piggybacking the slot info with
the WAL currently it sends. Having said that, I don't know the
feasibility of it. Anyways, I'm not in favour of having a new bg
worker to just ship the slot info. The standby/subscriber, while
making connection to primary/publisher, can choose to get the
replication slot info.

As I said upthread, the problem I see with standby/subscriber pulling
the info is that: how frequently the standby/subscriber is going to
sync the slot info from primary/publisher? How can it ensure that the
latest information exists say when the subscriber is down/crashed
before it picks up the latest slot information?

IIUC, the initial idea proposed in this patch deals with only logical
replication slots not the physical replication slots, what I'm
thinking is to have a generic way to deal with both of them.

Note: In the above description, I used primary-standby and
publisher-subscriber to represent the physical and logical replication
slots respectively.

Regards,
Bharath Rupireddy.

---

From: Dilip Kumar <[email-masked]>
Date: Mon, 29 Nov 2021 11:14:23 +0530

> making connection to primary/publisher, can choose to get the

I think it is possible that the standby is restoring the WAL directly
from the archive location and there might not be any wal sender at
time. So I think the idea of standby pulling the WAL looks better to
me.

> before it picks up the latest slot information?

Yeah that is a good question that how frequently the subscriber should
fetch the slot information, I think that should be configurable
values.  And the time delay is more, the chances of losing the latest
slot is more.

-- 
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 29 Nov 2021 12:19:00 +0530

> time. So I think the idea of standby pulling the WAL looks better to

My point was that why can't we let the walreceiver (of course users
can configure it on the standby/subscriber) to choose whether or not
to receive the replication (both physical and logical) slot info from
the primary/publisher and if yes, the walsender(on the
primary/publisher) sending it probably as a new WAL record or just
piggybacking the replication slot info with any of the existing WAL
records.

Or simply a common bg worker (as opposed to the bg worker proposed
originally in this thread which, IIUC, works for logical replication)
running on standby/subscriber for getting both the physical and
logical replication slots info.

> values.  And the time delay is more, the chances of losing the latest

I agree that it should be configurable. Even if the primary/publisher
is down/crashed, one can still compare the latest slot info from both
the primary/publisher and standby/subscriber using a new tool
pg_replslotdata proposed at [1] and see how far and which slots missed
the latest replication slot info and probably drop those alone to
recreate and retain other slots as is.

[1] - https://www.postgresql.org/message-id/CALj2ACW0rV5gWK8A3m6_X62qH%2BVfaq5hznC%3Di0R5Wojt5%2Byhyw%40mail.gmail.com

Regards,
Bharath Rupireddy.

---

From: Dilip Kumar <[email-masked]>
Date: Mon, 29 Nov 2021 13:09:48 +0530

> piggybacking the replication slot info with any of the existing WAL

Okay, I thought your point was that the primary pushing is better over
standby pulling the slot info, but now it seems that you also agree
that standby pulling is better right?  Now it appears your point is
about whether we will use the same connection for pulling the slot
information which we are using for streaming the data or any other
connection?  I mean in this patch also we are creating a replication
connection and pulling the slot information over there, just point is
we are starting a separate worker for pulling the slot information,
and I think that approach is better as this will not impact the
performance of the other replication connection which we are using for
communicating the data.

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 29 Nov 2021 17:58:26 +0530

> performance of the other replication connection which we are using for

The easiest way to implement this feature so far, is to use a common
bg worker (as opposed to the bg worker proposed originally in this
thread which, IIUC, works for logical replication) running on standby
(in case of streaming replication with physical replication slots) or
subscriber (in case of logical replication with logical replication
slots) for getting both the physical and logical replication slots
info from the primary or publisher. This bg worker requires at least
two GUCs, 1) to enable/disable the worker 2) to define the slot sync
interval (the bg worker gets the slots info after every sync interval
of time).

Thoughts?

Regards,
Bharath Rupireddy.

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 14 Dec 2021 23:12:44 +0100

> configure it, and how it fits into an overall replication and HA 

Here is an updated patch.  The main changes are that I added two 
configuration parameters.  The first, synchronize_slot_names, is set on 
the physical standby to specify which slots to sync from the primary. 
By default, it is empty.  (This also fixes the recovery test failures 
that I had to disable in the previous patch version.)  The second, 
standby_slot_names, is set on the primary.  It holds back logical 
replication until the listed physical standbys have caught up.  That 
way, when failover is necessary, the promoted standby is not behind the 
logical replication consumers.

In principle, this works now, I think.  I haven't made much progress in 
creating more test cases for this; that's something that needs more 
attention.

It's worth pondering what the configuration language for 
standby_slot_names should be.  Right now, it's just a list of slots that 
all need to be caught up.  More complicated setups are conceivable. 
Maybe you have standbys S1 and S2 that are potential failover targets 
for logical replication consumers L1 and L2, and also standbys S3 and S4 
that are potential failover targets for logical replication consumers L3 
and L4.  Viewed like that, this setting could be a replication slot 
setting.  The setting might also have some relationship with 
synchronous_standby_names.  Like, if you have synchronous_standby_names 
set, then that's a pretty good indication that you also want some or all 
of those standbys in standby_slot_names.  (But note that one is slots 
and one is application names.)  So there are a variety of possibilities.

From f1d306bfe3eb657b5d14b0ef3024586083beb4ed Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <[email-masked]>
Date: Tue, 14 Dec 2021 22:20:59 +0100
Subject: [PATCH v2] Synchronize logical replication slots from primary to
 standby

Discussion: https://www.postgresql.org/message-id/flat/514f6f2f-6833-4539-39f1-96cd1e011f23%40enterprisedb.com
---
 doc/src/sgml/config.sgml                      |  34 ++
 src/backend/commands/subscriptioncmds.c       |   4 +-
 src/backend/postmaster/bgworker.c             |   3 +
 .../libpqwalreceiver/libpqwalreceiver.c       |  96 ++++
 src/backend/replication/logical/Makefile      |   1 +
 src/backend/replication/logical/launcher.c    | 202 ++++++---
 .../replication/logical/reorderbuffer.c       |  85 ++++
 src/backend/replication/logical/slotsync.c    | 412 ++++++++++++++++++
 src/backend/replication/logical/tablesync.c   |  13 +-
 src/backend/replication/repl_gram.y           |  32 +-
 src/backend/replication/repl_scanner.l        |   1 +
 src/backend/replication/slotfuncs.c           |   2 +-
 src/backend/replication/walsender.c           | 196 ++++++++-
 src/backend/utils/activity/wait_event.c       |   3 +
 src/backend/utils/misc/guc.c                  |  23 +
 src/backend/utils/misc/postgresql.conf.sample |   2 +
 src/include/commands/subscriptioncmds.h       |   3 +
 src/include/nodes/nodes.h                     |   1 +
 src/include/nodes/replnodes.h                 |   9 +
 src/include/replication/logicalworker.h       |   9 +
 src/include/replication/slot.h                |   4 +-
 src/include/replication/walreceiver.h         |  16 +
 src/include/replication/worker_internal.h     |   8 +-
 src/include/utils/wait_event.h                |   1 +
 src/test/recovery/t/030_slot_sync.pl          |  58 +++
 25 files changed, 1148 insertions(+), 70 deletions(-)
 create mode 100644 src/backend/replication/logical/slotsync.c
 create mode 100644 src/test/recovery/t/030_slot_sync.pl

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index afbb6c35e3..2b2a21a251 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4406,6 +4406,23 @@ <title>Primary Server</title>
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-standby-slot-names" xreflabel="standby_slot_names">
+      <term><varname>standby_slot_names</varname> (<type>string</type>)
+      <indexterm>
+       <primary><varname>standby_slot_names</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        List of physical replication slots that logical replication waits for.
+        If a logical replication connection is meant to switch to a physical
+        standby after the standby is promoted, the physical replication slot
+        for the standby should be listed here.  This ensures that logical
+        replication is not ahead of the physical standby.
+       </para>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
     </sect2>
 
@@ -4794,6 +4811,23 @@ <title>Standby Servers</title>
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-synchronize_slot_names" xreflabel="synchronize_slot_names">
+      <term><varname>synchronize_slot_names</varname> (<type>string</type>)
+      <indexterm>
+       <primary><varname>synchronize_slot_names</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies a list of logical replication slots that a physical standby
+        should synchronize from the primary server.  This is necessary to be
+        able to retarget those logical replication connections to this standby
+        if it gets promoted.  Specify <literal>*</literal> to synchronize all
+        logical replication slots.  The default is empty.
+       </para>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
     </sect2>
 
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 2b658080fe..7cdea20207 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -737,7 +737,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)

 
 				/*
 				 * For READY state, we would have already dropped the
@@ -1239,7 +1239,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	{
 		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);

 
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index c08e599eef..57489a0d11 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -33,6 +33,7 @@
 #include "utils/memutils.h"
 #include "utils/pg_lsn.h"
 #include "utils/tuplestore.h"
+#include "utils/varlena.h"
 
 PG_MODULE_MAGIC;
 
@@ -58,6 +59,7 @@ static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
 									char **sender_host, int *sender_port);
 static char *libpqrcv_identify_system(WalReceiverConn *conn,
 									  TimeLineID *primary_tli);
+static List *libpqrcv_list_slots(WalReceiverConn *conn, const char *slot_names);
 static int	libpqrcv_server_version(WalReceiverConn *conn);
 static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
 											 TimeLineID tli, char **filename,
@@ -89,6 +91,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
 	libpqrcv_get_conninfo,
 	libpqrcv_get_senderinfo,
 	libpqrcv_identify_system,
+	libpqrcv_list_slots,
 	libpqrcv_server_version,
 	libpqrcv_readtimelinehistoryfile,
 	libpqrcv_startstreaming,
@@ -397,6 +400,99 @@ libpqrcv_server_version(WalReceiverConn *conn)
 	return PQserverVersion(conn->streamConn);
 }
 
+/*
+ * Get list of slots from primary.
+ */
+static List *
+libpqrcv_list_slots(WalReceiverConn *conn, const char *slot_names)
+{
+	PGresult   *res;
+	List	   *slotlist = NIL;
+	int			ntuples;
+	StringInfoData s;
+	WalRecvReplicationSlotData *slot_data;
+
+	initStringInfo(&s);
+	appendStringInfoString(&s, "LIST_SLOTS");
+
+	if (strcmp(slot_names, "") != 0 && strcmp(slot_names, "*") != 0)
+	{
+		char	   *rawname;
+		List	   *namelist;
+		ListCell   *lc;
+
+		appendStringInfoChar(&s, ' ');
+		rawname = pstrdup(slot_names);
+		SplitIdentifierString(rawname, ',', &namelist);
+		foreach (lc, namelist)
+		{
+			if (lc != list_head(namelist))
+				appendStringInfoChar(&s, ',');
+			appendStringInfo(&s, "%s",
+							 quote_identifier(lfirst(lc)));
+		}
+	}
+
+	res = libpqrcv_PQexec(conn->streamConn, s.data);
+	pfree(s.data);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("could not receive list of slots the primary server: %s",
+						pchomp(PQerrorMessage(conn->streamConn)))));
+	}
+	if (PQnfields(res) < 10)
+	{
+		int			nfields = PQnfields(res);
+
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("invalid response from primary server"),
+				 errdetail("Could not get list of slots: got %d fields, expected %d or more fields.",
+						   nfields, 10)));
+	}
+
+	ntuples = PQntuples(res);
+	for (int i = 0; i < ntuples; i++)
+	{
+		char	   *slot_type;
+
+		slot_data = palloc0(sizeof(WalRecvReplicationSlotData));
+		namestrcpy(&slot_data->name, PQgetvalue(res, i, 0));
+		if (!PQgetisnull(res, i, 1))
+			namestrcpy(&slot_data->plugin, PQgetvalue(res, i, 1));
+		slot_type = PQgetvalue(res, i, 2);
+		if (!PQgetisnull(res, i, 3))
+			slot_data->database = atooid(PQgetvalue(res, i, 3));
+		if (strcmp(slot_type, "physical") == 0)
+		{
+			if (OidIsValid(slot_data->database))
+				elog(ERROR, "unexpected physical replication slot with database set");
+		}
+		if (pg_strtoint32(PQgetvalue(res, i, 5)) == 1)
+			slot_data->persistency = RS_TEMPORARY;
+		else
+			slot_data->persistency = RS_PERSISTENT;
+		if (!PQgetisnull(res, i, 6))
+			slot_data->xmin = atooid(PQgetvalue(res, i, 6));
+		if (!PQgetisnull(res, i, 7))
+			slot_data->catalog_xmin = atooid(PQgetvalue(res, i, 7));
+		if (!PQgetisnull(res, i, 8))
+			slot_data->restart_lsn = pg_strtouint64(PQgetvalue(res, i, 8),
+													NULL, 10);
+		if (!PQgetisnull(res, i, 9))
+			slot_data->confirmed_flush = pg_strtouint64(PQgetvalue(res, i, 9),
+														NULL, 10);
+
+		slotlist = lappend(slotlist, slot_data);
+	}
+
+	PQclear(res);
+
+	return slotlist;
+}
+
 /*
  * Start streaming WAL data from given streaming options.
  *
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb71..bc3f23b5a2 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -24,6 +24,7 @@ OBJS = \
 	proto.o \
 	relation.o \
 	reorderbuffer.o \
+	slotsync.o \
 	snapbuild.o \
 	tablesync.o \
 	worker.o
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 3fb4caa803..207ef9bc8b 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -22,6 +22,7 @@
 #include "access/htup_details.h"
 #include "access/tableam.h"
 #include "access/xact.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
 #include "funcapi.h"
@@ -212,7 +213,7 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
  * subscription id and relid.
  */
 LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running)
 {
 	int			i;
 	LogicalRepWorker *res = NULL;
@@ -224,8 +225,8 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];

-	ereport(DEBUG1,
-			(errmsg_internal("starting logical replication worker for subscription \"%s\"",
-							 subname)));
+	if (OidIsValid(subid))
+		ereport(DEBUG1,
+				(errmsg_internal("starting logical replication worker for subscription \"%s\"",
+								 subname)));
+	else
+		ereport(DEBUG1,
+				(errmsg_internal("starting replication slot synchronization worker")));

 

 

 
 	if (worker)
@@ -795,6 +809,116 @@ ApplyLauncherWakeup(void)
 
+static void
+	if (strcmp(synchronize_slot_names, "") == 0)
+	slots = walrcv_list_slots(wrconn, synchronize_slot_names);
+	foreach(lc, slots)
@@ -822,14 +946,12 @@ ApplyLauncherMain(Datum main_arg)

@@ -841,42 +963,10 @@ ApplyLauncherMain(Datum main_arg)
 		if (TimestampDifferenceExceeds(last_start_time, now,
 									   wal_retrieve_retry_interval))
 		{
-			/* Use temporary context for the database list and worker info. */
-			subctx = AllocSetContextCreate(TopMemoryContext,
-										   "Logical Replication Launcher sublist",
-										   ALLOCSET_DEFAULT_SIZES);
-			oldctx = MemoryContextSwitchTo(subctx);
-
-			/* search for subscriptions to start or stop. */
-			sublist = get_subscription_list();
-
-			/* Start the missing workers for enabled subscriptions. */
-			foreach(lc, sublist)
-			{
-				Subscription *sub = (Subscription *) lfirst(lc);
-				LogicalRepWorker *w;
-
-				if (!sub->enabled)
-					continue;
-
-				LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-				w = logicalrep_worker_find(sub->oid, InvalidOid, false);
-				LWLockRelease(LogicalRepWorkerLock);
-
-				if (w == NULL)
-				{
-					last_start_time = now;
-					wait_time = wal_retrieve_retry_interval;
-
-					logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
-											 sub->owner, InvalidOid);
-				}
-			}
-
-			/* Switch back to original memory context. */
-			MemoryContextSwitchTo(oldctx);
-			/* Clean the temporary memory. */
-			MemoryContextDelete(subctx);
+			if (!RecoveryInProgress())
+				ApplyLauncherStartSubs(&last_start_time, &wait_time);
+			else
+				ApplyLauncherStartSlotSync(&last_start_time, &wait_time);
 		}
 		else
 		{
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 7aa5647a2c..f0b3b9ad87 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -95,11 +95,13 @@
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "replication/logical.h"
+#include "replication/logicalworker.h"
 #include "replication/reorderbuffer.h"
 #include "replication/slot.h"
 #include "replication/snapbuild.h"	/* just for SnapBuildSnapDecRefcount */
 #include "storage/bufmgr.h"
 #include "storage/fd.h"
+#include "storage/ipc.h"
 #include "storage/sinval.h"
 #include "utils/builtins.h"
 #include "utils/combocid.h"
@@ -107,6 +109,7 @@
 #include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/relfilenodemap.h"
+#include "utils/varlena.h"

 /* entry for a hash table we use to map from xid to our transaction state */
@@ -2006,6 +2009,85 @@ ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	}
 }
 
+static void
+wait_for_standby_confirmation(XLogRecPtr commit_lsn)
+{
+	char	   *rawname;
+	List	   *namelist;
+	ListCell   *lc;
+	XLogRecPtr	flush_pos = InvalidXLogRecPtr;
+
+	if (strcmp(standby_slot_names, "") == 0)
+		return;
+
+	rawname = pstrdup(standby_slot_names);
+	SplitIdentifierString(rawname, ',', &namelist);
+
+	while (true)
+	{
+		int			wait_slots_remaining;
+		XLogRecPtr	oldest_flush_pos = InvalidXLogRecPtr;
+		int			rc;
+
+		wait_slots_remaining = list_length(namelist);
+
+		LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+		for (int i = 0; i < max_replication_slots; i++)
+		{
+			ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+			bool		inlist;
+
+			if (!s->in_use)
+				continue;
+
+			inlist = false;
+			foreach (lc, namelist)
+			{
+				char *name = lfirst(lc);
+				if (strcmp(name, NameStr(s->data.name)) == 0)
+				{
+					inlist = true;
+					break;
+				}
+			}
+			if (!inlist)
+				continue;
+
+			SpinLockAcquire(&s->mutex);
+
+			if (s->data.database == InvalidOid)
+				/* Physical slots advance restart_lsn on flush and ignore confirmed_flush_lsn */
+				flush_pos = s->data.restart_lsn;
+			else
+				/* For logical slots we must wait for commit and flush */
+				flush_pos = s->data.confirmed_flush;
+
+			SpinLockRelease(&s->mutex);
+
+			/* We want to find out the min(flush pos) over all named slots */
+			if (oldest_flush_pos == InvalidXLogRecPtr
+				|| oldest_flush_pos > flush_pos)
+				oldest_flush_pos = flush_pos;
+
+			if (flush_pos >= commit_lsn && wait_slots_remaining > 0)
+				wait_slots_remaining --;
+		}
+		LWLockRelease(ReplicationSlotControlLock);
+
+		if (wait_slots_remaining == 0)
+			return;
+
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   1000L, PG_WAIT_EXTENSION);
+
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+		CHECK_FOR_INTERRUPTS();
+	}
+}
+
 /*
  * Helper function for ReorderBufferReplay and ReorderBufferStreamTXN.
  *
@@ -2434,6 +2516,9 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 			 * Call either PREPARE (for two-phase transactions) or COMMIT (for
 			 * regular ones).
 			 */
+
+			wait_for_standby_confirmation(commit_lsn);
+
 			if (rbtxn_prepared(txn))
 				rb->prepare(rb, txn, commit_lsn);
 			else
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
new file mode 100644
index 0000000000..654ac154ea
--- /dev/null
+++ b/src/backend/replication/logical/slotsync.c
@@ -0,0 +1,412 @@
+/*-------------------------------------------------------------------------
+ * slotsync.c
+ *	   PostgreSQL worker for synchronizing slots to a standby from primary
+ *
+ * Copyright (c) 2016-2018, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/slotsync.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "commands/dbcommands.h"
+#include "pgstat.h"
+#include "postmaster/bgworker.h"
+#include "replication/logicalworker.h"
+#include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "storage/procarray.h"
+#include "utils/builtins.h"
+#include "utils/pg_lsn.h"
+#include "utils/varlena.h"
+
+char	   *synchronize_slot_names;
+char	   *standby_slot_names;
+
+/*
+ * Wait for remote slot to pass localy reserved position.
+ */
+static void
+wait_for_primary_slot_catchup(WalReceiverConn *wrconn, char *slot_name,
+							  XLogRecPtr min_lsn)
+{
+	WalRcvExecResult *res;
+	TupleTableSlot *slot;
+	Oid			slotRow[1] = {LSNOID};
+	StringInfoData cmd;
+	bool		isnull;
+	XLogRecPtr	restart_lsn;
+
+	for (;;)
+	{
+		int			rc;
+
+		CHECK_FOR_INTERRUPTS();
+
+		initStringInfo(&cmd);
+		appendStringInfo(&cmd,
+						 "SELECT restart_lsn"
+						 "  FROM pg_catalog.pg_replication_slots"
+						 " WHERE slot_name = %s",
+						 quote_literal_cstr(slot_name));
+		res = walrcv_exec(wrconn, cmd.data, 1, slotRow);
+
+		if (res->status != WALRCV_OK_TUPLES)
+			ereport(ERROR,
+					(errmsg("could not fetch slot info for slot \"%s\" from primary: %s",
+							slot_name, res->err)));
+
+		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+		if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+			ereport(ERROR,
+					(errmsg("slot \"%s\" disapeared from provider",
+							slot_name)));
+
+		restart_lsn = DatumGetLSN(slot_getattr(slot, 1, &isnull));
+		Assert(!isnull);
+
+		ExecClearTuple(slot);
+		walrcv_clear_result(res);
+
+		if (restart_lsn >= min_lsn)
+			break;
+
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   wal_retrieve_retry_interval,
+					   WAIT_EVENT_REPL_SLOT_SYNC_MAIN);
+
+		ResetLatch(MyLatch);
+
+		/* emergency bailout if postmaster has died */
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+	}
+}
+
+/*
+ * Synchronize single slot to given position.
+ *
+ * This optionally creates new slot if there is no existing one.
+ */
+static void
+synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database,
+					 char *plugin_name, XLogRecPtr target_lsn)
+{
+	bool		found = false;
+	XLogRecPtr	endlsn;
+
+	/* Search for the named slot and mark it active if we find it. */
+	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+	for (int i = 0; i < max_replication_slots; i++)
+	{
+		ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+		if (!s->in_use)
+			continue;
+
+		if (strcmp(NameStr(s->data.name), slot_name) == 0)
+		{
+			found = true;
+			break;
+		}
+	}
+	LWLockRelease(ReplicationSlotControlLock);
+
+	StartTransactionCommand();
+
+	/* Already existing slot, acquire */
+	if (found)
+	{
+		ReplicationSlotAcquire(slot_name, true);
+
+		if (target_lsn < MyReplicationSlot->data.confirmed_flush)
+		{
+			elog(DEBUG1,
+				 "not synchronizing slot %s; synchronization would move it backward",
+				 slot_name);
+
+			ReplicationSlotRelease();
+			CommitTransactionCommand();
+			return;
+		}
+	}
+	/* Otherwise create the slot first. */
+	else
+	{
+		TransactionId xmin_horizon = InvalidTransactionId;
+		ReplicationSlot *slot;
+
+		ReplicationSlotCreate(slot_name, true, RS_EPHEMERAL, false);
+		slot = MyReplicationSlot;
+
+		SpinLockAcquire(&slot->mutex);
+		slot->data.database = get_database_oid(database, false);
+		namestrcpy(&slot->data.plugin, plugin_name);
+		SpinLockRelease(&slot->mutex);
+
+		ReplicationSlotReserveWal();
+
+		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+		xmin_horizon = GetOldestSafeDecodingTransactionId(true);
+		slot->effective_catalog_xmin = xmin_horizon;
+		slot->data.catalog_xmin = xmin_horizon;
+		ReplicationSlotsComputeRequiredXmin(true);
+		LWLockRelease(ProcArrayLock);
+
+		if (target_lsn < MyReplicationSlot->data.restart_lsn)
+		{
+			ereport(LOG,
+					errmsg("waiting for remote slot \"%s\" LSN (%X/%X) to pass local slot LSN (%X/%X)",
+						   slot_name,
+						   LSN_FORMAT_ARGS(target_lsn), LSN_FORMAT_ARGS(MyReplicationSlot->data.restart_lsn)));
+
+			wait_for_primary_slot_catchup(wrconn, slot_name,
+										  MyReplicationSlot->data.restart_lsn);
+		}
+
+		ReplicationSlotPersist();
+	}
+
+	endlsn = pg_logical_replication_slot_advance(target_lsn);
+
+	elog(DEBUG3, "synchronized slot %s to lsn (%X/%X)",
+		 slot_name, LSN_FORMAT_ARGS(endlsn));
+
+	ReplicationSlotRelease();
+	CommitTransactionCommand();
+}
+
+static void
+synchronize_slots(void)
+{
+	WalRcvExecResult *res;
+	WalReceiverConn *wrconn = NULL;
+	TupleTableSlot *slot;
+	Oid			slotRow[3] = {TEXTOID, TEXTOID, LSNOID};
+	StringInfoData s;
+	char	   *database;
+	char	   *err;
+	MemoryContext oldctx = CurrentMemoryContext;
+
+	if (!WalRcv)
+		return;
+
+	/* syscache access needs a transaction env. */
+	StartTransactionCommand();
+	/* make dbname live outside TX context */
+	MemoryContextSwitchTo(oldctx);
+
+	database = get_database_name(MyDatabaseId);
+	initStringInfo(&s);
+	appendStringInfo(&s, "%s dbname=%s", PrimaryConnInfo, database);
+	wrconn = walrcv_connect(s.data, true, "slot_sync", &err);
+
+	if (wrconn == NULL)
+		ereport(ERROR,
+				(errmsg("could not connect to the primary server: %s", err)));
+
+	resetStringInfo(&s);
+	appendStringInfo(&s,
+					 "SELECT slot_name, plugin, confirmed_flush_lsn"
+					 "  FROM pg_catalog.pg_replication_slots"
+					 " WHERE database = %s",
+					 quote_literal_cstr(database));
+	if (strcmp(synchronize_slot_names, "") != 0 && strcmp(synchronize_slot_names, "*") != 0)
+	{
+		char	   *rawname;
+		List	   *namelist;
+		ListCell   *lc;
+
+		rawname = pstrdup(synchronize_slot_names);
+		SplitIdentifierString(rawname, ',', &namelist);
+
+		appendStringInfoString(&s, " AND slot_name IN (");
+		foreach (lc, namelist)
+		{
+			if (lc != list_head(namelist))
+				appendStringInfoChar(&s, ',');
+			appendStringInfo(&s, "%s",
+							 quote_literal_cstr(lfirst(lc)));
+		}
+		appendStringInfoChar(&s, ')');
+	}
+
+	res = walrcv_exec(wrconn, s.data, 3, slotRow);
+	pfree(s.data);
+
+	if (res->status != WALRCV_OK_TUPLES)
+		ereport(ERROR,
+				(errmsg("could not fetch slot info from primary: %s",
+						res->err)));
+
+	CommitTransactionCommand();
+	/* CommitTransactionCommand switches to TopMemoryContext */
+	MemoryContextSwitchTo(oldctx);
+
+	slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+	while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+	{
+		char	   *slot_name;
+		char	   *plugin_name;
+		XLogRecPtr	confirmed_flush_lsn;
+		bool		isnull;
+
+		slot_name = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+		Assert(!isnull);
+
+		plugin_name = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+		Assert(!isnull);
+
+		confirmed_flush_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull));
+		Assert(!isnull);
+
+		synchronize_one_slot(wrconn, slot_name, database, plugin_name,
+							 confirmed_flush_lsn);
+
+		ExecClearTuple(slot);
+	}
+
+	walrcv_clear_result(res);
+	pfree(database);
+
+	walrcv_disconnect(wrconn);
+}
+
+/*
+ * The main loop of our worker process.
+ */
+void
+ReplSlotSyncMain(Datum main_arg)
+{
+	int			worker_slot = DatumGetInt32(main_arg);
+
+	/* Attach to slot */
+	logicalrep_worker_attach(worker_slot);
+
+	/* Establish signal handlers. */
+	BackgroundWorkerUnblockSignals();
+
+	/* Load the libpq-specific functions */
+	load_file("libpqwalreceiver", false);
+
+	/* Connect to our database. */
+	BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
+											  MyLogicalRepWorker->userid,
+											  0);
+
+	StartTransactionCommand();
+	ereport(LOG,
+			(errmsg("replication slot synchronization worker for database \"%s\" has started",
+					get_database_name(MyLogicalRepWorker->dbid))));
+	CommitTransactionCommand();
+
+	/* Main wait loop. */
+	for (;;)
+	{
+		int			rc;
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!RecoveryInProgress())
+			return;
+
+		if (strcmp(synchronize_slot_names, "") == 0)
+			return;
+
+		synchronize_slots();
+
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   wal_retrieve_retry_interval,
+					   WAIT_EVENT_REPL_SLOT_SYNC_MAIN);
+
+		ResetLatch(MyLatch);
+
+		/* emergency bailout if postmaster has died */
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+	}
+}
+
+/*
+ * Routines for handling the GUC variable(s)
+ */
+
+bool
+check_synchronize_slot_names(char **newval, void **extra, GucSource source)
+{
+	/* Special handling for "*" which means all. */
+	if (strcmp(*newval, "*") == 0)
+	{
+		return true;
+	}
+	else
+	{
+		char	   *rawname;
+		List	   *namelist;
+		ListCell   *lc;
+
+		/* Need a modifiable copy of string */
+		rawname = pstrdup(*newval);
+
+		/* Parse string into list of identifiers */
+		if (!SplitIdentifierString(rawname, ',', &namelist))
+		{
+			/* syntax error in name list */
+			GUC_check_errdetail("List syntax is invalid.");
+			pfree(rawname);
+			list_free(namelist);
+			return false;
+		}
+
+		foreach(lc, namelist)
+		{
+			char	   *curname = (char *) lfirst(lc);
+
+			ReplicationSlotValidateName(curname, ERROR);
+		}
+
+		pfree(rawname);
+		list_free(namelist);
+	}
+
+	return true;
+}
+
+
+bool
+check_standby_slot_names(char **newval, void **extra, GucSource source)
+{
+	char	   *rawname;
+	List	   *namelist;
+	ListCell   *lc;
+
+	/* Need a modifiable copy of string */
+	rawname = pstrdup(*newval);
+
+	/* Parse string into list of identifiers */
+	if (!SplitIdentifierString(rawname, ',', &namelist))
+	{
+		/* syntax error in name list */
+		GUC_check_errdetail("List syntax is invalid.");
+		pfree(rawname);
+		list_free(namelist);
+		return false;
+	}
+
+	foreach(lc, namelist)
+	{
+		char	   *curname = (char *) lfirst(lc);
+
+		ReplicationSlotValidateName(curname, ERROR);
+	}
+
+	pfree(rawname);
+	list_free(namelist);
+
+	return true;
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a43c..0e0593f716 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -100,6 +100,7 @@
 #include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_type.h"
 #include "commands/copy.h"
+#include "commands/subscriptioncmds.h"
 #include "miscadmin.h"
 #include "parser/parse_relation.h"
 #include "pgstat.h"
@@ -151,7 +152,8 @@ finish_sync_worker(void)
 	CommitTransactionCommand();

 	/* Stop gracefully */
 	proc_exit(0);

 
 			if (syncworker)
index dcb1108579..902841efe6 100644
 
 %type <node>	command
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
-				read_replication_slot timeline_history show sql_cmd
+				read_replication_slot timeline_history show sql_cmd list_slots
 %type <list>	base_backup_legacy_opt_list generic_option_list
 %type <defelt>	base_backup_legacy_opt generic_option
 %type <uintval>	opt_timeline
@@ -106,6 +107,7 @@ static SQLCmd *make_sqlcmd(void);
 %type <boolval>	opt_temporary
 %type <list>	create_slot_options create_slot_legacy_opt_list
 %type <defelt>	create_slot_legacy_opt
+%type <list>	slot_name_list slot_name_list_opt
 
 %%
 
@@ -129,6 +131,7 @@ command:
 			| read_replication_slot
 			| timeline_history
 			| show
+			| list_slots
 			| sql_cmd
 			;
 
@@ -142,6 +145,33 @@ identify_system:
 				}
 			;
 
+slot_name_list:
+			IDENT
+				{
+					$$ = list_make1($1);
+				}
+			| slot_name_list ',' IDENT
+				{
+					$$ = lappend($1, $3);
+				}
+
+slot_name_list_opt:
+			slot_name_list			{ $$ = $1; }
+			| /* EMPTY */			{ $$ = NIL; }
+		;
+
+/*
+ * LIST_SLOTS
+ */
+list_slots:
+			K_LIST_SLOTS slot_name_list_opt
+				{
+					ListSlotsCmd *cmd = makeNode(ListSlotsCmd);
+					cmd->slot_names = $2;
+					$$ = (Node *) cmd;
+				}
+			;
+
 /*
  * READ_REPLICATION_SLOT %s
  */
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index 1b599c255e..9ee638355d 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -85,6 +85,7 @@ identifier		{ident_start}{ident_cont}*
 BASE_BACKUP			{ return K_BASE_BACKUP; }
 FAST			{ return K_FAST; }
 IDENTIFY_SYSTEM		{ return K_IDENTIFY_SYSTEM; }
+LIST_SLOTS		{ return K_LIST_SLOTS; }
 READ_REPLICATION_SLOT	{ return K_READ_REPLICATION_SLOT; }
 SHOW		{ return K_SHOW; }
 LABEL			{ return K_LABEL; }
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index d11daeb1fc..e93fea55ad 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -484,7 +484,7 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto)
  * WAL and removal of old catalog tuples.  As decoding is done in fast_forward
  * mode, no changes are generated anyway.
  */
-static XLogRecPtr
+XLogRecPtr
 pg_logical_replication_slot_advance(XLogRecPtr moveto)
 {
 	LogicalDecodingContext *ctx;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 84915ed95b..2fce290ed6 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -456,6 +456,194 @@ IdentifySystem(void)
 	end_tup_output(tstate);
 }
 
+static int
+pg_qsort_namecmp(const void *a, const void *b)
+{
+	return strncmp(NameStr(*(Name) a), NameStr(*(Name) b), NAMEDATALEN);
+}
+
+/*
+ListSlots(ListSlotsCmd *cmd)
+	NameData   *slot_names;
+	int			numslot_names;
+	numslot_names = list_length(cmd->slot_names);
+	if (numslot_names)
+		int			i = 0;
+		slot_names = palloc(numslot_names * sizeof(NameData));
+		foreach(lc, cmd->slot_names)
+			char	   *slot_name = lfirst(lc);
+			ReplicationSlotValidateName(slot_name, ERROR);
+			namestrcpy(&slot_names[i++], slot_name);
+		qsort(slot_names, numslot_names, sizeof(NameData), pg_qsort_namecmp);
+	for (int slotno = 0; slotno < max_replication_slots; slotno++)
+		if (numslot_names &&
+			!bsearch((void *) &slot_name, (void *) slot_names,
+					 numslot_names, sizeof(NameData), pg_qsort_namecmp))
+					 LSN_FORMAT_ARGS(restart_lsn));
+					 "%X/%X", LSN_FORMAT_ARGS(confirmed_flush_lsn));
@@ -554,7 +742,6 @@ ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
 	end_tup_output(tstate);
 }
 
-
 /*
  * Handle TIMELINE_HISTORY command.
  */
@@ -1749,6 +1936,13 @@ exec_replication_command(const char *cmd_string)
 			EndReplicationCommand(cmdtag);
 			break;
 
+		case T_ListSlotsCmd:
+			cmdtag = "LIST_SLOTS";
+			set_ps_display(cmdtag);
+			ListSlots((ListSlotsCmd *) cmd_node);
+			EndReplicationCommand(cmdtag);
+			break;
+
 		case T_StartReplicationCmd:
 			{
 				StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 4d53f040e8..6922353c94 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -230,6 +230,9 @@ pgstat_get_wait_activity(WaitEventActivity w)
 		case WAIT_EVENT_LOGICAL_LAUNCHER_MAIN:
 			event_name = "LogicalLauncherMain";
 			break;
+		case WAIT_EVENT_REPL_SLOT_SYNC_MAIN:
+			event_name = "ReplSlotSyncMain";
+			break;
 		case WAIT_EVENT_PGSTAT_MAIN:
 			event_name = "PgStatMain";
 			break;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index f736e8d872..ff1eab0207 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -75,6 +75,7 @@
 #include "postmaster/syslogger.h"
 #include "postmaster/walwriter.h"
 #include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
 #include "replication/reorderbuffer.h"
 #include "replication/slot.h"
 #include "replication/syncrep.h"
@@ -4638,6 +4639,28 @@ static struct config_string ConfigureNamesString[] =
 		check_backtrace_functions, assign_backtrace_functions, NULL
 	},
 
+	{
+		{"synchronize_slot_names", PGC_SIGHUP, REPLICATION_STANDBY,
+			gettext_noop("Sets the names of replication slots which to synchronize from primary to standby."),
+			gettext_noop("Value of \"*\" means all."),
+			GUC_LIST_INPUT | GUC_LIST_QUOTE
+		},
+		&synchronize_slot_names,
+		"",
+		check_synchronize_slot_names, NULL, NULL
+	},
+
+	{
+		{"standby_slot_names", PGC_SIGHUP, REPLICATION_PRIMARY,
+			gettext_noop("List of physical slots that must confirm changes before changes are sent to logical replication consumers."),
+			NULL,
+			GUC_LIST_INPUT | GUC_LIST_QUOTE
+		},
+		&standby_slot_names,
+		"",
+		check_standby_slot_names, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, NULL, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index a1acd46b61..e8b5f76125 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -315,6 +315,7 @@
 				# and comma-separated list of application_name
 				# from standby(s); '*' = all
 #vacuum_defer_cleanup_age = 0	# number of xacts by which cleanup is delayed
+#standby_slot_names = ''	# physical standby slot names that logical replication waits for
 
 # - Standby Servers -
 
@@ -343,6 +344,7 @@
 #wal_retrieve_retry_interval = 5s	# time to wait before retrying to
 					# retrieve WAL after a failed attempt
 #recovery_min_apply_delay = 0		# minimum delay for applying changes during recovery
+#synchronize_slot_names = ''		# logical replication slots to sync to standby
 
 # - Subscribers -

 

 	/*
 	 * TAGS FOR RANDOM OTHER STUFF
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index a746fafc12..3f81409e50 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -33,6 +33,15 @@ typedef struct IdentifySystemCmd
 	NodeTag		type;
 } IdentifySystemCmd;
 
+	List	   *slot_names;
+} ListSlotsCmd;
 
index 2ad61a001a..f5b5ef07e8 100644
@@ -12,8 +12,17 @@
 #ifndef LOGICALWORKER_H
 #define LOGICALWORKER_H
 
+#include "utils/guc.h"
+
+extern char *synchronize_slot_names;
+extern char *standby_slot_names;
+
 extern void ApplyWorkerMain(Datum main_arg);
+extern void ReplSlotSyncMain(Datum main_arg);

+extern bool check_synchronize_slot_names(char **newval, void **extra, GucSource source);
+extern bool check_standby_slot_names(char **newval, void **extra, GucSource source);
+
 #endif							/* LOGICALWORKER_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 53d773ccff..a29e517707 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -216,7 +216,6 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid);
 extern bool InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno);
 extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
 extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, int szslot);
-extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);

index 0b607ed777..dbeb447e7a 100644
 
+/*

+/*
+ * TODO
+ */
+typedef List *(*walrcv_list_slots_fn) (WalReceiverConn *conn, const char *slots);
+
 /*
+#define walrcv_list_slots(conn, slots) \
+	WalReceiverFunctions->walrcv_list_slots(conn, slots)
index 9d29849d80..5de7f80e79 100644
@@ -58,7 +58,7 @@ typedef struct LogicalRepWorker
 
 	/* Stats. */
 	XLogRecPtr	last_lsn;
@@ -81,13 +81,13 @@ extern LogicalRepWorker *MyLogicalRepWorker;
 extern bool in_remote_transaction;

index 8785a8e12c..3274eade53 100644
index 0000000000..c87e7dc016
@@ -0,0 +1,58 @@
+$node_primary->append_conf('postgresql.conf', "standby_slot_names = 'pslot1'");
+$node_primary->psql('postgres', q{SELECT pg_create_physical_replication_slot('pslot1');});
+$node_phys_standby->append_conf('postgresql.conf', "synchronize_slot_names = '*'");
+$node_phys_standby->append_conf('postgresql.conf', "primary_slot_name = 'pslot1'");
+	"SELECT slot_name, plugin, database FROM pg_replication_slots");
+$node_primary->safe_psql('postgres', "INSERT INTO t1 VALUES (4), (5), (6)");
+$node_primary->wait_for_catchup('sub1');

base-commit: ece8c76192fee0b78509688325631ceabca44ff5
-- 
2.34.1

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 14 Dec 2021 23:13:43 +0100

> discussed this topic but I think we need it at least for testing

This has been addressed by patch v2 that adds such a setting.

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 14 Dec 2021 23:19:05 +0100

> Is there a case to be made about doing the same thing for physical

It has been considered.  At the moment, I'm not doing it, because it 
would add more code and complexity and it's not that important.  But it 
could be added in the future.

> needing to brush up on the topic.

FWIW, the way the replication command parser works, if there is a parse 
error, it tries to interpret the command as a plain SQL command.  But 
that only works for logical replication connections.  So in physical 
replication, if you try to run anything that does not parse, you will 
get this error.  But that has nothing to do with this feature.  The 
above command works for me, so maybe something else went wrong in your 
situation.

> the moment), well then, should we continue to allow that?

This has been added in v2.

> want to kill it publically very early then ;-)

I don't know what you mean by this.

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 14 Dec 2021 23:24:02 +0100

> it support logical replication slots and to get more information from

I looked at that but didn't see an obvious way to consolidate them. 
This is something we could look at again later.

> subscriber is down/crashed before it picks up the latest slot

The interval is currently hardcoded, but could be a configuration 
setting.  In the v2 patch, there is a new setting that orders physical 
replication before logical so that the logical subscribers cannot get 
ahead of the physical standby.

> the walsender, if yes, isn't it going to be much simpler?

This sounds like the failover slot feature, which was rejected.

---

From: "Hsu, John" <[email-masked]>
Date: Thu, 16 Dec 2021 02:15:42 +0000

Hello,

I started taking a brief look at the v2 patch, and it does appear to work for the basic case. Logical slot is synchronized across and I can connect to the promoted standby and stream changes afterwards.

It's not clear to me what the correct behavior is when a logical slot that has been synced to the replica and then it gets deleted on the writer. Would we expect this to be propagated or leave it up to the end-user to manage?

> +               for (int i = 0; i < max_replication_slots; i++)

Even though standby_slot_names is PGC_SIGHUP, we never reload/re-process the value. If we have a wrong entry in there, the backend becomes stuck until we re-establish the logical connection. Adding "postmaster/interrupt.h" with ConfigReloadPending / ProcessConfigFile does seem to work.

Another thing I noticed is that once it starts waiting in this block, Ctrl+C doesn't seem to terminate the backend?

pg_recvlogical -d postgres -p 5432 --slot regression_slot --start -f -
..
^Cpg_recvlogical: error: unexpected termination of replication stream: 

The logical backend connection is still present:

ps aux | grep 51263
   hsuchen 51263 80.7  0.0 320180 14304 ?        Rs   01:11   3:04 postgres: walsender hsuchen [local] START_REPLICATION

pstack 51263
#0  0x00007ffee99e79a5 in clock_gettime ()
#1  0x00007f8705e88246 in clock_gettime () from /lib64/libc.so.6
#2  0x000000000075f141 in WaitEventSetWait ()
#3  0x000000000075f565 in WaitLatch ()
#4  0x0000000000720aea in ReorderBufferProcessTXN ()
#5  0x00000000007142a6 in DecodeXactOp ()
#6  0x000000000071460f in LogicalDecodingProcessRecord ()

It can be terminated with a pg_terminate_backend though.

If we have a physical slot with name foo on the standby, and then a logical slot is created on the writer with the same slot_name it does error out on the replica although it prevents other slots from being synchronized which is probably fine.

2021-12-16 02:10:29.709 UTC [73788] LOG:  replication slot synchronization worker for database "postgres" has started
2021-12-16 02:10:29.713 UTC [73788] ERROR:  cannot use physical replication slot for logical decoding
2021-12-16 02:10:29.714 UTC [73037] DEBUG:  unregistering background worker "replication slot synchronization worker"

On 12/14/21, 2:26 PM, "Peter Eisentraut" <[email-masked]> wrote:

    CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe.

---

From: Ashutosh Sharma <[email-masked]>
Date: Mon, 7 Feb 2022 13:38:38 +0530

Hi Andres,

Are you talking about this scenario - what if the logical replication
slot on the publisher is dropped, but is being referenced by the
standby where the slot is synchronized? Should the redo function for
the drop replication slot have the capability to drop it on standby
and its subscribers (if any) as well?

--
With Regards,
Ashutosh Sharma.

On Sun, Feb 6, 2022 at 1:29 AM Andres Freund <[email-masked]> wrote:
>
> Hi,
>
> On 2022-01-03 14:46:52 +0100, Peter Eisentraut wrote:
> > From ec00dc6ab8bafefc00e9b1c78ac9348b643b8a87 Mon Sep 17 00:00:00 2001
> > Date: Mon, 3 Jan 2022 14:43:36 +0100
> > Subject: [PATCH v3] Synchronize logical replication slots from primary to
> >  standby
>
> I've just skimmed the patch and the related threads. As far as I can tell this
> cannot be safely used without the conflict handling in [1], is that correct?
>
>
> Greetings,
>
> Andres Freund
>
> [1] https://postgr.es/m/CA%2BTgmoZd-JqNL1-R3RJ0jQRD%2B-dc94X0nPJgh%2BdwdDF0rFuE3g%40mail.gmail.com
>
>

---

From: Andres Freund <[email-masked]>
Date: Mon, 7 Feb 2022 12:32:22 -0800

Hi,

On 2022-02-07 13:38:38 +0530, Ashutosh Sharma wrote:
> standby where the slot is synchronized?

It's a bit hard to say, because neither in this thread nor in the patch I've
found a clear description of what the syncing needs to & tries to
guarantee. It might be that that was discussed in one of the precursor
threads, but...

Generally I don't think we can permit scenarios where a slot can be in a
"corrupt" state, i.e. missing required catalog entries, after "normal"
administrative commands (i.e. not mucking around in catalog entries / on-disk
files). Even if the sequence of commands may be a bit weird. All such cases
need to be either prevented or detected.

As far as I can tell, the way this patch keeps slots on physical replicas
"valid" is solely by reorderbuffer.c blocking during replay via
wait_for_standby_confirmation().

Which means that if e.g. the standby_slot_names GUC differs from
synchronize_slot_names on the physical replica, the slots synchronized on the
physical replica are not going to be valid.  Or if the primary drops its
logical slots.

> Should the redo function for the drop replication slot have the capability
> to drop it on standby and its subscribers (if any) as well?

Slots are not WAL logged (and shouldn't be).

I think you pretty much need the recovery conflict handling infrastructure I
referenced upthread, which recognized during replay if a record has a conflict
with a slot on a standby.  And then ontop of that you can build something like
this patch.

Greetings,

Andres Freund

---

From: Ashutosh Sharma <[email-masked]>
Date: Tue, 8 Feb 2022 20:27:32 +0530

> with a slot on a standby.  And then ontop of that you can build something like

OK. Understood, thanks Andres.

--
With Regards,
Ashutosh Sharma.

---

From: Bruce Momjian <[email-masked]>
Date: Thu, 10 Feb 2022 16:47:15 -0500

> OK. Understood, thanks Andres.

I would love to see this feature in PG 15.  Can someone explain its
current status?  Thanks.

-- 
  Bruce Momjian  <[email-masked]>        https://momjian.us
  EDB                                      https://enterprisedb.com

  If only the physical world exists, free will is an illusion.

---

From: Peter Eisentraut <[email-masked]>
Date: Fri, 11 Feb 2022 15:26:03 +0100

> I would love to see this feature in PG 15.  Can someone explain its

The way I understand it:

1. This feature (probably) depends on the "Minimal logical decoding on 
standbys" patch.  The details there aren't totally clear (to me).  That 
patch had some activity lately but I don't see it in a state that it's 
nearing readiness.

2. I think the way this (my) patch is currently written needs some 
refactoring about how we launch and manage workers.  Right now, it's all 
mangled together with logical replication, since that is a convenient 
way to launch and manage workers, but it really doesn't need to be tied 
to logical replication, since it can also be used for other logical slots.

3. It's an open question how to configure this.  My patch show a very 
minimal configuration that allows you to keep all logical slots always 
behind one physical slot, which addresses one particular use case.  In 
general, you might have things like, one set of logical slots should 
stay behind one physical slot, another set behind another physical slot, 
another set should not care, etc.  This could turn into something like 
the synchronous replication feature, where it ends up with its own 
configuration language.

Each of these are clearly significant jobs on their own.

---

From: James Coleman <[email-masked]>
Date: Fri, 18 Feb 2022 10:25:44 -0500

> > I would love to see this feature in PG 15.  Can someone explain its

Hi Peter,

I'm starting to review this patch, and last time I checked I noticed
it didn't seem to apply cleanly to master anymore. Would you be able
to send a rebased version?

Thanks,
James Coleman

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Tue, 15 Nov 2022 10:02:17 +0100

Hi,

> patch had some activity lately but I don't see it in a state that it's 

FWIW, a proposal has been submitted in [1] to add information in the WAL 
records in preparation for logical slot conflict handling.

[1]: 
https://www.postgresql.org/message-id/[email-masked]

Regards,

-- 
Bertrand Drouvot
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Fri, 14 Apr 2023 15:22:26 +0200

Hi,

On 11/15/22 10:02 AM, Drouvot, Bertrand wrote:
>> 1. This feature (probably) depends on the "Minimal logical decoding on standbys" patch.  The details there aren't totally clear (to me).  That patch had some activity lately but I don't see it in a state that it's nearing readiness.
>>
> 
> FWIW, a proposal has been submitted in [1] to add information in the WAL records in preparation for logical slot conflict handling.
> 
> [1]: https://www.postgresql.org/message-id/[email-masked]
> 

Now that the "Minimal logical decoding on standby" patch series (mentioned up-thread) has been
committed, I think we can resume working on this one ("Synchronizing slots from primary to standby").

I'll work on a rebase and share it once done (unless someone already started working on a rebase).

Regards,

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Mon, 17 Apr 2023 16:05:58 +0200

Hi,

> I'll work on a rebase and share it once done (unless someone already started working on a rebase).

Please find attached V5 (a rebase of V4 posted up-thread).

In addition to the "rebasing" work, the TAP test adds a test about conflict handling (logical slot invalidation)
relying on the work done in the "Minimal logical decoding on standby" patch series.

I did not look more at the patch (than what's was needed for the rebase) but plan to do so.

Regards,

From 655359eedf37d8f2e522aeb1ec8c48adfc1759b1 Mon Sep 17 00:00:00 2001
From: Bertrand Drouvot <[email-masked]>
Date: Thu, 13 Apr 2023 11:32:28 +0000
Subject: [PATCH v5] Synchronize logical replication slots from primary to
 standby

---
 doc/src/sgml/config.sgml                      |  34 ++
 src/backend/commands/subscriptioncmds.c       |   4 +-
 src/backend/postmaster/bgworker.c             |   3 +
 .../libpqwalreceiver/libpqwalreceiver.c       |  95 ++++
 src/backend/replication/logical/Makefile      |   1 +
 src/backend/replication/logical/launcher.c    | 263 +++++++----
 src/backend/replication/logical/meson.build   |   1 +
 .../replication/logical/reorderbuffer.c       |  86 ++++
 src/backend/replication/logical/slotsync.c    | 413 ++++++++++++++++++
 src/backend/replication/logical/tablesync.c   |  13 +-
 src/backend/replication/logical/worker.c      |   3 +-
 src/backend/replication/repl_gram.y           |  32 +-
 src/backend/replication/repl_scanner.l        |   2 +
 src/backend/replication/slotfuncs.c           |   2 +-
 src/backend/replication/walsender.c           | 195 +++++++++
 src/backend/utils/activity/wait_event.c       |   3 +
 src/backend/utils/misc/guc_tables.c           |  26 ++
 src/backend/utils/misc/postgresql.conf.sample |   2 +
 src/include/commands/subscriptioncmds.h       |   3 +
 src/include/nodes/replnodes.h                 |   9 +
 src/include/replication/logicallauncher.h     |   2 +
 src/include/replication/logicalworker.h       |   9 +
 src/include/replication/slot.h                |   5 +-
 src/include/replication/walreceiver.h         |  20 +
 src/include/replication/worker_internal.h     |   8 +-
 src/include/utils/wait_event.h                |   1 +
 src/test/recovery/meson.build                 |   1 +
 src/test/recovery/t/037_slot_sync.pl          | 130 ++++++
 28 files changed, 1272 insertions(+), 94 deletions(-)
   3.8% doc/src/sgml/
   7.1% src/backend/replication/libpqwalreceiver/
  54.7% src/backend/replication/logical/
  14.9% src/backend/replication/
   3.3% src/backend/
   4.0% src/include/replication/
  10.9% src/test/recovery/t/

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 091a79d4f3..1360885208 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4466,6 +4466,23 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
       </listitem>
      </varlistentry>

@@ -4649,6 +4666,23 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       </listitem>
      </varlistentry>

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 3251d89ba8..8721706b79 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -991,7 +991,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,

 
 				/*
 				 * For READY state, we would have already dropped the
@@ -1589,7 +1589,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	{
 		LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);

diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 0dd22b2351..a89d1f10a1 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -129,6 +129,9 @@ static const struct
 	{
 		"ApplyWorkerMain", ApplyWorkerMain
 	},
+	{
+		"ReplSlotSyncMain", ReplSlotSyncMain
+	},
 	{
 		"ParallelApplyWorkerMain", ParallelApplyWorkerMain
 	}
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 052505e46f..4f7417c49a 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -34,6 +34,7 @@
 #include "utils/memutils.h"
 #include "utils/pg_lsn.h"
 #include "utils/tuplestore.h"
+#include "utils/varlena.h"
 
 PG_MODULE_MAGIC;
 
@@ -58,6 +59,7 @@ static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
 									char **sender_host, int *sender_port);
 static char *libpqrcv_identify_system(WalReceiverConn *conn,
 									  TimeLineID *primary_tli);
+static List *libpqrcv_list_slots(WalReceiverConn *conn, const char *slot_names);
 static int	libpqrcv_server_version(WalReceiverConn *conn);
 static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
 											 TimeLineID tli, char **filename,
@@ -96,6 +98,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
 	.walrcv_receive = libpqrcv_receive,
 	.walrcv_send = libpqrcv_send,
 	.walrcv_create_slot = libpqrcv_create_slot,
+	.walrcv_list_slots = libpqrcv_list_slots,
 	.walrcv_get_backend_pid = libpqrcv_get_backend_pid,
 	.walrcv_exec = libpqrcv_exec,
 	.walrcv_disconnect = libpqrcv_disconnect
@@ -409,6 +412,98 @@ libpqrcv_server_version(WalReceiverConn *conn)
 	return PQserverVersion(conn->streamConn);
 }
 
+/*
+ * Get list of slots from primary.
+ */
+static List *
+libpqrcv_list_slots(WalReceiverConn *conn, const char *slot_names)
+{
+	PGresult   *res;
+	List	   *slotlist = NIL;
+	int			ntuples;
+	StringInfoData s;
+	WalRecvReplicationSlotData *slot_data;
+
+	initStringInfo(&s);
+	appendStringInfoString(&s, "LIST_SLOTS");
+
+	if (strcmp(slot_names, "") != 0 && strcmp(slot_names, "*") != 0)
+	{
+		char	   *rawname;
+		List	   *namelist;
+		ListCell   *lc;
+
+		appendStringInfoChar(&s, ' ');
+		rawname = pstrdup(slot_names);
+		SplitIdentifierString(rawname, ',', &namelist);
+		foreach (lc, namelist)
+		{
+			if (lc != list_head(namelist))
+				appendStringInfoChar(&s, ',');
+			appendStringInfo(&s, "%s",
+							 quote_identifier(lfirst(lc)));
+		}
+	}
+
+	res = libpqrcv_PQexec(conn->streamConn, s.data);
+	pfree(s.data);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("could not receive list of slots the primary server: %s",
+						pchomp(PQerrorMessage(conn->streamConn)))));
+	}
+	if (PQnfields(res) < 10)
+	{
+		int			nfields = PQnfields(res);
+
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("invalid response from primary server"),
+				 errdetail("Could not get list of slots: got %d fields, expected %d or more fields.",
+						   nfields, 10)));
+	}
+
+	ntuples = PQntuples(res);
+	for (int i = 0; i < ntuples; i++)
+	{
+		char	   *slot_type;
+
+		slot_data = palloc0(sizeof(WalRecvReplicationSlotData));
+		namestrcpy(&slot_data->persistent_data.name, PQgetvalue(res, i, 0));
+		if (!PQgetisnull(res, i, 1))
+			namestrcpy(&slot_data->persistent_data.plugin, PQgetvalue(res, i, 1));
+		slot_type = PQgetvalue(res, i, 2);
+		if (!PQgetisnull(res, i, 3))
+			slot_data->persistent_data.database = atooid(PQgetvalue(res, i, 3));
+		if (strcmp(slot_type, "physical") == 0)
+		{
+			if (OidIsValid(slot_data->persistent_data.database))
+				elog(ERROR, "unexpected physical replication slot with database set");
+		}
+		if (pg_strtoint32(PQgetvalue(res, i, 5)) == 1)
+			slot_data->persistent_data.persistency = RS_TEMPORARY;
+		else
+			slot_data->persistent_data.persistency = RS_PERSISTENT;
+		if (!PQgetisnull(res, i, 6))
+			slot_data->persistent_data.xmin = atooid(PQgetvalue(res, i, 6));
+		if (!PQgetisnull(res, i, 7))
+			slot_data->persistent_data.catalog_xmin = atooid(PQgetvalue(res, i, 7));
+		if (!PQgetisnull(res, i, 8))
+			slot_data->persistent_data.restart_lsn = strtou64(PQgetvalue(res, i, 8), NULL, 10);
+		if (!PQgetisnull(res, i, 9))
+			slot_data->persistent_data.confirmed_flush = strtou64(PQgetvalue(res, i, 9), NULL, 10);
+
+		slot_data->last_sync_time = 0;
+		slotlist = lappend(slotlist, slot_data);
+	}
+
+	PQclear(res);
+
+	return slotlist;
+}
+
 /*
  * Start streaming WAL data from given streaming options.
  *
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 2dc25e37bb..ba03eeff1c 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -25,6 +25,7 @@ OBJS = \
 	proto.o \
 	relation.o \
 	reorderbuffer.o \
+	slotsync.o \
 	snapbuild.o \
 	tablesync.o \
 	worker.o
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 970d170e73..14af724639 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -22,6 +22,7 @@
 #include "access/htup_details.h"
 #include "access/tableam.h"
 #include "access/xact.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
 #include "funcapi.h"
@@ -246,7 +247,7 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
  * We are only interested in the leader apply worker or table sync worker.
  */
 LogicalRepWorker *
-logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
+logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running)
 {
 	int			i;
 	LogicalRepWorker *res = NULL;
@@ -262,8 +263,8 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
 		if (isParallelApplyWorker(w))
 			continue;
 
-		if (w->in_use && w->subid == subid && w->relid == relid &&
-			(!only_running || w->proc))
+		if (w->in_use && w->dbid == dbid && w->subid == subid &&
+			w->relid == relid && (!only_running || w->proc))
 		{
 			res = w;
 			break;
@@ -320,9 +321,13 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
 	/* Sanity check - tablesync worker cannot be a subworker */
 	Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
 
-	ereport(DEBUG1,
 
@@ -359,7 +364,9 @@ retry:
 
@@ -455,15 +462,20 @@ retry:
 
-	if (is_parallel_apply_worker)
+	if (!OidIsValid(subid))
+		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ReplSlotSyncMain");
+	else if (is_parallel_apply_worker)
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
 	else
 		snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
 
-	if (OidIsValid(relid))
+	if (!OidIsValid(subid))
+		snprintf(bgw.bgw_name, BGW_MAXLEN,
+				 "replication slot synchronization worker");
+	else if (OidIsValid(relid))
 		snprintf(bgw.bgw_name, BGW_MAXLEN,
 				 "logical replication worker for subscription %u sync %u", subid, relid);
 	else if (is_parallel_apply_worker)
@@ -591,13 +603,13 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
  * Stop the logical replication worker for subid/relid, if any.
  */
 void
-logicalrep_worker_stop(Oid subid, Oid relid)
+logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid)
 {
 	LogicalRepWorker *worker;

 
 	if (worker)
 	{
@@ -640,13 +652,13 @@ logicalrep_pa_worker_stop(int slot_no, uint16 generation)

 
 	if (worker)
@@ -888,7 +900,7 @@ ApplyLauncherRegister(void)
@@ -1071,6 +1083,157 @@ ApplyLauncherWakeup(void)
 
+static void
+ApplyLauncherStartSlotSync(long *wait_time)
+{
+	WalReceiverConn *wrconn;
+	char	   *err;
+	List	   *slots;
+	ListCell   *lc;
+	MemoryContext tmpctx;
+	MemoryContext oldctx;
+
+	if (strcmp(synchronize_slot_names, "") == 0)
+		return;
+
+	wrconn = walrcv_connect(PrimaryConnInfo, false, false,
+							"Logical Replication Launcher", &err);
+	if (!wrconn)
+		ereport(ERROR,
+				(errmsg("could not connect to the primary server: %s", err)));
+
+	/* Use temporary context for the slot list and worker info. */
+	tmpctx = AllocSetContextCreate(TopMemoryContext,
+								   "Logical Replication Launcher slot sync ctx",
+								   ALLOCSET_DEFAULT_SIZES);
+	oldctx = MemoryContextSwitchTo(tmpctx);
+
+	slots = walrcv_list_slots(wrconn, synchronize_slot_names);
+
+	foreach(lc, slots)
+	{
+		WalRecvReplicationSlotData *slot_data = lfirst(lc);
+		LogicalRepWorker *w;
+		TimestampTz last_sync;
+		TimestampTz	now;
+		long		elapsed;
+
+		if (!OidIsValid(slot_data->persistent_data.database))
+			continue;
+
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+		w = logicalrep_worker_find(slot_data->persistent_data.database, InvalidOid,
+								   InvalidOid, false);
+		LWLockRelease(LogicalRepWorkerLock);
+
+		if (w != NULL)
+			continue;		/* worker is running already */
+
+		/*
+		 * If the worker is eligible to start now, launch it.  Otherwise,
+		 * adjust wait_time so that we'll wake up as soon as it can be
+		 * started.
+		 *
+		 * Each apply worker can only be restarted once per
+		 * wal_retrieve_retry_interval, so that errors do not cause us to
+		 * repeatedly restart the worker as fast as possible.
+		 */
+		last_sync = slot_data->last_sync_time;
+		now = GetCurrentTimestamp();
+		if (last_sync == 0 ||
+			(elapsed = TimestampDifferenceMilliseconds(last_sync, now)) >= wal_retrieve_retry_interval)
+		{
+			slot_data->last_sync_time = now;
+			logicalrep_worker_launch(slot_data->persistent_data.database,
+									 InvalidOid, NULL,
+									 BOOTSTRAP_SUPERUSERID, InvalidOid,
+									 DSM_HANDLE_INVALID);
+		}
+		else
+		{
+			*wait_time = Min(*wait_time,
+							wal_retrieve_retry_interval - elapsed);
+		}
+	}
+
+	/* Switch back to original memory context. */
+	MemoryContextSwitchTo(oldctx);
+	/* Clean the temporary memory. */
+	MemoryContextDelete(tmpctx);
+
+	walrcv_disconnect(wrconn);
+}
+
+static void
+ApplyLauncherStartSubs(long *wait_time)
+{
+	List	   *sublist;
+	ListCell   *lc;
+	MemoryContext subctx;
+	MemoryContext oldctx;
+
+	/* Use temporary context to avoid leaking memory across cycles. */
+	subctx = AllocSetContextCreate(TopMemoryContext,
+								   "Logical Replication Launcher sublist",
+								   ALLOCSET_DEFAULT_SIZES);
+	oldctx = MemoryContextSwitchTo(subctx);
+
+	/* Start the missing workers for enabled subscriptions. */
+	sublist = get_subscription_list();
+	foreach(lc, sublist)
+	{
+		Subscription *sub = (Subscription *) lfirst(lc);
+		LogicalRepWorker *w;
+		TimestampTz last_start;
+		TimestampTz now;
+		long		elapsed;
+
+		if (!sub->enabled)
+			continue;
+
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+		w = logicalrep_worker_find(sub->dbid, sub->oid, InvalidOid, false);
+		LWLockRelease(LogicalRepWorkerLock);
+
+		if (w != NULL)
+			continue;		/* worker is running already */
+
+		/*
+		 * If the worker is eligible to start now, launch it.  Otherwise,
+		 * adjust wait_time so that we'll wake up as soon as it can be
+		 * started.
+		 *
+		 * Each subscription's apply worker can only be restarted once per
+		 * wal_retrieve_retry_interval, so that errors do not cause us to
+		 * repeatedly restart the worker as fast as possible.  In cases
+		 * where a restart is expected (e.g., subscription parameter
+		 * changes), another process should remove the last-start entry
+		 * for the subscription so that the worker can be restarted
+		 * without waiting for wal_retrieve_retry_interval to elapse.
+		 */
+		last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
+		now = GetCurrentTimestamp();
+		if (last_start == 0 ||
+			(elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
+		{
+			ApplyLauncherSetWorkerStartTime(sub->oid, now);
+			logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
+									 sub->owner, InvalidOid,
+									 DSM_HANDLE_INVALID);
+		}
+		else
+		{
+			*wait_time = Min(*wait_time,
+							wal_retrieve_retry_interval - elapsed);
+		}
+	}
+
+	/* Switch back to original memory context. */
+	MemoryContextSwitchTo(oldctx);
+	/* Clean the temporary memory. */
+	MemoryContextDelete(subctx);
+}
+
 /*
  * Main loop for the apply launcher process.
  */
@@ -1096,78 +1259,20 @@ ApplyLauncherMain(Datum main_arg)
 	 */
 	BackgroundWorkerInitializeConnection(NULL, NULL, 0);

 		CHECK_FOR_INTERRUPTS();
 
-		/* Use temporary context to avoid leaking memory across cycles. */
-		subctx = AllocSetContextCreate(TopMemoryContext,
-									   "Logical Replication Launcher sublist",
-									   ALLOCSET_DEFAULT_SIZES);
-		oldctx = MemoryContextSwitchTo(subctx);
-
-		/* Start any missing workers for enabled subscriptions. */
-		sublist = get_subscription_list();
-		foreach(lc, sublist)
-		{
-			Subscription *sub = (Subscription *) lfirst(lc);
-			LogicalRepWorker *w;
-			TimestampTz last_start;
-			TimestampTz now;
-			long		elapsed;
-
-			if (!sub->enabled)
-				continue;
-
-			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-			w = logicalrep_worker_find(sub->oid, InvalidOid, false);
-			LWLockRelease(LogicalRepWorkerLock);
-
-			if (w != NULL)
-				continue;		/* worker is running already */
-
-			/*
-			 * If the worker is eligible to start now, launch it.  Otherwise,
-			 * adjust wait_time so that we'll wake up as soon as it can be
-			 * started.
-			 *
-			 * Each subscription's apply worker can only be restarted once per
-			 * wal_retrieve_retry_interval, so that errors do not cause us to
-			 * repeatedly restart the worker as fast as possible.  In cases
-			 * where a restart is expected (e.g., subscription parameter
-			 * changes), another process should remove the last-start entry
-			 * for the subscription so that the worker can be restarted
-			 * without waiting for wal_retrieve_retry_interval to elapse.
-			 */
-			last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
-			now = GetCurrentTimestamp();
-			if (last_start == 0 ||
-				(elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
-			{
-				ApplyLauncherSetWorkerStartTime(sub->oid, now);
-				logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
-										 sub->owner, InvalidOid,
-										 DSM_HANDLE_INVALID);
-			}
-			else
-			{
-				wait_time = Min(wait_time,
-								wal_retrieve_retry_interval - elapsed);
-			}
-		}
-
-		/* Switch back to original memory context. */
-		MemoryContextSwitchTo(oldctx);
-		/* Clean the temporary memory. */
-		MemoryContextDelete(subctx);
+		if (!RecoveryInProgress())
+			ApplyLauncherStartSubs(&wait_time);
+		else
+			ApplyLauncherStartSlotSync(&wait_time);
 
 		/* Wait for more work. */
 		rc = WaitLatch(MyLatch,
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index d48cd4c590..9e52ec421f 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -11,6 +11,7 @@ backend_sources += files(
   'proto.c',
   'relation.c',
   'reorderbuffer.c',
+  'slotsync.c',
   'snapbuild.c',
   'tablesync.c',
   'worker.c',
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 9f44974473..1519b0ec64 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -95,11 +95,14 @@
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "replication/logical.h"
+#include "replication/logicalworker.h"
 #include "replication/reorderbuffer.h"
 #include "replication/slot.h"
 #include "replication/snapbuild.h"	/* just for SnapBuildSnapDecRefcount */
 #include "storage/bufmgr.h"
 #include "storage/fd.h"
+#include "storage/ipc.h"
+#include "storage/latch.h"
 #include "storage/sinval.h"
 #include "utils/builtins.h"
 #include "utils/combocid.h"
@@ -107,6 +110,7 @@
 #include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/relfilenumbermap.h"
+#include "utils/varlena.h"

 /* entry for a hash table we use to map from xid to our transaction state */
@@ -2053,6 +2057,85 @@ ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	}
 }
 
+static void
@@ -2502,6 +2585,9 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
index 0000000000..529ddb21ae
@@ -0,0 +1,413 @@
+#include "replication/logicallauncher.h"
+	wrconn = walrcv_connect(s.data, true, false, "slot_sync", &err);
index 6dce355633..2307d187e4 100644
 #include "nodes/makefuncs.h"
@@ -155,7 +156,8 @@ finish_sync_worker(void)
 
 	/* Find the leader apply worker and signal it. */
-	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
+	logicalrep_worker_wakeup(MyLogicalRepWorker->dbid,
+							 MyLogicalRepWorker->subid, InvalidOid);
 
 	/* Stop gracefully */
 	proc_exit(0);
@@ -195,7 +197,8 @@ wait_for_relation_state_change(Oid relid, char expected_state)
 
@@ -242,7 +245,8 @@ wait_for_worker_state_change(char expected_state)
@@ -508,7 +512,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)

 			if (syncworker)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 3d58910c14..b9354bd023 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1600,7 +1600,8 @@ apply_handle_stream_start(StringInfo s)
 				 * Signal the leader apply worker, as it may be waiting for
 				 * us.
 				 */
-				logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
+				logicalrep_worker_wakeup(MyLogicalRepWorker->dbid,
+										 MyLogicalRepWorker->subid, InvalidOid);
 			}
 
 			parallel_stream_nchanges = 0;
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 0c874e33cf..12a4b74368 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -76,11 +76,12 @@ Node *replication_parse_result;
 %token K_EXPORT_SNAPSHOT
 %token K_NOEXPORT_SNAPSHOT
 %token K_USE_SNAPSHOT
+%token K_LIST_SLOTS
 
 %type <node>	command
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
-				read_replication_slot timeline_history show
+				read_replication_slot timeline_history show list_slots
 %type <list>	generic_option_list
 %type <defelt>	generic_option
 %type <uintval>	opt_timeline
@@ -91,6 +92,7 @@ Node *replication_parse_result;
 %type <boolval>	opt_temporary
 %type <list>	create_slot_options create_slot_legacy_opt_list
 %type <defelt>	create_slot_legacy_opt
+%type <list>	slot_name_list slot_name_list_opt
 
 %%
 
@@ -114,6 +116,7 @@ command:
 			| read_replication_slot
 			| timeline_history
 			| show
+			| list_slots
 			;
 
 /*
@@ -126,6 +129,33 @@ identify_system:
 				}
 			;
 
+slot_name_list:
+			IDENT
+				{
+					$$ = list_make1($1);
+				}
+			| slot_name_list ',' IDENT
+				{
+					$$ = lappend($1, $3);
+				}
+
+slot_name_list_opt:
+			slot_name_list			{ $$ = $1; }
+			| /* EMPTY */			{ $$ = NIL; }
+		;
+
+/*
+ * LIST_SLOTS
+ */
+list_slots:
+			K_LIST_SLOTS slot_name_list_opt
+				{
+					ListSlotsCmd *cmd = makeNode(ListSlotsCmd);
+					cmd->slot_names = $2;
+					$$ = (Node *) cmd;
+				}
+			;
+
 /*
  * READ_REPLICATION_SLOT %s
  */
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index cb467ca46f..9501df38eb 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -128,6 +128,7 @@ DROP_REPLICATION_SLOT		{ return K_DROP_REPLICATION_SLOT; }
 TIMELINE_HISTORY	{ return K_TIMELINE_HISTORY; }
 PHYSICAL			{ return K_PHYSICAL; }
 RESERVE_WAL			{ return K_RESERVE_WAL; }
+LIST_SLOTS			{ return K_LIST_SLOTS; }
 LOGICAL				{ return K_LOGICAL; }
 SLOT				{ return K_SLOT; }
 TEMPORARY			{ return K_TEMPORARY; }
@@ -304,6 +305,7 @@ replication_scanner_is_replication_command(void)
 		case K_READ_REPLICATION_SLOT:
 		case K_TIMELINE_HISTORY:
 		case K_SHOW:
+		case K_LIST_SLOTS:
 			/* Yes; push back the first token so we can parse later. */
 			repl_pushed_back_token = first_token;
 			return true;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 6035cf4816..83ada6db6a 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -467,7 +467,7 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto)
  * WAL and removal of old catalog tuples.  As decoding is done in fast_forward
  * mode, no changes are generated anyway.
  */
-static XLogRecPtr
+XLogRecPtr
 pg_logical_replication_slot_advance(XLogRecPtr moveto)
 {
 	LogicalDecodingContext *ctx;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 45b8b3684f..0d01b8967a 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -473,6 +473,194 @@ IdentifySystem(void)
 	end_tup_output(tstate);
 }
 
+static int
@@ -1820,6 +2008,13 @@ exec_replication_command(const char *cmd_string)
 
+		case T_ListSlotsCmd:
+			cmdtag = "LIST_SLOTS";
+			set_ps_display(cmdtag);
+			ListSlots((ListSlotsCmd *) cmd_node);
+			EndReplicationCommand(cmdtag);
+			break;
+
 		case T_StartReplicationCmd:
 			{
 				StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 7940d64639..f2a9517091 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -230,6 +230,9 @@ pgstat_get_wait_activity(WaitEventActivity w)
 		case WAIT_EVENT_LOGICAL_LAUNCHER_MAIN:
 			event_name = "LogicalLauncherMain";
 			break;
+		case WAIT_EVENT_REPL_SLOT_SYNC_MAIN:
+			event_name = "ReplSlotSyncMain";
+			break;
 		case WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN:
 			event_name = "LogicalParallelApplyMain";
 			break;
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index cab3ddbe11..0ee7ad1348 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -63,8 +63,12 @@
 #include "postmaster/syslogger.h"
 #include "postmaster/walwriter.h"
 #include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
+#include "replication/reorderbuffer.h"
 #include "replication/slot.h"
 #include "replication/syncrep.h"
+#include "replication/walreceiver.h"
+#include "replication/walsender.h"
 #include "storage/bufmgr.h"
 #include "storage/large_object.h"
 #include "storage/pg_shmem.h"
@@ -4587,6 +4591,28 @@ struct config_string ConfigureNamesString[] =
 		check_io_direct, assign_io_direct, NULL
 	},
 
+	{
index dce5049bc2..2ff2188c02 100644
@@ -330,6 +330,7 @@
 
 # - Standby Servers -
 
@@ -357,6 +358,7 @@
 
 # - Subscribers -
 
diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index 214dc6c29e..0e77f9ee5c 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -17,6 +17,7 @@

 extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 										bool isTopLevel);
@@ -28,4 +29,6 @@ extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId);
 
 extern char defGetStreamingMode(DefElem *def);
 
index 4321ba8f86..980e0b2ee2 100644

 /* ----------------------
  *		BASE_BACKUP command
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index a07c9cb311..80fdbf9657 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -31,4 +31,6 @@ extern bool IsLogicalLauncher(void);
 
 extern pid_t GetLeaderApplyWorkerPid(pid_t pid);
 
+extern PGDLLIMPORT char *PrimaryConnInfo;
+
 #endif							/* LOGICALLAUNCHER_H */
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 39588da79f..6408753557 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -14,10 +14,16 @@
 
 #include <signal.h>
 
+#include "utils/guc.h"
+
+extern char *synchronize_slot_names;
+extern char *standby_slot_names;
+
 extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending;
 
 extern void ApplyWorkerMain(Datum main_arg);
 extern void ParallelApplyWorkerMain(Datum main_arg);
+extern void ReplSlotSyncMain(Datum main_arg);
 
 extern bool IsLogicalWorker(void);
 extern bool IsLogicalParallelApplyWorker(void);
@@ -29,4 +35,7 @@ extern void LogicalRepWorkersWakeupAtCommit(Oid subid);
 
 extern void AtEOXact_LogicalRepWorkers(bool isCommit);
 
+extern bool check_synchronize_slot_names(char **newval, void **extra, GucSource source);
+extern bool check_standby_slot_names(char **newval, void **extra, GucSource source);
+
 #endif							/* LOGICALWORKER_H */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index a8a89dc784..5dc2e0d30d 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -15,7 +15,6 @@
 #include "storage/lwlock.h"
 #include "storage/shmem.h"
 #include "storage/spin.h"
-#include "replication/walreceiver.h"
 
 /*
  * Behaviour of replication slots, upon release or crash.
@@ -238,7 +237,6 @@ extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_l
 extern int	ReplicationSlotIndex(ReplicationSlot *slot);
 extern bool ReplicationSlotName(int index, Name name);
 extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot);
-extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
 
@@ -246,4 +244,7 @@ extern void CheckPointReplicationSlots(void);
 
index 281626fa6f..9e9d64faf2 100644
@@ -20,6 +20,7 @@
@@ -191,6 +192,17 @@ typedef struct
 	}			proto;
 } WalRcvStreamOptions;
 
+/*
+ * Slot information receiver from remote.
+ *
+ * Currently same as ReplicationSlotPersistentData except last_sync_time
+ */
+typedef struct WalRecvReplicationSlotData
+{
+	ReplicationSlotPersistentData persistent_data;
+	TimestampTz last_sync_time;
+} WalRecvReplicationSlotData;
+
 struct WalReceiverConn;
 typedef struct WalReceiverConn WalReceiverConn;
 
@@ -280,6 +292,11 @@ typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn,
 typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
 											TimeLineID *primary_tli);
 
+/*
+ * TODO
+ */
+typedef List *(*walrcv_list_slots_fn) (WalReceiverConn *conn, const char *slots);
+
 /*
  * walrcv_server_version_fn
  *
@@ -393,6 +410,7 @@ typedef struct WalReceiverFunctionsType
 	walrcv_get_conninfo_fn walrcv_get_conninfo;
 	walrcv_get_senderinfo_fn walrcv_get_senderinfo;
 	walrcv_identify_system_fn walrcv_identify_system;
+	walrcv_list_slots_fn walrcv_list_slots;
 	walrcv_server_version_fn walrcv_server_version;
 	walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;
 	walrcv_startstreaming_fn walrcv_startstreaming;
@@ -417,6 +435,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 	WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
 #define walrcv_identify_system(conn, primary_tli) \
 	WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
+#define walrcv_list_slots(conn, slots) \
+	WalReceiverFunctions->walrcv_list_slots(conn, slots)
 #define walrcv_server_version(conn) \
 	WalReceiverFunctions->walrcv_server_version(conn)
 #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index dce71d2c50..5b4fda2fd9 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -65,7 +65,7 @@ typedef struct LogicalRepWorker
 	 * would be created for each transaction which will be deleted after the
 	 * transaction is finished.
 	 */
-	FileSet    *stream_fileset;
+	struct FileSet    *stream_fileset;
 
 	/*
 	 * PID of leader apply worker if this slot is used for a parallel apply
@@ -226,15 +226,15 @@ extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
 extern PGDLLIMPORT bool in_remote_transaction;
 
 extern void logicalrep_worker_attach(int slot);
-extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
+extern LogicalRepWorker *logicalrep_worker_find(Oid dbid, Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running);
 extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
-extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid);
 extern void logicalrep_pa_worker_stop(int slot_no, uint16 generation);
-extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
+extern void logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
 extern int	logicalrep_sync_worker_count(Oid subid);
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index 518d3b0a1f..cccd4d9d32 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -43,6 +43,7 @@ typedef enum
 	WAIT_EVENT_LOGICAL_APPLY_MAIN,
 	WAIT_EVENT_LOGICAL_LAUNCHER_MAIN,
 	WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN,
+	WAIT_EVENT_REPL_SLOT_SYNC_MAIN,
 	WAIT_EVENT_RECOVERY_WAL_STREAM,
 	WAIT_EVENT_SYSLOGGER_MAIN,
 	WAIT_EVENT_WAL_RECEIVER_MAIN,
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 2008958010..b6fcc8704e 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -42,6 +42,7 @@ tests += {
       't/034_create_database.pl',
       't/035_standby_logical_decoding.pl',
       't/036_truncated_dropped.pl',
+      't/037_slot_sync.pl',
     ],
   },
 }
diff --git a/src/test/recovery/t/037_slot_sync.pl b/src/test/recovery/t/037_slot_sync.pl
new file mode 100644
index 0000000000..0520042d96
--- /dev/null
+++ b/src/test/recovery/t/037_slot_sync.pl
@@ -0,0 +1,130 @@
+
+# Copyright (c) 2021, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $node_primary = PostgreSQL::Test::Cluster->new('primary');
+my $node_phys_standby = PostgreSQL::Test::Cluster->new('phys_standby');
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+
+# find $pat in logfile of $node after $off-th byte
+sub find_in_log
+{
+	my ($node, $pat, $off) = @_;
+
+	$off = 0 unless defined $off;
+	my $log = PostgreSQL::Test::Utils::slurp_file($node->logfile);
+	return 0 if (length($log) <= $off);
+
+	$log = substr($log, $off);
+
+	return $log =~ m/$pat/;
+}
+
+# Check invalidation in the logfile
+sub check_for_invalidation
+{
+	my ($log_start, $test_name) = @_;
+
+	# message should be issued
+	ok( find_in_log(
+		$node_phys_standby,
+        "invalidating obsolete replication slot \"sub1\"", $log_start),
+        "sub1 slot invalidation is logged $test_name");
+}
+
+# Check conflicting status in pg_replication_slots.
+sub check_slots_conflicting_status
+{
+	my $res = $node_phys_standby->safe_psql(
+				'postgres', qq(
+				select bool_and(conflicting) from pg_replication_slots;));
+
+	is($res, 't',
+		"Logical slot is reported as conflicting");
+}
+
+$node_primary->init(allows_streaming => 'logical');
+$node_primary->append_conf('postgresql.conf', "standby_slot_names = 'pslot1'");
+$node_primary->start;
+$node_primary->psql('postgres', q{SELECT pg_create_physical_replication_slot('pslot1');});
+
+$node_primary->backup('backup');
+
+$node_phys_standby->init_from_backup($node_primary, 'backup', has_streaming => 1);
+$node_phys_standby->append_conf('postgresql.conf', q{
+synchronize_slot_names = '*'
+primary_slot_name = 'pslot1'
+hot_standby_feedback = off
+});
+
+$node_phys_standby->start;
+
+$node_primary->safe_psql('postgres', "CREATE TABLE t1 (a int PRIMARY KEY)");
+$node_primary->safe_psql('postgres', "INSERT INTO t1 VALUES (1), (2), (3)");
+
+# Some tests need to wait for VACUUM to be replayed. But vacuum does not flush
+# WAL. An insert into flush_wal outside transaction does guarantee a flush.
+$node_primary->psql('postgres', q[CREATE TABLE flush_wal();]);
+
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+$node_subscriber->safe_psql('postgres', "CREATE TABLE t1 (a int PRIMARY KEY)");
+
+$node_primary->safe_psql('postgres', "CREATE PUBLICATION pub1 FOR TABLE t1");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub1 CONNECTION '" . ($node_primary->connstr . ' dbname=postgres') . "' PUBLICATION pub1");
+
+# Wait for initial sync of all subscriptions
+my $synced_query =
+  "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+my $result = $node_primary->safe_psql('postgres',
+	"SELECT slot_name, plugin, database FROM pg_replication_slots WHERE slot_type = 'logical'");
+
+is($result, qq(sub1|pgoutput|postgres), 'logical slot on primary');
+
+# FIXME: standby needs restart to pick up new slots
+$node_phys_standby->restart;
+sleep 3;
+
+$result = $node_phys_standby->safe_psql('postgres',
+	"SELECT slot_name, plugin, database FROM pg_replication_slots");
+
+is($result, qq(sub1|pgoutput|postgres), 'logical slot on standby');
+
+$node_primary->safe_psql('postgres', "INSERT INTO t1 VALUES (4), (5), (6)");
+$node_primary->wait_for_catchup('sub1');
+
+$node_primary->wait_for_catchup($node_phys_standby->name);
+
+# Logical subscriber and physical replica are caught up at this point.
+
+# Drop the subscription so that catalog_xmin is unknown on the primary
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
+
+# This should trigger a conflict as hot_standby_feedback is off on the standby
+$node_primary->safe_psql('postgres', qq[
+  CREATE TABLE conflict_test(x integer, y text);
+  DROP TABLE conflict_test;
+  VACUUM full pg_class;
+  INSERT INTO flush_wal DEFAULT VALUES; -- see create table flush_wal
+]);
+
+# Ensure physical replay catches up
+$node_primary->wait_for_catchup($node_phys_standby);
+
+# Check invalidation in the logfile
+check_for_invalidation(1, 'with vacuum FULL on pg_class');
+
+# Check conflicting status in pg_replication_slots.
+check_slots_conflicting_status();
+
+done_testing();
-- 
2.34.1

---

From: Amit Kapila <[email-masked]>
Date: Fri, 16 Jun 2023 15:26:05 +0530

> I did not look more at the patch (than what's was needed for the rebase) but plan to do so.

Are you still planning to continue working on this? Some miscellaneous
comments while going through this patch are as follows?

1. Can you please try to explain the functionality of the overall
patch somewhere in the form of comments and or commit message?

2. It seems that the initially synchronized list of slots is only used
to launch a per-database worker to synchronize all the slots
corresponding to that database. If so, then why do we need to fetch
all the slot-related information via LIST_SLOTS command?

3. As mentioned in the initial email, I think it would be better to
replace LIST_SLOTS command with a SELECT query.

4. How the limit of sync_slot workers is decided? Can we document such
a piece of information? Do we need a new GUC to decide the number of
workers? Ideally, it would be better to avoid GUC, can we use any
existing logical replication workers related GUC?

5. Can we separate out the functionality related to standby_slot_names
in a separate patch, probably the first one? I think that will patch
easier to review.

6. In libpqrcv_list_slots(), two-phase related slot information is not
retrieved. Is there a reason for the same?

7.
+static void

Some comments atop this function would make it easier to review.

8.

The copyright notice is out-of-date.

9. Why synchronize_one_slot() compares
MyReplicationSlot->data.restart_lsn with the value of
confirmed_flush_lsn passed to it? Also, why it does only for new slots
but not existing slots?

10. Can we somehow test if the restart_lsn is advanced properly after
sync? I think it is important to ensure that because otherwise after
standby's promotion, the subscriber can start syncing from the wrong
position.

-- 
With Regards,
Amit Kapila.

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Mon, 19 Jun 2023 08:02:53 +0200

Hi,

On 6/16/23 11:56 AM, Amit Kapila wrote:
> Are you still planning to continue working on this? 

Yes, I think it would be great to have such a feature in core.

> Some miscellaneous

Thanks! I'll look at them and will try to come back to you by
mid of next week.

Also I think we need to handle the case of invalidated replication slot(s): should
we drop/recreate it/them? (as the main goal is to have sync slot(s) on the standby).

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Mon, 19 Jun 2023 15:33:37 +0530

> we drop/recreate it/them? (as the main goal is to have sync slot(s) on the standby).

Do you intend to ask what happens to logical slots invalidated (due to
say max_slot_wal_keep_size) on publisher? I think those should be
invalidated on standby too. Another thought whether there is chance
that the slot on standby gets invalidated due to conflict (say
required rows removed on primary)? I think in such cases the slot on
primary/publisher should have been dropped/invalidated by that time.
BTW, does the patch handles drop of logical slots on standby when the
same slot is dropped on publisher/primary?

-- 
With Regards,
Amit Kapila.

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Mon, 19 Jun 2023 18:26:52 +0200

Hi,

On 6/19/23 12:03 PM, Amit Kapila wrote:
> invalidated on standby too. 

Agree that it should behave that way.

> Another thought whether there is chance
> required rows removed on primary)?

That's the scenario I had in mind when asking the question above.

> I think in such cases the slot on

I don't think so.

For example, such a scenario could occur:

- there is no physical slot between the standby and the primary
- the standby is shut down
- logical decoding on the primary is moving forward and now there is vacuum
operations that will conflict on the standby
- the standby starts and reports the logical slot being invalidated (while it is
not on the primary)

In such a case (slot valid on the primary but invalidated on the standby) then I think we
could drop and recreate the invalidated slot on the standby.

> same slot is dropped on publisher/primary?

from what I've seen, yes it looks like it behaves that way (will look closer).

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Tue, 20 Jun 2023 15:52:17 +0530

> could drop and recreate the invalidated slot on the standby.

Will it be safe? Because after recreating the slot, it will reserve
the new WAL location and build the snapshot based on that which might
miss some important information in the snapshot. For example, to
update the slot's position with new information from the primary, the
patch uses pg_logical_replication_slot_advance() which means it will
process all records and update the snapshot via
DecodeCommit->SnapBuildCommitTxn().

The other related thing is that do we somehow need to ensure that WAL
is replayed on standby before moving the slot's position to the target
location received from the primary?

> from what I've seen, yes it looks like it behaves that way (will look closer).

Okay, I have asked because I don't see a call to ReplicationSlotDrop()
in the patch.

-- 
With Regards,
Amit Kapila.

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Mon, 26 Jun 2023 07:45:28 +0200

Hi,

> DecodeCommit->SnapBuildCommitTxn().

Your concern is that the slot could have been consumed on the standby?

I mean, if we suppose the "synchronized" slot can't be consumed on the standby then
drop/recreate such an invalidated slot would be ok?
  
Asking, because I'm not sure we should allow consumption of a "synchronized" slot
until the standby gets promoted.

When the patch has been initially proposed, logical decoding from a standby
was not implemented yet.
  
> location received from the primary?

Yeah, will check if this is currently done that way in the patch proposal.

> Okay, I have asked because I don't see a call to ReplicationSlotDrop()

Right. I'd need to look closer to understand how it works (for the moment
the "only" thing I've done was the re-base shared up-thread).

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Mon, 26 Jun 2023 16:04:01 +0530

> drop/recreate such an invalidated slot would be ok?

That also may not be sufficient because as soon as the slot is
invalidated/dropped, the required WAL could be removed on standby.

-- 
With Regards,
Amit Kapila.

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Wed, 28 Jun 2023 08:49:50 +0200

Hi,

> invalidated/dropped, the required WAL could be removed on standby.

Yeah, I think once the slot is dropped we just have to wait for the slot to
be re-created on the standby according to the new synchronize_slot_names GUC.

Assuming the initial slot "creation" on the standby (coming from the synchronize_slot_names usage)
is working "correctly" then it should also work "correctly" once the slot is dropped.

If we agree that a synchronized slot can not/should not be consumed (will implement this behavior) then
I think the proposed scenario above should make sense, do you agree?

Regards,

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Thu, 29 Jun 2023 10:22:06 +0000

Dear Drouvot,

Hi, I'm also interested in the feature. Followings are my high-level comments.
I did not mention some detailed notations because this patch is not at the stage.
And very sorry that I could not follow all of this discussions.

1. I thought that we should not reuse logical replication launcher for another purpose.
   The background worker should have only one task. I wanted to ask opinions some other people...
2. I want to confirm the reason why new replication command is added. IIUC the
   launcher connects to primary by using primary_conninfo connection string, but
   it establishes the physical replication connection so that any SQL cannot be executed.
   Is it right? Another approach not to use is to specify the target database via
   GUC, whereas not smart. How do you think?
3. You chose the per-db worker approach, however, it is difficult to extend the
   feature to support physical slots. This may be problematic. Was there any
   reasons for that? I doubted ReplicationSlotCreate() or advance functions might
   not be used from other databases and these may be reasons, but not sure.
   If these operations can do without connecting to specific database, I think
   the architecture can be changed.
4. Currently the launcher establishes the connection every time. Isn't it better
   to reuse the same one instead?

Following comments are assumed the configuration, maybe the straightfoward:

primary->standby
   |->subscriber

5. After constructing the system, I dropped the subscription on the subscriber.
   In this case the logical slot on primary was removed, but that was not replicated
   to standby server. Did you support the workload or not?

```
$ psql -U postgres -p $port_sub -c "DROP SUBSCRIPTION sub"
NOTICE:  dropped replication slot "sub" on publisher
DROP SUBSCRIPTION

$ psql -U postgres -p $port_primary -c "SELECT * FROM pg_replication_slots"
slot_name |  plugin  | slot_type | datoid | database |...
-----------+----------+-----------+--------+----------+... 
(0 rows)

$ psql -U postgres -p $port_standby -c "SELECT * FROM pg_replication_slots"
 slot_name |  plugin  | slot_type | datoid | database |...
-----------+----------+-----------+--------+----------+...
 sub       | pgoutput | logical   |      5 | postgres |...
(1 row)

```

6. Current approach may delay the startpoint of sync.

Assuming that physical replication system is created first, and then the
subscriber connects to the publisher node. In this case the launcher connects to
primary earlier than the apply worker, and reads the slot. At that time there are
no slots on primary, so launcher disconnects from primary and waits a time period (up to 3min).
Even if the apply worker creates the slot on publisher, but the launcher on standby
cannot notice that. The synchronization may start 3 min later.

I'm not sure how to fix or it could be acceptable. Thought?

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Amit Kapila <[email-masked]>
Date: Thu, 29 Jun 2023 16:06:39 +0530

> is working "correctly" then it should also work "correctly" once the slot is dropped.

I also think so.

> I think the proposed scenario above should make sense, do you agree?

Yeah, I also can't think of a use case for this. So, we can probably
disallow it and document the same. I guess if we came across a use
case for this, we can rethink allowing to consume the changes from
synchronized slots.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Thu, 29 Jun 2023 16:15:35 +0530

>    The background worker should have only one task. I wanted to ask opinions some other people...

IIUC, the launcher will launch the sync slot workers corresponding to
slots that need sync on standby and apply workers for active
subscriptions on primary (which will be a subscriber in this context).
If this is correct, then do you expect to launch a separate kind of
standby launcher for sync slots?

-- 
With Regards,
Amit Kapila.

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Fri, 30 Jun 2023 08:09:40 +0200

Hi Kuroda-san,

> And very sorry that I could not follow all of this discussions.

Thanks for looking at it and your feedback!

All I've done so far is to provide a re-based version in April of the existing patch.

I'll have a closer look at the code, at your feedback and Amit's one while working on the new version that will:

- take care of slot invalidation
- ensure that synchronized slot cant' be consumed until the standby gets promoted

as discussed up-thread.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Sat, 1 Jul 2023 07:12:48 +0530

On Thu, Jun 29, 2023 at 3:52 PM Hayato Kuroda (Fujitsu)
<[email-masked]> wrote:
>
> 2. I want to confirm the reason why new replication command is added.
>

Are you referring LIST_SLOTS command? If so, I don't think that is
required and instead, we can use a query to fetch the required
information.

> IIUC the

I think this point needs some investigation but do we want just one
worker that syncs all the slots? That may lead to lag in keeping the
slots up-to-date. We probably need some tests.

>    to reuse the same one instead?

I feel it is not the launcher but a separate sync slot worker that
establishes the connection. It is not clear to me what exactly you
have in mind. Can you please explain a bit more?

>    to standby server. Did you support the workload or not?

This should work.

> cannot notice that. The synchronization may start 3 min later.

I feel this should be based on some GUC like
'wal_retrieve_retry_interval' which we are already using in the
launcher or probably a new one if that doesn't seem to match.

-- 
With Regards,
Amit Kapila.

---

From: Bharath Rupireddy <[email-masked]>
Date: Sun, 9 Jul 2023 13:00:00 +0530

> patch somewhere in the form of comments and or commit message?

IIUC, there are 2 core ideas of the feature:

1) It will never let the logical replication subscribers go ahead of
physical replication standbys specified in standby_slot_names. It
implements this by delaying decoding of commit records on the
walsenders corresponding to logical replication subscribers on the
primary until all the specified standbys confirm receiving the commit
LSN.

2) The physical replication standbys will synchronize data of the
specified logical replication slots (in synchronize_slot_names) from
the primary, creating the logical replication slots if necessary.
Since the logical replication subscribers will never go out of
physical replication standbys, the standbys can safely synchronize the
slots and keep the data necessary for subscribers to connect to it and
work seamlessly even after a failover.

If my understanding is right, I have few thoughts here:

1. All the logical walsenders are delayed on the primary - per
wait_for_standby_confirmation() despite the user being interested in
only a few of them via synchronize_slot_names. Shouldn't the delay be
for just the slots specified in synchronize_slot_names?
2. I think we can split the patch like this - 0001 can be the logical
walsenders delaying decoding on the primary unless standbys confirm,
0002 standby synchronizing the logical slots.
3. I think we need to change the GUC standby_slot_names to better
reflect what it is used for - wait_for_replication_slot_names or
wait_for_
4. It allows specifying logical slots in standby_slot_names, meaning,
it can disallow logical slots getting ahead of other logical slots
specified in standby_slot_names. Should we allow this case with the
thinking that if there's anyone using logical replication for failover
(well, will anybody do that in production?).
5. Similar to above, it allows specifying physical slots in
synchronize_slot_names. Should we disallow?

I'm attaching the v6 patch, a rebased version of v5.

-- 
Bharath Rupireddy

---

From: Amit Kapila <[email-masked]>
Date: Mon, 10 Jul 2023 09:06:22 +0530

On Sun, Jul 9, 2023 at 1:01 PM Bharath Rupireddy
<[email-masked]> wrote:
> If my understanding is right,
>

This matches my understanding as well.

> I have few thoughts here:
>

Agreed with the above two points.

> reflect what it is used for - wait_for_replication_slot_names or

I feel at this stage we can focus on getting the design and
implementation correct. We can improve GUC names later once we are
confident that the functionality is correct.

> (well, will anybody do that in production?).

I think on the contrary we should prohibit this case. We can always
extend this functionality later.

> synchronize_slot_names. Should we disallow?

We should prohibit that as well.

-- 
With Regards,
Amit Kapila.

---

From: shveta malik <[email-masked]>
Date: Thu, 20 Jul 2023 17:04:44 +0530

> replace LIST_SLOTS command with a SELECT query.

I had a look at this thread. I am interested to work on this and can
spend some time addressing the comments given here.

I tried to replace LIST_SLOTS command with a SELECT query. Attached
rebased patch and PoC patch for LIST_SLOTS removal. For LIST_SLOTs cmd
removal, below are the points where more analysis is needed.

1) I could not use the exposed libpqwalreceiver's functions
walrcv_exec/libpqrcv_exec in LogicalRepLauncher to run select query
instead of LIST_SLOTS cmd. This is because libpqrcv_exec() needs
database connection but since in LogicalReplauncher, we do not have
any (MyDatabseId is not set), so the API gives an error. Thus to make
it work for the time-being, I used 'libpqrcv_PQexec' which is not
dependent upon database connection. But since it is not exposed "yet"
to other layers, I temporarily added the new code to
libpqwalreceiver.c itself. In fact I reused the existing function
wrapper libpqrcv_list_slots and changed the functionality to get info
using select query rather than list_slots.

2) While using connect API walrcv_connect/libpqrcv_connect(), we need
to tell it whether it is for logical or physical replication. In the
existing patch, where we were using LIST_SLOTS cmd, we have this
connection made with logical=false. But now since we need to run
select query to get the same info, using connection with logical=false
gives error on primary while executing select query. "ERROR:  cannot
execute SQL commands in WAL sender for physical replication".
And thus in ApplyLauncherStartSlotSync(), I have changed connect API
to use logical=true for the time being.  I noticed that in the
existing patch, it was using logical=false in
ApplyLauncherStartSlotSync() while logical=true in
synchronize_slots(). Possibly due to the same fact that logical=false
connection will not allow synchronize_slots() to run select query on
primary while it worked for ApplyLauncherStartSlotSync() as it was
running list_slots cmd instead of select query.

I am exploring further on these points to figure out which one is the
correct way to deal with these. Meanwhile posting this WIP patch for
early feedback. I will try addressing other comments as well in next
versions.

thanks
Shveta

---

From: Bharath Rupireddy <[email-masked]>
Date: Fri, 21 Jul 2023 11:35:55 +0530

> spend some time addressing the comments given here.

Thanks for your interest. Coincidentally, I started to split the patch
into 2 recently - 0001 making the specified logical wal senders wait
for specified standbys to ack, 0002 synchronize logical slots. I think
I'll have these patches ready by early next week. For 0002, I'll
consider your latest changes having LIST_SLOTS removed.

---

From: shveta malik <[email-masked]>
Date: Fri, 21 Jul 2023 17:16:24 +0530

> consider your latest changes having LIST_SLOTS removed.

Thanks Bharat for letting us know. It is okay to split the patch, it
may definitely help to understand the modules better but shall we take
a step back and try to reevaluate the design first before moving to
other tasks?

I analyzed more on the issues stated in [1] for replacing LIST_SLOTS
with SELECT query. On rethinking, it might not be a good idea to
replace this cmd with SELECT in Launcher code-path, because we do not
have any database-connection in launcher and 'select' query needs one
and thus we need to supply dbname to it. We may take the
primary-dbname info in a new GUC from users, but I feel retaining LIST
cmd is a better idea over adding a new GUC. But I do not see a reason
why we should get complete replication-slots info in LIST command. The
logic in Launcher is to get distinct database-ids info out of all the
slots and start worker per database-id. Since we are only interested
in database-id, I think we should change LIST_SLOTS to something like
LIST_DBID_FOR_LOGICAL_SLOTS. This new command may get only unique
database-ids for all the logical-slots (or the ones mentioned in
synchronize_slot_names) from primary. By doing so, we can avoid huge
network traffic in cases where the number of replication slots is
quite high considering that max_replication_slots can go upto
MAX_BACKENDS:2^18-1.  So I plan to make this change where we retain
LIST cmd over SELECT query but make this cmd's output restricted to
only database-Ids. Thoughts?

Secondly, I was thinking if the design proposed in the patch is the
best one. No doubt, it is the most simplistic design and thus may
prove very efficient for scenarios where we have a reasonable number
of workers starting and each one actively busy in
slots-synchronisation, handling almost equivalent load.  But since we
are starting one worker per database id, it may not be most efficient
for cases where not all the databases are actively being used. We may
have some workers (started for databases not in use) just waking up
and sending queries to primary and then going back to sleep and in the
process generating network traffic, while others may be heavily loaded
to deal with large numbers of active slots for a heavily loaded
database. I feel the design should be adaptable to load conditions
i.e. if we have more number of actively used slots, then we should
have more workers spawned to handle it and when the work is less then
the number of spawned workers should be less.
I have not thought it thoroughly yet and also not sure whether it will
actually come out as a better one, but it or more such designs should
be considered before we start fixing bugs in this patch. Kindly let me
know if there are already discussions around it which I might have
missed.  Any feedback is appreciated.

[1] : https://www.postgresql.org/message-id/CAJpy0uCMNz3XERP-Vzp-7rHFztJgc6d%2BxsmUVCqsxWPkZvQz0Q%40mail.gmail.com

thanks
Shveta

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 24 Jul 2023 08:02:56 +0530

> a step back and try to reevaluate the design first before moving to

Agree that design comes first. FWIW, I'm attaching the v9 patch set
that I have with me. It can't be a perfect patch set unless the design
is finalized.

> replace this cmd with SELECT in Launcher code-path

I think there are open fundamental design aspects, before optimizing
LIST_SLOTS, see below. I'm sure we can come back to this later.

> .......... Any feedback is appreciated.

Here are my thoughts about this feature:

Current design:

1. On primary, never allow walsenders associated with logical
replication slots to go ahead of physical standbys that are candidates
for future primary after failover. This enables subscribers to connect
to new primary after failover.
2. On all candidate standbys, periodically sync logical slots from
primary (creating the slots if necessary) with one slot sync worker
per logical slot.

Important considerations:

1. Does this design guarantee the row versions required by subscribers
aren't removed on candidate standbys as raised here -
https://www.postgresql.org/message-id/20220218222319.yozkbhren7vkjbi5%40alap3.anarazel.de?

It seems safe with logical decoding on standbys feature. Also, a
test-case from upthread is already in patch sets (in v9 too)
https://www.postgresql.org/message-id/CAAaqYe9FdKODa1a9n%3Dqj%2Bw3NiB9gkwvhRHhcJNginuYYRCnLrg%40mail.gmail.com.
However, we need to verify the use cases extensively.

2. All candidate standbys will start one slot sync worker per logical
slot which might not be scalable. Is having one (or a few more - not
necessarily one for each logical slot) worker for all logical slots
enough?

It seems safe to have one worker for all logical slots - it's not a
problem even if the worker takes a bit of time to get to sync a
logical slot on a candidate standby, because the standby is ensured to
retain all the WAL and row versions required to decode and send to the
logical slots.

3. Indefinite waiting of logical walsenders for candidate standbys may
not be a good idea. Is having a timeout for logical walsenders a good
idea?

A problem with timeout is that it can make logical slots unusable
after failover.

4. All candidate standbys retain WAL required by logical slots. Amount
of WAL retained may be huge if there's a replication lag with logical
replication subscribers.

This turns out to be a typical problem with replication, so there's
nothing much this feature can do to prevent WAL file accumulation
except for asking one to monitor replication lag and WAL file growth.

5. Logical subscribers replication lag will depend on all candidate
standbys replication lag. If candidate standbys are too far from
primary and logical subscribers are too close, still logical
subscribers will have replication lag. There's nothing much this
feature can do to prevent this except for calling it out in
documentation.

6. This feature might need to prevent the GUCs from deviating on
primary and the candidate standbys - there's no point in syncing a
logical slot on candidate standbys if logical walsender related to it
on primary isn't keeping itself behind all the candidate standbys. If
preventing this from happening proves to be tough, calling it out in
documentation to keep GUCs the same is a good start.

7. There are some important review comments provided upthread as far
as this design and patches are concerned -
https://www.postgresql.org/message-id/20220207204557.74mgbhowydjco4mh%40alap3.anarazel.de
and https://www.postgresql.org/message-id/20220207203222.22aktwxrt3fcllru%40alap3.anarazel.de.
I'm sure we can come to these once the design is clear.

Please feel free to add the list if I'm missing anything.

Thoughts?

---

From: Amit Kapila <[email-masked]>
Date: Mon, 24 Jul 2023 09:00:15 +0530

> However, we need to verify the use cases extensively.

Agreed.

> slot which might not be scalable.
>

Yeah, that doesn't sound like a good idea but IIRC, the proposed patch
is using one worker per database (for all slots corresponding to a
database).

> Is having one (or a few more - not

I guess for a large number of slots the is a possibility of a large
gap in syncing the slots which probably means we need to retain
corresponding WAL for a much longer time on the primary. If we can
prove that the gap won't be large enough to matter then this would be
probably worth considering otherwise, I think we should find a way to
scale the number of workers to avoid the large gap.

-- 
With Regards,
Amit Kapila.

---

From: shveta malik <[email-masked]>
Date: Wed, 26 Jul 2023 10:31:32 +0530

> scale the number of workers to avoid the large gap.

How about this:

1) On standby, spawn 1 worker per database in the start (as it is
doing currently).

2) Maintain statistics on activity against each primary's database on
standby by any means. Could be by maintaining 'last_synced_time' and
'last_activity_seen time'.  The last_synced_time is updated every time
we sync/recheck slots for that particular database. The
'last_activity_seen_time' changes only if we get any slot on that
database where actually confirmed_flush or say restart_lsn has changed
from what was maintained already.

3) If at any moment, we find that 'last_synced_time' -
'last_activity_seen' goes beyond a threshold, that means that DB is
not active currently. Add it to list of inactive DB

4) Launcher on the other hand is always checking if it needs to spawn
any other extra worker for any new DB. It will additionally check if
number of inactive databases (maintained on standby) has gone higher
(> some threshold), then it brings down the workers for those and
starts a common worker which takes care of all such inactive databases
(or merge all in 1), while workers for active databases remain as such
(i.e. one per db). Each worker maintains the list of DBs which it is
responsible for.

5) If in the list of these inactive databases, we again find any
active database using the above logic, then the launcher will spawn a
separate worker for that.

Pros:
Lesser workers on standby as per the load on primary.
Lesser poking of primary by standby i.e. standby will send queries to
get slot info for all inactive DBs in 1 run instead of each worker
sending such queries separately.

Cons:  We might see spawning and freeing of workers more frequently.

Please let me know your thoughts on this.

thanks
Shveta

---

From: shveta malik <[email-masked]>
Date: Wed, 26 Jul 2023 17:55:17 +0530

> that I have with me. It can't be a perfect patch set unless the design

Thanks for the patch and summarizing all the issues here. I was going
through the patch and found that now we need to maintain
'synchronize_slot_names' on both primary and standby unlike the old
way where it was maintained only on standby. I am aware of the problem
in earlier implementation where each logical walsender/slot  needed to
wait for all standbys to catch-up before sending changes to logical
subscribers even though that particular slot is not even needed to be
synced by any of the standbys. Now it is more restrictive. But now, is
this 'synchronize_slot_names'  per standby? If there are multiple
standbys each having different  'synchronize_slot_names' requirements,
then how primary is going to keep track of that?
Please let me know if that scenario can never arise where standbys can
have different 'synchronize_slot_names'.

thanks
Shveta

---

From: Amit Kapila <[email-masked]>
Date: Thu, 27 Jul 2023 10:54:57 +0530

> not active currently. Add it to list of inactive DB

I think we should also increase the next_sync_time if in current sync,
there is no update.

> separate worker for that.

I wonder if we anyway some sort of design like this because we
shouldn't allow to spawn as many workers as the number of databases.
There has to be some existing or new GUC like max_sync_slot_workers
which decided the number of workers.

Overall, this sounds to be a more workload-adaptive approach as
compared to the current one.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Thu, 27 Jul 2023 12:24:19 +0530

> have different 'synchronize_slot_names'.

Can we think of sending 'synchronize_slot_names' from standby to
primary at the time of connection? I think we also need to ensure that
if the user changes this value then we need to restart the sync slot
worker to allow this information to be sent to the primary. We do
something similar for apply worker in logical replication.

-- 
With Regards,
Amit Kapila.

---

From: Bharath Rupireddy <[email-masked]>
Date: Fri, 28 Jul 2023 20:09:27 +0530

> is using one worker per database (for all slots corresponding to a

Right. It's based on one worker for each database.

> scale the number of workers to avoid the large gap.

I think the gap is largely determined by the time taken to advance
each slot and the amount of WAL that each logical slot moves ahead on
primary. I've measured the time it takes for
pg_logical_replication_slot_advance with different amounts WAL on my
system. It took 2595ms/5091ms/31238ms to advance the slot by
3.7GB/7.3GB/13GB respectively. To put things into perspective here,
imagine there are 3 logical slots to sync for a single slot sync
worker and each of them are in need of advancing the slot by
3.7GB/7.3GB/13GB of WAL. The slot sync worker gets to slot 1 again
after 2595ms+5091ms+31238ms (~40sec), gets to slot 2 again after
advance time of slot 1 with amount of WAL that the slot has moved
ahead on primary during 40sec, gets to slot 3 again after advance time
of slot 1 and slot 2 with amount of WAL that the slot has moved ahead
on primary and so on. If WAL generation on the primary is pretty fast,
and if the logical slot moves pretty fast on the primary, the time it
takes for a single sync worker to sync a slot can increase.

Now, let's think what happens if there's a large gap, IOW, a logical
slot on standby is behind X amount of WAL from that of the logical
slot on primary. The standby needs to retain more WAL for sure. IIUC,
primary doesn't need to retain the WAL required for a logical slot on
standby, no?

---

From: Bharath Rupireddy <[email-masked]>
Date: Fri, 28 Jul 2023 20:54:38 +0530

> which decided the number of workers.

It seems reasonable to not have one slot sync worker for each
database. IMV, the slot sync workers must be generic and independently
manageable - generic in the sense that given a database and primary
conninfo, each worker must sync all the slots related to the given
database, independently mangeable in the sense that separate GUC for
number of sync workers, launchable directly by logical replication
launcher dynamically. The work division amongst the sync workers can
be simple, the logical replication launcher builds a shared memory
structure based on number of slots to sync and starts the sync workers
dynamically, and each sync worker picks {dboid, slot name, conninfo}
from the shared memory, syncs it and proceeds with other slots.

Thoughts?

---

From: shveta malik <[email-masked]>
Date: Tue, 1 Aug 2023 17:01:36 +0530

On Fri, Jul 28, 2023 at 8:54 PM Bharath Rupireddy
<[email-masked]> wrote:
> launcher dynamically.

yes agreed. The patch v10-0003 attempts to do the same.

> The work division amongst the sync workers can

Do you mean  the logical replication launcher builds a shared memory
structure based
on the number of 'dbs' to sync as I understood from your initial comment?

thanks
Shveta

---

From: Bharath Rupireddy <[email-masked]>
Date: Thu, 3 Aug 2023 00:27:00 +0530

> on the number of 'dbs' to sync as I understood from your initial comment?

Yes. I haven't looked at the 0003 patch posted upthread. However, the
standby must do the following at a minimum:

- Make GUCs synchronize_slot_names and max_slot_sync_workers of
PGC_POSTMASTER type needing postmaster restart when changed as they
affect the number of slot sync workers.
- LR (logical replication) launcher connects to primary to fetch the
logical slots specified in synchronize_slot_names. This is a one-time
task.
- LR launcher prepares a dynamic shared memory (created via
dsm_create) with some state like locks for IPC and an array of
{slot_name, dboid_associated_with_slot, is_sync_in_progress} - maximum
number of elements in the array is the number of slots specified in
synchronize_slot_names. This is a one-time task.
- LR launcher decides the *best* number of slot sync workers - (based
on some perf numbers) it can just launch, say, one worker per 2 or 4
or 8 etc. slots.
- Each slot sync worker then picks up a slot from the DSM, connects to
primary using primary conn info, syncs it, and moves to another slot.

Not having the capability of on-demand stop/launch of slot sync
workers makes the above design simple IMO.

Thoughts?

---

From: shveta malik <[email-masked]>
Date: Thu, 3 Aug 2023 16:19:40 +0530

> affect the number of slot sync workers.

I agree that max_slot_sync_workers should be allowed to change only
during startup but I strongly feel that synchronize_slot_names should
be runtime modifiable. We should give that flexibility to the user.

> logical slots specified in synchronize_slot_names. This is a one-time

if synchronize_slot_names='*', we need to fetch slots info at regular
intervals even if it is not runtime modifiable. For a runtime
modifiable case, it is obvious to reftech it regular intervals.

> synchronize_slot_names. This is a one-time task.

yes, we need dynamic-shared-memory but it is not a
one-time-allocation. If it were a one-time allocation, then there was
no need for DSM, only shared memory allocation was enough. It is not a
one time allocation in any of the designs. If it is slot based design,
slots may keep on varying for '*' case and if it is DB based design,
then number of DBs may go beyond the initial memory allocated and we
may need reallocation and relaunch of worker and thus the need of DSM.

> primary using primary conn info, syncs it, and moves to another slot.

The design based on slots i.e. launcher dividing the slots among the
available workers, could prove beneficial over db based division for a
case where number of slots per DB varies largely and we end up
assigning all DBs with lesser slots to one worker
while all heavily loaded DBs to another. But other than this, I see
lot of pain points:

1) Since we are going to do slots based synching, query construction
will be complex. We will have a query with a long 'where' clause:
where slots in (slot1, slot2, slots...).

2) Number of pings to primary will be more as we are pinging it slot
based instead of DB based. So the information which we could have
fetched collectively in one query (if it was db based) is now splitted
to multiple queries assuming that there could be cases where slots
belonging to the same DBs end up getting splitted among different
workers.

3) if number of slots < max number of workers, how are we going to
assign the worker? One slot per worker or all in one worker. If it is
one slot per worker, it will again be not that efficient as it will
result in more network traffic. This needs more thoughts and case to
case varying design.

> workers makes the above design simple IMO.

We need to anyways relaunch workers when DSM is reallocated in case
Dbs (or sya slots) exceed some initial allocation limit.

thanks
Shveta

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Fri, 4 Aug 2023 09:40:22 +0200

Hi,

> Here are my thoughts about this feature:

Thanks for looking at it!

> However, we need to verify the use cases extensively.

Agree. We also discussed up-thread that we'd have to drop any "sync" slots if they
are invalidated. And they should be re-created based on the synchronize_slot_names.

> Please feel free to add the list if I'm missing anything.

We'd also have to ensure that "sync" slots can't be consumed on the standby (this has been
discussed up-thread).

Regards,

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Fri, 4 Aug 2023 11:14:33 +0200

Hi,

On 7/28/23 4:39 PM, Bharath Rupireddy wrote:
> primary. 

Sorry to be late, but I gave a second thought and I wonder if we really need this design.
(i.e start a logical replication background worker on the standby to sync the slots).

Wouldn't that be simpler to "just" update the sync slots "metadata"
as the https://github.com/EnterpriseDB/pg_failover_slots module (mentioned by Peter
up-thread) is doing?
(making use of LogicalConfirmReceivedLocation(), LogicalIncreaseXminForSlot()
and LogicalIncreaseRestartDecodingForSlot(), If I read synchronize_one_slot() correctly).

> I've measured the time it takes for

That would be way "faster" and we would probably not need to
worry that much about the number of "sync" workers (if it/they "just" has/have
to sync slot's "metadata") as proposed above.

Thoughts?

Regards,

---

From: shveta malik <[email-masked]>
Date: Fri, 4 Aug 2023 17:02:41 +0530

> and LogicalIncreaseRestartDecodingForSlot(), If I read synchronize_one_slot() correctly).

Agreed. It would be simpler to just update the metadata. I think you
have not got chance to review the latest posted patch ('v10-0003')
yet, it does the same.

But I do not quite get it as in how can we do it w/o starting a
background worker? Even the failover-slots extension starts one
background worker. The question here is how many background workers we
need to have. Will one be sufficient or do we need one per db (as done
earlier by the original patches in this thread) or are we good with
dividing work among some limited number of workers?

I feel syncing all slots in one worker may increase the lag between
subsequent syncs for a particular slot and if the number of slots are
huge, the chances of losing the slot-data is more in case of failure.
Starting one worker per db also might not be that efficient as it will
increase load on the system (both in terms of background worker and
network traffic) especially for a case where the number of dbs are
more. Thus starting max 'n' number of workers where 'n' is decided by
GUC and dividing the work/DBs among these looks a better option to me.
Please see the discussion in and around the email at [1]

[1]: https://www.postgresql.org/message-id/CAJpy0uCT%2BnpL4eUvCWiV_MBEri9ixcUgJVDdsBCJSqLd0oD1fQ%40mail.gmail.com

> to sync slot's "metadata") as proposed above.

Agreed, we need not to worry about delay due to
pg_logical_replication_slot_advance if we are only going to update a
few simple things using the function calls as mentioned above.

thanks
Shveta

---

From: "Drouvot, Bertrand" <[email-masked]>
Date: Mon, 7 Aug 2023 11:47:47 +0200

Hi,

> have not got chance to review the latest posted patch ('v10-0003')

Thanks for the feedback! Yeah, I did not look at v10 in details and was
looking at the email thread only.

Indeed, I now see that 0003 does update the metadata in local_slot_advance(),
that's great!

> background worker? 

Yeah, agree that we still need background workers.
What I meant was to avoid to use "logical replication background worker"
(aka through logicalrep_worker_launch()) to sync the slots.

> The question here is how many background workers we

Thanks for the link! If I read the email thread correctly, this discussion
was before V10 (which is the first version making use of LogicalConfirmReceivedLocation(),
LogicalIncreaseXminForSlot(), LogicalIncreaseRestartDecodingForSlot()) means
before the metadata sync only has been implemented.

While I agree that the approach to split the sync load among workers with the new
max_slot_sync_workers GUC seems reasonable without the sync only feature (pre V10),
I'm not sure that with the metadata sync only in place the extra complexity to manage multiple
sync workers is needed.

Maybe we should start some tests/benchmark with only one sync worker to get numbers
and start from there?

Regards,