Hi, On Thu, Jul 9, 2026 at 2:53 PM Masahiko Sawada <[email protected]> wrote: > > > I partially agree with your suggestion for case 2. My preference is to > > not add any blockers for vacuum. It does opportunistic XID-age > > invalidation, invalidates the slots it can take without waiting, and > > leaves the held ones to the checkpointer, which is the guaranteed > > path. It is also easier to reason about and to explain to users. > > > The tradeoff is that vacuum won't invalidate a held slot in the same > > pass, so the relation being vacuumed right then doesn't get the > > advanced horizon. Once the checkpointer has invalidated the slot, > > later relations pick it up, which may well be in the same vacuum cycle > > or next. I am fine with that. > > > I would skip having the vacuum explicitly wake the checkpointer when > > it sees a held slot. Imagine one walsender holding the slot's xmin and > > 8 autovacuum workers running. Each worker hits that same held slot and > > would signal the checkpointer, so we would get a signal per worker for > > a single slot, repeated every cycle until the checkpointer acts. If > > the checkpointer is already running, another request gets queued, and > > this can happen repeatedly, creating a flood of checkpoint requests. > > I agree that we should avoid requesting checkpoints just to invalidate > XID-aged slots. > > If we rely on the checkpointer for held slot cases, XID-aged slots > held by someone could be left for up to one day, the maximum value of > checkpoint_timeout, in the worst case. > > Given that it's not common for XID-aged slots to > be still held by someone, it would be okay to skip the held slots. The > same applies on replicas, where the restartpoint pass is the only > mechanism anway. So I agree to simply skip held slots.
Thank you! Please find the attached v12 patch with this change. I dropped the 0002 patch that was pushing the tests to reach a production-like XID wraparound since it takes a bit of time to run such tests and I understand that our test infrastructure and resources are not free. I folded the tests into simpler ones in 0001 itself. Tests now cover vacuum, autovacuum invalidating unheld slots, skipping held slots, checkpoint and restartpoint doing the invalidation, and both logical and physical replication slots. I believe this gives good coverage for the feature. -- Bharath Rupireddy Amazon Web Services: https://aws.amazon.com
From 0ee0f5f66b925e651be28f1cc2312f60b68b9f9d Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sun, 12 Jul 2026 19:16:29 +0000 Subject: [PATCH v12] 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. Invalidating such a slot removes one of the blockers of vacuum's oldest xmin, letting vacuum freeze XIDs to avoid transaction ID wraparound and prune dead rows to reduce bloat. The invalidation check occurs during both VACUUM (manual and autovacuum) and checkpoints. During vacuum, the check is performed on a per-relation basis. The XID-age based slot invalidation is considered only when slots' XIDs are holding back the oldest xmin for the current relation. That is, we invalidate slots only when doing so can advance the vacuum cutoff and allow dead tuple reclamation. Checking slot XIDs per relation would mean taking the proc array lock again for each relation to read the slot xmin values (xmin and catalog_xmin in the proc array). To avoid that, the vacuum cutoff calculation has been extended to return the oldest slot xmin and catalog_xmin alongside the global oldest xmin, both obtained from the same horizon computation. The per-relation invalidation check then needs no extra lock. Vacuum performs this invalidation without ever blocking. It invalidates only the aged slots it can acquire immediately and leaves any slot that is currently held by a live process alone; those are cleaned up by the checkpointer, which does the guaranteed terminate-and-wait invalidation on its regular pass. This keeps vacuum easy to reason about, and it avoids many autovacuum workers and backends piling onto a single slot's condition variable waiting for a slow walsender to release it. A slot that vacuum skips this way is invalidated by the checkpointer, and later relations (possibly in the same vacuum cycle) then pick up the advanced horizon. The slot invalidation path takes a flag telling it whether to wait: vacuum asks it not to wait, while the checkpointer and other callers let it terminate the owner and wait. On standby servers, where vacuum does not run, checkpoints (as restartpoints) are what invalidate aged slots. Note that slots currently being synced from the primary (i.e., synced = true) are exempt from this invalidation mechanism. Author: Bharath Rupireddy <[email protected]> Reviewed-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 | 64 ++++++ doc/src/sgml/logical-replication.sgml | 4 +- doc/src/sgml/system-views.sgml | 8 + src/backend/access/heap/vacuumlazy.c | 15 ++ src/backend/access/transam/xlog.c | 36 +++- src/backend/commands/vacuum.c | 77 ++++++- src/backend/replication/slot.c | 95 +++++++- 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 | 9 +- src/include/storage/procarray.h | 3 + src/test/recovery/t/019_replslot_limit.pl | 204 ++++++++++++++++++ 16 files changed, 572 insertions(+), 29 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 0848c18d329..44c4e8e5eb8 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4950,6 +4950,70 @@ 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. + A value of zero (the default) disables this feature. Users can set + this value anywhere from zero to 2.1 billion transactions. This parameter + can only be set in the <filename>postgresql.conf</filename> file or on + the server command line. + </para> + + <para> + Slot invalidation due to this limit occurs during vacuum (both the + <command>VACUUM</command> command and autovacuum) and during checkpoint. + During vacuum, only slots that are not currently in use by a replication + connection are invalidated; a slot that is still held by a connection is + left for the next checkpoint, which invalidates it even while it is in + use. Because vacuum and checkpoints happen at their own intervals, there + can be some lag between when a slot's <literal>xmin</literal> or + <literal>catalog_xmin</literal> age exceeds + <varname>max_slot_xid_age</varname> and when the slot invalidation is + actually triggered. To avoid such lags, users can force a checkpoint to + promptly invalidate the slot. + </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 vacuum back from freezing + XIDs and from pruning dead rows. This can lead to table and index bloat + that holds disk space that vacuum would otherwise reclaim, and + eventually to transaction ID wraparound. Invalidating such a slot + removes one of these blockers, letting vacuum freeze XIDs and reclaim + disk space again. See <xref linkend="routine-vacuuming"/> for more + details. Set <varname>max_slot_xid_age</varname> below + <xref linkend="guc-vacuum-failsafe-age"/>, so an aged slot 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 690598bff98..2065e72e0fb 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2674,7 +2674,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 5ea19d68622..a237cface0c 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -3103,6 +3103,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..0d06025456c 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -800,6 +800,21 @@ 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 */ + 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 b431a921e4b..263489e92de 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7410,6 +7410,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, false)) { /* * 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, false)) { /* * Some slots have been invalidated; recalculate the old-segment @@ -9216,7 +9241,8 @@ xlog_redo(XLogReaderState *record) */ InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, 0, InvalidOid, - InvalidTransactionId); + InvalidTransactionId, + InvalidTransactionId, false); } else if (sync_replication_slots) { diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index 38539a6fd3d..5438e282aad 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" @@ -1139,7 +1140,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 = + GetOldestNonRemovableTransactionIdAndSlotXmins(rel, + &cutoffs->OldestSlotXmin, + &cutoffs->OldestSlotCatalogXmin); Assert(TransactionIdIsNormal(cutoffs->OldestXmin)); @@ -2713,3 +2717,74 @@ 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 + * xmin and catalog_xmin, typically all computed together by a single + * ComputeXidHorizons() call. 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, so invalidate slots + * that have exceeded the XID age limit. + * + * We invalidate any aged slot regardless of type. For a non-catalog table + * only physical slots' xmin actually blocks vacuum, so invalidating + * logical slots there won't advance the cutoff, but distinguishing slot + * types by table kind would add complexity for little gain. + * + * Pass nowait = true so vacuum never blocks: it takes only the aged slots + * it can acquire immediately and leaves any slot held by a live process + * to the checkpointer. This also avoids many vacuum processes piling onto + * one slot's condition variable. + * + * Note: invalidating a slot does not guarantee oldest_xmin advances. + * Something else (a long-running transaction, or another slot) may hold + * the same xmin, in which case the slot is invalidated but the horizon is + * unchanged. + */ + if (TransactionIdPrecedes(oldest_xmin, xid_limit)) + return InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, + 0, InvalidOid, + InvalidTransactionId, + xid_limit, true); + + return false; +} diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index d7fb9f5a67f..a2502c91501 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. @@ -1790,7 +1797,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; @@ -1835,6 +1845,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(); } @@ -1873,6 +1906,23 @@ CanInvalidateIdleSlot(ReplicationSlot *s) !(RecoveryInProgress() && s->data.synced)); } +/* + * Can we invalidate an XID-aged replication slot? + * + * This is allowed only when max_slot_xid_age is set and the slot has a valid + * xmin or catalog_xmin. A slot being synced from the primary while the server + * is in recovery is exempt, since synced slots are always considered inactive + * because they don't perform logical decoding to produce changes. + */ +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. @@ -1884,6 +1934,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); @@ -1955,6 +2006,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; } @@ -1977,6 +2040,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, + TransactionId xidLimit, + bool nowait, bool *released_lock_out) { int last_signaled_pid = 0; @@ -2029,6 +2094,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, + xidLimit, &inactive_since, now); @@ -2095,6 +2161,14 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, if (active_proc != INVALID_PROC_NUMBER) { + /* + * The slot is held by a live process. When the caller asked us + * not to wait (nowait), we don't terminate the owner or block on + * its condition variable; we simply leave the slot alone. + */ + if (nowait) + break; + /* * Prepare the sleep on the slot's condition variable before * releasing the lock, to close a possible race condition if the @@ -2122,7 +2196,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), @@ -2175,7 +2250,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; @@ -2202,11 +2278,18 @@ 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" * parameter can be a MASK representing one or more of the defined causes. * + * If "nowait" is true, slots that are currently held by a live process are + * left untouched instead of terminating the owner and waiting for the slot to + * be released. Vacuum uses this for XID-age invalidation so it never blocks; + * held slots are cleaned up by the checkpointer, which always waits. + * * If it invalidates the last logical slot in the cluster, it requests to * disable logical decoding. * @@ -2215,7 +2298,9 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon) + TransactionId snapshotConflictHorizon, + TransactionId xidLimit, + bool nowait) { XLogRecPtr oldestLSN; bool invalidated = false; @@ -2254,7 +2339,7 @@ restart: if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, - &released_lock)) + xidLimit, nowait, &released_lock)) { Assert(released_lock); diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 60336b31803..78c0b462f2a 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_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 +GetOldestNonRemovableTransactionIdAndSlotXmins(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 7f011e04990..00148f7f421 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, false); } /* diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index d421cdbde76..4ee2cf9345e 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -2149,6 +2149,14 @@ max => 'MAX_KILOBYTES', }, +{ name => 'max_slot_xid_age', type => 'int', context => 'PGC_SIGHUP', group => 'REPLICATION_SENDING', + short_desc => 'Sets the maximum age of a replication slot\'s transaction ID before it is 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 7958653077b..69b5e1fe3d6 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 20b354aed56..3271d2b51af 100644 --- a/src/bin/pg_basebackup/pg_createsubscriber.c +++ b/src/bin/pg_basebackup/pg_createsubscriber.c @@ -1681,7 +1681,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 9b29444cbca..17a00f71cb1 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,9 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid); extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon); + TransactionId snapshotConflictHorizon, + TransactionId xidLimit, + bool nowait); 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 d718a5b542f..c9c327672fa 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -51,6 +51,9 @@ extern RunningTransactions GetRunningTransactionData(void); extern bool TransactionIdIsInProgress(TransactionId xid); extern TransactionId GetOldestNonRemovableTransactionId(Relation rel); +extern TransactionId GetOldestNonRemovableTransactionIdAndSlotXmins(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 a412faf51c6..d705bac1878 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -546,4 +546,208 @@ 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 + invalidation_reason = 'xid_aged'; + ]) or die "Timed out waiting for slot $slot_name to be invalidated"; +} + +# A small age lets slots reach the limit after just a few XIDs. +my $max_slot_xid_age = 100; +my $primary5 = PostgreSQL::Test::Cluster->new('primary5'); +$primary5->init(allows_streaming => 'logical'); +$primary5->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $max_slot_xid_age +autovacuum = off +}); +$primary5->start; + +# Consume XIDs, one per committed transaction, to age a slot's xmin. +$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; +}); + +$primary5->safe_psql('postgres', + "CREATE TABLE tab_int5 AS SELECT generate_series(1,10) AS a"); + +$backup_name = 'backup5'; +$primary5->backup($backup_name); + +# testcase 1: inactive logical slot, aged catalog_xmin, invalidated by VACUUM. +# The slot gets a catalog_xmin as soon as it is created. +$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_replication_slots + WHERE slot_name = 'lsub5_slot'; +]) or die "Timed out waiting for slot lsub5_slot catalog_xmin"; + +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $max_slot_xid_age)}); + +# A data table's OldestXmin excludes the slot's catalog_xmin, so vacuum does +# not even attempt invalidation here. +$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'); + +# A catalog table's OldestXmin includes the slot's catalog_xmin, so vacuum +# invalidates the unheld slot. +$primary5->safe_psql('postgres', "VACUUM pg_class"); +wait_for_xid_aged_invalidation($primary5, 'lsub5_slot'); +ok(1, "inactive logical slot invalidated by VACUUM"); + +# testcase 2: inactive physical slot, aged xmin, invalidated by autovacuum. +# A standby with hot_standby_feedback gives the slot an xmin; stopping the +# standby freezes it. Autovacuum is driven by dead tuples with naptime 1s, so +# it fires quickly without waiting for wraparound. +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', + "SELECT pg_create_physical_replication_slot('sb5_slot', true)"); +$standby5->start; +$primary5->wait_for_catchup($standby5); + +$primary5->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL FROM pg_replication_slots + WHERE slot_name = 'sb5_slot'; +]) or die "Timed out waiting for slot sb5_slot xmin from HS feedback"; + +$standby5->stop; + +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $max_slot_xid_age)}); + +# Turn autovacuum on and give it a table with dead tuples. +$primary5->append_conf( + 'postgresql.conf', q{ +autovacuum = on +autovacuum_naptime = 1s +log_autovacuum_min_duration = 0 +}); +$primary5->reload; +$primary5->safe_psql( + 'postgres', q{ + CREATE TABLE tab_av (a int); + INSERT INTO tab_av SELECT generate_series(1, 10000); + DELETE FROM tab_av; +}); +wait_for_xid_aged_invalidation($primary5, 'sb5_slot'); +ok(1, "inactive physical slot invalidated by autovacuum"); + +$primary5->append_conf('postgresql.conf', "autovacuum = off"); +$primary5->reload; + +# testcase 3: held physical slot, aged xmin. VACUUM must skip it (never blocks +# on a held slot); the checkpoint terminates the owner and invalidates it. +# The slot is kept active by a running standby, and its xmin is frozen by an +# open transaction on the standby that feedback keeps reporting. +$primary5->safe_psql('postgres', + "SELECT pg_create_physical_replication_slot('sb5_slot_held', true)"); + +my $standby5c = PostgreSQL::Test::Cluster->new('standby5c'); +$standby5c->init_from_backup($primary5, $backup_name, has_streaming => 1); +$standby5c->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb5_slot_held' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$standby5c->start; +$primary5->wait_for_catchup($standby5c); + +$primary5->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL FROM pg_replication_slots + WHERE slot_name = 'sb5_slot_held'; +]) or die "Timed out waiting for slot sb5_slot_held xmin from HS feedback"; + +# Open a transaction on the standby to pin its reported xmin. +my $held = $standby5c->background_psql('postgres'); +$held->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT 1;"); + +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $max_slot_xid_age)}); + +# VACUUM leaves the held slot for the checkpoint: still valid and active. +$primary5->safe_psql('postgres', "VACUUM tab_int5"); +is( $primary5->safe_psql( + 'postgres', + qq[SELECT invalidation_reason IS NULL AND active FROM pg_replication_slots WHERE slot_name = 'sb5_slot_held';] + ), + 't', + 'held physical slot not invalidated by VACUUM'); + +# The checkpoint terminates the owner and invalidates the slot. +$primary5->safe_psql('postgres', "CHECKPOINT"); +wait_for_xid_aged_invalidation($primary5, 'sb5_slot_held'); +ok(1, "held physical slot invalidated by checkpoint"); + +$held->quit; +$standby5c->stop; + +# testcase 4: inactive logical slot on a standby, aged catalog_xmin, +# invalidated by a restartpoint. Vacuum does not run on a standby, so the +# restartpoint is the only path. Disable max_slot_xid_age on the primary so +# only the standby's own logical slot ages out. +$primary5->safe_psql( + 'postgres', q{ +ALTER SYSTEM SET max_slot_xid_age = 0; +SELECT pg_reload_conf(); +}); +$primary5->safe_psql('postgres', + "SELECT pg_create_physical_replication_slot('sb5_slot_sb', true)"); +$standby5->append_conf( + 'postgresql.conf', qq{ +primary_slot_name = 'sb5_slot_sb' +max_slot_xid_age = $max_slot_xid_age +}); +$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_replication_slots + WHERE slot_name = 'sb5_logical_slot'; +]) or die "Timed out waiting for sb5_logical_slot catalog_xmin"; + +$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, "inactive logical slot on standby invalidated by restartpoint"); + +$standby5->stop; +$primary5->stop; + done_testing(); -- 2.47.3
