From ff800695324b0839aabbaff5721032c55aa9f37e Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Wed, 19 Aug 2026 22:53:09 +0800
Subject: [PATCH v71 1/2] Refactor apply worker's remote transaction tracking

The apply worker tracked the final LSN of the remote transaction being
applied in a separate global variable remote_final_lsn, even though
apply_error_callback_arg already tracked the same transaction's xid and
finish LSN for error context reporting, and both were updated together
at the same places.

Rename ApplyErrorCallbackArg to ApplyRemoteCtx and treat its
remote_xid/finish_lsn pair as the generic descriptor of the remote
transaction currently being applied, replacing all uses of
remote_final_lsn so the redundant global can be removed.  Rename
set_apply_error_context_xact() to set_remote_transaction_info() and
reset_apply_error_context_info() to reset_apply_remote_context() to
match the wider role.

There is no behavior change.  The only place that updated
remote_final_lsn without going through set_apply_error_context_xact()
was apply_spooled_messages(), but its leader apply worker callers set
the same LSN in the error context just before replaying the spooled
changes, and the parallel apply worker passed InvalidXLogRecPtr, which
matches the context's initial value there.  Conversely, a STREAM ABORT
for a subtransaction recorded the subxid and abort LSN only in the
error context; that difference was never observable, because no change
is applied between a STREAM ABORT and the STREAM START or STREAM
COMMIT/PREPARE that follows it, and each of those records the
transaction's own values again.
---
 .../replication/logical/applyparallelworker.c |   5 +-
 src/backend/replication/logical/worker.c      | 177 +++++++++++-------
 src/tools/pgindent/typedefs.list              |   2 +-
 3 files changed, 116 insertions(+), 68 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 012d55e9d3d..96bf499e49c 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -309,8 +309,9 @@ pa_can_start(void)
 	 * For streaming transactions that are being applied using a parallel
 	 * apply worker, we cannot decide whether to apply the change for a
 	 * relation that is not in the READY state (see
-	 * should_apply_changes_for_rel) as we won't know remote_final_lsn by that
-	 * time. So, we don't start the new parallel apply worker in this case.
+	 * should_apply_changes_for_rel) as we won't know the finish LSN of the
+	 * transaction by that time. So, we don't start the new parallel apply
+	 * worker in this case.
 	 */
 	if (!AllTablesyncsReady())
 		return false;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 92ea1d0df24..c33097ec18a 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -324,8 +324,23 @@ typedef struct ApplyExecutionData
 	PartitionTupleRouting *proute;	/* partition routing info */
 } ApplyExecutionData;
 
-/* Struct for saving and restoring apply errcontext information */
-typedef struct ApplyErrorCallbackArg
+/*
+ * Context describing the remote transaction whose changes are currently
+ * being applied, and the change within it.
+ *
+ * The remote transaction information (remote_xid and finish_lsn) is set when
+ * the transaction's changes begin to be applied and is used not only for
+ * error context reporting (see apply_error_callback), but also to decide
+ * whether a change should be applied at all (see
+ * should_apply_changes_for_rel) and to validate the transaction finish
+ * messages.  finish_lsn is invalid when the final LSN of the remote
+ * transaction is not yet known (e.g. while streaming an in-progress
+ * transaction).
+ *
+ * The remaining fields describe the individual change being applied and are
+ * used only for error context reporting.
+ */
+typedef struct ApplyRemoteCtx
 {
 	LogicalRepMsgType command;	/* 0 if invalid */
 	LogicalRepRelMapEntry *rel;
@@ -335,7 +350,7 @@ typedef struct ApplyErrorCallbackArg
 	TransactionId remote_xid;
 	XLogRecPtr	finish_lsn;
 	char	   *origin_name;
-} ApplyErrorCallbackArg;
+} ApplyRemoteCtx;
 
 /*
  * The action to be taken for the changes in the transaction.
@@ -460,8 +475,8 @@ typedef struct RetainDeadTuplesData
 #define MIN_XID_ADVANCE_INTERVAL 100
 #define MAX_XID_ADVANCE_INTERVAL 180000
 
-/* errcontext tracker */
-static ApplyErrorCallbackArg apply_error_callback_arg =
+/* Context of the remote transaction being applied */
+static ApplyRemoteCtx remote_ctx =
 {
 	.command = 0,
 	.rel = NULL,
@@ -488,7 +503,6 @@ static bool MySubscriptionValid = false;
 static List *on_commit_wakeup_workers_subids = NIL;
 
 bool		in_remote_transaction = false;
-static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
 
 /* fields valid only when processing streamed transaction */
 static bool in_streamed_transaction = false;
@@ -626,9 +640,9 @@ static void maybe_start_skipping_changes(XLogRecPtr finish_lsn);
 static void stop_skipping_changes(void);
 static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn);
 
