From a0f79192efe423cf88c1ecbe27edf431b34c373f Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Wed, 4 Nov 2026 20:17:34 +0900 Subject: [PATCH v7 3/3] Distinguish leader context in parallel apply worker errors Errors propagated from a parallel apply worker already contain the context of the operation that failed in that worker. The leader apply worker may be processing a different transaction when it receives such an error, so adding the leader's normal apply context without any identities makes the two operations difficult to distinguish. Add a separate error context callback for errors propagated from parallel apply workers. It labels the leader's current operation explicitly before adding its apply context. Install this callback only while rethrowing a parallel apply worker error, so errors raised directly by the leader apply worker are unaffected. --- .../replication/logical/applyparallelworker.c | 11 ++++++++--- src/backend/replication/logical/worker.c | 18 ++++++++++++++++++ src/include/replication/worker_internal.h | 1 + ..._streaming_parallel_apply_error_context.pl | 19 +++++++------------ 4 files changed, 34 insertions(+), 15 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 09dce307250..36a41d2cf1e 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -1029,6 +1029,7 @@ ProcessParallelApplyMessage(StringInfo msg) case PqMsg_ErrorResponse: { ErrorData edata; + ErrorContextCallback leader_errcallback; /* Parse ErrorResponse. */ pq_parse_errornotice(msg, &edata); @@ -1049,10 +1050,14 @@ ProcessParallelApplyMessage(StringInfo msg) edata.context = pstrdup(_("logical replication parallel apply worker")); /* - * Use the callbacks saved before the leader's apply error - * context callback was installed. + * Add the leader's current activity only while propagating an + * error received from a parallel apply worker. */ - error_context_stack = apply_loop_outer_error_context_stack; + leader_errcallback.callback = leader_apply_error_callback; + leader_errcallback.arg = NULL; + leader_errcallback.previous = + apply_loop_outer_error_context_stack; + error_context_stack = &leader_errcallback; /* Rethrow error from parallel apply worker */ ThrowErrorData(&edata); diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index a1452686ef1..a5dbb65c837 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -6418,6 +6418,24 @@ apply_error_callback(void *arg) } } +/* + * Error callback for errors propagated from parallel apply workers. + * + * Distinguish the leader's current operation from the operation that failed + * in the parallel apply worker. + */ +void +leader_apply_error_callback(void *arg) +{ + ApplyRemoteCtx *ctx = &remote_ctx; + + if (ctx->command == 0) + return; + + errcontext("leader apply worker was processing:"); + apply_error_callback(arg); +} + /* * Set information identifying the remote transaction currently being * applied, kept for the duration of that transaction. diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index febe255c902..fca9a21fbc8 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -327,6 +327,7 @@ extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn); /* Function for apply error callback */ extern void apply_error_callback(void *arg); extern void set_apply_error_context_origin(char *originname); +extern void leader_apply_error_callback(void *arg); /* Parallel apply worker setup and interactions */ extern void pa_allocate_worker(TransactionId xid); diff --git a/src/test/subscription/t/039_streaming_parallel_apply_error_context.pl b/src/test/subscription/t/039_streaming_parallel_apply_error_context.pl index 44d71f7cd3a..74ddb8ed524 100644 --- a/src/test/subscription/t/039_streaming_parallel_apply_error_context.pl +++ b/src/test/subscription/t/039_streaming_parallel_apply_error_context.pl @@ -1,9 +1,8 @@ # Copyright (c) 2026, PostgreSQL Global Development Group -# Test that an error relayed from a parallel apply worker (applying a -# streamed transaction) to the leader does not get a misleading extra -# CONTEXT line describing some other, unrelated transaction the leader -# happens to be applying directly at the same time. +# Test that an error relayed from a parallel apply worker clearly +# distinguishes the operation that failed in the parallel worker from +# the operation concurrently processed by the leader apply worker. # # This is the same underlying issue as reported upstream for the parallel # apply of non-streaming transactions patch set, but reproduced here @@ -158,15 +157,11 @@ like( 'the relayed error is the tab_a trigger exception' ); -# ... and it must not be polluted with an unrelated CONTEXT line describing -# whatever the leader itself happened to be doing at that moment (tab_b): -# the leader's own context-stack walk in ProcessParallelApplyMessage() -# currently appends exactly such a bogus line, so this fails today. -unlike( +# ... and the log also includes the leader apply worker context +like( $log_contents, - qr/relation "public\.tab_b"/, - 'the relayed error must not be annotated with the leader\'s own, ' - . 'unrelated tab_b activity as additional context' + qr/leader apply worker was processing:.*relation "public\.tab_b"/s, + 'the relayed error includes the leader apply worker context' ); # TX1 never got to commit (its worker errored out instead); errors make -- 2.52.0