v43-0007-Fix-apply-worker-empty-prepare.patch

From d1813c3d6024e71fce5a5cb4228776078df98fd8 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 25 Feb 2021 16:41:01 +1100
Subject: [PATCH v43] Fix apply worker empty prepare.

By sad timing of apply/tablesync workers it is possible to have a "consistent snapshot" that spans prepare/commit in such a way that the tablesync did not do the prepare (because snapshot not consistent) and the apply worker does the prepare ('b') but it skips all the prepared operations [e.g. inserts] while the tablesync was still busy (see the condition of should_apply_changes_for_rel). Later, at the commit prepared time when the apply worker does the commit prepare ('K'), there is nothing committed (because the inserts were skipped earlier).

This patch implements a two-part fix as suggested [1] on hackers.

Part 1 - The begin_prepare handler of apply will always wait for any busy tablesync workers to acheive SYNCDONE/READY state.

Part 2 - If (after Part 1) the apply-worker's prepare is found to be lagging behind any of the sync-workers then the subsequent prepared operations will be spooled to a file to be replayed at commit_prepared time.

Discussion:
[1] https://www.postgresql.org/message-id/CAA4eK1L%3DdhuCRvyDvrXX5wZgc7s1hLRD29CKCK6oaHtVCPgiFA%40mail.gmail.com
---
 src/backend/replication/logical/tablesync.c | 184 +++++++--
 src/backend/replication/logical/worker.c    | 595 +++++++++++++++++++++++++++-
 src/include/replication/worker_internal.h   |   3 +
 src/tools/pgindent/typedefs.list            |   1 +
 4 files changed, 747 insertions(+), 36 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index feb634e..8b20519 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -115,7 +115,10 @@
 #include "utils/memutils.h"
 #include "utils/snapmgr.h"
 
-static bool table_states_valid = false;
+static bool		table_states_valid = false;
+static List    *table_states_not_ready = NIL;
+static List	   *table_states_all = NIL;
+static void		FetchTableStates(bool *started_tx);
 
 StringInfo	copybuf = NULL;
 
@@ -359,7 +362,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 		Oid			relid;
 		TimestampTz last_start_time;
 	};
-	static List *table_states = NIL;
 	static HTAB *last_start_times = NULL;
 	ListCell   *lc;
 	bool		started_tx = false;
@@ -367,42 +369,14 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	Assert(!IsTransactionState());
 
 	/* We need up-to-date sync state info for subscription tables here. */
-	if (!table_states_valid)
-	{
-		MemoryContext oldctx;
-		List	   *rstates;
-		ListCell   *lc;
-		SubscriptionRelState *rstate;
-
-		/* Clean the old list. */
-		list_free_deep(table_states);
-		table_states = NIL;
-
-		StartTransactionCommand();
-		started_tx = true;
-
-		/* Fetch all non-ready tables. */
-		rstates = GetSubscriptionNotReadyRelations(MySubscription->oid);
-
-		/* Allocate the tracking info in a permanent memory context. */
-		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
-		foreach(lc, rstates)
-		{
-			rstate = palloc(sizeof(SubscriptionRelState));
-			memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
-			table_states = lappend(table_states, rstate);
-		}
-		MemoryContextSwitchTo(oldctx);
-
-		table_states_valid = true;
-	}
+	FetchTableStates(&started_tx);
 
 	/*
 	 * Prepare a hash table for tracking last start times of workers, to avoid
 	 * immediate restarts.  We don't need it if there are no tables that need
 	 * syncing.
 	 */
-	if (table_states && !last_start_times)
+	if (table_states_not_ready && !last_start_times)
 	{
 		HASHCTL		ctl;
 
@@ -416,7 +390,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	 * Clean up the hash table when we're done with all tables (just to
 	 * release the bit of memory).
 	 */