-/* Functions for apply error callback */
-static inline void set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn);
-static inline void reset_apply_error_context_info(void);
+/* Functions to maintain the context of the remote transaction being applied */
+static inline void set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn);
+static inline void reset_apply_remote_context(void);
 
 static TransApplyAction get_transaction_apply_action(TransactionId xid,
 													 ParallelApplyWorkerInfo **winfo);
@@ -671,13 +685,14 @@ ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
  * Note we need to do smaller or equals comparison for SYNCDONE state because
  * it might hold position of end of initial slot consistent point WAL
  * record + 1 (ie start of next record) and next record can be COMMIT of
- * transaction we are now processing (which is what we set remote_final_lsn
- * to in apply_handle_begin).
+ * transaction we are now processing (which is what we set the finish LSN of
+ * the remote transaction context to in apply_handle_begin).
  *
  * Note that for streaming transactions that are being applied in the parallel
  * apply worker, we disallow applying changes if the target table in the
  * subscription is not in the READY state, because we cannot decide whether to
- * apply the change as we won't know remote_final_lsn by that time.
+ * apply the change as we won't know the finish LSN of the transaction by
+ * that time.
  *
  * We already checked this in pa_can_start() before assigning the
  * streaming transaction to the parallel worker, but it also needs to be
@@ -708,7 +723,7 @@ should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
 		case WORKERTYPE_APPLY:
 			return (rel->state == SUBREL_STATE_READY ||
 					(rel->state == SUBREL_STATE_SYNCDONE &&
-					 rel->statelsn <= remote_final_lsn));
+					 rel->statelsn <= remote_ctx.finish_lsn));
 
 		case WORKERTYPE_SEQUENCESYNC:
 			/* Should never happen. */
@@ -1050,7 +1065,7 @@ slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel,
 			colvalue = &tupleData->colvalues[remoteattnum];
 
 			/* Set attnum for error callback */
-			apply_error_callback_arg.remote_attnum = remoteattnum;
+			remote_ctx.remote_attnum = remoteattnum;
 
 			if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT)
 			{
@@ -1099,7 +1114,7 @@ slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel,
 			}
 
 			/* Reset attnum for error callback */
-			apply_error_callback_arg.remote_attnum = -1;
+			remote_ctx.remote_attnum = -1;
 		}
 		else
 		{
@@ -1169,7 +1184,7 @@ slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot,
 			StringInfo	colvalue = &tupleData->colvalues[remoteattnum];
 
 			/* Set attnum for error callback */
-			apply_error_callback_arg.remote_attnum = remoteattnum;
+			remote_ctx.remote_attnum = remoteattnum;
 
 			if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT)
 			{
@@ -1214,7 +1229,7 @@ slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot,
 			}
 
 			/* Reset attnum for error callback */
-			apply_error_callback_arg.remote_attnum = -1;
+			remote_ctx.remote_attnum = -1;
 		}
 	}
 
@@ -1234,9 +1249,7 @@ apply_handle_begin(StringInfo s)
 	Assert(!TransactionIdIsValid(stream_xid));
 
 	logicalrep_read_begin(s, &begin_data);
-	set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
-
-	remote_final_lsn = begin_data.final_lsn;
+	set_remote_transaction_info(begin_data.xid, begin_data.final_lsn);
 
 	maybe_start_skipping_changes(begin_data.final_lsn);
 
@@ -1257,12 +1270,12 @@ apply_handle_commit(StringInfo s)
 
 	logicalrep_read_commit(s, &commit_data);
 
