From 667df7c75c0ed4d1890eac2ab33b268ba2dd98dc 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 v72 1/3] Remove the redundant remote_final_lsn variable from
 the apply worker.

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.

There is no behavior change.

Recording the remote transaction being applied in a single place should
also help the proposed conflict log table patch, which needs the same
information when logging a conflict.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: Amit Kapila <amit.kapila16@gmail.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Dilip Kumar <dilipbalaut@gmail.com>
Discussion: https://postgr.es/m/TY4PR01MB177186ED4D38C5356294CC51C94A42@TY4PR01MB17718.jpnprd01.prod.outlook.com
---
 .../replication/logical/applyparallelworker.c |   5 +-
 src/backend/replication/logical/worker.c      | 243 ++++++++++--------
 src/tools/pgindent/typedefs.list              |   2 +-
 3 files changed, 141 insertions(+), 109 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..9ba08ab3fb1 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -324,8 +324,19 @@ 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. 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 +346,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 +471,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 +499,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 +636,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 +681,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 +719,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 +1061,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 +1110,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 +1180,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 +1225,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 +1245,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 +1266,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 +1282,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 +1303,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 +1359,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 +1414,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 +1433,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 +1466,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 +1485,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 +1533,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 +1561,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 +1679,7 @@ apply_handle_stream_prepare(StringInfo s)
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -1768,7 +1775,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 +2006,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 +2122,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 +2253,7 @@ apply_handle_stream_abort(StringInfo s)
 			break;
 	}
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -2318,8 +2335,6 @@ apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
 
 	MemoryContextSwitchTo(oldcxt);
 
-	remote_final_lsn = lsn;
-
 	/*
 	 * Make sure the handle apply_dispatch methods are aware we're in a remote
 	 * transaction.
@@ -2421,7 +2436,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 +2526,7 @@ apply_handle_stream_commit(StringInfo s)
 
 	pgstat_report_activity(STATE_IDLE, NULL);
 
-	reset_apply_error_context_info();
+	reset_apply_remote_context();
 }
 
 /*
@@ -2692,7 +2707,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 +2738,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 +2860,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 +2926,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 +3083,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 +3125,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 +3820,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 +3913,7 @@ apply_dispatch(StringInfo s)
 	}
 
 	/* Reset the current command */
-	apply_error_callback_arg.command = saved_command;
+	remote_ctx.command = saved_command;
 }
 
 /*
@@ -6290,90 +6305,107 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
 void
 apply_error_callback(void *arg)
 {
-	ApplyErrorCallbackArg *errarg = &apply_error_callback_arg;
+	ApplyRemoteCtx *ctx = &remote_ctx;
 
-	if (apply_error_callback_arg.command == 0)
+	if (ctx->command == 0)
 		return;
 
-	Assert(errarg->origin_name);
+	Assert(ctx->origin_name);
 
-	if (errarg->rel == NULL)
+	if (ctx->rel == NULL)
 	{
-		if (!TransactionIdIsValid(errarg->remote_xid))
+		if (!TransactionIdIsValid(ctx->remote_xid))
 			errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
-					   errarg->origin_name,
-					   logicalrep_message_type(errarg->command));
-		else if (!XLogRecPtrIsValid(errarg->finish_lsn))
+					   ctx->origin_name,
+					   logicalrep_message_type(ctx->command));
+		else if (!XLogRecPtrIsValid(ctx->finish_lsn))
 			errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
-					   errarg->origin_name,
-					   logicalrep_message_type(errarg->command),
-					   errarg->remote_xid);
+					   ctx->origin_name,
+					   logicalrep_message_type(ctx->command),
+					   ctx->remote_xid);
 		else
 			errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%08X",
-					   errarg->origin_name,
-					   logicalrep_message_type(errarg->command),
-					   errarg->remote_xid,
-					   LSN_FORMAT_ARGS(errarg->finish_lsn));
+					   ctx->origin_name,
+					   logicalrep_message_type(ctx->command),
+					   ctx->remote_xid,
+					   LSN_FORMAT_ARGS(ctx->finish_lsn));
 	}
 	else
 	{
-		if (errarg->remote_attnum < 0)
+		if (ctx->remote_attnum < 0)
 		{
-			if (!XLogRecPtrIsValid(errarg->finish_lsn))
+			if (!XLogRecPtrIsValid(ctx->finish_lsn))
 				errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
-						   errarg->origin_name,
-						   logicalrep_message_type(errarg->command),
-						   errarg->rel->remoterel.nspname,
-						   errarg->rel->remoterel.relname,
-						   errarg->remote_xid);
+						   ctx->origin_name,
+						   logicalrep_message_type(ctx->command),
+						   ctx->rel->remoterel.nspname,
+						   ctx->rel->remoterel.relname,
+						   ctx->remote_xid);
 			else
 				errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u, finished at %X/%08X",
-						   errarg->origin_name,
-						   logicalrep_message_type(errarg->command),
-						   errarg->rel->remoterel.nspname,
-						   errarg->rel->remoterel.relname,
-						   errarg->remote_xid,
-						   LSN_FORMAT_ARGS(errarg->finish_lsn));
+						   ctx->origin_name,
+						   logicalrep_message_type(ctx->command),
+						   ctx->rel->remoterel.nspname,
+						   ctx->rel->remoterel.relname,
+						   ctx->remote_xid,
+						   LSN_FORMAT_ARGS(ctx->finish_lsn));
 		}
 		else
 		{
-			if (!XLogRecPtrIsValid(errarg->finish_lsn))
+			if (!XLogRecPtrIsValid(ctx->finish_lsn))
 				errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
-						   errarg->origin_name,
-						   logicalrep_message_type(errarg->command),
-						   errarg->rel->remoterel.nspname,
-						   errarg->rel->remoterel.relname,
-						   errarg->rel->remoterel.attnames[errarg->remote_attnum],
-						   errarg->remote_xid);
+						   ctx->origin_name,
+						   logicalrep_message_type(ctx->command),
+						   ctx->rel->remoterel.nspname,
+						   ctx->rel->remoterel.relname,
+						   ctx->rel->remoterel.attnames[ctx->remote_attnum],
+						   ctx->remote_xid);
 			else
 				errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u, finished at %X/%08X",
-						   errarg->origin_name,
-						   logicalrep_message_type(errarg->command),
-						   errarg->rel->remoterel.nspname,
-						   errarg->rel->remoterel.relname,
-						   errarg->rel->remoterel.attnames[errarg->remote_attnum],
-						   errarg->remote_xid,
-						   LSN_FORMAT_ARGS(errarg->finish_lsn));
+						   ctx->origin_name,
+						   logicalrep_message_type(ctx->command),
+						   ctx->rel->remoterel.nspname,
+						   ctx->rel->remoterel.relname,
+						   ctx->rel->remoterel.attnames[ctx->remote_attnum],
+						   ctx->remote_xid,
+						   LSN_FORMAT_ARGS(ctx->finish_lsn));
 		}
 	}
 }
 
-/* Set transaction information of apply error callback */
+/*
+ * Set information identifying the remote transaction currently being
+ * applied, kept for the duration of that transaction.
+ *
+ * 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) 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 +6464,7 @@ 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 6c366d3a523..cf0ac5b351b 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