-	else if (!table_states && last_start_times)
+	else if (!table_states_not_ready && last_start_times)
 	{
 		hash_destroy(last_start_times);
 		last_start_times = NULL;
@@ -425,7 +399,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 	/*
 	 * Process all tables that are being synchronized.
 	 */
-	foreach(lc, table_states)
+	foreach(lc, table_states_not_ready)
 	{
 		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 
@@ -1137,3 +1111,145 @@ copy_table_done:
 	wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
 	return slotname;
 }
+
+/*
+ * Common code to fetch the up-to-date sync state info into the static lists.
+ *
+ * XXX - Is there a potential timing problem here - e.g. if signal arrives
+ * while executing this then maybe we will set table_states_valid without
+ * refetching them?
+ */
+static void
+FetchTableStates(bool *started_tx)
+{
+	*started_tx = false;
+
+	if (!table_states_valid)
+	{
+		MemoryContext oldctx;
+		List	   *rstates;
+		ListCell   *lc;
+		SubscriptionRelState *rstate;
+
+		/* Clean the old lists. */
+		list_free_deep(table_states_all);
+		table_states_all = NIL;
+		list_free_deep(table_states_not_ready);
+		table_states_not_ready = NIL;
+
+		StartTransactionCommand();
+		*started_tx = true;
+
+		/* Fetch all tables. */
+		rstates = GetSubscriptionRelations(MySubscription->oid);
+
+		/* Allocate the tracking info in a permanent memory context. */
+		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+		foreach(lc, rstates)
+		{
+			SubscriptionRelState *cur_rstate = (SubscriptionRelState *) lfirst(lc);
+
+			/* List of all states */
+			rstate = palloc(sizeof(SubscriptionRelState));
+			memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState));
+			table_states_all = lappend(table_states_all, rstate);
+
+			/* List of only not-ready states */
+			if (cur_rstate->state != SUBREL_STATE_READY)
+			{
+				rstate = palloc(sizeof(SubscriptionRelState));
+				memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState));
+				table_states_not_ready = lappend(table_states_not_ready, rstate);
+			}
+		}
+		MemoryContextSwitchTo(oldctx);
+
+		table_states_valid = true;
+	}
+}
+
+/*
+ * Are there any tablesyncs which have still not yet reached SYNCDONE/READY state?
+ */
+bool
+BusyTablesyncs()
+{
+	bool		found_busy = false;
+	bool		started_tx = false;
+	int			count = 0;
+	ListCell   *lc;
+
+	/* We need up-to-date sync state info for subscription tables here. */
+	FetchTableStates(&started_tx);
+
+	/*
+	 * Process all not-READY tables to see if any are also not-SYNCDONE
+	 */
+	foreach (lc, table_states_not_ready)
+	{
+		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
+
+		count++;
+		elog(DEBUG1,
+			 "BusyTablesyncs: #%d. Table relid %u has state '%c'",
+			 count,
+			 rstate->relid,
+			 rstate->state);
+
+		/*
+		 * XXX - When the process_syncing_tables_for_sync changes the state
+		 * from SYNCDONE to READY, that change is actually written directly
+		 * into the list element of table_states_not_ready.
+		 *
+		 * So the "table_states_not_ready" list might end up having a READY
+		 * state event though there was nont initially. That is why we need
+		 * to check for it below.
+		 */
+		if (rstate->state != SUBREL_STATE_SYNCDONE &&
+			rstate->state != SUBREL_STATE_READY)
+		{
+			found_busy = true;
+			break;
+		}
+	}
+
+	if (started_tx)
+	{
+		CommitTransactionCommand();
+		pgstat_report_stat(false);
+	}
+
+	elog(DEBUG1,
+		 "BusyTablesyncs: Scanned %d tables, and found busy = %s",
+		 count,
+		 found_busy ? "true" : "false");
+
+	return found_busy;
+}
+
+/*
+ * What is the biggest LSN from the all the known tablesyncs?
+ */
+XLogRecPtr
+BiggestTablesyncLSN()
+{
+	XLogRecPtr	biggest_lsn = InvalidXLogRecPtr;
+	ListCell   *lc;
+	int			count = 0;
+
+	foreach (lc, table_states_all)
+	{
+		SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
+
+		count++;
+		if (rstate->lsn > biggest_lsn)
+			biggest_lsn = rstate->lsn;
+	}
+
+	elog(DEBUG1,
+		 "BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X",
+		 count,
+		 LSN_FORMAT_ARGS(biggest_lsn));
+
+	return biggest_lsn;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b80bf69..d9b7cfa 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -212,6 +212,45 @@ static void subxact_info_add(TransactionId xid);
 static inline void cleanup_subxact_info(void);
 
 /*
+ * The following are for the support of a spoolfile for prepared messages.
+ */
+
+/*
+ * A contest for the prepare spooling
+ */
+static MemoryContext PsfContext = NULL;
+
+/*
+ * A Prepare spoolfile hash entry. We create this entry in the psf_hash. This is
+ * for maintaining a mapping between the name of the prepared spoolfile, and the
+ * corresponding fileset handles of same.
+ */
+typedef struct PsfHashEntry
+{
+	char	name[MAXPGPATH];	/* Hash key --- must be first */
+	SharedFileSet *fileset;		/* shared file set for prepare spoolfile */
+} PsfHashEntry;
+
+/*
+ * Hash table for storing the Prepared spoolfile info along with shared fileset.
+ */
+static HTAB *psf_hash = NULL;
+
+/*
+ * The spoolfile handle is only valid between begin_prepare and prepare.
+ */
+static BufFile *psf_fd = NULL;
+
+static void prepare_spoolfile_create(char *path);
+static void prepare_spoolfile_write(char action, StringInfo s);
+static void prepare_spoolfile_close(void);
+static void prepare_spoolfile_cleanup(char *path);
+static bool prepare_spoolfile_exists(char *path);
+static void prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid);
+static int  prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn);
+static bool prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s);
+
+/*
  * Serialize and deserialize changes for a toplevel transaction.
  */
 static void stream_cleanup_files(Oid subid, TransactionId xid);