-	if (commit_data.commit_lsn != remote_final_lsn)
+	if (commit_data.commit_lsn != remote_ctx.finish_lsn)
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("incorrect commit LSN %X/%08X in commit message (expected %X/%08X)",
 								 LSN_FORMAT_ARGS(commit_data.commit_lsn),
-								 LSN_FORMAT_ARGS(remote_final_lsn))));
+								 LSN_FORMAT_ARGS(remote_ctx.finish_lsn))));
 
 	apply_handle_commit_internal(&commit_data);
 
@@ -1273,7 +1286,7 @@ apply_handle_commit(StringInfo s)
 	ProcessSyncingRelations(commit_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -1294,9 +1307,7 @@ apply_handle_begin_prepare(StringInfo s)
 	Assert(!TransactionIdIsValid(stream_xid));
 
 	logicalrep_read_begin_prepare(s, &begin_data);
-	set_apply_error_context_xact(begin_data.xid, begin_data.prepare_lsn);
-
-	remote_final_lsn = begin_data.prepare_lsn;
+	set_remote_transaction_info(begin_data.xid, begin_data.prepare_lsn);
 
 	maybe_start_skipping_changes(begin_data.prepare_lsn);
 
@@ -1352,12 +1363,12 @@ apply_handle_prepare(StringInfo s)
 
 	logicalrep_read_prepare(s, &prepare_data);
 
-	if (prepare_data.prepare_lsn != remote_final_lsn)
+	if (prepare_data.prepare_lsn != remote_ctx.finish_lsn)
 		ereport(ERROR,
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("incorrect prepare LSN %X/%08X in prepare message (expected %X/%08X)",
 								 LSN_FORMAT_ARGS(prepare_data.prepare_lsn),
-								 LSN_FORMAT_ARGS(remote_final_lsn))));
+								 LSN_FORMAT_ARGS(remote_ctx.finish_lsn))));
 
 	/*
 	 * Unlike commit, here, we always prepare the transaction even though no
@@ -1407,7 +1418,7 @@ apply_handle_prepare(StringInfo s)
 	clear_subscription_skip_lsn(prepare_data.prepare_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -1426,7 +1437,7 @@ apply_handle_commit_prepared(StringInfo s)
 	char		gid[GIDSIZE];
 
 	logicalrep_read_commit_prepared(s, &prepare_data);
-	set_apply_error_context_xact(prepare_data.xid, prepare_data.commit_lsn);
+	set_remote_transaction_info(prepare_data.xid, prepare_data.commit_lsn);
 
 	/* Compute GID for two_phase transactions. */
 	TwoPhaseTransactionGid(MySubscription->oid, prepare_data.xid,
@@ -1459,7 +1470,7 @@ apply_handle_commit_prepared(StringInfo s)
 	clear_subscription_skip_lsn(prepare_data.end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -1478,7 +1489,7 @@ apply_handle_rollback_prepared(StringInfo s)
 	char		gid[GIDSIZE];
 
 	logicalrep_read_rollback_prepared(s, &rollback_data);
-	set_apply_error_context_xact(rollback_data.xid, rollback_data.rollback_end_lsn);
+	set_remote_transaction_info(rollback_data.xid, rollback_data.rollback_end_lsn);
 
 	/* Compute GID for two_phase transactions. */
 	TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid,
@@ -1526,7 +1537,7 @@ apply_handle_rollback_prepared(StringInfo s)
 	ProcessSyncingRelations(rollback_data.rollback_end_lsn);
 
 	pgstat_report_activity(STATE_IDLE, NULL);
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -1554,7 +1565,7 @@ apply_handle_stream_prepare(StringInfo s)
 				 errmsg_internal("tablesync worker received a STREAM PREPARE message")));
 
 	logicalrep_read_stream_prepare(s, &prepare_data);
-	set_apply_error_context_xact(prepare_data.xid, prepare_data.prepare_lsn);
+	set_remote_transaction_info(prepare_data.xid, prepare_data.prepare_lsn);
 
 	apply_action = get_transaction_apply_action(prepare_data.xid, &winfo);
 
@@ -1672,7 +1683,7 @@ apply_handle_stream_prepare(StringInfo s)
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -1768,7 +1779,11 @@ apply_handle_stream_start(StringInfo s)
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("invalid transaction ID in streamed replication transaction")));
 
