On Fri, Sep 4, 2026 at 3:27 PM Zhijie Hou (Fujitsu)
<[email protected]> wrote:
>
> Thanks for reporting this. I think we should skip updating the slot.xmin in
> the
> cycle where the old worker hasn't stopped but retention has been resumed.
> Here's
> the updated patch that fixes this. To make it cleaner, I also refactored the
> logic slightly to centralize the decision of whether to update the slot.xmin.
>
I have made minor changes in the comments to make them clear. See attached.
> >
> > Both 4 and 5 are documented but will it be better to provide this
> > information explicitly, say via Logging when the conflcit_slot is not
> > getting advanced due to any of these reasons?
>
> We can do that. It seems to me we could make this improvement for PG20 at this
> stage. For PG19, I think we can improve the documentation to make it a bit
> clearer instead. Amit and I discussed this off-list, and See 0002 for the
> patch
> Amit shared to improve the doc.
>
Thanks for sharing the doc updates. Can you please fold those in 0001
and share a combined patch?
--
With Regards,
Amit Kapila.
diff --git a/src/backend/replication/logical/launcher.c
b/src/backend/replication/logical/launcher.c
index e011789fe36..ee4cb870667 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -1318,16 +1318,16 @@ ApplyLauncherMain(Datum main_arg)
can_update_xmin &= sub->enabled;
/*
- * Reset the slot's xmin to the safe
decoding horizon if
- * it is not valid, or if this database
newly appears
- * among the retained databases: the
xmin may be newer
- * than the new database's oldest
active transaction ID,
- * violating the per-database invariant
checked in
- * get_candidate_xid(). The set is
keyed on
- * subretentionactive, so that a
subscription resuming
- * retention is also treated as a new
arrival. One call
- * per cycle is enough, as the horizon
is a safe seed for
- * every database.
+ * A retaining apply worker starts with
the slot's xmin as
+ * its oldest_nonremovable_xid (see
+ * logicalrep_worker_launch()), so a
database whose oldest
+ * active transaction ID precedes that
xmin would trip the
+ * assertion in get_candidate_xid().
Reset the xmin before
+ * starting the worker, which also
seeds it on the first
+ * cycle. The set of retained databases
is keyed on
+ * subretentionactive, so a
subscription resuming retention
+ * is also treated as a new arrival.
One call per cycle is
+ * enough, as a single reset covers all
such databases.
*/
if (!reset_done &&
(!TransactionIdIsValid(MyReplicationSlot->data.xmin) ||
@@ -1349,9 +1349,7 @@ ApplyLauncherMain(Datum main_arg)
/*
* Compute the minimum xmin required to protect dead
tuples
* required for conflict detection among all running
apply
- * workers. One subscription that cannot contribute is
enough to
- * prevent the slot's xmin from advancing, so stop
collecting once
- * that happens.
+ * workers.
*/
can_update_xmin = can_update_xmin &&
collect_min_nonremovable_xid(sub, w, &xmin);
@@ -1461,13 +1459,14 @@ ApplyLauncherMain(Datum main_arg)
}
/*
- * Fold this subscription's minimum non-removable transaction ID into *xmin,
- * for subscriptions that are actively retaining conflict information.
+ * Determine the minimum non-removable transaction ID required for conflict
+ * detection, accumulating the contribution of the given subscription in
+ * *xmin. Subscriptions that are not actively retaining conflict information
+ * are ignored.
*
* Returns false if the slot's xmin cannot be advanced in this cycle, which is
* the case when the apply worker is not running (worker is NULL) or does not
- * yet hold a valid oldest_nonremovable_xid. Subscriptions that are not
- * retaining place no constraint on the xmin and return true.
+ * yet hold a valid oldest_nonremovable_xid.
*
* The caller must hold LogicalRepWorkerLock, to prevent accessing invalid
* worker data in scenarios where a worker might exit and reset its state
@@ -1572,14 +1571,11 @@ update_conflict_slot_xmin(TransactionId new_xmin)
/*
* Reset the xmin of the conflict detection slot to the cluster-wide safe
- * decoding horizon, which is a safe seed for an apply worker in any
- * database.
+ * decoding horizon, which accounts for all running transactions and is
+ * therefore safe for an apply worker in any database. Called when the slot is
+ * created and when a database newly needs retention.
*
- * This is used both for the initial setup and when a database newly appears
- * among the databases with actively-retaining subscriptions; see
- * ApplyLauncherMain(). An already-valid xmin is only moved backwards, as
- * advancing it here would bypass the advancement protocol; regressing is
- * safe, and the workers will advance the xmin again in later cycles.
+ * The xmin is left unchanged if it is already at or before the horizon.
*/
static void
reset_conflict_slot_xmin_to_safe_horizon(void)
@@ -1597,7 +1593,10 @@ reset_conflict_slot_xmin_to_safe_horizon(void)
/*
* Nothing to do if the current xmin is valid and not newer than the
- * horizon.
+ * horizon. The xmin must not be advanced here, as advancing is only
safe
+ * once the apply workers confirm that all concurrent transactions have
+ * been applied. Regressing it only retains more than necessary, and the
+ * workers will advance it again in later cycles.
*/
if (TransactionIdIsValid(old_xmin) &&
TransactionIdPrecedesOrEquals(old_xmin, xmin_horizon))
