From 53960949d0d49560d9af7e14fc183e8bbf1a1b28 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Tue, 19 Mar 2024 18:38:52 +0000 Subject: [PATCH v12 7/7] Add XID age based replication slot invalidation Up until now, postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set an XID age (age of slot's xmin or catalog_xmin) of say 1 or 1.5 billion, after which the slots get invalidated. To achieve the above, postgres uses replication slot 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), and a new GUC max_slot_xid_age. The checkpointer then looks at all replication slots invalidating the slots based on the age set. --- doc/src/sgml/config.sgml | 21 +++++ doc/src/sgml/system-views.sgml | 8 ++ src/backend/access/transam/xlog.c | 10 ++ src/backend/replication/slot.c | 49 +++++++++- src/backend/utils/misc/guc_tables.c | 10 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/t/050_invalidate_slots.pl | 92 +++++++++++++++++++ 8 files changed, 193 insertions(+), 1 deletion(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 65a6e6c408..6dd54ffcb7 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4544,6 +4544,27 @@ 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. 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. + + + + track_commit_timestamp (boolean) diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index 9821c6f77a..4400274b04 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2595,6 +2595,14 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx inactive_timeout parameter. + + + xid_aged means that the slot's + xmin or catalog_xmin + has reached the age specified by + parameter. + + diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index ea4ece22de..3e61617abc 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7172,6 +7172,11 @@ CreateCheckPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, InvalidOid, InvalidTransactionId); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Make more log segments if needed. (Do this after recycling old log * segments, since that may supply some of the needed files.) @@ -7643,6 +7648,11 @@ CreateRestartPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, InvalidOid, InvalidTransactionId); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Make more log segments if needed. (Do this after recycling old log * segments, since that may supply some of the needed files.) diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 26121e939d..8999eb23c9 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -108,10 +108,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", [RS_INVAL_INACTIVE_TIMEOUT] = "inactive_timeout", + [RS_INVAL_XID_AGE] = "xid_aged", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_INACTIVE_TIMEOUT +#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -141,6 +142,7 @@ ReplicationSlot *MyReplicationSlot = NULL; /* GUC variables */ int max_replication_slots = 10; /* the maximum number of replication * slots */ +int max_slot_xid_age = 0; /* * This GUC lists streaming replication standby server slot names that @@ -159,6 +161,7 @@ static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr; static void ReplicationSlotShmemExit(int code, Datum arg); static void ReplicationSlotDropPtr(ReplicationSlot *slot); +static bool IsSlotXIDAged(TransactionId xmin); /* internal persistency functions */ static void RestoreSlotFromDisk(const char *name); @@ -1554,6 +1557,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_INACTIVE_TIMEOUT: appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by slot's inactive_timeout parameter.")); break; + case RS_INVAL_XID_AGE: + appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1570,6 +1576,31 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, pfree(err_detail.data); } +/* + * Returns true if slot's passed in xmin/catalog_xmin age is more than + * max_slot_xid_age. + */ +static bool +IsSlotXIDAged(TransactionId xmin) +{ + TransactionId xid_cur; + TransactionId xid_limit; + + if (!TransactionIdIsNormal(xmin)) + return false; + + xid_cur = ReadNextTransactionId(); + xid_limit = xmin + max_slot_xid_age; + + if (xid_limit < FirstNormalTransactionId) + xid_limit += FirstNormalTransactionId; + + if (TransactionIdFollowsOrEquals(xid_cur, xid_limit)) + return true; + + return false; +} + /* * Helper for InvalidateObsoleteReplicationSlots * @@ -1685,6 +1716,21 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, conflict = cause; } break; + case RS_INVAL_XID_AGE: + { + if (IsSlotXIDAged(s->data.xmin)) + { + conflict = cause; + break; + } + + if (IsSlotXIDAged(s->data.catalog_xmin)) + { + conflict = cause; + break; + } + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1839,6 +1885,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs + * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 57d9de4dd9..056533a059 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2943,6 +2943,16 @@ struct config_int ConfigureNamesInt[] = check_max_slot_wal_keep_size, NULL, NULL }, + { + {"max_slot_xid_age", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Age of the transaction ID at which a replication slot gets invalidated."), + gettext_noop("The transaction is the oldest transaction (including the one affecting the system catalogs) that a replication slot needs the database to retain.") + }, + &max_slot_xid_age, + 0, 0, 2000000000, + NULL, NULL, NULL + }, + { {"wal_sender_timeout", PGC_USERSET, REPLICATION_SENDING, gettext_noop("Sets the maximum time to wait for WAL replication."), diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 2244ee52f7..b4c928b826 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -334,6 +334,7 @@ #wal_sender_timeout = 60s # in milliseconds; 0 disables #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) +#max_slot_xid_age = 0 # - Primary Server - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 2efb9490c1..385c96aa09 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -55,6 +55,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_WAL_LEVEL, /* inactive slot timeout has occurred */ RS_INVAL_INACTIVE_TIMEOUT, + /* slot's xmin or catalog_xmin has reached the age */ + RS_INVAL_XID_AGE, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -235,6 +237,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; /* GUCs */ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; +extern PGDLLIMPORT int max_slot_xid_age; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl index d046e1d5d7..6eb2dcd5b5 100644 --- a/src/test/recovery/t/050_invalidate_slots.pl +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -165,4 +165,96 @@ ok($invalidated, 'check that slot lsub1_slot invalidation has been logged'); # Testcase end: Invalidate logical subscriber's slot due to inactive_timeout # ============================================================================= +# ============================================================================= +# Testcase start: Invalidate streaming standby's slot due to max_slot_xid_age +# + +# Create a standby linking to the primary using the replication slot +my $standby2 = PostgreSQL::Test::Cluster->new('standby2'); +$standby2->init_from_backup($primary, $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. +$standby2->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb2_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); + +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot(slot_name := 'sb2_slot'); +]); + +$standby2->start; + +# Create some content on primary to move xmin +$primary->safe_psql('postgres', + "CREATE TABLE tab_int AS SELECT generate_series(1,10) AS a"); + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby2); + +$primary->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'sb2_slot'; +]) or die "Timed out waiting for slot xmin to advance"; + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 500; +]); +$primary->reload; + +# Stop standby to make the replication slot's xmin on primary to age +$standby2->stop; + +$logstart = -s $primary->logfile; + +# Do some work to advance xmin +$primary->safe_psql( + 'postgres', q{ +do $$ +begin + for i in 10000..11000 loop + -- use an exception block so that each iteration eats an XID + begin + insert into tab_int values (i); + exception + when division_by_zero then null; + end; + end loop; +end$$; +}); + +$invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb2_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb2_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb2_slot' AND + invalidation_reason = 'xid_aged'; +]) + or die + "Timed out while waiting for replication slot sb2_slot to be invalidated"; + +# Testcase end: Invalidate streaming standby's slot due to max_slot_xid_age +# ============================================================================= + done_testing(); -- 2.34.1