-	set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
+	/*
+	 * The final LSN of the streamed transaction is known only when its commit
+	 * record arrives.
+	 */
+	set_remote_transaction_info(stream_xid, InvalidXLogRecPtr);
 
 	/* Try to allocate a worker for the streaming transaction. */
 	if (first_segment)
@@ -1995,7 +2010,7 @@ apply_handle_stream_stop(StringInfo s)
 	else
 		pgstat_report_activity(STATE_IDLE, NULL);
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -2111,7 +2126,13 @@ apply_handle_stream_abort(StringInfo s)
 	subxid = abort_data.subxid;
 	toplevel_xact = (xid == subxid);
 
-	set_apply_error_context_xact(subxid, abort_data.abort_lsn);
+	/*
+	 * Record the xid of the (sub)transaction being aborted, so that the error
+	 * context names whatever failed.  Note this is the top-level xid itself
+	 * when a top-level transaction aborts, and a subxid only when a
+	 * subtransaction rolls back.  See set_remote_transaction_info().
+	 */
+	set_remote_transaction_info(subxid, abort_data.abort_lsn);
 
 	apply_action = get_transaction_apply_action(xid, &winfo);
 
@@ -2236,7 +2257,7 @@ apply_handle_stream_abort(StringInfo s)
 			break;
 	}
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -2318,7 +2339,12 @@ apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
 
 	MemoryContextSwitchTo(oldcxt);
 
-	remote_final_lsn = lsn;
+	/*
+	 * No need to record the finish LSN here: the leader apply worker has
+	 * already set it in the remote transaction context when processing the
+	 * transaction finish message, and in the parallel apply worker it is
+	 * invalid, just like the lsn passed by the caller there.
+	 */
 
 	/*
 	 * Make sure the handle apply_dispatch methods are aware we're in a remote
@@ -2421,7 +2447,7 @@ apply_handle_stream_commit(StringInfo s)
 				 errmsg_internal("STREAM COMMIT message without STREAM STOP")));
 
 	xid = logicalrep_read_stream_commit(s, &commit_data);
-	set_apply_error_context_xact(xid, commit_data.commit_lsn);
+	set_remote_transaction_info(xid, commit_data.commit_lsn);
 
 	apply_action = get_transaction_apply_action(xid, &winfo);
 
@@ -2511,7 +2537,7 @@ apply_handle_stream_commit(StringInfo s)
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -2692,7 +2718,7 @@ apply_handle_insert(StringInfo s)
 		SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt);
 
 	/* Set relation for error callback */
-	apply_error_callback_arg.rel = rel;
+	remote_ctx.rel = rel;
 
 	/* Initialize the executor state. */
 	edata = create_edata_for_relation(rel);
@@ -2723,7 +2749,7 @@ apply_handle_insert(StringInfo s)
 	finish_edata(edata);
 
 	/* Reset relation for error callback */
-	apply_error_callback_arg.rel = NULL;
+	remote_ctx.rel = NULL;
 
 	if (!run_as_owner)
 		RestoreUserContext(&ucxt);
@@ -2845,7 +2871,7 @@ apply_handle_update(StringInfo s)
 	}
 
 	/* Set relation for error callback */
-	apply_error_callback_arg.rel = rel;
+	remote_ctx.rel = rel;
 
 	/* Check if we can do the update. */
 	check_relation_updatable(rel);
@@ -2911,7 +2937,7 @@ apply_handle_update(StringInfo s)
 	finish_edata(edata);
 
 	/* Reset relation for error callback */
-	apply_error_callback_arg.rel = NULL;
+	remote_ctx.rel = NULL;
 
 	if (!run_as_owner)
 		RestoreUserContext(&ucxt);
@@ -3068,7 +3094,7 @@ apply_handle_delete(StringInfo s)
 	}
 
 	/* Set relation for error callback */
-	apply_error_callback_arg.rel = rel;
+	remote_ctx.rel = rel;
 
 	/* Check if we can do the delete. */
 	check_relation_updatable(rel);
@@ -3110,7 +3136,7 @@ apply_handle_delete(StringInfo s)
 	finish_edata(edata);
 
 	/* Reset relation for error callback */
