From b4a743194d7b59b1d6a19a442f3f277206a931bc Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Wed, 7 Oct 2026 15:56:06 +0900 Subject: [PATCH v7 2/3] Avoid adding leader apply context to parallel worker errors Errors from parallel apply workers are relayed through the leader apply worker. Before reporting such an error, ProcessParallelApplyMessage() replaces error_context_stack with a stack saved by LogicalRepApplyLoop(). This avoids running callbacks installed temporarily by whatever the leader is doing when the error arrives. Commit 216a784829c saved that stack after pushing apply_error_callback, intentionally retaining the error context callbacks in effect inside LogicalRepApplyLoop(). However, apply_error_callback describes the leader's own apply state, including its current relation, remote transaction, and replication message. If the leader is applying another transaction when a parallel worker reports an error, the callback therefore adds a misleading CONTEXT line describing the leader's unrelated work. The error relayed from the parallel worker already contains the context of the operation that failed, so running the leader's apply_error_callback provides no useful information. Reverse this part of the choice made by 216a784829c by saving error_context_stack before pushing apply_error_callback. This preserves callbacks outside LogicalRepApplyLoop() while excluding the leader's unrelated apply context. top-up: use ThrowErrorData --- .../replication/logical/applyparallelworker.c | 22 +++++++++---------- src/backend/replication/logical/worker.c | 13 ++++++++--- src/include/replication/worker_internal.h | 2 +- ..._streaming_parallel_apply_error_context.pl | 4 ++-- 4 files changed, 24 insertions(+), 17 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 96bf499e49c..09dce307250 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -1033,6 +1033,9 @@ ProcessParallelApplyMessage(StringInfo msg) /* Parse ErrorResponse. */ pq_parse_errornotice(msg, &edata); + /* Death of a worker isn't enough justification for suicide. */ + edata.elevel = Min(edata.elevel, ERROR); + /* * If desired, add a context line to show that this is a * message propagated from a parallel apply worker. Otherwise, @@ -1046,19 +1049,16 @@ ProcessParallelApplyMessage(StringInfo msg) edata.context = pstrdup(_("logical replication parallel apply worker")); /* - * Context beyond that should use the error context callbacks - * that were in effect in LogicalRepApplyLoop(). + * Use the callbacks saved before the leader's apply error + * context callback was installed. */ - error_context_stack = apply_error_context_stack; + error_context_stack = apply_loop_outer_error_context_stack; - /* - * The actual error must have been reported by the parallel - * apply worker. - */ - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("logical replication parallel apply worker exited due to error"), - errcontext("%s", edata.context))); + /* Rethrow error from parallel apply worker */ + ThrowErrorData(&edata); + + /* Not reachable */ + Assert(false); } /* diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 37b37500213..a1452686ef1 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -483,7 +483,12 @@ static ApplyRemoteCtx remote_ctx = .origin_name = NULL, }; -ErrorContextCallback *apply_error_context_stack = NULL; +/* + * Error context stack outside LogicalRepApplyLoop(), before + * apply_error_callback is pushed. Used when the leader reports an error + * propagated from a parallel apply worker. + */ +ErrorContextCallback *apply_loop_outer_error_context_stack = NULL; MemoryContext ApplyMessageContext = NULL; MemoryContext ApplyContext = NULL; @@ -4081,6 +4086,9 @@ LogicalRepApplyLoop(XLogRecPtr last_received) /* mark as idle, before starting to loop */ pgstat_report_activity(STATE_IDLE, NULL); + /* Preserve the error context stack outside the apply loop. */ + apply_loop_outer_error_context_stack = error_context_stack; + /* * Push apply error context callback. Fields will be filled while applying * a change. @@ -4088,7 +4096,6 @@ LogicalRepApplyLoop(XLogRecPtr last_received) errcallback.callback = apply_error_callback; errcallback.previous = error_context_stack; error_context_stack = &errcallback; - apply_error_context_stack = error_context_stack; /* This outer loop iterates once per wait. */ for (;;) @@ -4357,7 +4364,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received) /* Pop the error context stack */ error_context_stack = errcallback.previous; - apply_error_context_stack = error_context_stack; + apply_loop_outer_error_context_stack = error_context_stack; /* All done */ walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli); diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 88cb7c1e252..febe255c902 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -238,7 +238,7 @@ extern PGDLLIMPORT MemoryContext ApplyContext; extern PGDLLIMPORT MemoryContext ApplyMessageContext; -extern PGDLLIMPORT ErrorContextCallback *apply_error_context_stack; +extern PGDLLIMPORT ErrorContextCallback *apply_loop_outer_error_context_stack; extern PGDLLIMPORT ParallelApplyWorkerShared *MyParallelShared; 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 a2acb2dd83a..44d71f7cd3a 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 @@ -3,7 +3,7 @@ # 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. This currently fails. +# happens to be applying directly at the same time. # # This is the same underlying issue as reported upstream for the parallel # apply of non-streaming transactions patch set, but reproduced here @@ -146,7 +146,7 @@ $node_subscriber->safe_psql('postgres', "SELECT injection_points_wakeup('parallel-worker-before-stream-start')"); $node_subscriber->wait_for_log( - qr/logical replication parallel apply worker exited due to error/, + qr/logical replication parallel apply worker/, $log_offset); my $log_contents = slurp_file($node_subscriber->logfile, $log_offset); -- 2.52.0