From 4729d90908bb70219df9a9d23c44e5c9fa4f7fd7 Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Mon, 20 Jul 2026 15:08:45 +0530 Subject: [PATCH v63-topup] Insert conflict log tuple in ReportApplyConflict() for ERROR conflicts. Previously an ERROR-level conflict raised the error immediately, and the apply worker's PG_CATCH block aborted the transaction and inserted the conflict log tuple in a new transaction before re-throwing. Running that transactional work inside a catch block is an unusual pattern. Instead, ReportApplyConflict() now resets the origin, aborts the apply transaction, inserts the conflict log tuple in a fresh transaction, and only then raises the error. The report strings are captured beforehand because the abort frees the executor tuples and closes the conflict log relation. As the insert no longer runs during error recovery, the worker error handlers are simplified: start_apply()'s catch returns to plain teardown and ParallelApplyWorkerMain() no longer needs a PG_CATCH. The pa_wait_for_xact_finish() wait loop is retained; ReportApplyConflict() sets PARALLEL_TRANS_ERROR before the abort so the leader still waits for the worker to finish writing the conflict row. --- .../replication/logical/applyparallelworker.c | 61 +------- src/backend/replication/logical/conflict.c | 135 +++++++++++------- src/backend/replication/logical/worker.c | 31 +--- 3 files changed, 86 insertions(+), 141 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 55a0ee4831a..0261b7bff73 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 diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index d53b95a4f28..045ae98c8e4 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,87 @@ 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, and (for a parallel apply worker) tell the + * leader we errored before the abort releases the transaction lock + * it waits on in pa_wait_for_xact_finish(). Then abort the failed + * apply transaction and insert the conflict tuple in a fresh + * transaction. + */ + replorigin_xact_clear(true); + if (am_parallel_apply_worker()) + pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_ERROR); + 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 +496,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/worker.c b/src/backend/replication/logical/worker.c index 042c7203619..68f5ec707e4 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(); -- 2.54.0