From 2739ebc8aeca2cea18c4b84ba1e12653e7a182e6 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Wed, 23 Sep 2026 17:18:35 +0000 Subject: [PATCH v15 2/3] Allow vacuum to invalidate XID-aged replication slots. Commit XXX added support for invalidating a replication slot once the age of its xmin or catalog_xmin is beyond the max_slot_xid_age GUC. That check runs during checkpoints, and on a standby during restartpoints. Because checkpoints happen at their own interval, it doesn't always help when vacuum needs it the most, that is when such a slot is what holds vacuum back from freezing XIDs and from pruning dead rows. This commit implements non-blocking invalidation of XID-aged slots during vacuum, to be precise when vacuum computes its xmin cutoffs, so that a vacuum held back by an aged slot can invalidate that slot and unblock itself, proceeding to freeze XIDs and prune dead rows without waiting for the next checkpoint. The cutoffs are recomputed once a slot is invalidated, so the vacuum in progress is the one that benefits. This applies to both the VACUUM command and autovacuum (but not to VACUUM FULL or REPACK). The check runs per relation, and only when a replication slot is what holds that relation's oldest xmin back and has aged past the limit, so the extra work happens only where invalidating the slot can actually let vacuum freeze more XIDs and remove more rows. It reuses the cutoff computation vacuum already does for the relation, which now reports the oldest slot xmin and catalog_xmin alongside the oldest xmin, so no additional proc array scan is needed per relation. A logical slot holds vacuum back only on system catalogs, through its catalog_xmin, so vacuuming a user table does not invalidate it; such a slot is invalidated when a system catalog is vacuumed, or at a checkpoint. A physical slot holds back the removal of rows in both user tables and system catalogs, through its xmin, and so can be invalidated by vacuuming any table. Vacuum never blocks on this. It invalidates only the aged slots it can acquire immediately, and leaves any slot that is still in use to the next checkpoint, where the invalidation does terminate the process that owns the slot and wait for the slot to be released. This keeps vacuum simple to reason about and avoids many autovacuum workers and backends piling up on one slot waiting for a slow walsender. A slot that vacuum skips this way is invalidated at that later checkpoint, and relations vacuumed after that pick up the advanced cutoffs. Author: Bharath Rupireddy Reviewed-by: John Hsu Reviewed-by: Masahiko Sawada Reviewed-by: Hayato Kuroda Reviewed-by: Satya Narlapuram Reviewed-by: Amit Kapila Reviewed-by: Bertrand Drouvot Reviewed-by: Nisha Moond Reviewed-by: Surya Poondla Discussion: https://postgr.es/m/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com Discussion: https://postgr.es/m/CALj2ACUmPbkcj4y4oeXvzUkBejG68QDtrFF7QHDC_qz2vQcTCg@mail.gmail.com --- doc/src/sgml/config.sgml | 42 ++++-- src/backend/access/heap/vacuumlazy.c | 16 +++ src/backend/access/transam/xlog.c | 12 +- src/backend/commands/vacuum.c | 6 +- src/backend/postmaster/autovacuum.c | 11 ++ src/backend/replication/slot.c | 102 +++++++++++++- src/backend/storage/ipc/procarray.c | 64 +++++++-- src/backend/storage/ipc/standby.c | 4 +- src/include/commands/vacuum.h | 13 ++ src/include/replication/slot.h | 8 +- src/include/storage/procarray.h | 4 + .../t/099_invalidate_xid_aged_slots.pl | 125 +++++++++++++++++- 12 files changed, 370 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index a36344f369a..2b827c06b2b 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -5039,17 +5039,37 @@ HINT: If it is safe for all REPLICATION users to use this library as an output - Slot invalidation due to this limit occurs during checkpoint. Because - checkpoints happen at their own interval, there can be some lag between - when a slot's xmin or catalog_xmin - age exceeds max_slot_xid_age and when the slot - invalidation is actually triggered. To avoid such lags, users can force - a checkpoint to promptly invalidate the slot. On a standby, invalidation - happens at a restartpoint, and a restartpoint occurs only after the - standby has replayed a checkpoint record from the primary. The lag on a - standby therefore depends on the primary's checkpoint interval, and - forcing a checkpoint on the standby does not invalidate a slot until - such a record has been replayed. + Slot invalidation due to XID age occurs during vacuum (both the + VACUUM command and autovacuum, but not + VACUUM FULL or REPACK) and + during checkpoint. During vacuum, only the slots that can be acquired + immediately are invalidated, so that vacuum never blocks; a slot that + is still in use is left for the next checkpoint, where the + invalidation terminates the process that owns the slot and waits for + the slot to be released. Because vacuum and checkpoints happen at + their own intervals, there can be some lag between when a slot's + xmin or catalog_xmin age exceeds + max_slot_xid_age and when the slot invalidation is + actually triggered. To avoid such lags, users can force a checkpoint + to promptly invalidate the slot. On a standby, invalidation happens at + a restartpoint, and a restartpoint occurs only after the standby has + replayed a checkpoint record from the primary. The lag on a standby + therefore depends on the primary's checkpoint interval, and forcing a + checkpoint on the standby does not invalidate a slot until such a + record has been replayed. + + + + During vacuum, a slot is invalidated only when it is holding vacuum + of the current relation back. A logical replication slot holds back + only the removal of system catalog rows (through its + catalog_xmin), so vacuuming a user table does + not invalidate it, even when its age has exceeded + max_slot_xid_age; such a slot is invalidated when a + system catalog is vacuumed or at the next checkpoint. A physical + replication slot holds back the removal of rows in both user tables + and system catalogs (through its xmin), and so can + be invalidated by vacuuming any table. diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 8e1f660bc2f..1e081a4f34d 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" @@ -799,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 a replication slot whose XID age exceeds the limit is holding the + * vacuum cutoff back, invalidate it and recompute the cutoffs. + */ + if (InvalidateXidAgedReplicationSlots(vacrel->cutoffs.OldestXmin, + vacrel->cutoffs.SlotXmin, + vacrel->cutoffs.SlotCatalogXmin, + vacrel->cutoffs.SlotCatalogXminRelevant)) + { + /* 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 e816239bfa5..ad4a8d6ebbb 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -8189,7 +8189,9 @@ CreateCheckPoint(int flags) KeepLogSeg(recptr, &_logSegNo); if (InvalidateObsoleteReplicationSlots(possible_causes, _logSegNo, InvalidOid, - InvalidTransactionId)) + InvalidTransactionId, + false, /* nowait */ + true)) /* check_catalog_xmin */ { /* * Some slots have been invalidated; recalculate the old-segment @@ -8737,7 +8739,9 @@ CreateRestartPoint(int flags) if (InvalidateObsoleteReplicationSlots(possible_causes, _logSegNo, InvalidOid, - InvalidTransactionId)) + InvalidTransactionId, + false, /* nowait */ + true)) /* check_catalog_xmin */ { /* * Some slots have been invalidated; recalculate the old-segment @@ -9670,7 +9674,9 @@ xlog_redo(XLogReaderState *record) */ InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, 0, InvalidOid, - InvalidTransactionId); + InvalidTransactionId, + false, /* nowait */ + true); /* check_catalog_xmin */ } else if (sync_replication_slots) { diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index d8c2f33c615..c3066a351fc 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -1159,7 +1159,11 @@ 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->SlotXmin, + &cutoffs->SlotCatalogXmin, + &cutoffs->SlotCatalogXminRelevant); Assert(TransactionIdIsNormal(cutoffs->OldestXmin)); diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c index 60ebe828900..f1abc28d936 100644 --- a/src/backend/postmaster/autovacuum.c +++ b/src/backend/postmaster/autovacuum.c @@ -89,6 +89,7 @@ #include "postmaster/autovacuum.h" #include "postmaster/interrupt.h" #include "postmaster/postmaster.h" +#include "replication/slot.h" #include "storage/aio_subsys.h" #include "storage/bufmgr.h" #include "storage/ipc.h" @@ -2559,6 +2560,16 @@ do_autovacuum(void) /* this resets ProcGlobal->statusFlags[i] too */ AbortOutOfAnyTransaction(); + + /* + * This worker may still hold a replication slot, from an error + * thrown while invalidating an XID-aged slot during vacuum. The + * transaction abort above does not release it, so release it here + * before moving on to the next table. + */ + if (MyReplicationSlot != NULL) + ReplicationSlotRelease(); + FlushErrorState(); MemoryContextReset(PortalContext); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index cdf31c0f1e3..88a3f65ee6c 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -2002,7 +2002,8 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, TimestampTz *inactive_since, TimestampTz now, TransactionId xidLimit, TransactionId *slot_xmin, - TransactionId *slot_catalog_xmin) + TransactionId *slot_catalog_xmin, + bool check_catalog_xmin) { Assert(possible_causes != RS_INVAL_NONE); @@ -2095,12 +2096,16 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, * Record each of xmin and catalog_xmin that has aged past the limit, * so the invalidation message names the xids that actually triggered * it. Either one alone is enough to invalidate the slot. + * + * catalog_xmin is considered only when the caller asks for it; see + * InvalidateXidAgedReplicationSlots() for when vacuum leaves it out. */ if (TransactionIdIsValid(effective_xmin) && TransactionIdPrecedes(effective_xmin, xidLimit)) *slot_xmin = effective_xmin; - if (TransactionIdIsValid(effective_catalog_xmin) && + if (check_catalog_xmin && + TransactionIdIsValid(effective_catalog_xmin) && TransactionIdPrecedes(effective_catalog_xmin, xidLimit)) *slot_catalog_xmin = effective_catalog_xmin; @@ -2132,6 +2137,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, TransactionId xidLimit, + bool nowait, + bool check_catalog_xmin, bool *released_lock_out) { int last_signaled_pid = 0; @@ -2190,7 +2197,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, now, xidLimit, &slot_xmin, - &slot_catalog_xmin); + &slot_catalog_xmin, + check_catalog_xmin); /* if there's no invalidation, we're done */ if (invalidation_cause == RS_INVAL_NONE) @@ -2255,6 +2263,10 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, if (active_proc != INVALID_PROC_NUMBER) { + /* A nowait caller leaves an active slot untouched. */ + if (nowait) + break; + /* * Prepare the sleep on the slot's condition variable before * releasing the lock, to close a possible race condition if the @@ -2371,6 +2383,14 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, * 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, a slot that is still in use is skipped instead of + * terminating the process that owns it and waiting for the slot to be + * released. Vacuum uses this for XID-age invalidation so that it never + * blocks; a slot skipped that way is left for the next checkpoint, which + * does wait. + * + * "check_catalog_xmin" applies only to XID-age invalidation. + * * If it invalidates the last logical slot in the cluster, it requests to * disable logical decoding. * @@ -2379,7 +2399,9 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon) + TransactionId snapshotConflictHorizon, + bool nowait, + bool check_catalog_xmin) { XLogRecPtr oldestLSN; TransactionId xidLimit = InvalidTransactionId; @@ -2423,7 +2445,7 @@ restart: if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, - xidLimit, + xidLimit, nowait, check_catalog_xmin, &released_lock)) { Assert(released_lock); @@ -2478,6 +2500,76 @@ restart: return invalidated; } +/* + * Invalidate replication slots whose XID age exceeds the limit. + * + * The caller passes the vacuum cutoff computed for the relation, plus the + * oldest xmin and catalog_xmin of any replication slot and whether that + * catalog_xmin is relevant for the relation, all as reported by + * GetOldestNonRemovableTransactionIdAndSlotXmins(). If a replication slot is + * not what holds that cutoff back, or the cutoff has not yet aged past the + * limit, there is nothing to do. + * + * slot_catalog_xmin_relevant tells whether a slot's catalog_xmin can hold this + * relation's cutoff back, which is true for catalog and shared relations, + * whose cutoff is computed from both the slot xmin and catalog_xmin. When it + * is false, a slot holding only a catalog_xmin cannot be blocking this vacuum, + * so such slots are neither considered here nor invalidated: even if one is + * aged, invalidating it would not advance the cutoff, and the slot may yet + * advance on its own before a catalog vacuum or a checkpoint acts on it. + * + * Returns true if at least one slot was invalidated. + */ +bool +InvalidateXidAgedReplicationSlots(TransactionId oldest_xmin, + TransactionId slot_xmin, + TransactionId slot_catalog_xmin, + bool slot_catalog_xmin_relevant) +{ + TransactionId xid_limit; + bool slot_holds_oldest_xmin; + + Assert(TransactionIdIsNormal(oldest_xmin)); + + /* + * Check if a replication slot's xmin, or its catalog_xmin when that is + * relevant for this relation, is what's holding the oldest xmin back. If + * not, skip the unnecessary work. + */ + slot_holds_oldest_xmin = + (TransactionIdIsValid(slot_xmin) && + TransactionIdEquals(oldest_xmin, slot_xmin)) || + (slot_catalog_xmin_relevant && + TransactionIdIsValid(slot_catalog_xmin) && + TransactionIdEquals(oldest_xmin, slot_catalog_xmin)); + + if (!slot_holds_oldest_xmin) + return false; + + /* Nothing to do if the age limit is disabled */ + xid_limit = GetSlotXidAgeLimit(); + if (!TransactionIdIsValid(xid_limit)) + return false; + + /* + * A replication slot holds the oldest xmin back, so invalidate any slot + * that has aged past the limit. + * + * Vacuum never blocks on this. It invalidates only the slots it can + * acquire immediately and leaves any slot still in use to the next + * checkpoint, so that autovacuum workers and backends do not pile up + * waiting on one slot. + */ + if (TransactionIdPrecedes(oldest_xmin, xid_limit)) + return InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, + 0, InvalidOid, + InvalidTransactionId, + true, /* nowait */ + slot_catalog_xmin_relevant); + + return false; +} + /* * Flush all replication slots to disk. * diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index b7e03134ed8..26afbd763a4 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -1929,6 +1929,30 @@ GlobalVisHorizonKindForRel(Relation rel) return VISHORIZON_TEMP; } +/* + * Return the oldest non-removable XID for the given relation, out of the + * horizons already computed by ComputeXidHorizons(). + */ +static 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 +1971,34 @@ 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 reports the oldest + * replication slot xmin and catalog_xmin, and whether that catalog_xmin is + * relevant for this relation, from the same ComputeXidHorizons() call. This + * avoids a second ProcArrayLock acquisition for a caller that needs them all. + * See InvalidateXidAgedReplicationSlots() for what makes a catalog_xmin + * relevant. + */ +TransactionId +GetOldestNonRemovableTransactionIdAndSlotXmins(Relation rel, + TransactionId *slot_xmin, + TransactionId *slot_catalog_xmin, + bool *slot_catalog_xmin_relevant) +{ + ComputeXidHorizonsResult horizons; + GlobalVisHorizonKind kind = GlobalVisHorizonKindForRel(rel); + + ComputeXidHorizons(&horizons); + + *slot_xmin = horizons.slot_xmin; + *slot_catalog_xmin = horizons.slot_catalog_xmin; + *slot_catalog_xmin_relevant = (kind == VISHORIZON_CATALOG || + kind == VISHORIZON_SHARED); + + return GetOldestNonRemovableTransactionIdFromHorizons(&horizons, rel); } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 7f011e04990..7b12f8ca431 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -504,7 +504,9 @@ ResolveRecoveryConflictWithSnapshot(TransactionId snapshotConflictHorizon, */ if (IsLogicalDecodingEnabled() && isCatalogRel) InvalidateObsoleteReplicationSlots(RS_INVAL_HORIZON, 0, locator.dbOid, - snapshotConflictHorizon); + snapshotConflictHorizon, + false, /* nowait */ + true); /* check_catalog_xmin */ } /* diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h index 6e3c912bf5c..f271f9b15fe 100644 --- a/src/include/commands/vacuum.h +++ b/src/include/commands/vacuum.h @@ -295,6 +295,19 @@ struct VacuumCutoffs */ TransactionId FreezeLimit; MultiXactId MultiXactCutoff; + + /* + * SlotXmin and SlotCatalogXmin are the oldest xmin and catalog_xmin of + * any replication slot, from the same ComputeXidHorizons() call that + * computed OldestXmin. + * + * SlotCatalogXminRelevant is whether a slot's catalog_xmin can hold + * OldestXmin back, which is true for catalog and shared relations. See + * InvalidateXidAgedReplicationSlots(). + */ + TransactionId SlotXmin; + TransactionId SlotCatalogXmin; + bool SlotCatalogXminRelevant; }; /* diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index ab264c8c09a..806ba37cab6 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -367,7 +367,13 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid); extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon); + TransactionId snapshotConflictHorizon, + bool nowait, + bool check_catalog_xmin); +extern bool InvalidateXidAgedReplicationSlots(TransactionId oldest_xmin, + TransactionId slot_xmin, + TransactionId slot_catalog_xmin, + bool slot_catalog_xmin_relevant); 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..f41e039b091 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -51,6 +51,10 @@ 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, + bool *slot_catalog_xmin_relevant); extern TransactionId GetOldestTransactionIdConsideredRunning(void); extern TransactionId GetOldestActiveTransactionId(bool inCommitOnly, bool allDbs); diff --git a/src/test/recovery/t/099_invalidate_xid_aged_slots.pl b/src/test/recovery/t/099_invalidate_xid_aged_slots.pl index 5459f8a4cee..d6db34554d4 100644 --- a/src/test/recovery/t/099_invalidate_xid_aged_slots.pl +++ b/src/test/recovery/t/099_invalidate_xid_aged_slots.pl @@ -18,6 +18,27 @@ sub wait_for_slot or die "Timed out waiting for slot $slot_name: $cond"; } +# Vacuum the given relation, then check the slot's invalidation reason and +# whether the relation's dead tuples could be removed. +sub vacuum_and_check +{ + my ($node, $relname, $slot_name, $reason, $dead_removed) = @_; + + $node->safe_psql('postgres', "VACUUM $relname"); + is( $node->safe_psql('postgres', + "SELECT coalesce(invalidation_reason, 'none') FROM pg_replication_slots WHERE slot_name = '$slot_name'" + ), + $reason, + "slot $slot_name reads $reason after vacuuming $relname"); + is( $node->safe_psql('postgres', + "SELECT n_dead_tup = 0 FROM pg_stat_all_tables WHERE relname = '$relname'" + ), + $dead_removed ? 't' : 'f', + "vacuum " + . ($dead_removed ? "removes" : "leaves") + . " the dead tuples in $relname"); +} + # A small age lets slots reach the limit after just a few XIDs my $slot_xid_age = 100; @@ -40,16 +61,19 @@ my $consume_xid_proc = qq{ my $primary = PostgreSQL::Test::Cluster->new('primary'); $primary->init(allows_streaming => 'logical'); -# No checkpoints and no autovacuum, so that a slot is invalidated only where a -# testcase asks for it. +# No checkpoints, autovacuum or walsender timeouts, so that nothing invalidates +# a slot or advances its horizon behind a testcase's back. $primary->append_conf( 'postgresql.conf', qq{ max_slot_xid_age = $slot_xid_age autovacuum = off checkpoint_timeout = 1h +wal_sender_timeout = 0 }); $primary->start; $primary->safe_psql('postgres', $consume_xid_proc); +$primary->safe_psql('postgres', + "CREATE TABLE tbl_user AS SELECT generate_series(1,10) AS a"); # Testcase 1: an inactive logical slot with an aged catalog_xmin is invalidated # at a checkpoint. @@ -114,6 +138,103 @@ $running_xact->quit; # The terminated backend took its psql down too, so just reap the process $export->{run}->finish; +# Testcase 3: the VACUUM command skips an active logical slot with an aged +# catalog_xmin, rather than terminating its owner to invalidate it. +$primary->safe_psql('postgres', + "SELECT pg_create_logical_replication_slot('logical_active_slot', 'test_decoding')" +); + +# Dead catalog rows that only this slot's catalog_xmin holds back +$primary->safe_psql('postgres', + "CREATE TABLE tbl_tmp(a int); DROP TABLE tbl_tmp;"); + +# No status messages, so the client's feedback cannot advance the slot's +# catalog_xmin while the testcase ages it. +my ($stdout, $stderr); +my $recvlogical = IPC::Run::start( + [ + 'pg_recvlogical', + '--dbname' => $primary->connstr('postgres'), + '--slot' => 'logical_active_slot', + '--status-interval' => 0, + '--file' => '-', + '--no-loop', + '--start', + ], + '>' => \$stdout, + '2>' => \$stderr, + IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default)); +wait_for_slot($primary, 'logical_active_slot', 'active_pid IS NOT NULL'); + +$primary->safe_psql('postgres', qq{CALL consume_xid(2 * $slot_xid_age)}); + +# Fail rather than pass vacuously, should the slot's horizon have moved anyway +is( $primary->safe_psql('postgres', + "SELECT age(catalog_xmin) > $slot_xid_age FROM pg_replication_slots WHERE slot_name = 'logical_active_slot'" + ), + 't', + 'active slot is aged past the limit'); + +vacuum_and_check($primary, 'pg_class', 'logical_active_slot', 'none', 0); + +# Testcase 4: the VACUUM command invalidates that same slot once it is +# inactive, and then removes the rows it was holding back. + +# End the client's session to make the slot inactive (portable way) +$primary->safe_psql('postgres', + "SELECT pg_terminate_backend(active_pid) FROM pg_replication_slots WHERE slot_name = 'logical_active_slot'" +); +wait_for_slot($primary, 'logical_active_slot', 'active_pid IS NULL'); +$recvlogical->finish; + +# A logical slot holds only a catalog_xmin, so a user table's cutoff is not its +# to hold back. +$primary->safe_psql('postgres', "VACUUM tbl_user"); +is( $primary->safe_psql('postgres', + "SELECT invalidation_reason IS NULL FROM pg_replication_slots WHERE slot_name = 'logical_active_slot'" + ), + 't', + 'logical slot not invalidated by vacuuming a user table'); + +vacuum_and_check($primary, 'pg_class', 'logical_active_slot', 'xid_aged', 1); +$primary->safe_psql('postgres', + "SELECT pg_drop_replication_slot('logical_active_slot')"); + +# Testcase 5: the VACUUM command invalidates an inactive physical slot with an +# aged xmin. Feedback from a standby is what gives such a slot an xmin, and +# stopping the standby freezes it. +my $backup_name = 'backup'; +$primary->backup($backup_name); + +my $standby = PostgreSQL::Test::Cluster->new('standby'); +$standby->init_from_backup($primary, $backup_name, has_streaming => 1); + +$primary->safe_psql('postgres', + "SELECT pg_create_physical_replication_slot('phys_slot', true)"); +$standby->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'phys_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$standby->start; +$primary->wait_for_catchup($standby); +wait_for_slot($primary, 'phys_slot', 'xmin IS NOT NULL'); +$standby->stop; + +# Dead rows from an XID that the now frozen xmin holds back +$primary->safe_psql('postgres', "DELETE FROM tbl_user"); + +$log_offset = -s $primary->logfile; +$primary->safe_psql('postgres', qq{CALL consume_xid(2 * $slot_xid_age)}); +vacuum_and_check($primary, 'tbl_user', 'phys_slot', 'xid_aged', 1); + +# The slot holds an xmin alone, so only its age is reported +ok( $primary->log_contains( + qr/invalidating obsolete replication slot "phys_slot"\n.*DETAIL:.*The slot's xmin age of \d+ transactions exceeds the configured "max_slot_xid_age" of $slot_xid_age\./, + $log_offset), + 'aged xmin is reported on invalidation'); + $primary->stop; done_testing();