@@ -759,6 +798,76 @@ apply_handle_begin_prepare(StringInfo s)
 		return;
 	}
 
+	/*
+	 * A Problem:
+	 *
+	 * By sad timing of apply/tablesync workers it is possible to have a
+	 * “consistent snapshot” that spans prepare/commit in such a way that the
+	 * tablesync did not do the prepare (because snapshot not consistent) and
+	 * the apply worker does the prepare (‘b’) but it skips all the prepared
+	 * operations [e.g. inserts] while the tablesync was still busy (see the
+	 * condition of should_apply_changes_for_rel). Later at the commit prepared
+	 * time when the apply worker does the commit prepare (‘K’), there is
+	 * nothing in it (because the inserts were skipped earlier).
+	 *
+	 * The following code has a 2-part workaround for that scenario.
+	 */
+	if (!am_tablesync_worker())
+	{
+		/*
+		 * Workaround Part 1 of 2:
+		 *
+		 * Make sure every tablesync has reached at least SYNCDONE state before
+		 * letting the apply worker proceed.
+		 */
+		elog(DEBUG1,
+			 "apply_handle_begin_prepare, begin_data.end_lsn = %X/%X, lstate_lsn = %X/%X",
+			 LSN_FORMAT_ARGS(begin_data.end_lsn),
+			 LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn));
+
+		while (BusyTablesyncs())
+		{
+			elog(DEBUG1, "apply_handle_begin_prepare - waiting for all sync workers to be DONE/READY");
+
+			process_syncing_tables(begin_data.end_lsn);
+
+			/* This latch is to prevent 100% CPU looping. */
+			(void) WaitLatch(MyLatch,
+							 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+							 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+			ResetLatch(MyLatch);
+		}
+
+		/*
+		 * Workaround Part 2 of 2:
+		 *
+		 * If (when getting to SYNCDONE/READY state) some tablesync went beyond
+		 * this begin_prepare LSN then set all messages (until prepared) will be
+		 * saved to a spoolfile for replay later at commit_prepared time.
+		 */
+		if (begin_data.end_lsn < BiggestTablesyncLSN()
+#if 0
+			|| true /* XXX - Add this line to force psf (for easier debugging) */
+#endif
+		   )
+		{
+			char psfpath[MAXPGPATH];
+
+			/* The begin_prepare's LSN has been overtaken. */
+
+			/*
+			 * We need a transaction for handling the buffile, used for serializing
+			 * prepared messages. This transaction lasts until the commit_prepared/
+			 * rollback_prepared.
+			 */
+			ensure_transaction();
+
+			prepare_spoolfile_name(psfpath, sizeof(psfpath),
+								   MyLogicalRepWorker->subid, begin_data.gid);
+			prepare_spoolfile_create(psfpath);
+		}
+	}
+
 	remote_final_lsn = begin_data.final_lsn;
 
 	in_remote_transaction = true;