-	apply_error_callback_arg.rel = NULL;
+	remote_ctx.rel = NULL;
 
 	if (!run_as_owner)
 		RestoreUserContext(&ucxt);
@@ -3805,8 +3831,8 @@ apply_dispatch(StringInfo s)
 	 * called recursively when applying spooled changes, save the current
 	 * command.
 	 */
-	saved_command = apply_error_callback_arg.command;
-	apply_error_callback_arg.command = action;
+	saved_command = remote_ctx.command;
+	remote_ctx.command = action;
 
 	switch (action)
 	{
@@ -3898,7 +3924,7 @@ apply_dispatch(StringInfo s)
 	}
 
 	/* Reset the current command */
-	apply_error_callback_arg.command = saved_command;
+	remote_ctx.command = saved_command;
 }
 
 /*
@@ -6290,9 +6316,9 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 void
 apply_error_callback(void *arg)
 {
-	ApplyErrorCallbackArg *errarg = &apply_error_callback_arg;
+	ApplyRemoteCtx *errarg = &remote_ctx;
 
-	if (apply_error_callback_arg.command == 0)
+	if (remote_ctx.command == 0)
 		return;
 
 	Assert(errarg->origin_name);
@@ -6358,22 +6384,43 @@ apply_error_callback(void *arg)
 	}
 }
 
-/* Set transaction information of apply error callback */
+/*
+ * Set information identifying the remote transaction currently being
+ * applied, kept for the duration of that transaction.  Besides naming the
+ * transaction in the error context, this decides whether a change is applied
+ * at all (see should_apply_changes_for_rel) and validates the transaction
+ * finish messages.
+ *
+ * This must be called for every message type that begins or resumes applying
+ * a remote transaction's changes (BEGIN, BEGIN PREPARE, STREAM START, STREAM
+ * COMMIT, STREAM PREPARE), since interleaved transactions (possible only for
+ * streaming, where other transactions can be applied to completion in
+ * between one transaction's streamed chunks) would otherwise leave stale
+ * values from whichever transaction last called this.
+ *
+ * Callers normally pass the top-level transaction's xid.  The exception is a
+ * STREAM ABORT, which passes the xid of the (sub)transaction being aborted so
+ * that the error context names whatever failed; that is a subxid only when a
+ * subtransaction rolls back, and the top-level xid otherwise.  Nothing else
+ * observes a subxid recorded this way, because no change is applied between a
+ * STREAM ABORT and the STREAM START or STREAM COMMIT/PREPARE that follows it,
+ * and each of those calls this again with the transaction's own values.
+ */
 static inline void
-set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn)
+set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn)
 {
-	apply_error_callback_arg.remote_xid = xid;
-	apply_error_callback_arg.finish_lsn = lsn;
+	remote_ctx.remote_xid = xid;
+	remote_ctx.finish_lsn = lsn;
 }
 
-/* Reset all information of apply error callback */
+/* Reset all information of the remote transaction context */
 static inline void
-reset_apply_error_context_info(void)
+reset_apply_remote_context(void)
 {
-	apply_error_callback_arg.command = 0;
-	apply_error_callback_arg.rel = NULL;
-	apply_error_callback_arg.remote_attnum = -1;
-	set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
+	remote_ctx.command = 0;
+	remote_ctx.rel = NULL;
+	remote_ctx.remote_attnum = -1;
+	set_remote_transaction_info(InvalidTransactionId, InvalidXLogRecPtr);
 }
 
 /*
@@ -6432,8 +6479,8 @@ AtEOXact_LogicalRepWorkers(bool isCommit)
 void
 set_apply_error_context_origin(char *originname)
 {
-	apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
-															   originname);
+	remote_ctx.origin_name = MemoryContextStrdup(ApplyContext,
+												 originname);
 }
 
 /*
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 298a3d586e7..665f8155a93 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -132,8 +132,8 @@ AppendPath
 AppendPathInput
 AppendRelInfo
 AppendState
-ApplyErrorCallbackArg
 ApplyExecutionData
+ApplyRemoteCtx
 ApplySubXactData
 Archive
 ArchiveCheckConfiguredCB
-- 
2.49.0

