Hi, On Wed, Apr 15, 2026 at 10:03 PM Bharath Rupireddy <[email protected]> wrote: > > Hi, > > On Mon, Apr 6, 2026 at 10:42 AM Bharath Rupireddy > <[email protected]> wrote: > > > > On Mon, Apr 6, 2026 at 1:45 AM Masahiko Sawada <[email protected]> > > wrote: > > > > > > > I took a look at the v10 patch and it LGTM. I tested it - make > > > > check-world passes, pgindent doesn't complain. > > > > > > While reviewing the patch, I found that with this patch, backend > > > processes and autovacuum workers can simultaneously attempt to > > > invalidate the same slot for the same reason. When invalidating a > > > slot, we send a signal to the process owning the slot and wait for it > > > to exit and release the slot. If the process takes a long time to exit > > > for some reason, subsequent autovacuum workers attempting to > > > invalidate the same slot will also send a SIGTERM and get stuck at > > > InvalidatePossiblyObsoleteSlot(). In the worst case, this could result > > > in all autovacuum activity being blocked. I think we need to address > > > this problem. > > > > Thank you! > > > > You're right that multiple autovacuum workers can wait on the same > > slot for SIGTERM to take effect on the process (mainly walsenders) > > holding the slot. Once the process holding the slot exits, one worker > > finishes the invalidation and the others see it's done and move on. > > > > However, IMHO, this is unlikely to be a problem in practice. > > > > First, SIGTERM must take a long time to terminate the process holding > > the slot. This seems unlikely unless I'm missing some cases. > > > > Second, the slot's xmin must be very old (past XID age) while the > > process is still running but slow to exit. If we set max_slot_xid_age > > close to vacuum_failsafe_age (e.g., 1.6 billion. I've added this note > > in the docs), it seems unlikely that the replication connection would > > still be active at that point. > > > > Also, concurrent invalidation can already happen today between the > > startup process and checkpointer on standby. > > > > If needed, we could add a flag to skip extra invalidation attempts > > based on field experience. Since this feature is off by default, I'd > > prefer to keep things simple, but I'm open to other approaches. > > > > Thoughts? > > Thank you Sawada-san. I've been thinking more about it and I agree we > need to address this. While I still think the scenario is unlikely in > practice (SIGTERM would have to take a long time, the slot's xmin > would have to be very old while the walsender is still running, etc.), > I think it's worth handling. > > I can think of a couple of approaches: > > 1. Use ConditionVariableTimedSleep instead of ConditionVariableSleep > when called from an autovacuum worker. Workers don't block forever, > but they still wait for the timeout duration, still send redundant > SIGTERMs, and a correct timeout value needs to be chosen. When it > expires, the worker either retries (still stuck) or gives up (same as > approach 2). > > 2. Make the vacuum path non-blocking when another process is already > invalidating the same slot. The first process to attempt invalidation > proceeds normally: it sends SIGTERM and waits on > ConditionVariableSleep for the process holding the slot to exit. But > if a subsequent autovacuum worker finds that another process has > already initiated invalidation of this slot, it skips the slot and > proceeds with vacuum instead of waiting on the same > ConditionVariableSleep. > > I think approach 2 is simple. If another process is already > invalidating the slot, there's no reason for the autovacuum worker to > also block. The tradeoff is that this vacuum cycle's OldestXmin won't > move forward and it will need another cycle for this relation. But > that's fine given that the scenario as explained above is unlikely to > happen in practice. > > Please let me know if my thinking sounds reasonable. I'm open to other > ideas too. > > Thoughts?
I implemented the approach 2 (patch 0003). I added an injection point to mimic the walsender taking time to process SIGTERM, so that the process invalidating the slot waits on the slot's CV. Please have a look and share your thoughts. Thank you! -- Bharath Rupireddy Amazon Web Services: https://aws.amazon.com
From 0d879d4dc2812ed5c071d53a06d936b09bd62957 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 17 Apr 2026 18:29:41 +0000 Subject: [PATCH v11 1/3] Introduce max_slot_xid_age to invalidate old replication slots This commit introduces a new GUC parameter, max_slot_xid_age. It invalidates replication slots whose xmin or catalog_xmin age exceeds the configured limit. While time-based slot invalidation is useful for cleaning up inactive slots, an XID-age-based limit acts as a critical backstop to directly prevent transaction ID wraparound and severe bloat caused by orphaned slots. The invalidation check occurs during both VACUUM (manual and autovacuum) and checkpoints. During vacuum, the check is performed on a per-relation basis. Crucially, the XID-age based slot invalidation is considered only when slots' XIDs are actively holding back the OldestXmin for the current relation. In other words, we only invalidate slots when doing so has the potential to advance the vacuum cutoff and allow dead tuple reclamation. Checking slot XIDs per relation could introduce significant performance overhead if it required acquiring ProcArrayLock repeatedly to get replication slot xmin values (xmin and catalog_xmin in procArray). To avoid this, the vacuum cutoff calculation has been optimized. A new function GetOldestNonRemovableTransactionIdWithSlotXids() is introduced to return the global OldestXmin along with the oldest slot xmin and catalog_xmin values. This allows the per-relation invalidation check to be performed with zero additional lock acquisitions. In addition to vacuum, slots are also checked and invalidated during checkpoints. This is particularly important for standby servers where vacuum does not run. Note that on standbys, slots that are currently being synced from the primary (i.e., synced = true) are exempt from this invalidation mechanism. Author: Bharath Rupireddy <[email protected]> Co-authored-by: John Hsu <[email protected]> Reviewed-by: Masahiko Sawada <[email protected]> Reviewed-by: Bertrand Drouvot <[email protected]> Reviewed-by: shveta malik <[email protected]> Reviewed-by: Amit Kapila <[email protected]> Reviewed-by: SATYANARAYANA NARLAPURAM <[email protected]> Discussion: https://postgr.es/m/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com Discussion: https://postgr.es/m/CA+-JvFsMHckBMzsu5Ov9HCG3AFbMh056hHy1FiXazBRtZ9pFBg@mail.gmail.com --- doc/src/sgml/config.sgml | 54 ++++++ doc/src/sgml/logical-replication.sgml | 4 +- doc/src/sgml/system-views.sgml | 8 + src/backend/access/heap/vacuumlazy.c | 18 ++ src/backend/access/transam/xlog.c | 34 +++- src/backend/commands/vacuum.c | 80 +++++++- src/backend/commands/vacuumparallel.c | 26 +++ src/backend/replication/slot.c | 82 +++++++- src/backend/storage/ipc/procarray.c | 61 ++++-- src/backend/storage/ipc/standby.c | 3 +- src/backend/utils/misc/guc_parameters.dat | 8 + src/backend/utils/misc/postgresql.conf.sample | 2 + src/bin/pg_basebackup/pg_createsubscriber.c | 2 +- src/include/commands/vacuum.h | 10 + src/include/replication/slot.h | 8 +- src/include/storage/procarray.h | 3 + src/test/recovery/t/019_replslot_limit.pl | 175 ++++++++++++++++++ 17 files changed, 550 insertions(+), 28 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 67da9a1de66..ff24e259a43 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4948,6 +4948,60 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-max-slot-xid-age" xreflabel="max_slot_xid_age"> + <term><varname>max_slot_xid_age</varname> (<type>integer</type>) + <indexterm> + <primary><varname>max_slot_xid_age</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots whose <structfield>xmin</structfield> age + or <structfield>catalog_xmin</structfield> age in the + <link linkend="view-pg-replication-slots">pg_replication_slots</link> + view has exceeded the age specified by this setting. Slot invalidation + due to this limit occurs during vacuum (both <command>VACUUM</command> + command and autovacuum) and during checkpoint. + A value of zero (the default) disables this feature. Users can set + this value anywhere from zero to two billion transactions. This parameter + can only be set in the <filename>postgresql.conf</filename> file or on + the server command line. + </para> + + <para> + The current age of a slot's <literal>xmin</literal> and + <literal>catalog_xmin</literal> can be monitored by applying the + <function>age</function> function to the corresponding columns in the + <link linkend="view-pg-replication-slots">pg_replication_slots</link> + view. + </para> + + <para> + Idle or forgotten replication slots can hold back vacuum, leading to + bloat and eventually transaction ID wraparound. This setting avoids + that by invalidating slots that have fallen too far behind. + See <xref linkend="routine-vacuuming"/> for more details. + </para> + + <para> + It is recommended to set <varname>max_slot_xid_age</varname> + to a value equal to or slightly less than + <xref linkend="guc-vacuum-failsafe-age"/>, so that the slot holding the + vacuum back is invalidated before vacuum enters failsafe mode. + </para> + + <para> + Note that this invalidation mechanism is not applicable for slots + on the standby server that are being synced from the primary server + (i.e., standby slots having + <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>synced</structfield> + value <literal>true</literal>). Synced slots are always considered to + be inactive because they don't perform logical decoding to produce + changes. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-wal-sender-timeout" xreflabel="wal_sender_timeout"> <term><varname>wal_sender_timeout</varname> (<type>integer</type>) <indexterm> diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 598e23ad4f5..b20f2133d99 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2649,7 +2649,9 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER <para> Logical replication slots are also affected by - <link linkend="guc-idle-replication-slot-timeout"><varname>idle_replication_slot_timeout</varname></link>. + <link linkend="guc-idle-replication-slot-timeout"><varname>idle_replication_slot_timeout</varname></link> + and + <link linkend="guc-max-slot-xid-age"><varname>max_slot_xid_age</varname></link>. </para> <para> diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index 2ebec6928d5..5862f9841c0 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -3102,6 +3102,14 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx <xref linkend="guc-idle-replication-slot-timeout"/> duration. </para> </listitem> + <listitem> + <para> + <literal>xid_aged</literal> means that the slot's + <literal>xmin</literal> or <literal>catalog_xmin</literal> + has reached the age specified by + <xref linkend="guc-max-slot-xid-age"/> parameter. + </para> + </listitem> </itemizedlist> </para></entry> </row> diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 39395aed0d5..19b107a9b70 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -800,6 +800,24 @@ heap_vacuum_rel(Relation rel, const VacuumParams *params, * to increase the number of dead tuples it can prune away.) */ vacrel->aggressive = vacuum_get_cutoffs(rel, params, &vacrel->cutoffs); + + /* + * If the current vacuum cutoff (OldestXmin) is being held back by a + * replication slot that has exceeded max_slot_xid_age, attempt to + * invalidate such slots. + */ + if (maybe_invalidate_xid_aged_slots(vacrel->cutoffs.OldestXmin, + vacrel->cutoffs.OldestSlotXmin, + vacrel->cutoffs.OldestSlotCatalogXmin)) + { + /* + * Some slots have been invalidated; re-compute the vacuum cutoffs and + * aggressiveness. + */ + vacrel->aggressive = vacuum_get_cutoffs(rel, params, + &vacrel->cutoffs); + } + vacrel->rel_pages = orig_rel_pages = RelationGetNumberOfBlocks(rel); vacrel->vistest = GlobalVisTestFor(rel); diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f85b5286086..823ebadab62 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7405,6 +7405,8 @@ CreateCheckPoint(int flags) VirtualTransactionId *vxids; int nvxids; int oldXLogAllowed = 0; + uint32 slotInvalidationCauses; + TransactionId slotXidLimit; /* * An end-of-recovery checkpoint is really a shutdown checkpoint, just @@ -7849,9 +7851,20 @@ CreateCheckPoint(int flags) */ XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size); KeepLogSeg(recptr, &_logSegNo); - if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT, + + slotInvalidationCauses = RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT; + slotXidLimit = InvalidTransactionId; + if (max_slot_xid_age > 0) + { + slotInvalidationCauses |= RS_INVAL_XID_AGE; + slotXidLimit = TransactionIdRetreatedBy(ReadNextTransactionId(), + max_slot_xid_age); + } + + if (InvalidateObsoleteReplicationSlots(slotInvalidationCauses, _logSegNo, InvalidOid, - InvalidTransactionId)) + InvalidTransactionId, + slotXidLimit)) { /* * Some slots have been invalidated; recalculate the old-segment @@ -8138,6 +8151,8 @@ CreateRestartPoint(int flags) XLogRecPtr endptr; XLogSegNo _logSegNo; TimestampTz xtime; + uint32 slotInvalidationCauses; + TransactionId slotXidLimit; /* Concurrent checkpoint/restartpoint cannot happen */ Assert(!IsUnderPostmaster || MyBackendType == B_CHECKPOINTER); @@ -8316,9 +8331,19 @@ CreateRestartPoint(int flags) INJECTION_POINT("restartpoint-before-slot-invalidation", NULL); - if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT, + slotInvalidationCauses = RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT; + slotXidLimit = InvalidTransactionId; + if (max_slot_xid_age > 0) + { + slotInvalidationCauses |= RS_INVAL_XID_AGE; + slotXidLimit = TransactionIdRetreatedBy(ReadNextTransactionId(), + max_slot_xid_age); + } + + if (InvalidateObsoleteReplicationSlots(slotInvalidationCauses, _logSegNo, InvalidOid, - InvalidTransactionId)) + InvalidTransactionId, + slotXidLimit)) { /* * Some slots have been invalidated; recalculate the old-segment @@ -9234,6 +9259,7 @@ xlog_redo(XLogReaderState *record) */ InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, 0, InvalidOid, + InvalidTransactionId, InvalidTransactionId); } else if (sync_replication_slots) diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index 99d0db82ed7..73439a163ef 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -48,6 +48,7 @@ #include "postmaster/autovacuum.h" #include "postmaster/bgworker_internals.h" #include "postmaster/interrupt.h" +#include "replication/slot.h" #include "storage/bufmgr.h" #include "storage/lmgr.h" #include "storage/pmsignal.h" @@ -1134,7 +1135,10 @@ vacuum_get_cutoffs(Relation rel, const VacuumParams *params, * that only one vacuum process can be working on a particular table at * any time, and that each vacuum is always an independent transaction. */ - cutoffs->OldestXmin = GetOldestNonRemovableTransactionId(rel); + cutoffs->OldestXmin = + GetOldestNonRemovableTransactionIdWithSlotXids(rel, + &cutoffs->OldestSlotXmin, + &cutoffs->OldestSlotCatalogXmin); Assert(TransactionIdIsNormal(cutoffs->OldestXmin)); @@ -2708,3 +2712,77 @@ vac_tid_reaped(ItemPointer itemptr, void *state) return TidStoreIsMember(dead_items, itemptr); } + +/* + * Invalidate replication slots whose XID age exceeds max_slot_xid_age. + * + * The caller provides the overall oldest xmin along with the oldest + * slot and catalog_xmin, typically all obtained from a single consistent + * snapshot via ComputeXidHorizons(). These values are used to avoid + * unnecessary work: if the global oldest_xmin is held back by something + * other than a replication slot (e.g., a long-running transaction), + * invalidating slots would not advance the horizon and is therefore + * skipped. Similarly, no action is taken if the current horizons have + * not yet exceeded the threshold. + * + * Returns true if at least one slot was invalidated. + */ +bool +maybe_invalidate_xid_aged_slots(TransactionId oldest_xmin, + TransactionId oldest_slot_xmin, + TransactionId oldest_slot_catalog_xmin) +{ + TransactionId xid_limit; + bool slot_holds_oldest_xmin; + + if (max_slot_xid_age == 0) + return false; + + Assert(TransactionIdIsNormal(oldest_xmin)); + + /* + * Check if a replication slot's xmin or catalog_xmin is what's holding + * back oldest_xmin. If not, skip the unnecessary work. + */ + slot_holds_oldest_xmin = + (TransactionIdIsValid(oldest_slot_xmin) && + TransactionIdEquals(oldest_xmin, oldest_slot_xmin)) || + (TransactionIdIsValid(oldest_slot_catalog_xmin) && + TransactionIdEquals(oldest_xmin, oldest_slot_catalog_xmin)); + + if (!slot_holds_oldest_xmin) + return false; + + xid_limit = TransactionIdRetreatedBy(ReadNextTransactionId(), + max_slot_xid_age); + + /* + * A replication slot is holding back oldest_xmin. We invalidate slots + * that have exceeded the XID age limit. + * + * Note that while a non-catalog vacuum is technically only blocked by + * physical slots' xmin values, we invalidate logical slots too that + * exceed the XID age limit if we trigger the XID-age based slot + * invalidation. One might think that this is unnecessary for non-catalog + * tables as invalidating logical slots while vacuuming a non-catalog + * table doesn't help advance vacuum cutoffs. But performing invalidation + * trials for physical and logical slots would add complexity. + * + * In practice, XID-age-based invalidation is lightweight (e.g., it does + * not require process termination). This unified approach keeps the API + * simple by avoiding the need to distinguish between catalog and + * non-catalog tables here. + * + * Note: Invalidating a slot does not guarantee that the oldest xmin will + * advance. Due to a race condition, a long-running transaction might be + * holding the same xmin as the slot. In such cases, the slot is + * invalidated, but the global horizon remains unchanged. + */ + if (TransactionIdPrecedes(oldest_xmin, xid_limit)) + return InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, + 0, InvalidOid, + InvalidTransactionId, + xid_limit); + + return false; +} diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index 979c2be4abd..deffc537269 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/src/backend/commands/vacuumparallel.c @@ -47,6 +47,7 @@ #include "storage/proc.h" #include "tcop/tcopprot.h" #include "utils/lsyscache.h" +#include "utils/ps_status.h" #include "utils/rel.h" /* @@ -1097,6 +1098,28 @@ parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation indrel, pvs->indname = pstrdup(RelationGetRelationName(indrel)); pvs->status = indstats->status; + /* + * Update the ps display to show which index this worker is currently + * processing, along with the table and index OIDs. This makes it easy + * to identify which index a parallel vacuum worker is stuck on via + * "ps -ef". For example: + * "parallel worker for PID 12345: vacuuming index idx_foo (table OID 16384, index OID 16385)" + */ + { + char ps_suffix[128]; + const char *phase; + + phase = (indstats->status == PARALLEL_INDVAC_STATUS_NEED_BULKDELETE) + ? "vacuuming" : "cleaning up"; + snprintf(ps_suffix, sizeof(ps_suffix), + ": %s index \"%s\" (table OID %u, index OID %u)", + phase, + RelationGetRelationName(indrel), + RelationGetRelid(pvs->heaprel), + RelationGetRelid(indrel)); + set_ps_display_suffix(ps_suffix); + } + switch (indstats->status) { case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE: @@ -1144,6 +1167,9 @@ parallel_vacuum_process_one_index(ParallelVacuumState *pvs, Relation indrel, pfree(pvs->indname); pvs->indname = NULL; + /* Clear the ps display suffix now that this index is done */ + set_ps_display_suffix(""); + /* * Call the parallel variant of pgstat_progress_incr_param so workers can * report progress of index vacuum to the leader. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 83fcde74718..fe43f5c8820 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -118,6 +118,7 @@ static const SlotInvalidationCauseMap SlotInvalidationCauses[] = { {RS_INVAL_HORIZON, "rows_removed"}, {RS_INVAL_WAL_LEVEL, "wal_level_insufficient"}, {RS_INVAL_IDLE_TIMEOUT, "idle_timeout"}, + {RS_INVAL_XID_AGE, "xid_aged"}, }; /* @@ -169,6 +170,12 @@ int max_repack_replication_slots = 5; /* the maximum number of slots */ int idle_replication_slot_timeout_secs = 0; +/* + * Invalidate replication slots that have xmin or catalog_xmin older + * than the specified age; '0' disables it. + */ +int max_slot_xid_age = 0; + /* * This GUC lists streaming replication standby server slot names that * logical WAL sender processes will wait for. @@ -1794,7 +1801,10 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon, - long slot_idle_seconds) + long slot_idle_seconds, + TransactionId xmin, + TransactionId catalog_xmin, + TransactionId xidLimit) { StringInfoData err_detail; StringInfoData err_hint; @@ -1839,6 +1849,29 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, "idle_replication_slot_timeout"); break; } + + case RS_INVAL_XID_AGE: + { + TransactionId slot_xid = TransactionIdIsValid(xmin) ? xmin : catalog_xmin; + int32 exceeded_by = (int32) (xidLimit - slot_xid); + int32 slot_age = (int32) max_slot_xid_age + exceeded_by; + + /* Either the slot's xmin or catalog_xmin must be valid */ + Assert(TransactionIdIsValid(slot_xid)); + + /* translator: %s is a GUC variable name */ + appendStringInfo(&err_detail, + TransactionIdIsValid(xmin) + ? _("The slot's xmin age of %d exceeds the configured \"%s\" of %d by %d transactions") + : _("The slot's catalog xmin age of %d exceeds the configured \"%s\" of %d by %d transactions"), + slot_age, "max_slot_xid_age", max_slot_xid_age, exceeded_by); + + /* translator: %s is a GUC variable name */ + appendStringInfo(&err_hint, _("You might need to increase \"%s\"."), + "max_slot_xid_age"); + break; + } + case RS_INVAL_NONE: pg_unreachable(); } @@ -1877,6 +1910,25 @@ CanInvalidateIdleSlot(ReplicationSlot *s) !(RecoveryInProgress() && s->data.synced)); } +/* + * Can we invalidate an XID-aged replication slot? + * + * XID-aged based invalidation is allowed to the given slot when: + * + * 1. Max XID-age is set + * 2. Slot has valid xmin or catalog_xmin + * 3. The slot is not being synced from the primary while the server is in + * recovery. + */ +static inline bool +CanInvalidateXidAgedSlot(ReplicationSlot *s) +{ + return (max_slot_xid_age != 0 && + (TransactionIdIsValid(s->data.xmin) || + TransactionIdIsValid(s->data.catalog_xmin)) && + !(RecoveryInProgress() && s->data.synced)); +} + /* * DetermineSlotInvalidationCause - Determine the cause for which a slot * becomes invalid among the given possible causes. @@ -1888,6 +1940,7 @@ static ReplicationSlotInvalidationCause DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, + TransactionId xidLimit, TimestampTz *inactive_since, TimestampTz now) { Assert(possible_causes != RS_INVAL_NONE); @@ -1959,6 +2012,18 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, } } + /* Check if the slot needs to be invalidated due to max_slot_xid_age GUC */ + if ((possible_causes & RS_INVAL_XID_AGE) && CanInvalidateXidAgedSlot(s)) + { + Assert(TransactionIdIsValid(xidLimit)); + + if ((TransactionIdIsValid(s->data.xmin) && + TransactionIdPrecedes(s->data.xmin, xidLimit)) || + (TransactionIdIsValid(s->data.catalog_xmin) && + TransactionIdPrecedes(s->data.catalog_xmin, xidLimit))) + return RS_INVAL_XID_AGE; + } + return RS_INVAL_NONE; } @@ -1981,6 +2046,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, + TransactionId xidLimit, bool *released_lock_out) { int last_signaled_pid = 0; @@ -2033,6 +2099,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, + xidLimit, &inactive_since, now); @@ -2126,7 +2193,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReportSlotInvalidation(invalidation_cause, true, active_pid, slotname, restart_lsn, oldestLSN, snapshotConflictHorizon, - slot_idle_secs); + slot_idle_secs, s->data.xmin, + s->data.catalog_xmin, xidLimit); if (MyBackendType == B_STARTUP) (void) SignalRecoveryConflict(GetPGProcByNumber(active_proc), @@ -2179,7 +2247,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReportSlotInvalidation(invalidation_cause, false, active_pid, slotname, restart_lsn, oldestLSN, snapshotConflictHorizon, - slot_idle_secs); + slot_idle_secs, s->data.xmin, + s->data.catalog_xmin, xidLimit); /* done with this slot for now */ break; @@ -2206,6 +2275,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, * logical. * - RS_INVAL_IDLE_TIMEOUT: has been idle longer than the configured * "idle_replication_slot_timeout" duration. + * - RS_INVAL_XID_AGE: slot xid age is older than the configured + * "max_slot_xid_age" age. * * Note: This function attempts to invalidate the slot for multiple possible * causes in a single pass, minimizing redundant iterations. The "cause" @@ -2219,7 +2290,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon) + TransactionId snapshotConflictHorizon, + TransactionId xidLimit) { XLogRecPtr oldestLSN; bool invalidated = false; @@ -2258,7 +2330,7 @@ restart: if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, - &released_lock)) + xidLimit, &released_lock)) { Assert(released_lock); diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 9299bcebbda..a352fd7ff3a 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -1929,6 +1929,31 @@ GlobalVisHorizonKindForRel(Relation rel) return VISHORIZON_TEMP; } +/* + * A helper function to return the appropriate oldest non-removable + * TransactionId from the pre-computed horizons, based on the relation + * type. + */ +static pg_attribute_always_inline TransactionId +GetOldestNonRemovableTransactionIdFromHorizons(ComputeXidHorizonsResult *horizons, + Relation rel) +{ + switch (GlobalVisHorizonKindForRel(rel)) + { + case VISHORIZON_SHARED: + return horizons->shared_oldest_nonremovable; + case VISHORIZON_CATALOG: + return horizons->catalog_oldest_nonremovable; + case VISHORIZON_DATA: + return horizons->data_oldest_nonremovable; + case VISHORIZON_TEMP: + return horizons->temp_oldest_nonremovable; + } + + /* just to prevent compiler warnings */ + return InvalidTransactionId; +} + /* * Return the oldest XID for which deleted tuples must be preserved in the * passed table. @@ -1947,20 +1972,30 @@ GetOldestNonRemovableTransactionId(Relation rel) ComputeXidHorizons(&horizons); - switch (GlobalVisHorizonKindForRel(rel)) - { - case VISHORIZON_SHARED: - return horizons.shared_oldest_nonremovable; - case VISHORIZON_CATALOG: - return horizons.catalog_oldest_nonremovable; - case VISHORIZON_DATA: - return horizons.data_oldest_nonremovable; - case VISHORIZON_TEMP: - return horizons.temp_oldest_nonremovable; - } + return GetOldestNonRemovableTransactionIdFromHorizons(&horizons, rel); +} - /* just to prevent compiler warnings */ - return InvalidTransactionId; +/* + * Same as GetOldestNonRemovableTransactionId(), but also returns the + * replication slot xmin and catalog_xmin from the same ComputeXidHorizons() + * call. This avoids a separate ProcArrayLock acquisition when the caller + * needs both values. + */ +TransactionId +GetOldestNonRemovableTransactionIdWithSlotXids(Relation rel, + TransactionId *slot_xmin, + TransactionId *slot_catalog_xmin) +{ + ComputeXidHorizonsResult horizons; + + ComputeXidHorizons(&horizons); + + if (slot_xmin) + *slot_xmin = horizons.slot_xmin; + if (slot_catalog_xmin) + *slot_catalog_xmin = horizons.slot_catalog_xmin; + + return GetOldestNonRemovableTransactionIdFromHorizons(&horizons, rel); } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 29af7733948..d54d6cc7544 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -504,7 +504,8 @@ ResolveRecoveryConflictWithSnapshot(TransactionId snapshotConflictHorizon, */ if (IsLogicalDecodingEnabled() && isCatalogRel) InvalidateObsoleteReplicationSlots(RS_INVAL_HORIZON, 0, locator.dbOid, - snapshotConflictHorizon); + snapshotConflictHorizon, + InvalidTransactionId); } /* diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 83af594d4af..4738c84c6c5 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -2132,6 +2132,14 @@ max => 'MAX_KILOBYTES', }, +{ name => 'max_slot_xid_age', type => 'int', context => 'PGC_SIGHUP', group => 'REPLICATION_SENDING', + short_desc => 'Age of the transaction ID at which a replication slot gets invalidated.', + variable => 'max_slot_xid_age', + boot_val => '0', + min => '0', + max => '2100000000', +}, + # We use the hopefully-safely-small value of 100kB as the compiled-in # default for max_stack_depth. InitializeGUCOptions will increase it # if possible, depending on the actual platform-specific stack limit. diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index ac38cddaaf9..358b6edc9f1 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -361,6 +361,8 @@ #wal_keep_size = 0 # in megabytes; 0 disables #max_slot_wal_keep_size = -1 # in megabytes; -1 disables #idle_replication_slot_timeout = 0 # in seconds; 0 disables +#max_slot_xid_age = 0 # maximum XID age before a replication slot + # gets invalidated; 0 disables #wal_sender_timeout = 60s # in milliseconds; 0 disables #wal_sender_shutdown_timeout = -1 # in milliseconds; -1 disables #track_commit_timestamp = off # collect timestamp of transaction commit diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c index 15e06e5686e..f75dc79fc03 100644 --- a/src/bin/pg_basebackup/pg_createsubscriber.c +++ b/src/bin/pg_basebackup/pg_createsubscriber.c @@ -1678,7 +1678,7 @@ start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_ appendPQExpBufferStr(pg_ctl_cmd, " -s -o \"-c sync_replication_slots=off\""); /* Prevent unintended slot invalidation */ - appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c idle_replication_slot_timeout=0\""); + appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c idle_replication_slot_timeout=0 -c max_slot_xid_age=0\""); if (restricted_access) { diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h index 956d9cea36d..7d81e3d1906 100644 --- a/src/include/commands/vacuum.h +++ b/src/include/commands/vacuum.h @@ -287,6 +287,13 @@ struct VacuumCutoffs */ TransactionId FreezeLimit; MultiXactId MultiXactCutoff; + + /* + * Oldest xmin and catalog xmin of any replication slot obtained from the + * same ComputeXidHorizons() call that computed OldestXmin. + */ + TransactionId OldestSlotXmin; + TransactionId OldestSlotCatalogXmin; }; /* @@ -399,6 +406,9 @@ extern IndexBulkDeleteResult *vac_bulkdel_one_index(IndexVacuumInfo *ivinfo, VacDeadItemsInfo *dead_items_info); extern IndexBulkDeleteResult *vac_cleanup_one_index(IndexVacuumInfo *ivinfo, IndexBulkDeleteResult *istat); +extern bool maybe_invalidate_xid_aged_slots(TransactionId oldest_xmin, + TransactionId oldest_slot_xmin, + TransactionId oldest_slot_catalog_xmin); /* In postmaster/autovacuum.c */ extern void AutoVacuumUpdateCostLimit(void); diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 77c8d0975b6..cad1d89b05b 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -66,10 +66,12 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_WAL_LEVEL = (1 << 2), /* idle slot timeout has occurred */ RS_INVAL_IDLE_TIMEOUT = (1 << 3), + /* slot's xmin or catalog_xmin has reached max xid age */ + RS_INVAL_XID_AGE = (1 << 4), } ReplicationSlotInvalidationCause; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES 4 +#define RS_INVAL_MAX_CAUSES 5 /* * When the slot synchronization worker is running, or when @@ -327,6 +329,7 @@ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT int max_repack_replication_slots; extern PGDLLIMPORT char *synchronized_standby_slots; extern PGDLLIMPORT int idle_replication_slot_timeout_secs; +extern PGDLLIMPORT int max_slot_xid_age; /* management of individual slots */ extern void ReplicationSlotCreate(const char *name, bool db_specific, @@ -364,7 +367,8 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid); extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon); + TransactionId snapshotConflictHorizon, + TransactionId xidLimit); extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock); extern int ReplicationSlotIndex(ReplicationSlot *slot); extern bool ReplicationSlotName(int index, Name name); diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index ec89c448220..b1dee4ad889 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -51,6 +51,9 @@ extern RunningTransactions GetRunningTransactionData(Oid dbid); extern bool TransactionIdIsInProgress(TransactionId xid); extern TransactionId GetOldestNonRemovableTransactionId(Relation rel); +extern TransactionId GetOldestNonRemovableTransactionIdWithSlotXids(Relation rel, + TransactionId *slot_xmin, + TransactionId *slot_catalog_xmin); extern TransactionId GetOldestTransactionIdConsideredRunning(void); extern TransactionId GetOldestActiveTransactionId(bool inCommitOnly, bool allDbs); diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl index 7b253e64d9c..6b7e4818bf4 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -540,4 +540,179 @@ is( $publisher4->safe_psql( $publisher4->stop; $subscriber4->stop; +# Wait for the given slot to be invalidated with reason 'xid_aged' +sub wait_for_xid_aged_invalidation +{ + my ($node, $slot_name) = @_; + $node->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = '$slot_name' AND + active = false AND + invalidation_reason = 'xid_aged'; + ]) or die "Timed out waiting for slot $slot_name to be invalidated"; +} + +# ===================================================================== +# Testcase start: Invalidate physical slot due to max_slot_xid_age GUC + +# Initialize primary node for XID age tests +my $primary5 = PostgreSQL::Test::Cluster->new('primary5'); +$primary5->init(allows_streaming => 'logical'); + +# Disable autovacuum so checkpointer triggers the invalidation +my $max_slot_xid_age = 100; +$primary5->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $max_slot_xid_age +autovacuum = off +}); + +$primary5->start; + +# Create a procedure to consume XIDs +$primary5->safe_psql( + 'postgres', qq{ + CREATE PROCEDURE consume_xid(cnt int) + AS \$\$ + DECLARE + i int; + BEGIN + FOR i IN 1..cnt LOOP + EXECUTE 'SELECT pg_current_xact_id()'; + COMMIT; + END LOOP; + END; + \$\$ LANGUAGE plpgsql; +}); + +# Take a backup for creating standby +$backup_name = 'backup5'; +$primary5->backup($backup_name); + +# Create standby with HS feedback so the slot gains an xmin +my $standby5 = PostgreSQL::Test::Cluster->new('standby5'); +$standby5->init_from_backup($primary5, $backup_name, has_streaming => 1); +$standby5->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb5_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$primary5->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot(slot_name := 'sb5_slot', immediately_reserve := true); +]); +$standby5->start; + +# Create some content on primary to move xmin +$primary5->safe_psql('postgres', + "CREATE TABLE tab_int5 AS SELECT generate_series(1,10) AS a"); +$primary5->wait_for_catchup($standby5); + +# Wait for the physical slot to get xmin via hot_standby_feedback +$primary5->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'sb5_slot'; +]) or die "Timed out waiting for slot sb5_slot xmin from HS feedback"; + +# Stop standby so the slot becomes inactive with its xmin frozen +$standby5->stop; + +# Advance XIDs past 2x max_slot_xid_age so the slot's xmin is stale enough +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $max_slot_xid_age)}); +$primary5->safe_psql('postgres', "CHECKPOINT"); +wait_for_xid_aged_invalidation($primary5, 'sb5_slot'); +ok(1, "physical slot invalidated due to XID age (via checkpoint)"); + +# Testcase end: Invalidate physical slot due to max_slot_xid_age GUC +# =================================================================== + +# ==================================================================== +# Testcase start: Invalidate logical slot due to max_slot_xid_age GUC + +# Create a logical slot directly on the primary (no subscriber needed). +# The slot gets a catalog_xmin immediately upon creation. +$primary5->safe_psql('postgres', + "SELECT pg_create_logical_replication_slot('lsub5_slot', 'pgoutput')"); + +$primary5->poll_query_until( + 'postgres', qq[ + SELECT catalog_xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'lsub5_slot'; +]) or die "Timed out waiting for slot lsub5_slot catalog_xmin"; + +# Advance XIDs past 2x max_slot_xid_age so the slot's catalog_xmin is stale enough +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $max_slot_xid_age)}); + +# Vacuum a user table so OldestXmin does not include the slot's catalog_xmin, +# skipping the invalidation of the slot. +$primary5->safe_psql('postgres', "VACUUM tab_int5"); +is( $primary5->safe_psql( + 'postgres', + qq[SELECT invalidation_reason IS NULL FROM pg_replication_slots WHERE slot_name = 'lsub5_slot';] + ), + 't', + 'logical slot not invalidated after vacuuming a data table'); + +# Vacuum a catalog table so OldestXmin includes the slot's catalog_xmin, +# triggering invalidation of the slot. +$primary5->safe_psql('postgres', "VACUUM pg_class"); +wait_for_xid_aged_invalidation($primary5, 'lsub5_slot'); +ok(1, "logical slot invalidated due to XID age (via vacuum)"); + +# Testcase end: Invalidate logical slot due to max_slot_xid_age GUC +# ================================================================== + +# =============================================================================== +# Testcase start: Invalidate logical slot on standby due to max_slot_xid_age GUC + +# Disable max_slot_xid_age on primary and recreate the streaming slot +$primary5->safe_psql( + 'postgres', + q{ +ALTER SYSTEM SET max_slot_xid_age = 0; +SELECT pg_reload_conf(); +}); +$primary5->safe_psql('postgres', + "SELECT pg_drop_replication_slot('sb5_slot')"); +$primary5->safe_psql('postgres', + "SELECT pg_create_physical_replication_slot('sb5_slot', true)"); +$standby5->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $max_slot_xid_age +autovacuum = off +}); +$standby5->start; + +$primary5->wait_for_catchup($standby5); + +$standby5->create_logical_slot_on_standby($primary5, 'sb5_logical_slot', + 'postgres'); + +$standby5->poll_query_until( + 'postgres', qq[ + SELECT catalog_xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'sb5_logical_slot'; +]) or die "Timed out waiting for sb5_logical_slot catalog_xmin"; + +# Advance XIDs on primary, replay on standby, then restartpoint to invalidate +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $max_slot_xid_age)}); +$primary5->safe_psql('postgres', "CHECKPOINT"); +$primary5->wait_for_replay_catchup($standby5); +$standby5->safe_psql('postgres', "CHECKPOINT"); + +wait_for_xid_aged_invalidation($standby5, 'sb5_logical_slot'); +ok(1, "logical (standby) slot invalidated due to XID age (via restartpoint)"); + +$standby5->stop; +$primary5->stop; + +# Testcase end: Invalidate logical slot on standby due to max_slot_xid_age GUC +# ============================================================================= + done_testing(); -- 2.47.3
From 60084a4d54284c4a67d548830900a53ddd10450b Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 17 Apr 2026 18:31:10 +0000 Subject: [PATCH v11 2/3] Add more tests for XID age slot invalidation Consume XIDs up to wraparound WARNING limits with max_slot_xid_age matching vacuum_failsafe_age (1.6B). Verify that autovacuum invalidates the inactive replication slot (XID-age-based invalidation), unblocks datfrozenxid advancement, and prevents wraparound without any intervention. --- src/test/recovery/Makefile | 3 +- src/test/recovery/t/019_replslot_limit.pl | 130 ++++++++++++++++++++++ 2 files changed, 132 insertions(+), 1 deletion(-) diff --git a/src/test/recovery/Makefile b/src/test/recovery/Makefile index d41aaaf8ae1..5c3d2c89941 100644 --- a/src/test/recovery/Makefile +++ b/src/test/recovery/Makefile @@ -12,7 +12,8 @@ EXTRA_INSTALL=contrib/pg_prewarm \ contrib/pg_stat_statements \ contrib/test_decoding \ - src/test/modules/injection_points + src/test/modules/injection_points \ + src/test/modules/xid_wraparound subdir = src/test/recovery top_builddir = ../../.. diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl index 6b7e4818bf4..6f92034f3d5 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -715,4 +715,134 @@ $primary5->stop; # Testcase end: Invalidate logical slot on standby due to max_slot_xid_age GUC # ============================================================================= +# ================================================================================= +# Testcase start: XID-age-based slot invalidation with autovacuum (production-like) + +# Standby sets slot xmin via HS feedback, disconnects, XIDs are consumed. +# max_slot_xid_age is set to vacuum_failsafe_age (1.6B) so autovacuum +# invalidates the slot before entering failsafe mode, unblocking +# datfrozenxid advancement and avoiding XID wraparound without manual +# VACUUM or downtime. + +# Verify server log shows slot invalidation by autovacuum worker +sub verify_slot_xid_aged_invalidation_in_server_log +{ + my ($node, $slot_name, $max_age, $consumed_xids) = @_; + + my $log = slurp_file($node->logfile); + + # Verify the invalidation was performed by an autovacuum worker + like($log, + qr/autovacuum worker\[\d+\] LOG:\s+invalidating obsolete replication slot "$slot_name"/, + "server log: $slot_name invalidated by autovacuum worker"); + + # Verify DETAIL shows the xmin age exceeding max_slot_xid_age + like($log, + qr/autovacuum worker\[\d+\] DETAIL:\s+The slot's (?:catalog )?xmin age of (\d+) exceeds the configured "max_slot_xid_age" of $max_age by (\d+) transactions/, + "server log: DETAIL shows xmin age exceeds max_slot_xid_age $max_age"); + + # Extract xid age from the log and report for diagnostics + $log =~ + /The slot's (?:catalog )?xmin age of (\d+) exceeds the configured "max_slot_xid_age" of $max_age by (\d+)/; + my $log_xid_age = $1 // 'N/A'; + my $exceeded_by = $2 // 'N/A'; + diag "xid_age from server log=$log_xid_age, exceeded_by=$exceeded_by, max_slot_xid_age=$max_age, consumed=$consumed_xids XIDs"; +} + +# Verify slot invalidation and wait for autovacuum to advance datfrozenxid +sub verify_invalidation_and_recovery +{ + my ($node, $slot_name, $max_age, $consumed_xids) = @_; + + return if $max_age == 0; + + wait_for_xid_aged_invalidation($node, $slot_name); + ok(1, 'autovacuum invalidated slot due to xid_aged'); + + verify_slot_xid_aged_invalidation_in_server_log($node, $slot_name, + $max_age, $consumed_xids); + + # Wait for autovacuum to advance datfrozenxid in all databases past the + # wraparound threshold. + $node->poll_query_until( + 'postgres', qq[ + SELECT NOT EXISTS ( + SELECT 1 FROM pg_database + WHERE age(datfrozenxid) > 2000000000 + ); + ]) or die "Timed out waiting for autovacuum to advance datfrozenxid in all databases"; +} + +my $primary6 = PostgreSQL::Test::Cluster->new('primary6'); +$primary6->init(allows_streaming => 'logical'); + +$max_slot_xid_age = 1600000000; # matches vacuum_failsafe_age default +$primary6->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $max_slot_xid_age +autovacuum_naptime = 1s +}); + +$primary6->start; +$primary6->safe_psql('postgres', "CREATE EXTENSION xid_wraparound"); + +$backup_name = 'backup6'; +$primary6->backup($backup_name); + +my $standby6 = PostgreSQL::Test::Cluster->new('standby6'); +$standby6->init_from_backup($primary6, $backup_name, has_streaming => 1); +$standby6->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb6_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); + +$primary6->safe_psql('postgres', + "SELECT pg_create_physical_replication_slot('sb6_slot', true)"); + +$standby6->start; + +$primary6->safe_psql('postgres', + "CREATE TABLE tab_int6 AS SELECT generate_series(1,10) AS a"); +$primary6->wait_for_catchup($standby6); + +$primary6->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL FROM pg_replication_slots + WHERE slot_name = 'sb6_slot'; +]) or die "Timed out waiting for sb6_slot xmin from HS feedback"; + +# Stop standby; slot xmin persists and holds back datfrozenxid +$standby6->stop; + +# Consume XIDs in 50M chunks; autovacuum (naptime=1s) will invalidate the +# slot once xmin age exceeds max_slot_xid_age. +my $logstart6 = -s $primary6->logfile; +my $chunk = 50_000_000; +my $max_xids = 2_200_000_000; +my $consumed = 0; + +while ($consumed < $max_xids) +{ + $primary6->safe_psql('postgres', "SELECT consume_xids($chunk)"); + $consumed += $chunk; + my $remaining = $max_xids - $consumed; + diag "consumed $consumed / $max_xids XIDs ($remaining remaining)"; +} + +verify_invalidation_and_recovery($primary6, 'sb6_slot', + $max_slot_xid_age, $consumed); + +# Consume 1B more XIDs — combining with the 2.2B consumed above, the total +# of 3.2B exceeds the 2^31 (~2.1B) usable XID space (xidStopLimit), i.e. +# more than one full wraparound cycle, proving the system is healthy. +$primary6->safe_psql('postgres', "SELECT consume_xids(1000000000)"); +ok(1, 'writes succeed after autovacuum invalidated the slot'); + +$primary6->stop; + +# Testcase end: XID-age-based slot invalidation with autovacuum (production-like) +# ================================================================================ + done_testing(); -- 2.47.3
From 7e17543594722054a9310a15da9b250d476e436e Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 17 Apr 2026 18:31:35 +0000 Subject: [PATCH v11 3/3] Avoid concurrent XID-age slot invalidation attempts Multiple processes (autovacuum workers, backends running VACUUM) can concurrently attempt to invalidate the same replication slot due to XID age, causing them to wait on the same ConditionVariable while waiting for a slow walsender to release the slot. Add an invalidating_proc field to ReplicationSlot to track which process is currently attempting invalidation. When a process sees that another live process is already working on a slot, it skips the slot and defers to a subsequent cycle. This prevents unnecessary blocking during XID-age based slot invalidation. Added an injection point in the walsender to mimic slow SIGTERM processing and a TAP test for the concurrent slot invalidation. --- src/backend/replication/slot.c | 75 ++++++++++ src/backend/tcop/postgres.c | 11 ++ src/include/replication/slot.h | 7 + src/test/recovery/t/019_replslot_limit.pl | 161 ++++++++++++++++++++++ 4 files changed, 254 insertions(+) diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index fe43f5c8820..7cef325a888 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -233,6 +233,7 @@ ReplicationSlotsShmemInit(void *arg) /* everything else is zeroed by the memset above */ slot->active_proc = INVALID_PROC_NUMBER; + slot->invalidating_proc = INVALID_PROC_NUMBER; SpinLockInit(&slot->mutex); LWLockInitialize(&slot->io_in_progress_lock, LWTRANCHE_REPLICATION_SLOT_IO); @@ -501,6 +502,7 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->last_saved_restart_lsn = InvalidXLogRecPtr; slot->inactive_since = 0; slot->slotsync_skip_reason = SS_SKIP_NONE; + slot->invalidating_proc = INVALID_PROC_NUMBER; /* * Create the slot on disk. We haven't actually marked the slot allocated @@ -2052,6 +2054,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, int last_signaled_pid = 0; bool released_lock = false; bool invalidated = false; + bool am_invalidating = false; TimestampTz inactive_since = 0; for (;;) @@ -2112,6 +2115,59 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, break; } + /* + * Skip XID-age invalidation if another process is already + * invalidating this slot. + * + * Check if another process is already trying to invalidate this slot. + * If so, skip it to avoid multiple processes blocking on the same CV + * sleep. The first process will complete the invalidation attempt; + * others defer to a subsequent cycle. + * + * We handle this only for XID-age invalidation because multiple + * processes (autovacuum workers, backends running VACUUM, the + * checkpointer) can attempt it concurrently, making it likely that + * several end up blocking on the same ConditionVariable while waiting + * for a slow walsender to release the slot. Invalidation due to other + * causes can also involve multiple processes (e.g., on a standby, the + * checkpointer and the startup process may attempt to invalidate a + * slot for RS_INVAL_WAL_LEVEL and RS_INVAL_HORIZON respectively), but + * such concurrent attempts are rare in practice. + */ + if (invalidation_cause == RS_INVAL_XID_AGE && + s->invalidating_proc != INVALID_PROC_NUMBER) + { + int invalidating_pid; + + invalidating_pid = GetPGProcByNumber(s->invalidating_proc)->pid; + + if (invalidating_pid != 0 && + s->invalidating_proc != MyProcNumber) + { + /* Another live process is already invalidating this slot */ + SpinLockRelease(&s->mutex); + if (released_lock) + LWLockRelease(ReplicationSlotControlLock); + break; + } + + /* + * The previously recorded process has exited (pid == 0) or it's + * us. Reset and proceed with invalidation. + */ + s->invalidating_proc = INVALID_PROC_NUMBER; + } + + /* + * If the slot is active (we'll need to signal and wait), record + * ourselves as the invalidating process. + */ + if (s->active_proc != INVALID_PROC_NUMBER) + { + s->invalidating_proc = MyProcNumber; + am_invalidating = true; + } + slotname = s->data.name; active_proc = s->active_proc; @@ -2131,6 +2187,12 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, s->active_proc = MyProcNumber; s->data.invalidated = invalidation_cause; + /* + * Clear the invalidating process since we have completed the + * invalidation (acquired the slot and marked it invalid). + */ + s->invalidating_proc = INVALID_PROC_NUMBER; + /* * XXX: We should consider not overwriting restart_lsn and instead * just rely on .invalidated. @@ -2255,6 +2317,19 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, } } + /* + * If we set invalidating_proc but exited without completing the + * invalidation (e.g., the slot caught up while we were waiting, or the + * slot was dropped), clear it so other processes don't skip this slot. + */ + if (am_invalidating && !invalidated) + { + SpinLockAcquire(&s->mutex); + if (s->invalidating_proc == MyProcNumber) + s->invalidating_proc = INVALID_PROC_NUMBER; + SpinLockRelease(&s->mutex); + } + Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock)); *released_lock_out = released_lock; diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 2c1f14b7889..846ee613243 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -3375,6 +3375,17 @@ ProcessInterrupts(void) ProcDieSenderUid = 0; QueryCancelPending = false; /* ProcDie trumps QueryCancel */ LockErrorCleanup(); + +#ifdef USE_INJECTION_POINTS + /* + * Injection point used to simulate a walsender that is slow to + * respond to SIGTERM, allowing tests to verify concurrent slot + * invalidation behavior. + */ + if (am_walsender) + INJECTION_POINT("walsender-before-sigterm-exit", NULL); +#endif + /* As in quickdie, don't risk sending to client during auth */ if (ClientAuthInProgress && whereToSendOutput == DestRemote) whereToSendOutput = DestNone; diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index cad1d89b05b..fb085447d23 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -284,6 +284,13 @@ typedef struct ReplicationSlot * slotsync_skip_reason provides no practical benefit. */ SlotSyncSkipReason slotsync_skip_reason; + + /* + * Process currently attempting to invalidate this slot. + * INVALID_PROC_NUMBER means no invalidation is in progress. Protected by + * the slot's mutex. + */ + ProcNumber invalidating_proc; } ReplicationSlot; #define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid) diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl index 6f92034f3d5..0897f4388d7 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -845,4 +845,165 @@ $primary6->stop; # Testcase end: XID-age-based slot invalidation with autovacuum (production-like) # ================================================================================ +# =============================================================================== +# Testcase start: Concurrent slot invalidation due to max_slot_xid_age GUC +# +# Two concurrent VACUUMs both try to invalidate the same active logical slot. +# An injection point delays the walsender's SIGTERM processing so that vacuum1 +# blocks on the CV waiting for the slot to be released. When vacuum2 runs, it +# sees that vacuum1 is already invalidating the same slot and skips without +# blocking. After the walsender is woken, vacuum1 completes the invalidation. + +# Skip if injection points are not available. +if ($ENV{enable_injection_points} ne 'yes') +{ + done_testing(); + exit; +} + +my $primary7 = PostgreSQL::Test::Cluster->new('primary7'); +$primary7->init(allows_streaming => 'logical'); + +my $max_slot_xid_age7 = 100; +$primary7->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $max_slot_xid_age7 +autovacuum = off +shared_preload_libraries = 'injection_points' +}); + +$primary7->start; + +# Check if injection_points extension is available. +if (!$primary7->check_extension('injection_points')) +{ + $primary7->stop; + done_testing(); + exit; +} + +$primary7->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + +# Helper to consume XIDs. +$primary7->safe_psql( + 'postgres', qq{ + CREATE PROCEDURE consume_xid(cnt int) + AS \$\$ + DECLARE + i int; + BEGIN + FOR i IN 1..cnt LOOP + EXECUTE 'SELECT pg_current_xact_id()'; + COMMIT; + END LOOP; + END; + \$\$ LANGUAGE plpgsql; +}); + + +# Create a logical slot (gets catalog_xmin immediately). +$primary7->safe_psql('postgres', + "SELECT pg_create_logical_replication_slot('lslot7', 'test_decoding')"); + +# Hold the slot active via pg_recvlogical. +my $pg_recvlog_stdout7 = ''; +my $pg_recvlog_stderr7 = ''; +my $connstr7 = $primary7->connstr('postgres'); +my $pg_recvlogical_handle7 = IPC::Run::start( + [ + 'pg_recvlogical', '-d', $connstr7, + '--slot', 'lslot7', '--start', + '-f', '/dev/null', '--no-loop' + ], + '>', \$pg_recvlog_stdout7, + '2>', \$pg_recvlog_stderr7); + +# Wait for the slot to become active. +$primary7->poll_query_until( + 'postgres', qq[ + SELECT active FROM pg_replication_slots WHERE slot_name = 'lslot7'; +]) or die "Timed out waiting for slot lslot7 to become active"; + +# Make the walsender block before processing SIGTERM. +$primary7->safe_psql('postgres', + "SELECT injection_points_attach('walsender-before-sigterm-exit', 'wait')"); + +# Make the slot's catalog_xmin stale. +$primary7->safe_psql('postgres', + qq{CALL consume_xid(2 * $max_slot_xid_age7)}); + +# Launch vacuum1 on a catalog table: the logical slot holds catalog_xmin, +# so only catalog VACUUMs see it as OldestXmin and trigger invalidation. +# vacuum1 will SIGTERM the walsender, then block on the CV. +my $vacuum1 = $primary7->background_psql('postgres'); +my $vacuum2 = $primary7->background_psql('postgres'); + +$vacuum1->query_until( + qr/starting_vacuum/, + q(\echo starting_vacuum +VACUUM pg_class; +\echo vacuum1_done +)); + +# Wait for the walsender to hit the injection point. +$primary7->wait_for_event('walsender', 'walsender-before-sigterm-exit'); + +# Verify vacuum1 is blocked on the CV. +$primary7->poll_query_until( + 'postgres', qq[ + SELECT count(*) = 1 FROM pg_stat_activity + WHERE wait_event = 'ReplicationSlotDrop' + AND backend_type = 'client backend'; +]) or die "Timed out waiting for vacuum1 to block on slot CV"; + +# Launch vacuum2 on a different catalog table: it also computes the catalog +# horizon and sees the slot needs invalidation, but finds vacuum1 is already +# invalidating the same slot (via invalidating_proc) and skips. +$vacuum2->query_until( + qr/starting_vacuum/, + q(\echo starting_vacuum +VACUUM pg_type; +\echo vacuum2_done +)); + +# vacuum2 completes without blocking. +$vacuum2->query_until(qr/vacuum2_done/, ''); + +# Verify only vacuum1 is waiting on ReplicationSlotDrop. +$result = $primary7->safe_psql( + 'postgres', qq[ + SELECT count(*) FROM pg_stat_activity + WHERE wait_event = 'ReplicationSlotDrop' + AND backend_type = 'client backend'; +]); +is($result, '1', + 'only vacuum1 blocks on CV; vacuum2 skips via invalidating_proc'); + +# Wake up the walsender so it can exit and release the slot. +$primary7->safe_psql('postgres', + "SELECT injection_points_wakeup('walsender-before-sigterm-exit')"); + +# Detach the injection point. +$primary7->safe_psql('postgres', + "SELECT injection_points_detach('walsender-before-sigterm-exit')"); + +# Wait for pg_recvlogical to exit. +$pg_recvlogical_handle7->finish; + +# Wait for vacuum1 to complete now that the walsender has released the slot. +$vacuum1->query_until(qr/vacuum1_done/, ''); + +# Verify the slot was invalidated. +wait_for_xid_aged_invalidation($primary7, 'lslot7'); +ok(1, + "concurrent VACUUM: vacuum1 blocks on CV, vacuum2 skips via invalidating_proc" +); + +$vacuum1->quit; +$vacuum2->quit; +$primary7->stop; + +# Testcase end: Concurrent slot invalidation due to max_slot_xid_age GUC +# =============================================================================== + done_testing(); -- 2.47.3
