From 12679264b359cebc2129c0b5a80a3c995981cf3a Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Tue, 4 Aug 2026 12:10:08 +0530 Subject: [PATCH v66 2/3] Perform conflict log tuple insertion directly in ReportApplyConflict Previously, when an ERROR-level logical replication conflict was detected, ReportApplyConflict() prepared the conflict log tuple and immediately raised the error via ereport(ERROR). The actual insertion of the deferred conflict log tuple (ProcessPendingConflictLogTuple) was executed inside PG_CATCH() blocks in the apply worker (start_apply) and parallel apply worker (ParallelApplyWorkerMain) after saving error data and aborting the failed apply transaction. This patch refactors the conflict reporting workflow by handling the deferred conflict log insertion directly inside ReportApplyConflict() before raising the error. When an ERROR-level conflict is reported and table logging is enabled, ReportApplyConflict() stashes necessary detail strings into ApplyContext, clears the origin replication state, aborts the failed apply transaction, and calls ProcessPendingConflictLogTuple() to insert the conflict tuple in a fresh transaction before emitting the final error. This eliminates error-handling from the error recovery paths in worker.c and applyparallelworker.c. Instead of the worker setting PARALLEL_TRANS_ERROR before aborting, the leader now waits in pa_wait_for_xact_finish() for the worker to either report its error or exit, detecting exit via logicalrep_pa_worker_running() checking its LogicalRepWorker slot. This drops the worker-side flag and its spinlock from the error path. --- .../replication/logical/applyparallelworker.c | 101 ++------------ src/backend/replication/logical/conflict.c | 130 ++++++++++-------- src/backend/replication/logical/launcher.c | 29 ++++ src/backend/replication/logical/worker.c | 31 +---- src/include/replication/worker_internal.h | 4 +- 5 files changed, 120 insertions(+), 175 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 55a0ee4831a..93e72bc4572 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -986,66 +986,7 @@ ParallelApplyWorkerMain(Datum main_arg) set_apply_error_context_origin(originname); - PG_TRY(); - { - LogicalParallelApplyLoop(mqh); - } - PG_CATCH(); - { - MemoryContext oldcontext; - ErrorData *edata; - - /* - * Reset the origin state to prevent the advancement of origin - * progress if we fail to apply. Otherwise, this will result in - * transaction loss as that transaction won't be sent again by the - * server. - */ - replorigin_xact_clear(true); - - /* - * Copy the error and recover to an idle state so we can insert the - * deferred conflict log tuple (if any) before re-throwing. Copy the - * error into a longer-lived context first, as it may have been raised - * under ErrorContext. Also reset the error context stack: the - * callbacks in effect when the error was thrown belong to unwound - * stack frames, and the deferred insert installs its own. - */ - oldcontext = MemoryContextSwitchTo(TopMemoryContext); - edata = CopyErrorData(); - MemoryContextSwitchTo(oldcontext); - - FlushErrorState(); - error_context_stack = NULL; - - /* - * Tell the leader we failed and are about to report the error and log - * the conflict. This must be set before AbortOutOfAnyTransaction() - * below releases the transaction lock that the leader waits on in - * pa_wait_for_xact_finish(); otherwise the leader would see a - * non-finished state, assume the connection was lost, and tear this - * worker down while it is still writing the conflict log tuple. - */ - pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_ERROR); - - AbortOutOfAnyTransaction(); - - /* - * Insert the deferred conflict log tuple before re-throwing. - * Re-throwing is what reports the error to the leader (via the error - * queue set up above), so the insertion must happen first: otherwise - * the leader could start tearing down this worker while it is still - * writing the conflict log tuple. If the insertion itself fails, - * that error (annotated with the conflict context, see - * InsertConflictLogTuple) propagates to the leader instead of the - * original. - */ - ProcessPendingConflictLogTuple(); - - /* Re-throw the original error, which reports it to the leader. */ - ReThrowError(edata); - } - PG_END_TRY(); + LogicalParallelApplyLoop(mqh); /* * The parallel apply worker must not get here because the parallel apply @@ -1342,6 +1283,11 @@ pa_wait_for_xact_state(ParallelApplyWorkerInfo *winfo, /* An interrupt may have occurred while we were waiting. */ CHECK_FOR_INTERRUPTS(); + + if (!logicalrep_pa_worker_running(winfo)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("lost connection to the logical replication parallel apply worker"))); } } @@ -1368,38 +1314,11 @@ pa_wait_for_xact_finish(ParallelApplyWorkerInfo *winfo) pa_unlock_transaction(winfo->shared->xid, AccessShareLock); /* - * Check if the state becomes PARALLEL_TRANS_FINISHED in case the parallel - * apply worker failed while applying changes causing the lock to be - * released. + * Wait for the transaction state to reach PARALLEL_TRANS_FINISHED. The wait + * function handles the case where the parallel apply worker errors out + * before updating the state. */ - if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED) - { - /* - * If the worker signalled that it errored (PARALLEL_TRANS_ERROR), it - * is logging the conflict and will report the actual error via the - * error queue before exiting. Wait for that rather than reporting a - * generic lost connection: CHECK_FOR_INTERRUPTS() drives - * ProcessParallelApplyMessages(), which raises the real error on the - * worker's ErrorResponse (or "lost connection" if the worker died - * without reporting). Waiting here also keeps the worker alive long - * enough to finish writing the conflict log tuple. - */ - while (pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_ERROR) - { - CHECK_FOR_INTERRUPTS(); - - (void) WaitLatch(MyLatch, - WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, - 10L, - WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE); - - ResetLatch(MyLatch); - } - - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("lost connection to the logical replication parallel apply worker"))); - } + pa_wait_for_xact_state(winfo, PARALLEL_TRANS_FINISHED); } /* diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index d53b95a4f28..63841827645 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -27,6 +27,7 @@ #include "funcapi.h" #include "pgstat.h" #include "replication/conflict.h" +#include "replication/origin.h" #include "replication/worker_internal.h" #include "storage/lmgr.h" #include "utils/array.h" @@ -340,6 +341,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, Relation conflictlogrel; bool log_dest_table; bool log_dest_logfile; + char *logdetail = NULL; pgstat_report_subscription_conflict(MySubscription->oid, type); @@ -386,12 +388,9 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, } /* - * Report the conflict to the server log before inserting it into the - * conflict log table. Emitting it first guarantees the conflict is - * recorded even if the table insert below fails; it is also what raises - * the error for ERROR-level conflicts. When the server log is one of the - * destinations we emit the full details, otherwise (table-only) we emit a - * shorter message since the details are captured in the table. + * Build the server-log detail now, while the executor tuples are still + * available. For an ERROR-level conflict we abort below (which frees + * them) before raising the error. */ if (log_dest_logfile) { @@ -409,21 +408,82 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, conflicttuple->ts, &err_detail); - /* Standard reporting with full internal details. */ + logdetail = err_detail.data; + } + + /* + * For an ERROR-level conflict, insert the conflict log tuple in its own + * transaction and then raise the error, so that no transactional work runs + * in the apply worker's error (PG_CATCH) path. When the table is a + * destination, AbortOutOfAnyTransaction() below frees the executor tuples + * and closes the conflict log relation, so the strings the report needs are + * first captured into a context that survives the abort. + */ + if (elevel >= ERROR) + { + if (log_dest_table) + { + MemoryContext oldctx; + char *qualname; + char *clt_relname = NULL; + + oldctx = MemoryContextSwitchTo(ApplyContext); + qualname = pstrdup(RelationGetQualifiedRelationName(localrel)); + if (logdetail) + logdetail = pstrdup(logdetail); + else /* table is the only destination */ + clt_relname = pstrdup(RelationGetRelationName(conflictlogrel)); + MemoryContextSwitchTo(oldctx); + + /* + * Reset the origin so the insert's commit does not advance + * replication progress. Then abort the failed apply transaction + * and insert the conflict tuple in a fresh transaction. + */ + replorigin_xact_clear(true); + AbortOutOfAnyTransaction(); + ProcessPendingConflictLogTuple(); + + if (logdetail) + ereport(elevel, + errcode_apply_conflict(type), + errmsg("conflict detected on relation \"%s\": conflict=%s", + qualname, ConflictTypeNames[type]), + errdetail_internal("%s", logdetail)); + else + ereport(elevel, + errcode_apply_conflict(type), + errmsg("conflict detected on relation \"%s\": conflict=%s", + qualname, ConflictTypeNames[type]), + errdetail("Conflict details are logged to the conflict log table: %s", + clt_relname)); + } + else + { + /* Server log only; no abort needed, executor tuples still valid. */ + ereport(elevel, + errcode_apply_conflict(type), + errmsg("conflict detected on relation \"%s\": conflict=%s", + RelationGetQualifiedRelationName(localrel), + ConflictTypeNames[type]), + errdetail_internal("%s", logdetail)); + } + + pg_unreachable(); + } + + /* + * Below ERROR the apply transaction continues. Report the conflict, then + * insert the tuple immediately in the same transaction if requested. + */ + if (log_dest_logfile) ereport(elevel, errcode_apply_conflict(type), errmsg("conflict detected on relation \"%s\": conflict=%s", RelationGetQualifiedRelationName(localrel), ConflictTypeNames[type]), - errdetail_internal("%s", err_detail.data)); - } + errdetail_internal("%s", logdetail)); else if (log_dest_table) - { - /* - * Not logging conflict details to the server log; report the conflict - * but omit raw tuple data since it is captured in the conflict log - * table. - */ ereport(elevel, errcode_apply_conflict(type), errmsg("conflict detected on relation \"%s\": conflict=%s", @@ -431,48 +491,10 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, ConflictTypeNames[type]), errdetail("Conflict details are logged to the conflict log table: %s", RelationGetRelationName(conflictlogrel))); - } - /* - * Insert into the conflict log table if requested. For conflicts below - * ERROR the apply transaction continues, so insert immediately; for - * ERROR-level conflicts the ereport() above already raised the error and - * the insertion is deferred to a new transaction - * (ProcessPendingConflictLogTuple) so that it is not rolled back. - */ if (log_dest_table) { - if (elevel < ERROR) - { - PG_TRY(); - { - InsertConflictLogTuple(conflictlogrel); - } - PG_CATCH(); - { - /* - * The insert failed, so the apply transaction will abort and - * the error will propagate to the worker's error handler. The - * conflict was already reported to the server log above, so - * it is not lost. Discard the prepared tuple so that the - * deferred insertion path (ProcessPendingConflictLogTuple) - * does not retry this same failing insert. - */ - if (pending_conflict_log.tuple != NULL) - { - heap_freetuple(pending_conflict_log.tuple); - pending_conflict_log.tuple = NULL; - } - if (pending_conflict_log.errcontext_str != NULL) - { - pfree(pending_conflict_log.errcontext_str); - pending_conflict_log.errcontext_str = NULL; - } - PG_RE_THROW(); - } - PG_END_TRY(); - } - + InsertConflictLogTuple(conflictlogrel); table_close(conflictlogrel, RowExclusiveLock); } } diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 313e31ff2e3..71fb3321d3e 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -725,6 +725,35 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo) LWLockRelease(LogicalRepWorkerLock); } +/* + * Is the given parallel apply worker still running? + * + * The generation guards against the slot having been reused by a different + * worker after ours exited. + */ +bool +logicalrep_pa_worker_running(ParallelApplyWorkerInfo *winfo) +{ + int slot_no; + uint16 generation; + LogicalRepWorker *worker; + bool running; + + SpinLockAcquire(&winfo->shared->mutex); + generation = winfo->shared->logicalrep_worker_generation; + slot_no = winfo->shared->logicalrep_worker_slot_no; + SpinLockRelease(&winfo->shared->mutex); + + Assert(slot_no >= 0 && slot_no < max_logical_replication_workers); + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + worker = &LogicalRepCtx->workers[slot_no]; + running = (worker->generation == generation && worker->proc != NULL); + LWLockRelease(LogicalRepWorkerLock); + + return running; +} + /* * Wake up (using latch) any logical replication worker that matches the * specified worker type, subscription id, and relation id. diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index e57e1523d0f..a43535e7dda 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -5660,9 +5660,6 @@ start_apply(XLogRecPtr origin_startpos) } PG_CATCH(); { - MemoryContext oldcontext; - ErrorData *edata; - /* * Reset the origin state to prevent the advancement of origin * progress if we fail to apply. Otherwise, this will result in @@ -5676,34 +5673,14 @@ start_apply(XLogRecPtr origin_startpos) else { /* - * Save the error and recover to an idle state so we can insert - * the deferred conflict log tuple (if any) before re-throwing. - * Copy the error into a long-lived context first, as it may have - * been raised under ErrorContext. Also reset the error context - * stack: the callbacks in effect when the error was thrown belong - * to unwound stack frames, and the deferred insert installs its - * own. + * Report the worker failed while applying changes. Abort the + * current transaction so that the stats message is sent in an + * idle state. */ - oldcontext = MemoryContextSwitchTo(TopMemoryContext); - edata = CopyErrorData(); - MemoryContextSwitchTo(oldcontext); - - FlushErrorState(); - error_context_stack = NULL; AbortOutOfAnyTransaction(); pgstat_report_subscription_error(MySubscription->oid); - /* - * Insert the deferred conflict log tuple in its own transaction. - * If this fails, that error (annotated with the conflict context, - * see InsertConflictLogTuple) propagates instead of the original; - * such failures are expected to be rare and persistent (e.g. out - * of disk space). - */ - ProcessPendingConflictLogTuple(); - - /* Re-throw the original error. */ - ReThrowError(edata); + PG_RE_THROW(); } } PG_END_TRY(); diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 394f4c6265e..1a74a366a44 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -121,9 +121,6 @@ typedef enum ParallelTransState PARALLEL_TRANS_UNKNOWN, PARALLEL_TRANS_STARTED, PARALLEL_TRANS_FINISHED, - PARALLEL_TRANS_ERROR, /* worker failed; it will report the error - * (and log the conflict, if any) before - * exiting */ } ParallelTransState; /* @@ -276,6 +273,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype, extern void logicalrep_worker_stop(LogicalRepWorkerType wtype, Oid subid, Oid relid); extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo); +extern bool logicalrep_pa_worker_running(ParallelApplyWorkerInfo *winfo); extern void logicalrep_worker_wakeup(LogicalRepWorkerType wtype, Oid subid, Oid relid); extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker); -- 2.54.0