@@ -788,6 +897,27 @@ apply_handle_prepare(StringInfo s)
 		return;
 	}
 
+	if (psf_fd)
+	{
+		/*
+		 * The psf_fd is meaningful only between begin_prepare and prepared.
+		 * So close it now. If we had been writing any messages to the psf_fd
+		 * (the spoolfile) then those will be applied later during
+		 * handle_commit_prepared.
+		 */
+		prepare_spoolfile_close();
+
+		/*
+		 * And end the transaction that was created by begin_prepare for
+		 * working with the psf buffiles.
+		 */
+		Assert(IsTransactionState());
+		CommitTransactionCommand();
+
+		in_remote_transaction = false;
+		return;
+	}
+
 	Assert(prepare_data.prepare_lsn == remote_final_lsn);
 
 	if (IsTransactionState())
@@ -835,6 +965,7 @@ static void
 apply_handle_commit_prepared(StringInfo s)
 {
 	LogicalRepPreparedTxnData prepare_data;
+	char psfpath[MAXPGPATH];
 
 	/*
 	 * We don't expect any other transaction data while skipping a prepared
@@ -844,6 +975,55 @@ apply_handle_commit_prepared(StringInfo s)
 
 	logicalrep_read_commit_prepared(s, &prepare_data);
 
+	/*
+	 * If this prepare's messages were being spooled to a file,
+	 * then replay them all now, and afterwards cleanup the spoolfile.
+	 */
+	prepare_spoolfile_name(psfpath, sizeof(psfpath),
+						   MyLogicalRepWorker->subid, prepare_data.gid);
+	if (prepare_spoolfile_exists(psfpath))
+	{
+		int nchanges;
+
+		/*
+		 * 1. replay the spooled messages
+		 */
+
+		ensure_transaction();
+
+		nchanges = prepare_spoolfile_replay_messages(psfpath, prepare_data.end_lsn);
+		elog(DEBUG1,
+			 "apply_handle_commit_prepared: replayed %d (all) changes.",
+			 nchanges);
+
+		prepare_spoolfile_cleanup(psfpath);
+
+		/*
+		 * 2. mark as PREPARED
+		 */
+
+		/*
+		 * BeginTransactionBlock is necessary to balance the
+		 * EndTransactionBlock called within the PrepareTransactionBlock
+		 * below.
+		 */
+		BeginTransactionBlock();
+		CommitTransactionCommand();
+
+		/*
+		 * Update origin state so we can restart streaming from correct
+		 * position in case of crash.
+		 */
+		replorigin_session_origin_lsn = prepare_data.end_lsn;
+		replorigin_session_origin_timestamp = prepare_data.preparetime;
+
+		PrepareTransactionBlock(prepare_data.gid);
+		CommitTransactionCommand();
+		pgstat_report_stat(false);
+
+		store_flush_position(prepare_data.end_lsn);
+	}
+
 	/* there is no transaction when COMMIT PREPARED is called */
 	ensure_transaction();
 
@@ -874,6 +1054,8 @@ static void
 apply_handle_rollback_prepared(StringInfo s)
 {
 	LogicalRepRollbackPreparedTxnData rollback_data;
+	bool using_psf;
+	char psfpath[MAXPGPATH];
 
 	/*
 	 * We don't expect any other transaction data while skipping a prepared
@@ -884,11 +1066,46 @@ apply_handle_rollback_prepared(StringInfo s)
 	logicalrep_read_rollback_prepared(s, &rollback_data);
 
 	/*
+	 * If this prepare's messages were being spooled to a file, then cleanup
+	 * the file.
+	 */
+	prepare_spoolfile_name(psfpath, sizeof(psfpath),
+						   MyLogicalRepWorker->subid, rollback_data.gid);
+	using_psf = prepare_spoolfile_exists(psfpath);
+	if (using_psf)
+	{
+		if (psf_fd)
+		{
+			/* XXX - For some reason it is currently possible (due to bug?) it
+			 * is possibe to get here, after a restart, when there was a
+			 * begin_prepare but there was NO prepare. Since there was no
+			 * prepare, the psf_fd and the transaction are still lingering so
+			 * they need to be cleaned up now.
+			 */
+			prepare_spoolfile_close();
+
+			/*
+			 * And end the transaction that was created by the begin_prepare
+			 * for working with psf buffiles.
+			 */
+			Assert(IsTransactionState());
+			AbortCurrentTransaction();
+		}
+
+		prepare_spoolfile_cleanup(psfpath);
+	}
+
+	/*
 	 * It is possible that we haven't received prepare because it occurred
 	 * before walsender reached a consistent point in which case we need to
 	 * skip rollback prepared.
+	 *
+	 * And we also skip the FinishPreparedTransaction if we're using the
+	 * Prepare Spoolfile (using_psf) because in that case there is no matching
+	 * PrepareTransactionBlock done yet.
 	 */
-	if (LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn,
+	if (!using_psf &&
+		LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn,
 					rollback_data.preparetime))
 	{
 		/*
@@ -941,6 +1158,26 @@ apply_handle_stream_prepare(StringInfo s)
 	elog(DEBUG1, "received prepare for streamed transaction %u", xid);
 
 	/*
+	 * Wait for all the sync workers to read a SYNCDONE/READY state.
+	 *
+	 * This is same waiting logic as in appy_handle_begin_prepare function (see
+	 * that function for more details comments about this).
+	 */
+	if (!am_tablesync_worker())
+	{
+		while (BusyTablesyncs())
+		{
+			process_syncing_tables(prepare_data.end_lsn);
+
+			/* This latch is to prevent 100% CPU looping. */
+			(void) WaitLatch(MyLatch,
+							 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+							 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+			ResetLatch(MyLatch);
+		}
+	}
+
+	/*
 	 *
 	 * --------------------------------------------------------------------------
 	 * 1. Replay all the spooled operations - Similar code as for
@@ -1007,6 +1244,7 @@ apply_handle_origin(StringInfo s)
 	 * remote transaction and before any actual writes.
 	 */
 	if (!in_streamed_transaction &&
+		psf_fd == NULL &&
 		(!in_remote_transaction ||
 		 (IsTransactionState() && !am_tablesync_worker())))
 		ereport(ERROR,
@@ -1468,6 +1706,9 @@ apply_handle_insert(StringInfo s)
 	if (skip_prepared_txn)
 		return;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_INSERT, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s))
 		return;
 
@@ -1592,6 +1833,9 @@ apply_handle_update(StringInfo s)
 	if (skip_prepared_txn)
 		return;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_UPDATE, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, s))
 		return;
 
@@ -1753,6 +1997,9 @@ apply_handle_delete(StringInfo s)
 	if (skip_prepared_txn)
 		return;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_DELETE, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, s))
 		return;
 
