From c19dd30a46379f8030ea41357c74bc4734e17083 Mon Sep 17 00:00:00 2001 From: John Hsu Date: Tue, 4 Jun 2024 20:27:47 +0000 Subject: [PATCH] Introduce a new guc 'standby_slot_names_from_syncrep' If synchronous replication is enabled, this patch allows logical subscribers to wait for changes to be replicated to synchronous replicas before consuming the changes. In the event that both 'standby_slot_names' and 'standby_slot_names_from_syncrep' are set, the former takes precedence. --- src/backend/replication/slot.c | 261 ++++++++++-------- src/backend/replication/syncrep.c | 11 +- src/backend/utils/misc/guc_tables.c | 14 + src/include/replication/slot.h | 1 + src/include/replication/syncrep.h | 2 + .../t/040_standby_failover_slots_sync.pl | 137 +++++++++ 6 files changed, 313 insertions(+), 113 deletions(-) diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 0e54ea5bb9..b5b69a0d6f 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -49,6 +49,7 @@ #include "postmaster/interrupt.h" #include "replication/slotsync.h" #include "replication/slot.h" +#include "replication/syncrep.h" #include "replication/walsender_private.h" #include "storage/fd.h" #include "storage/ipc.h" @@ -146,6 +147,7 @@ int max_replication_slots = 10; /* the maximum number of replication * logical WAL sender processes will wait for. */ char *standby_slot_names; +bool standby_slot_names_from_syncrep; /* This is the parsed and cached configuration for standby_slot_names */ static StandbySlotNamesConfigData *standby_slot_names_config; @@ -2572,8 +2574,8 @@ SlotExistsInStandbySlotNames(const char *slot_name) } /* - * Return true if the slots specified in standby_slot_names have caught up to - * the given WAL location, false otherwise. + * Return true if the slots specified in standby_slot_names or synchronous_commit settings, + * have caught up to the given WAL location, false otherwise. * * The elevel parameter specifies the error level used for logging messages * related to slots that do not exist, are invalidated, or are inactive. @@ -2587,9 +2589,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel) /* * Don't need to wait for the standbys to catch up if there is no value in - * standby_slot_names. + * standby_slot_names and not waiting for synchronous replication. */ - if (standby_slot_names_config == NULL) + if (standby_slot_names_config == NULL && !standby_slot_names_from_syncrep) return true; /* @@ -2600,136 +2602,172 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel) return true; /* - * Don't need to wait for the standbys to catch up if they are already - * beyond the specified WAL location. - */ - if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) && - ss_oldest_flush_lsn >= wait_for_lsn) - return true; - - /* - * To prevent concurrent slot dropping and creation while filtering the - * slots, take the ReplicationSlotControlLock outside of the loop. + * In the event that both standby_slot_names_config and standby_slot_names_from_syncrep is enabled, + * have the former take precedence. */ - LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); - - name = standby_slot_names_config->slot_names; - for (int i = 0; i < standby_slot_names_config->nslotnames; i++) + if (standby_slot_names_config != NULL) { - XLogRecPtr restart_lsn; - bool invalidated; - bool inactive; - ReplicationSlot *slot; + /* + * Don't need to wait for the standbys to catch up if they are already + * beyond the specified WAL location. + */ + if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) && + ss_oldest_flush_lsn >= wait_for_lsn) + return true; - slot = SearchNamedReplicationSlot(name, false); + /* + * To prevent concurrent slot dropping and creation while filtering the + * slots, take the ReplicationSlotControlLock outside of the loop. + */ + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); - if (!slot) + name = standby_slot_names_config->slot_names; + for (int i = 0; i < standby_slot_names_config->nslotnames; i++) { - /* - * If a slot name provided in standby_slot_names does not exist, - * report a message and exit the loop. A user can specify a slot - * name that does not exist just before the server startup. The - * GUC check_hook(validate_standby_slots) cannot validate such a - * slot during startup as the ReplicationSlotCtl shared memory is - * not initialized at that time. It is also possible for a user to - * drop the slot in standby_slot_names afterwards. - */ - ereport(elevel, - errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("replication slot \"%s\" specified in parameter %s does not exist", - name, "standby_slot_names"), - errdetail("Logical replication is waiting on the standby associated with \"%s\".", - name), - errhint("Consider creating the slot \"%s\" or amend parameter %s.", - name, "standby_slot_names")); - break; - } + XLogRecPtr restart_lsn; + bool invalidated; + bool inactive; + ReplicationSlot *slot; - if (SlotIsLogical(slot)) - { - /* - * If a logical slot name is provided in standby_slot_names, - * report a message and exit the loop. Similar to the non-existent - * case, a user can specify a logical slot name in - * standby_slot_names before the server startup, or drop an - * existing physical slot and recreate a logical slot with the - * same name. - */ - ereport(elevel, - errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("cannot have logical replication slot \"%s\" in parameter %s", - name, "standby_slot_names"), - errdetail("Logical replication is waiting for correction on \"%s\".", - name), - errhint("Consider removing logical slot \"%s\" from parameter %s.", - name, "standby_slot_names")); - break; - } + slot = SearchNamedReplicationSlot(name, false); - SpinLockAcquire(&slot->mutex); - restart_lsn = slot->data.restart_lsn; - invalidated = slot->data.invalidated != RS_INVAL_NONE; - inactive = slot->active_pid == 0; - SpinLockRelease(&slot->mutex); + if (!slot) + { + /* + * If a slot name provided in standby_slot_names does not exist, + * report a message and exit the loop. A user can specify a slot + * name that does not exist just before the server startup. The + * GUC check_hook(validate_standby_slots) cannot validate such a + * slot during startup as the ReplicationSlotCtl shared memory is + * not initialized at that time. It is also possible for a user to + * drop the slot in standby_slot_names afterwards. + */ + ereport(elevel, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("replication slot \"%s\" specified in parameter %s does not exist", + name, "standby_slot_names"), + errdetail("Logical replication is waiting on the standby associated with \"%s\".", + name), + errhint("Consider creating the slot \"%s\" or amend parameter %s.", + name, "standby_slot_names")); + break; + } - if (invalidated) - { - /* Specified physical slot has been invalidated */ - ereport(elevel, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("physical slot \"%s\" specified in parameter %s has been invalidated", - name, "standby_slot_names"), - errdetail("Logical replication is waiting on the standby associated with \"%s\".", - name), - errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.", - name, "standby_slot_names")); - break; - } + if (SlotIsLogical(slot)) + { + /* + * If a logical slot name is provided in standby_slot_names, + * report a message and exit the loop. Similar to the non-existent + * case, a user can specify a logical slot name in + * standby_slot_names before the server startup, or drop an + * existing physical slot and recreate a logical slot with the + * same name. + */ + ereport(elevel, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot have logical replication slot \"%s\" in parameter %s", + name, "standby_slot_names"), + errdetail("Logical replication is waiting for correction on \"%s\".", + name), + errhint("Consider removing logical slot \"%s\" from parameter %s.", + name, "standby_slot_names")); + break; + } - if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn) - { - /* Log a message if no active_pid for this physical slot */ - if (inactive) + SpinLockAcquire(&slot->mutex); + restart_lsn = slot->data.restart_lsn; + invalidated = slot->data.invalidated != RS_INVAL_NONE; + inactive = slot->active_pid == 0; + SpinLockRelease(&slot->mutex); + + if (invalidated) + { + /* Specified physical slot has been invalidated */ ereport(elevel, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid", - name, "standby_slot_names"), + errmsg("physical slot \"%s\" specified in parameter %s has been invalidated", + name, "standby_slot_names"), errdetail("Logical replication is waiting on the standby associated with \"%s\".", - name), - errhint("Consider starting standby associated with \"%s\" or amend parameter %s.", + name), + errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.", name, "standby_slot_names")); + break; + } - /* Continue if the current slot hasn't caught up. */ - break; + if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn) + { + /* Log a message if no active_pid for this physical slot */ + if (inactive) + ereport(elevel, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid", + name, "standby_slot_names"), + errdetail("Logical replication is waiting on the standby associated with \"%s\".", + name), + errhint("Consider starting standby associated with \"%s\" or amend parameter %s.", + name, "standby_slot_names")); + + /* Continue if the current slot hasn't caught up. */ + break; + } + + Assert(restart_lsn >= wait_for_lsn); + + if (XLogRecPtrIsInvalid(min_restart_lsn) || + min_restart_lsn > restart_lsn) + min_restart_lsn = restart_lsn; + + caught_up_slot_num++; + + name += strlen(name) + 1; } - Assert(restart_lsn >= wait_for_lsn); + LWLockRelease(ReplicationSlotControlLock); - if (XLogRecPtrIsInvalid(min_restart_lsn) || - min_restart_lsn > restart_lsn) - min_restart_lsn = restart_lsn; + /* + * Return false if not all the standbys have caught up to the specified + * WAL location. + */ + if (caught_up_slot_num != standby_slot_names_config->nslotnames) + return false; + + /* The ss_oldest_flush_lsn must not retreat. */ + Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) || + min_restart_lsn >= ss_oldest_flush_lsn); - caught_up_slot_num++; + ss_oldest_flush_lsn = min_restart_lsn; - name += strlen(name) + 1; + return true; } + else + { + volatile WalSndCtlData *walsndctl = WalSndCtl; + static XLogRecPtr lsn[NUM_SYNC_REP_WAIT_MODE] = {InvalidXLogRecPtr}; + int mode = SyncRepWaitMode; + int i; - LWLockRelease(ReplicationSlotControlLock); + /* + * Don't wait if synchronous_commit is not configured properly + */ + if (mode == SYNC_REP_NO_WAIT) + return true; - /* - * Return false if not all the standbys have caught up to the specified - * WAL location. - */ - if (caught_up_slot_num != standby_slot_names_config->nslotnames) - return false; + if (!XLogRecPtrIsInvalid(lsn[mode]) && lsn[mode] >= wait_for_lsn) + return true; - /* The ss_oldest_flush_lsn must not retreat. */ - Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) || - min_restart_lsn >= ss_oldest_flush_lsn); + LWLockAcquire(SyncRepLock, LW_SHARED); + for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++) + { + lsn[i] = walsndctl->lsn[i]; + } - ss_oldest_flush_lsn = min_restart_lsn; + LWLockRelease(SyncRepLock); - return true; + if (!XLogRecPtrIsInvalid(lsn[mode]) && lsn[mode] >= wait_for_lsn) + return true; + + return false; + } } /* @@ -2744,13 +2782,12 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn) /* * Don't need to wait for the standby to catch up if the current acquired * slot is not a logical failover slot, or there is no value in - * standby_slot_names. + * standby_slot_names and standby_slot_names_from_syncrep. */ - if (!MyReplicationSlot->data.failover || !standby_slot_names_config) + if (!MyReplicationSlot->data.failover || !(standby_slot_names_config || standby_slot_names_from_syncrep)) return; ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv); - for (;;) { CHECK_FOR_INTERRUPTS(); diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index fa5988c824..0b85f3c3b8 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -78,6 +78,7 @@ #include "common/int.h" #include "miscadmin.h" #include "pgstat.h" +#include "replication/slot.h" #include "replication/syncrep.h" #include "replication/walsender.h" #include "replication/walsender_private.h" @@ -95,7 +96,7 @@ char *SyncRepStandbyNames; static bool announce_next_takeover = true; SyncRepConfigData *SyncRepConfig = NULL; -static int SyncRepWaitMode = SYNC_REP_NO_WAIT; +int SyncRepWaitMode = SYNC_REP_NO_WAIT; static void SyncRepQueueInsert(int mode); static void SyncRepCancelWait(void); @@ -523,6 +524,14 @@ SyncRepReleaseWaiters(void) LWLockRelease(SyncRepLock); + /* + * Only wake if standby_slot_names_from_syncrep is the only value set + */ + if (standby_slot_names_from_syncrep && strcmp(standby_slot_names, "") == 0) + { + ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv); + } + elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X", numwrite, LSN_FORMAT_ARGS(writePtr), numflush, LSN_FORMAT_ARGS(flushPtr), diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 46c258be28..eb4ec4800c 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2025,6 +2025,20 @@ struct config_bool ConfigureNamesBool[] = NULL, NULL, NULL }, + { + {"standby_slot_names_from_syncrep", PGC_SIGHUP, REPLICATION_PRIMARY, + gettext_noop("Logical WAL sender processes will wait on the current" + "synchronous replication settings."), + gettext_noop("Logical WAL sender processes will send decoded " + "changes to plugins only after the changes have " + "been synchronously replicated."), + GUC_NOT_IN_SAMPLE + }, + &standby_slot_names_from_syncrep, + false, + NULL, NULL, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, false, NULL, NULL, NULL diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 1bc80960ef..77052cb6a5 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -230,6 +230,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; /* GUCs */ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; +extern PGDLLIMPORT bool standby_slot_names_from_syncrep; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h index ea439e6da6..fb783576b5 100644 --- a/src/include/replication/syncrep.h +++ b/src/include/replication/syncrep.h @@ -96,6 +96,8 @@ extern int SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys); /* called by checkpointer */ extern void SyncRepUpdateSyncStandbysDefined(void); +extern int SyncRepWaitMode; + /* * Internal functions for parsing synchronous_standby_names grammar, * in syncrep_gram.y and syncrep_scanner.l diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl index 3b6dddba71..94f3a5b54c 100644 --- a/src/test/recovery/t/040_standby_failover_slots_sync.pl +++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl @@ -710,6 +710,143 @@ $result = $subscriber1->safe_psql('postgres', is($result, 't', "subscriber1 gets data from primary after standby1 acknowledges changes"); +################################################## +# Test that logical failover replication slots wait for the specified +# synchronous replicas to receive the changes first. It uses the +# following set up: +# +# (synchronous physical standbys) +# | ----> standby1 (application_name = standby1) +# | ----> standby2 (application_name = standby2) +# primary ----- | +# (logical replication) +# | ----> subscriber1 (failover = true, slot_name = lsub1_slot) +# | ----> subscriber2 (failover = false, slot_name = lsub2_slot) +# +# synchronous_commit = 'on' +# synchronous_standby_names = 'ANY 2 (standby1, standby2)' +# standby_slot_names_from_syncrep = 'true' +# +# The setup is configured in such a way that the logical slot of subscriber1 is +# enabled for failover, and thus the subscriber1 will wait for the changes to have +# been synchronously replicated before receiving the decoded changes. +################################################## + +$primary->safe_psql('postgres', "TRUNCATE tab_int;"); +# Setup synchronous replication +$primary->append_conf( + 'postgresql.conf', qq( + synchronous_commit = 'on' + synchronous_standby_names = 'ANY 2 (standby1, standby2)' + standby_slot_names_from_syncrep = 'true' +)); + +$primary->reload; + +$standby1->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby1'"); +$standby1->reload; + +$standby2->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby2'"); +$standby2->reload; + +# Check that synchronous replication is setup properly +$standby2->stop; + +# Create some data on the primary +$primary_row_count = 10; + +my $sync_back_q = $primary->background_psql( + 'postgres', + on_error_stop => 0, + timeout => $PostgreSQL::Test::Utils::timeout_default); + +$sync_back_q->query_until(qr/insert_blocked_on_sync_rep/, q( + \echo insert_blocked_on_sync_rep + INSERT INTO tab_int SELECT generate_series(1, 10); +)); + +$result = $primary->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); + +is($result, 'f', "primary row count is not updated due to synchronous replication"); + +# Verify the standby specified in standby_slot_names (sb1_slot aka standby1) +# catches up with the primary. +$primary->wait_for_replay_catchup($standby1); +$result = $standby1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', "standby1 gets data from primary"); + +# Validate that standby_slot_names overrides standby_slot_names_from_syncrep +# since the slot specified (sb1_slot) has received the changes, primary can send +# the decoded changes to the subscription enabled for failover +# (i.e. regress_mysub1). +$primary->wait_for_catchup('regress_mysub1'); +$result = $subscriber1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', + "subscriber1 gets data from primary after standby1 acknowledges changes, overriding standby_slot_names_from_syncrep"); + +$standby2->start; +$primary->wait_for_replay_catchup($standby2); + +# Unset standby_slot_names to test standby_slot_names_from_syncrep +# blocks primary from sending logical decoded changes to failover slots until +# changes have been synchronously replicated. +$primary->append_conf( + 'postgresql.conf', qq( + standby_slot_names = '')); +$primary->reload; + +$primary_row_count = 20; +$standby2->stop; + +$sync_back_q->query_until( + qr/insert_blocked_on_additional_sync_rep/, q( + \echo insert_blocked_on_additional_sync_rep + INSERT INTO tab_int SELECT generate_series(11, 20); +)); + +# Since $standby2 has not received the changes, validate that subscriber1 (failover = true) +# has not received the decoded changes, but subscriber2 (failover = false) has. +$primary->wait_for_catchup('regress_mysub2'); + +$result = $subscriber2->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', + "subscriber2 gets data from primary even if the changes have not been synchronously acknowledged."); + +$result = $subscriber1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 'f', + "subscriber1 does not get data from primary since changes have not been synchronously acknowledged."); + +# Start standby2 to allow the changes to be acknowledged by all the synchronous standbys. +$standby2->start; +$primary->wait_for_replay_catchup($standby2); +$primary->wait_for_catchup('regress_mysub1'); + +# Now that the changes have been replicated to all synchronous nodes, +# primary can send the decoded changes to the subscription enabled for failover +# (i.e. regress_mysub1). While the standby was down, regress_mysub1 didn't +# receive any data from the primary. i.e. the primary didn't allow it to go +# ahead of requirements of synchronous commit. +$result = $subscriber1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', + "subscriber1 gets data from primary since changes have been syhcronously acknowledged."); +# No longer need to run the background session. +$sync_back_q->quit; + +# Reset standby_slot_names and synchronous commit for below test cases +$primary->append_conf( + 'postgresql.conf', qq( +standby_slot_names = 'sb1_slot' +synchronous_standby_names = '' +standby_slot_names_from_syncrep = 'false' +)); +$primary->reload; + ################################################## # Verify that when using pg_logical_slot_get_changes to consume changes from a # logical failover slot, it will also wait for the slots specified in -- 2.40.1