From 8d271cbba9088aa32950f199865401bfe96a8464 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Mon, 10 Aug 2026 16:13:00 +0000 Subject: [PATCH v13 1/2] Invalidate XID-aged replication slots. An inactive or forgotten replication slot holds vacuum back from freezing XIDs and from pruning dead rows, through the xmin or catalog_xmin it retains. This can lead to table and index bloat and, left unchecked, eventually to transaction ID wraparound. Until now the only way to bound this was to notice the slot and drop it by hand. This commit adds a GUC, max_slot_xid_age, that invalidates a replication slot once the age of its xmin or catalog_xmin exceeds the configured number of transactions. A value of zero, the default, disables the feature. The invalidation check runs during checkpoints, and on a standby during the restartpoints that stand in for them. This is the same place the WAL and idle-timeout slot invalidations already run. It terminates the process that owns a slot still in use and waits for the slot to be released. Because checkpoints happen at their own interval, there can be lag between when a slot ages past the limit and when it is invalidated; a manual CHECKPOINT triggers it promptly. Slots on a standby that are being synced from the primary are exempt, since they do not perform logical decoding to produce changes. An upcoming commit adds a non-blocking invalidation path in vacuum, so that a vacuum held back by an aged slot can invalidate that slot itself and proceed to freeze XIDs and prune dead rows, without waiting for the next checkpoint. Author: Bharath Rupireddy Reviewed-by: John Hsu Reviewed-by: Masahiko Sawada Reviewed-by: Hayato Kuroda Reviewed-by: Satya Narlapuram Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com Discussion: https://www.postgresql.org/message-id/CALj2ACUmPbkcj4y4oeXvzUkBejG68QDtrFF7QHDC_qz2vQcTCg@mail.gmail.com Discussion: https://www.postgresql.org/message-id/CALj2ACVD0_DhCQ_QOAa7F=nFv8+ZGsHR8SbOc-FmuV8ZrV92HQ@mail.gmail.com --- doc/src/sgml/config.sgml | 57 ++++++++ doc/src/sgml/logical-replication.sgml | 4 +- doc/src/sgml/maintenance.sgml | 5 +- doc/src/sgml/system-views.sgml | 8 ++ src/backend/access/transam/xlog.c | 34 ++++- src/backend/replication/slot.c | 121 +++++++++++++++- src/backend/storage/ipc/standby.c | 3 +- src/backend/utils/misc/guc_parameters.dat | 8 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/bin/pg_basebackup/pg_createsubscriber.c | 2 +- src/include/replication/slot.h | 8 +- src/test/recovery/t/019_replslot_limit.pl | 133 ++++++++++++++++++ 12 files changed, 368 insertions(+), 16 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 10b304122ef..6f61dd6ec28 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -5020,6 +5020,63 @@ HINT: If it is safe for all REPLICATION users to use this library as an output + + max_slot_xid_age (integer) + + max_slot_xid_age configuration parameter + + + + + Invalidate replication slots whose xmin or + catalog_xmin transaction age in the + pg_replication_slots + 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 postgresql.conf file or on + the server command line. + + + + 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. + + + + The current age of a slot's xmin and + catalog_xmin can be monitored by applying the + age function to the corresponding columns in the + pg_replication_slots + view. + + + + Inactive 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 + 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/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 3a61e2d6889..a2c772d478d 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2699,7 +2699,9 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER Logical replication slots are also affected by - idle_replication_slot_timeout. + idle_replication_slot_timeout + and + max_slot_xid_age. diff --git a/doc/src/sgml/maintenance.sgml b/doc/src/sgml/maintenance.sgml index 33ab4edf87c..0689fc7edf7 100644 --- a/doc/src/sgml/maintenance.sgml +++ b/doc/src/sgml/maintenance.sgml @@ -720,7 +720,10 @@ HINT: Execute a database-wide VACUUM in that database. is large. In many cases, such slots were created for replication to servers that no longer exist, or that have been down for a long time. If you drop a slot for a server that still exists and might still try to connect to that slot, that replica may - need to be rebuilt. + need to be rebuilt. Setting makes the + server invalidate such slots automatically once their age(xmin) + or age(catalog_xmin) exceeds the configured limit, + preventing them from holding vacuum back indefinitely. Execute VACUUM in the target database. A database-wide diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index 5ea19d68622..2b953ab37bd 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 duration. + + + xid_aged means that the slot's + xmin or catalog_xmin + has reached the transaction age specified by + parameter. + + diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index b23d8bbbdad..dda66bca9e9 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)) { /* * Some slots have been invalidated; recalculate the old-segment @@ -8145,6 +8158,8 @@ CreateRestartPoint(int flags) XLogRecPtr endptr; XLogSegNo _logSegNo; TimestampTz xtime; + uint32 slotInvalidationCauses; + TransactionId slotXidLimit; /* Concurrent checkpoint/restartpoint cannot happen */ Assert(!IsUnderPostmaster || MyBackendType == B_CHECKPOINTER); @@ -8323,9 +8338,19 @@ CreateRestartPoint(int flags) INJECTION_POINT("restartpoint-before-slot-invalidation", NULL); - if (InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT, + slotInvalidationCauses = RS_INVAL_WAL_REMOVED | RS_INVAL_IDLE_TIMEOUT; + slotXidLimit = InvalidTransactionId; + if (max_slot_xid_age > 0) + { + slotInvalidationCauses |= RS_INVAL_XID_AGE; + slotXidLimit = TransactionIdRetreatedBy(ReadNextTransactionId(), + max_slot_xid_age); + } + + if (InvalidateObsoleteReplicationSlots(slotInvalidationCauses, _logSegNo, InvalidOid, - InvalidTransactionId)) + InvalidTransactionId, + slotXidLimit)) { /* * Some slots have been invalidated; recalculate the old-segment @@ -9223,6 +9248,7 @@ xlog_redo(XLogReaderState *record) */ InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, 0, InvalidOid, + InvalidTransactionId, InvalidTransactionId); } else if (sync_replication_slots) diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 1a0ff682068..5d873198567 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 whose xmin or catalog_xmin transaction age + * has exceeded this setting; '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. @@ -1792,7 +1799,10 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon, - long slot_idle_seconds) + long slot_idle_seconds, + TransactionId xidLimit, + TransactionId slot_xmin, + TransactionId slot_catalog_xmin) { StringInfoData err_detail; StringInfoData err_hint; @@ -1837,6 +1847,45 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, "idle_replication_slot_timeout"); break; } + + case RS_INVAL_XID_AGE: + { + /* + * The caller passes only the xmin or catalog_xmin that has + * aged past the limit (or both, in the rare case that both + * have), so report whichever is valid. exceeded_by is + * positive because the reported xid precedes xidLimit. Note + * that at least one of them is always valid here. + */ + if (TransactionIdIsValid(slot_xmin)) + { + int32 exceeded_by = (int32) (xidLimit - slot_xmin); + int32 slot_age = (int32) max_slot_xid_age + exceeded_by; + + /* translator: %s is a GUC variable name */ + appendStringInfo(&err_detail, _("The slot's 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); + } + + if (TransactionIdIsValid(slot_catalog_xmin)) + { + int32 exceeded_by = (int32) (xidLimit - slot_catalog_xmin); + int32 slot_age = (int32) max_slot_xid_age + exceeded_by; + + if (err_detail.len > 0) + appendStringInfoChar(&err_detail, ' '); + + /* translator: %s is a GUC variable name */ + appendStringInfo(&err_detail, _("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(); } @@ -1875,6 +1924,26 @@ CanInvalidateIdleSlot(ReplicationSlot *s) !(RecoveryInProgress() && s->data.synced)); } +/* + * Can we invalidate an XID-aged replication slot? + * + * XID age invalidation is allowed only when: + * + * 1. XID age limit is set + * 2. Slot has a valid xmin or catalog_xmin + * 3. The slot is not being synced from the primary while the server is in + * recovery. This is because synced slots are always considered to be + * 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. @@ -1886,7 +1955,10 @@ static ReplicationSlotInvalidationCause DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, - TimestampTz *inactive_since, TimestampTz now) + TimestampTz *inactive_since, TimestampTz now, + TransactionId xidLimit, + TransactionId *slot_xmin, + TransactionId *slot_catalog_xmin) { Assert(possible_causes != RS_INVAL_NONE); @@ -1957,6 +2029,30 @@ 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)); + + /* + * Record whichever of xmin and catalog_xmin has aged past the limit, + * so the invalidation message names the xid that actually triggered + * it. Both can have aged in the rare case of a physical slot that + * also holds a catalog_xmin for cascaded logical decoding. + */ + if (TransactionIdIsValid(s->data.xmin) && + TransactionIdPrecedes(s->data.xmin, xidLimit)) + *slot_xmin = s->data.xmin; + + if (TransactionIdIsValid(s->data.catalog_xmin) && + TransactionIdPrecedes(s->data.catalog_xmin, xidLimit)) + *slot_catalog_xmin = s->data.catalog_xmin; + + if (TransactionIdIsValid(*slot_xmin) || + TransactionIdIsValid(*slot_catalog_xmin)) + return RS_INVAL_XID_AGE; + } + return RS_INVAL_NONE; } @@ -1979,6 +2075,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, + TransactionId xidLimit, bool *released_lock_out) { int last_signaled_pid = 0; @@ -1995,6 +2092,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE; TimestampTz now = 0; long slot_idle_secs = 0; + TransactionId slot_xmin = InvalidTransactionId; + TransactionId slot_catalog_xmin = InvalidTransactionId; Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED)); @@ -2032,7 +2131,10 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, dboid, snapshotConflictHorizon, &inactive_since, - now); + now, + xidLimit, + &slot_xmin, + &slot_catalog_xmin); /* if there's no invalidation, we're done */ if (invalidation_cause == RS_INVAL_NONE) @@ -2124,7 +2226,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReportSlotInvalidation(invalidation_cause, true, active_pid, slotname, restart_lsn, oldestLSN, snapshotConflictHorizon, - slot_idle_secs); + slot_idle_secs, xidLimit, + slot_xmin, slot_catalog_xmin); if (MyBackendType == B_STARTUP) (void) SignalRecoveryConflict(GetPGProcByNumber(active_proc), @@ -2177,7 +2280,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReportSlotInvalidation(invalidation_cause, false, active_pid, slotname, restart_lsn, oldestLSN, snapshotConflictHorizon, - slot_idle_secs); + slot_idle_secs, xidLimit, + slot_xmin, slot_catalog_xmin); /* done with this slot for now */ break; @@ -2204,6 +2308,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: has an xmin or catalog_xmin whose age exceeds the + * configured "max_slot_xid_age". * * Note: This function attempts to invalidate the slot for multiple possible * causes in a single pass, minimizing redundant iterations. The "cause" @@ -2217,7 +2323,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon) + TransactionId snapshotConflictHorizon, + TransactionId xidLimit) { XLogRecPtr oldestLSN; bool invalidated = false; @@ -2226,6 +2333,7 @@ InvalidateObsoleteReplicationSlots(uint32 possible_causes, Assert(!(possible_causes & RS_INVAL_HORIZON) || TransactionIdIsValid(snapshotConflictHorizon)); Assert(!(possible_causes & RS_INVAL_WAL_REMOVED) || oldestSegno > 0); + Assert(!(possible_causes & RS_INVAL_XID_AGE) || TransactionIdIsValid(xidLimit)); Assert(possible_causes != RS_INVAL_NONE); if (max_replication_slots == 0 && max_repack_replication_slots == 0) @@ -2256,6 +2364,7 @@ restart: if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN, dboid, snapshotConflictHorizon, + xidLimit, &released_lock)) { Assert(released_lock); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 7f011e04990..6cbf456459f 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 3c5e16ad1e7..1cf82729135 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 transaction age of a replication slot\'s xmin or catalog_xmin 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 c6311837e67..66fe631fd91 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -361,6 +361,7 @@ #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 # in transaction age; 0 disables #wal_sender_timeout = 60s # in milliseconds; 0 disables #wal_sender_shutdown_timeout = -1 # in milliseconds # -1 disables the timeout and waits for catch-up 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/replication/slot.h b/src/include/replication/slot.h index 9b29444cbca..8c77a61db8e 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 age exceeds the limit */ + RS_INVAL_XID_AGE = (1 << 4), } ReplicationSlotInvalidationCause; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES 4 +#define RS_INVAL_MAX_CAUSES 5 /* * When the slot synchronization worker is running, or when @@ -327,6 +329,7 @@ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT int max_repack_replication_slots; extern PGDLLIMPORT char *synchronized_standby_slots; extern PGDLLIMPORT int idle_replication_slot_timeout_secs; +extern PGDLLIMPORT int max_slot_xid_age; /* management of individual slots */ extern void ReplicationSlotCreate(const char *name, bool db_specific, @@ -364,7 +367,8 @@ extern void ReplicationSlotsDropDBSlots(Oid dboid); extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, - TransactionId snapshotConflictHorizon); + TransactionId snapshotConflictHorizon, + TransactionId xidLimit); extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock); extern int ReplicationSlotIndex(ReplicationSlot *slot); extern bool ReplicationSlotName(int index, Name name); diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl index a412faf51c6..52ff48034c8 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -546,4 +546,137 @@ is( $publisher4->safe_psql( $publisher4->stop; $subscriber4->stop; +# Wait for the given slot to be invalidated due to its xid age +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 $slot_xid_age = 100; + +# Defines a procedure that consumes XIDs, one per committed transaction, to +# age a slot's xmin or catalog_xmin. Created on each test primary below. +my $consume_xid_proc = 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; +}; + +# Tests where a checkpoint or restartpoint invalidates the slot +my $primary5 = PostgreSQL::Test::Cluster->new('primary5'); +$primary5->init(allows_streaming => 'logical'); +$primary5->append_conf( + 'postgresql.conf', qq{ +max_slot_xid_age = $slot_xid_age +autovacuum = off +checkpoint_timeout = 1h +}); +$primary5->start; +$primary5->safe_psql('postgres', $consume_xid_proc); +$primary5->safe_psql('postgres', + "CREATE TABLE tbl_user5 AS SELECT generate_series(1,10) AS a"); +$backup_name = 'backup5'; +$primary5->backup($backup_name); + +my $standby5 = PostgreSQL::Test::Cluster->new('standby5'); +$standby5->init_from_backup($primary5, $backup_name, has_streaming => 1); + +# Testcase 1: an active physical slot (aged xmin) is invalidated by the +# checkpoint, which terminates its owner. A running standby keeps the slot +# active; an open transaction there, reported via feedback, freezes its xmin. +$primary5->safe_psql('postgres', + "SELECT pg_create_physical_replication_slot('sb5_slot_a', true)"); + +$standby5->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb5_slot_a' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$standby5->start; +$primary5->wait_for_catchup($standby5); + +# Confirm streaming works +$primary5->safe_psql('postgres', + "INSERT INTO tbl_user5 SELECT generate_series(11,20)"); +$primary5->wait_for_replay_catchup($standby5); +is( $standby5->safe_psql( + 'postgres', "SELECT count(*) FROM tbl_user5"), + '20', + 'check streamed content on standby'); + +$primary5->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL FROM pg_replication_slots + WHERE slot_name = 'sb5_slot_a'; +]) or die "Timed out waiting for slot sb5_slot_a xmin from HS feedback"; + +# Open a transaction on the standby to pin its reported xmin +my $held = $standby5->background_psql('postgres'); +$held->query_safe("BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT 1;"); + +$primary5->safe_psql('postgres', qq{CALL consume_xid(2 * $slot_xid_age)}); + +# The checkpoint terminates the owner and invalidates the slot +$primary5->safe_psql('postgres', "CHECKPOINT"); +wait_for_xid_aged_invalidation($primary5, 'sb5_slot_a'); +ok(1, "held physical slot invalidated by checkpoint"); + +$held->quit; +$standby5->stop; + +# Testcase 2: an inactive logical slot on a standby (aged catalog_xmin) is +# invalidated by a restartpoint. The age limit is disabled 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_b', true)"); + +# Reuse the same standby, now with the age limit set on it +$standby5->append_conf( + 'postgresql.conf', qq{ +primary_slot_name = 'sb5_slot_b' +hot_standby_feedback = off +max_slot_xid_age = $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 * $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