@@ -2125,6 +2372,9 @@ apply_handle_truncate(StringInfo s)
 	if (skip_prepared_txn)
 		return;
 
+	if (prepare_spoolfile_handler(LOGICAL_REP_MSG_TRUNCATE, s))
+		return;
+
 	if (handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, s))
 		return;
 
@@ -2418,6 +2668,23 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	}
 
 	/*
+	 * Initialize the psf_hash table if we haven't yet. This will be used for
+	 * the entire duration of the apply worker so create it in permanent
+	 * context.
+	 */
+	if (psf_hash == NULL)
+	{
+		HASHCTL			hash_ctl;
+		PsfHashEntry   *hentry;
+
+		hash_ctl.keysize = sizeof(hentry->name);
+		hash_ctl.entrysize = sizeof(PsfHashEntry);
+		hash_ctl.hcxt = ApplyContext;
+		psf_hash = hash_create("PrepareSpoolfileHash", 1024, &hash_ctl,
+							  HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+	}
+
+	/*
 	 * Init the ApplyMessageContext which we clean up after each replication
 	 * protocol message.
 	 */
@@ -2433,6 +2700,14 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 													"LogicalStreamingContext",
 													ALLOCSET_DEFAULT_SIZES);
 
+	/*
+	 * This memory context is used when the prepare spooling is used. It
+	 * is reset at prepare commit/rollback time.
+	 */
+	PsfContext = AllocSetContextCreate(ApplyContext,
+									   "PsfContext",
+									   ALLOCSET_DEFAULT_SIZES);
+
 	/* mark as idle, before starting to loop */
 	pgstat_report_activity(STATE_IDLE, NULL);
 
