From 795a0c837329ec376f1214c4278ca95a2378e8f3 Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Mon, 18 Nov 2024 16:13:26 +0530 Subject: [PATCH v49 1/2] Add error handling while acquiring a replication slot In ReplicationSlotAcquire(), raise an error for invalid slots if caller specify error_if_invalid=true. --- .../replication/logical/logicalfuncs.c | 2 +- src/backend/replication/logical/slotsync.c | 4 +- src/backend/replication/slot.c | 48 +++++++++++++++++-- src/backend/replication/slotfuncs.c | 2 +- src/backend/replication/walsender.c | 4 +- src/backend/utils/adt/pg_upgrade_support.c | 2 +- src/include/replication/slot.h | 3 +- src/test/recovery/t/019_replslot_limit.pl | 2 +- 8 files changed, 55 insertions(+), 12 deletions(-) diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c index b4dd5cce75..56fc1a45a9 100644 --- a/src/backend/replication/logical/logicalfuncs.c +++ b/src/backend/replication/logical/logicalfuncs.c @@ -197,7 +197,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin else end_of_wal = GetXLogReplayRecPtr(NULL); - ReplicationSlotAcquire(NameStr(*name), true); + ReplicationSlotAcquire(NameStr(*name), true, true); PG_TRY(); { diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index d62186a510..69a32422bf 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -446,7 +446,7 @@ drop_local_obsolete_slots(List *remote_slot_list) if (synced_slot) { - ReplicationSlotAcquire(NameStr(local_slot->data.name), true); + ReplicationSlotAcquire(NameStr(local_slot->data.name), true, false); ReplicationSlotDropAcquired(); } @@ -665,7 +665,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid) * pre-check to ensure that at least one of the slot properties is * changed before acquiring the slot. */ - ReplicationSlotAcquire(remote_slot->name, true); + ReplicationSlotAcquire(remote_slot->name, true, false); Assert(slot == MyReplicationSlot); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 6828100cf1..0cbfd4fbaf 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -535,9 +535,12 @@ ReplicationSlotName(int index, Name name) * * An error is raised if nowait is true and the slot is currently in use. If * nowait is false, we sleep until the slot is released by the owning process. + * + * An error is raised if error_if_invalid is true and the slot has been + * invalidated previously. */ void -ReplicationSlotAcquire(const char *name, bool nowait) +ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid) { ReplicationSlot *s; int active_pid; @@ -615,6 +618,45 @@ retry: /* We made this slot active, so it's ours now. */ MyReplicationSlot = s; + /* + * An error is raised if error_if_invalid is true and the slot has been + * previously invalidated. + */ + if (error_if_invalid && + s->data.invalidated != RS_INVAL_NONE) + { + StringInfoData err_detail; + + initStringInfo(&err_detail); + appendStringInfo(&err_detail, _("This slot has been invalidated because ")); + + switch (s->data.invalidated) + { + case RS_INVAL_WAL_REMOVED: + appendStringInfo(&err_detail, _("the required WAL has been removed.")); + break; + + case RS_INVAL_HORIZON: + appendStringInfo(&err_detail, _("the required rows have been removed.")); + break; + + case RS_INVAL_WAL_LEVEL: + appendStringInfo(&err_detail, _("wal_level is insufficient for slot.")); + break; + + case RS_INVAL_NONE: + pg_unreachable(); + } + + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("can no longer get changes from replication slot \"%s\"", + NameStr(s->data.name)), + errdetail_internal("%s", err_detail.data)); + + pfree(err_detail.data); + } + /* * The call to pgstat_acquire_replslot() protects against stats for a * different slot, from before a restart or such, being present during @@ -785,7 +827,7 @@ ReplicationSlotDrop(const char *name, bool nowait) { Assert(MyReplicationSlot == NULL); - ReplicationSlotAcquire(name, nowait); + ReplicationSlotAcquire(name, nowait, false); /* * Do not allow users to drop the slots which are currently being synced @@ -812,7 +854,7 @@ ReplicationSlotAlter(const char *name, const bool *failover, Assert(MyReplicationSlot == NULL); Assert(failover || two_phase); - ReplicationSlotAcquire(name, false); + ReplicationSlotAcquire(name, false, false); if (SlotIsPhysical(MyReplicationSlot)) ereport(ERROR, diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index 488a161b3e..578cff64c8 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -536,7 +536,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS) moveto = Min(moveto, GetXLogReplayRecPtr(NULL)); /* Acquire the slot so we "own" it */ - ReplicationSlotAcquire(NameStr(*slotname), true); + ReplicationSlotAcquire(NameStr(*slotname), true, true); /* A slot whose restart_lsn has never been reserved cannot be advanced */ if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 371eef3ddd..b36ae90b2c 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -816,7 +816,7 @@ StartReplication(StartReplicationCmd *cmd) if (cmd->slotname) { - ReplicationSlotAcquire(cmd->slotname, true); + ReplicationSlotAcquire(cmd->slotname, true, true); if (SlotIsLogical(MyReplicationSlot)) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), @@ -1434,7 +1434,7 @@ StartLogicalReplication(StartReplicationCmd *cmd) Assert(!MyReplicationSlot); - ReplicationSlotAcquire(cmd->slotname, true); + ReplicationSlotAcquire(cmd->slotname, true, true); /* * Force a disconnect, so that the decoding code doesn't need to care diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c index 8a45b5827e..3322848e03 100644 --- a/src/backend/utils/adt/pg_upgrade_support.c +++ b/src/backend/utils/adt/pg_upgrade_support.c @@ -298,7 +298,7 @@ binary_upgrade_logical_slot_has_caught_up(PG_FUNCTION_ARGS) slot_name = PG_GETARG_NAME(0); /* Acquire the given slot */ - ReplicationSlotAcquire(NameStr(*slot_name), true); + ReplicationSlotAcquire(NameStr(*slot_name), true, false); Assert(SlotIsLogical(MyReplicationSlot)); diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 45582cf9d8..2c325fd942 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -249,7 +249,8 @@ extern void ReplicationSlotDropAcquired(void); extern void ReplicationSlotAlter(const char *name, const bool *failover, const bool *two_phase); -extern void ReplicationSlotAcquire(const char *name, bool nowait); +extern void ReplicationSlotAcquire(const char *name, bool nowait, + bool error_if_invalid); extern void ReplicationSlotRelease(void); extern void ReplicationSlotCleanup(bool synced_only); extern void ReplicationSlotSave(void); diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl index efb4ba3af1..333e040e7f 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -234,7 +234,7 @@ my $failed = 0; for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) { if ($node_standby->log_contains( - "requested WAL segment [0-9A-F]+ has already been removed", + "This slot has been invalidated because the required WAL has been removed", $logstart)) { $failed = 1; -- 2.34.1