From 572dafbacfceeb7a334a860c9723958c4e2e0aaa Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Mon, 30 Mar 2026 01:12:56 +0000 Subject: [PATCH v6 1/2] Add XID age based replication slot invalidation Introduce max_slot_xid_age, a GUC that invalidates replication slots whose xmin or catalog_xmin exceeds the specified age. Disabled by default. Idle or forgotten replication slots can hold back vacuum, leading to bloat and eventually XID wraparound. In the worst case this requires dropping the slot and single-user mode vacuuming. This setting avoids that by proactively invalidating slots that have fallen too far behind. Invalidation checks are performed once per relation during vacuum (both vacuum command and autovacuum), and also by the checkpointer during checkpoints and restartpoints. --- doc/src/sgml/config.sgml | 40 ++++ doc/src/sgml/system-views.sgml | 8 + src/backend/access/heap/vacuumlazy.c | 20 +- src/backend/access/transam/xlog.c | 20 +- src/backend/commands/cluster.c | 2 +- src/backend/commands/vacuum.c | 8 +- src/backend/replication/slot.c | 128 ++++++++++++- src/backend/storage/ipc/procarray.c | 19 ++ src/backend/storage/ipc/standby.c | 3 +- src/backend/utils/misc/guc_parameters.dat | 8 + src/backend/utils/misc/postgresql.conf.sample | 2 + src/include/commands/vacuum.h | 4 +- src/include/replication/slot.h | 10 +- src/include/storage/procarray.h | 3 + src/test/recovery/t/019_replslot_limit.pl | 175 ++++++++++++++++++ 15 files changed, 435 insertions(+), 15 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 229f41353eb..46aac59cb20 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4764,6 +4764,46 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows + + max_slot_xid_age (integer) + + max_slot_xid_age configuration parameter + + + + + Invalidate replication slots whose xmin (the oldest + transaction that this slot needs the database to retain) or + catalog_xmin (the oldest transaction affecting the + system catalogs that this slot needs the database to retain) has reached + the age specified by this setting. This invalidation check happens + during vacuum (both VACUUM command and autovacuum) + and during checkpoints. + A value of zero (which is default) disables this feature. Users can set + this value anywhere from zero to two billion. This parameter can only be + set in the postgresql.conf file or on the server + command line. + + + + 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 for more details. + + + + 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 + pg_replication_slots.synced + value true). Synced slots are always considered to + be inactive because they don't perform logical decoding to produce + changes. + + + + wal_sender_timeout (integer) diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index 9ee1a2bfc6a..1a507b430f9 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 duration. + + + xid_aged means that the slot's + xmin or catalog_xmin + has reached the age specified by + parameter. + + diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index f698c2d899b..2e8e61eb0e2 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -147,6 +147,7 @@ #include "pgstat.h" #include "portability/instr_time.h" #include "postmaster/autovacuum.h" +#include "replication/slot.h" #include "storage/bufmgr.h" #include "storage/freespace.h" #include "storage/latch.h" @@ -642,6 +643,8 @@ heap_vacuum_rel(Relation rel, const VacuumParams params, ErrorContextCallback errcallback; char **indnames = NULL; Size dead_items_max_bytes = 0; + TransactionId slot_xmin = InvalidTransactionId; + TransactionId slot_catalog_xmin = InvalidTransactionId; verbose = (params.options & VACOPT_VERBOSE) != 0; instrument = (verbose || (AmAutoVacuumWorkerProcess() && @@ -798,7 +801,22 @@ heap_vacuum_rel(Relation rel, const VacuumParams params, * want to teach lazy_scan_prune to recompute vistest from time to time, * to increase the number of dead tuples it can prune away.) */ - vacrel->aggressive = vacuum_get_cutoffs(rel, params, &vacrel->cutoffs); + vacrel->aggressive = vacuum_get_cutoffs(rel, params, &vacrel->cutoffs, + &slot_xmin, &slot_catalog_xmin); + + /* + * Try to invalidate XID-aged replication slots. Use the slot xmin values + * obtained from the same horizons computation that produced OldestXmin, + * avoiding an extra ProcArrayLock acquisition. + */ + if (MaybeInvalidateXIDAgedSlots(slot_xmin, slot_catalog_xmin)) + { + /* Recompute cutoffs after slot invalidation. */ + vacrel->aggressive = vacuum_get_cutoffs(rel, params, + &vacrel->cutoffs, + NULL, NULL); + } + 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 f5c9a34374d..cb943bdc2a9 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7019,6 +7019,7 @@ CreateCheckPoint(int flags) VirtualTransactionId *vxids; int nvxids; int oldXLogAllowed = 0; + uint32 possibleInvalidationCauses; /* * An end-of-recovery checkpoint is really a shutdown checkpoint, just @@ -7441,8 +7442,15 @@ CreateCheckPoint(int flags) */ XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size); KeepLogSeg(recptr, &_logSegNo); - if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT, + + possibleInvalidationCauses = RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT | + RS_INVAL_XID_AGE; + + if (InvalidateObsoleteReplicationSlots(possibleInvalidationCauses, _logSegNo, InvalidOid, + InvalidTransactionId, + max_slot_xid_age > 0 ? + ReadNextTransactionId() : InvalidTransactionId)) { /* @@ -7724,6 +7732,7 @@ CreateRestartPoint(int flags) XLogRecPtr endptr; XLogSegNo _logSegNo; TimestampTz xtime; + uint32 possibleInvalidationCauses; /* Concurrent checkpoint/restartpoint cannot happen */ Assert(!IsUnderPostmaster || MyBackendType == B_CHECKPOINTER); @@ -7898,8 +7907,14 @@ CreateRestartPoint(int flags) INJECTION_POINT("restartpoint-before-slot-invalidation", NULL); - if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT, + possibleInvalidationCauses = RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT | + RS_INVAL_XID_AGE; + + if (InvalidateObsoleteReplicationSlots(possibleInvalidationCauses, _logSegNo, InvalidOid, + InvalidTransactionId, + max_slot_xid_age > 0 ? + ReadNextTransactionId() : InvalidTransactionId)) { /* @@ -8764,6 +8779,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/cluster.c b/src/backend/commands/cluster.c index 09066db0956..118d5d28c1e 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -927,7 +927,7 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, bool verb * not to be aggressive about this. */ memset(¶ms, 0, sizeof(VacuumParams)); - vacuum_get_cutoffs(OldHeap, params, &cutoffs); + vacuum_get_cutoffs(OldHeap, params, &cutoffs, NULL, NULL); /* * FreezeXid will become the table's new relfrozenxid, and that mustn't go diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index bce3a2daa24..e85d53a0ecb 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -1098,7 +1098,9 @@ get_all_vacuum_rels(MemoryContext vac_context, int options) */ bool vacuum_get_cutoffs(Relation rel, const VacuumParams params, - struct VacuumCutoffs *cutoffs) + struct VacuumCutoffs *cutoffs, + TransactionId *slot_xmin, + TransactionId *slot_catalog_xmin) { int freeze_min_age, multixact_freeze_min_age, @@ -1133,7 +1135,9 @@ 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 = GetOldestNonRemovableTransactionIdExt(rel, + slot_xmin, + slot_catalog_xmin); Assert(TransactionIdIsNormal(cutoffs->OldestXmin)); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index a9092fc2382..c2ff7e464f0 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -117,6 +117,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"}, }; /* @@ -158,6 +159,12 @@ int max_replication_slots = 10; /* the maximum number of replication */ 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. @@ -1780,7 +1787,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 recentXid) { StringInfoData err_detail; StringInfoData err_hint; @@ -1825,6 +1835,30 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, "idle_replication_slot_timeout"); break; } + + case RS_INVAL_XID_AGE: + { + Assert(TransactionIdIsValid(xmin) || TransactionIdIsValid(catalog_xmin)); + + if (TransactionIdIsValid(xmin)) + { + /* translator: %s is a GUC variable name */ + appendStringInfo(&err_detail, _("The slot's xmin %u is %d transactions old, which exceeds the configured \"%s\" value of %d."), + xmin, (int32) (recentXid - xmin), "max_slot_xid_age", max_slot_xid_age); + } + else + { + /* translator: %s is a GUC variable name */ + appendStringInfo(&err_detail, _("The slot's catalog_xmin %u is %d transactions old, which exceeds the configured \"%s\" value of %d."), + catalog_xmin, (int32) (recentXid - catalog_xmin), "max_slot_xid_age", max_slot_xid_age); + } + + /* 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(); } @@ -1863,6 +1897,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. @@ -1874,6 +1927,7 @@ static ReplicationSlotInvalidationCause DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, + TransactionId recentXid, TimestampTz *inactive_since, TimestampTz now) { Assert(possible_causes != RS_INVAL_NONE); @@ -1945,6 +1999,22 @@ 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)) + { + TransactionId xidLimit; + + Assert(TransactionIdIsValid(recentXid)); + + xidLimit = TransactionIdRetreatedBy(recentXid, max_slot_xid_age); + + 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; } @@ -1967,6 +2037,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, + TransactionId recentXid, bool *released_lock_out) { int last_signaled_pid = 0; @@ -2019,6 +2090,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, + recentXid, &inactive_since, now); @@ -2112,7 +2184,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, recentXid); if (MyBackendType == B_STARTUP) (void) SignalRecoveryConflict(GetPGProcByNumber(active_proc), @@ -2165,7 +2238,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, recentXid); /* done with this slot for now */ break; @@ -2192,6 +2266,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" @@ -2205,7 +2281,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon) + TransactionId snapshotConflictHorizon, + TransactionId recentXid) { XLogRecPtr oldestLSN; bool invalidated = false; @@ -2244,7 +2321,7 @@ restart: if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, - &released_lock)) + recentXid, &released_lock)) { Assert(released_lock); @@ -3275,3 +3352,44 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn) ConditionVariableCancelSleep(); } + +/* + * Invalidate replication slots whose XID age exceeds the max_slot_xid_age + * GUC. + * + * The slot_xmin and slot_catalog_xmin are the replication slot xmin values + * obtained from the same ComputeXidHorizons() call that computed OldestXmin + * during vacuum. Using these avoids a separate ProcArrayLock acquisition. + * + * Returns true if at least one slot was invalidated. + */ +bool +MaybeInvalidateXIDAgedSlots(TransactionId slot_xmin, + TransactionId slot_catalog_xmin) +{ + TransactionId recentXid; + TransactionId xidLimit; + bool invalidated = false; + + if (max_slot_xid_age == 0) + return false; + + recentXid = ReadNextTransactionId(); + xidLimit = TransactionIdRetreatedBy(recentXid, max_slot_xid_age); + + /* + * Invalidate possibly obsolete slots based on XID-age, if either slot's + * xmin or catalog_xmin is older than the cutoff. + */ + if ((TransactionIdIsValid(slot_xmin) && + TransactionIdPrecedes(slot_xmin, xidLimit)) || + (TransactionIdIsValid(slot_catalog_xmin) && + TransactionIdPrecedes(slot_catalog_xmin, xidLimit))) + invalidated = InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, + 0, + InvalidOid, + InvalidTransactionId, + recentXid); + + return invalidated; +} diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index cc207cb56e3..18683ce5aea 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -1950,11 +1950,30 @@ GlobalVisHorizonKindForRel(Relation rel) */ TransactionId GetOldestNonRemovableTransactionId(Relation rel) +{ + return GetOldestNonRemovableTransactionIdExt(rel, NULL, NULL); +} + +/* + * 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 +GetOldestNonRemovableTransactionIdExt(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; + switch (GlobalVisHorizonKindForRel(rel)) { case VISHORIZON_SHARED: diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index de9092fdf5b..d60f39ec08e 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 0a862693fcd..ca3cc8417da 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -2089,6 +2089,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 => '2000000000', +}, + # 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 cf15597385b..055eba56bdf 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -351,6 +351,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 #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h index 1f45bca015c..c5ae9efe977 100644 --- a/src/include/commands/vacuum.h +++ b/src/include/commands/vacuum.h @@ -384,7 +384,9 @@ extern void vac_update_relstats(Relation relation, bool *minmulti_updated, bool in_outer_xact); extern bool vacuum_get_cutoffs(Relation rel, const VacuumParams params, - struct VacuumCutoffs *cutoffs); + struct VacuumCutoffs *cutoffs, + TransactionId *slot_xmin, + TransactionId *slot_catalog_xmin); extern bool vacuum_xid_failsafe_check(const struct VacuumCutoffs *cutoffs); extern void vac_update_datfrozenxid(void); extern void vacuum_delay_point(bool is_analyze); diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 4b4709f6e2c..0baa7112559 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 @@ -326,6 +328,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *synchronized_standby_slots; extern PGDLLIMPORT int idle_replication_slot_timeout_secs; +extern PGDLLIMPORT int max_slot_xid_age; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); @@ -367,7 +370,10 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid); extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon); + TransactionId snapshotConflictHorizon, + TransactionId recentXid); +extern bool MaybeInvalidateXIDAgedSlots(TransactionId slot_xmin, + TransactionId slot_catalog_xmin); 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 abdf021e66e..c198fd22515 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -53,6 +53,9 @@ extern RunningTransactions GetRunningTransactionData(void); extern bool TransactionIdIsInProgress(TransactionId xid); extern TransactionId GetOldestNonRemovableTransactionId(Relation rel); +extern TransactionId GetOldestNonRemovableTransactionIdExt(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..d0f58d8317f 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; +# Advance XIDs, run VACUUM, and wait for a slot to be invalidated due to XID age. +sub invalidate_slot_by_xid_age +{ + my ($node, $table_name, $slot_name, $slot_type, $nxids, $trigger) = @_; + + # Do some work to advance xids + $node->safe_psql( + 'postgres', qq[ + do \$\$ + begin + for i in 1..$nxids loop + -- use an exception block so that each iteration eats an XID + begin + insert into $table_name values (i); + exception + when division_by_zero then null; + end; + end loop; + end\$\$; + ]); + + if ($trigger eq 'checkpoint') + { + $node->safe_psql('postgres', "CHECKPOINT"); + } + else + { + $node->safe_psql('postgres', "VACUUM"); + } + + # Wait for the replication slot to be invalidated due to XID age. + $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 while waiting for slot $slot_name to be invalidated"; + + ok(1, "$slot_type replication slot invalidated due to XID age (via $trigger)"); +} + +# ============================================================================= +# Testcase start: Invalidate streaming standby's 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'); + +# Configure primary with XID age settings. Set autovacuum_naptime high so +# that the checkpointer (not vacuum) triggers the invalidation. +my $max_slot_xid_age = 500; +$primary5->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $max_slot_xid_age +autovacuum_naptime = '1h' +}); + +$primary5->start; + +# Take a backup for creating standby +$backup_name = 'backup5'; +$primary5->backup($backup_name); + +# Create a standby linking to the primary using the replication slot +my $standby5 = PostgreSQL::Test::Cluster->new('standby5'); +$standby5->init_from_backup($primary5, $backup_name, has_streaming => 1); + +# Enable HS feedback. The slot should gain an xmin. We set the status interval +# so we'll see the results promptly. +$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"); + +# Wait until standby has replayed enough data +$primary5->wait_for_catchup($standby5); + +# Wait for the slot to get xmin from 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 to make the replication slot on primary inactive. +# The slot's xmin persists and holds back datfrozenxid. +$standby5->stop; + +# Advance XIDs and wait for the slot to be invalidated due to XID age. +# Use 2x the max_slot_xid_age to ensure the slot's xmin age comfortably +# exceeds the configured limit. +invalidate_slot_by_xid_age($primary5, 'tab_int5', 'sb5_slot', 'physical', + 2 * $max_slot_xid_age, 'checkpoint'); + +# Testcase end: Invalidate streaming standby's slot due to max_slot_xid_age +# GUC (via checkpoint). +# ============================================================================= + +# ============================================================================= +# Testcase start: Invalidate logical subscriber's slot due to max_slot_xid_age +# GUC (via vacuum). + +# Reset autovacuum_naptime so that VACUUM-triggered invalidation works normally +$primary5->safe_psql('postgres', + "ALTER SYSTEM RESET autovacuum_naptime; SELECT pg_reload_conf();"); + +# Create a subscriber node +my $subscriber5 = PostgreSQL::Test::Cluster->new('subscriber5'); +$subscriber5->init(allows_streaming => 'logical'); +$subscriber5->start; + +# Create tables on both primary and subscriber +$primary5->safe_psql('postgres', "CREATE TABLE test_tbl5 (id int)"); +$subscriber5->safe_psql('postgres', "CREATE TABLE test_tbl5 (id int)"); + +# Insert some initial data +$primary5->safe_psql('postgres', + "INSERT INTO test_tbl5 VALUES (generate_series(1, 5));"); + +# Setup logical replication +my $primary5_connstr = $primary5->connstr . ' dbname=postgres'; +$primary5->safe_psql('postgres', + "CREATE PUBLICATION pub5 FOR TABLE test_tbl5"); + +$subscriber5->safe_psql('postgres', + "CREATE SUBSCRIPTION sub5 CONNECTION '$primary5_connstr' PUBLICATION pub5 WITH (slot_name = 'lsub5_slot')" +); + +# Wait for initial sync to complete +$subscriber5->wait_for_subscription_sync($primary5, 'sub5'); + +$result = $subscriber5->safe_psql('postgres', "SELECT count(*) FROM test_tbl5"); +is($result, qq(5), "check initial copy was done for logical replication (XID age test)"); + +# Wait for the logical slot to get catalog_xmin (logical slots use catalog_xmin, not xmin) +$primary5->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NULL AND 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 to advance"; + +# Stop subscriber to make the replication slot on primary inactive +$subscriber5->stop; + +# Advance XIDs and wait for the slot to be invalidated due to XID age. +# Use 2x the max_slot_xid_age to ensure the slot's catalog_xmin age +# comfortably exceeds the configured limit. +invalidate_slot_by_xid_age($primary5, 'test_tbl5', 'lsub5_slot', 'logical', + 2 * $max_slot_xid_age, 'vacuum'); + +$primary5->stop; + +# Testcase end: Invalidate logical subscriber's slot due to max_slot_xid_age +# GUC. +# ============================================================================= + done_testing(); -- 2.47.3