@@ -2537,7 +2812,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		/* confirm all writes so far */
 		send_feedback(last_received, false, false);
 
-		if (!in_remote_transaction && !in_streamed_transaction)
+		if (!in_remote_transaction && !in_streamed_transaction && psf_fd == NULL)
 		{
 			/*
 			 * If we didn't get any transactions for a while there might be
@@ -3462,3 +3737,319 @@ IsLogicalWorker(void)
 {
 	return MyLogicalRepWorker != NULL;
 }
+
+/*
+ * Handle the PREPARE spoolfile (if any)
+ *
+ * It can be necessary to redirect the PREPARE messages to a spoolfile (see
+ * apply_handle_begin_prepare) and then replay them back at the COMMIT PREPARED
+ * time. If needed, this is the common function to do that file redirection.
+ *
+ * Returns true if the message was redirected to the spoolfile, false
+ * otherwise (regular mode).
+ */
+static bool
+prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s)
+{
+	if (psf_fd == NULL)
+		return false;
+
+	Assert(!in_streamed_transaction);
+
+	/* write the change to the current file */
+	prepare_spoolfile_write(action, s);
+
+	return true;
+}
+
+/*
+ * Create the spoolfile used to serialize the prepare messages.
+ */
+static void
+prepare_spoolfile_create(char *path)
+{
+	MemoryContext	oldctx;
+	bool			found;
+	PsfHashEntry   *hentry;
+
+	elog(DEBUG1, "creating file \"%s\" for prepare changes", path);
+
+	Assert(psf_fd == NULL);
+
+	/* create or find the prepare spoolfile entry in the psf_hash */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_ENTER | HASH_FIND,
+										  &found);
+
+	/*
+	 * Create/open the bufFiles under the Prepare Spoolfile Context so that we
+	 * have those files until prepare commit/rollback.
+	 */
+	oldctx = MemoryContextSwitchTo(PsfContext);
+
+	if (!found)
+	{
+		MemoryContext	savectx;
+		SharedFileSet  *fileset;
+
+		elog(DEBUG1, "Not found file \"%s\"", path);
+		savectx = MemoryContextSwitchTo(ApplyContext);
+		fileset = palloc(sizeof(SharedFileSet));
+
+		SharedFileSetInit(fileset, NULL);
+		MemoryContextSwitchTo(savectx);
+
+		psf_fd = BufFileCreateShared(fileset, path);
+
+		/* Remember this path's fileset for the next time */
+		memcpy(hentry->name, path, MAXPGPATH);
+		hentry->fileset = fileset;
+	}
+	else
+	{
+		/*
+		 * Open the file and seek to the beginning because we always want to
+		 * create/overwrite this file.
+		 */
+		elog(DEBUG1, "Found file \"%s\". Overwrite it.", path);
+		psf_fd = BufFileOpenShared(hentry->fileset, path, O_RDWR);
+		BufFileSeek(psf_fd, 0, 0L, SEEK_SET);
+	}
+
+	MemoryContextSwitchTo(oldctx);
+
+	/* Sanity check */
+	Assert(prepare_spoolfile_exists(path));
+}
+
+/*
+ * Close the "current" spoolfile and unset the fd.
+ */
+static void
+prepare_spoolfile_close()
+{
+	if (psf_fd)
+		BufFileClose(psf_fd);
+	psf_fd = NULL;
+}
+
+/*
+ * Delete the specified psf spoolfile.
+ */
+static void
+prepare_spoolfile_cleanup(char *path)
+{
+	PsfHashEntry   *hentry;
+
+	/* The current psf should be closed already, but make sure anyway. */
+	prepare_spoolfile_close();
+
+	/* And remove the path entry from the psf_hash */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_REMOVE,
+										  NULL);
+
+	/* By this time we must have created the entry */
+	Assert(hentry != NULL);
+
+	/* Delete the file and release the fileset memory */
+	SharedFileSetDeleteAll(hentry->fileset);
+	pfree(hentry->fileset);
+	hentry->fileset = NULL;
+
+	/* Reset the memory context used during the file creation */
+	MemoryContextReset(PsfContext);
+}
+
+/*
+ * Serialize a change to the prepare spoolfile for the current toplevel transaction.
+ *
+ * The change is serialized in a simple format, with length (not including
+ * the length), action code (identifying the message type) and message
+ * contents (without the subxact TransactionId value).
+ */
+static void
+prepare_spoolfile_write(char action, StringInfo s)
+{
+	int			len;
+
+	Assert(psf_fd != NULL);
+
+	elog(DEBUG1, "prepare_spoolfile_write: writing action '%c'", action);
+
+	/* total on-disk size, including the action type character */
+	len = (s->len - s->cursor) + sizeof(char);
+
+	/* first write the size */
+	BufFileWrite(psf_fd, &len, sizeof(len));
+
+	/* then the action */
+	BufFileWrite(psf_fd, &action, sizeof(action));
+
+	/* and finally the remaining part of the buffer (after the XID) */
+	len = (s->len - s->cursor);
+
+	BufFileWrite(psf_fd, &s->data[s->cursor], len);
+}
+
+/*
+ * Is there a prepare spoolfile for the specified gid?
+ */
+static bool
+prepare_spoolfile_exists(char *path)
+{
+	bool	found;
+
+	/* Find the prepare spoolfile entry in the psf_hash */
+	hash_search(psf_hash,
+				path,
+				HASH_FIND,
+				&found);
+
+	elog(DEBUG1,
+		 "prepare_spoolfile_exists: Prepared spoolfile \"%s\" was %s",
+		 path,
+		 found ? "found" : "not found");
+
+	return found;
+}
+
+/*
+ * Replay (apply) all the prepared messages that are in the prepare spoolfile.
+ *
+ * [Note: this is mostly copied code from apply_spooled_messages function]
+ */
+static int
+prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn)
+{
+	StringInfoData	s2;
+	int				nchanges = 0;
+	char		   *buffer = NULL;
+	MemoryContext	oldctx, oldctx2;
+	bool			found = false;
+	PsfHashEntry   *hentry;
+	BufFile		   *fd;
+
+	elog(DEBUG1,
+		 "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"",
+		 path);
+
+	/*
+	 * Allocate memory required to process all the messages in
+	 * TopTransactionContext to avoid it getting reset after each message is
+	 * processed.
+	 */
+	oldctx = MemoryContextSwitchTo(TopTransactionContext);
+
+	/* Open the spool file */
+	hentry = (PsfHashEntry *) hash_search(psf_hash,
+										  path,
+										  HASH_FIND,
+										  &found);
+	Assert(found);
+	fd = BufFileOpenShared(hentry->fileset, path, O_RDONLY);
+
+	buffer = palloc(BLCKSZ);
+	initStringInfo(&s2);
+
+	MemoryContextSwitchTo(oldctx);
+
+	remote_final_lsn = lsn;
+
+	/*
+	 * Make sure the handle apply_dispatch methods are aware we're in a remote
+	 * transaction.
+	 */
+	in_remote_transaction = true;
+	pgstat_report_activity(STATE_RUNNING, NULL);
+
+	/*
+	 * Read the entries one by one and pass them through the same logic as in
+	 * apply_dispatch.
+	 */
+	while (true)
+	{
+		int	nbytes;
+		int	len;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* read length of the on-disk record */
+		nbytes = BufFileRead(fd, &len, sizeof(len));
+
+		/* have we reached end of the file? */
+		if (nbytes == 0)
+			break;
+
+		/* do we have a correct length? */
+		if (nbytes != sizeof(len))
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read from prepared spoolfile \"%s\": %m",
+							path)));
+
+		Assert(len > 0);
+
+		/* make sure we have sufficiently large buffer */
+		buffer = repalloc(buffer, len);
+
+		/* and finally read the data into the buffer */
+		if (BufFileRead(fd, buffer, len) != len)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not read from prepared spoolfile \"%s\": %m",
+							path)));
+
+		/* copy the buffer to the stringinfo and call apply_dispatch */
+		resetStringInfo(&s2);
+		appendBinaryStringInfo(&s2, buffer, len);
+
+		/* Ensure we are reading the data into our memory context. */
+		oldctx2 = MemoryContextSwitchTo(ApplyMessageContext);
+
+		apply_dispatch(&s2);
+
+		MemoryContextReset(ApplyMessageContext);
+
+		MemoryContextSwitchTo(oldctx2);
+
+		nchanges++;
+
+		if (nchanges % 1000 == 0)
+			elog(DEBUG1, "replayed %d changes from file '%s'",
+				 nchanges, path);
+	}
+
+	BufFileClose(fd);
+
+	pfree(buffer);
+	pfree(s2.data);
+
+	elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
+		 nchanges, path);
+
+	return nchanges;
+}
+
+/*
+ * Format the filename for the prepare spoolfile.
+ */
+static void
+prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid)
+{
+	PsfHashEntry *hentry;
+
+	/*
+	 * This name is used as the key in the psf_hash HTAB.
+	 *
+	 * Therefore, the name and the key must be exactly same lengths and padded
+	 * with '\0' so garbage does not impact the HTAB lookups.
+	 */
+	Assert(sizeof(hentry->name) == MAXPGPATH);
+	Assert(szpath == MAXPGPATH);
+	memset(path, '\0', MAXPGPATH);
+
+	snprintf(path, MAXPGPATH, "%u-%s.prep_changes", subid, gid);
+}
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 1cac75e..602b4f7 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -86,6 +86,9 @@ extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
 											  char *originname, int szorgname);
 extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);
 
+extern bool BusyTablesyncs(void);
+extern XLogRecPtr BiggestTablesyncLSN(void);
+
 void		process_syncing_tables(XLogRecPtr current_lsn);
 void		invalidate_syncing_table_states(Datum arg, int cacheid,
 											uint32 hashvalue);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 048681c..b7da03b 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1958,6 +1958,7 @@ ProtocolVersion
 PrsStorage
 PruneState
 PruneStepResult
+PsfContext
 PsqlScanCallbacks
 PsqlScanQuoteType
 PsqlScanResult
-- 
1.8.3.1