From 105aa65632eac049b8d7b0952570abd06775e19b Mon Sep 17 00:00:00 2001 From: Joao Foltran Date: Thu, 19 Mar 2026 12:30:06 -0300 Subject: [PATCH v5 1/5] Add auto-revalidation infrastructure for physical replication slots Physical replication slots that are invalidated (e.g., due to WAL removal or idle timeout) currently cannot be reacquired, requiring manual slot recreation. This patch adds the infrastructure for automatic revalidation of physical slots after a standby reconnects and confirms receipt of WAL verified by the current walsender session. A new per-slot persistent field 'auto_revalidate' (default: false) controls whether a physical slot is eligible for revalidation. When enabled, the slot can be acquired despite being invalidated. Revalidation is gated on a per-session target LSN, set once by XLogSendPhysical(): the end of the first WAL data message queued, or the current send position if the standby is already caught up (in which case the confirmed flush position is rechecked immediately, since a caught-up standby may send no further status updates). The invalidation is cleared atomically (under spinlock) with the restart_lsn update only when a flush ACK covers that target. A startpoint check is not sufficient: for wal_removed, invalidation zeroes restart_lsn, and walreceiver ACKs its replay position immediately on connect at a segment-aligned startpoint, so the first ACK can pass a startpoint gate before the session verified it can stream. A boolean "WAL was sent" flag is also insufficient, because a pre-send ACK buffered in the socket can be processed after the flag becomes true. The target is therefore invalid until XLogSendPhysical() establishes it. Only RS_INVAL_WAL_REMOVED and RS_INVAL_IDLE_TIMEOUT are revalidatable via an explicit allowlist, so future invalidation reasons are not automatically eligible. This patch adds the field and revalidation logic but does not yet provide a way to set auto_revalidate=true; that will be added in a subsequent patch. Bump SLOT_VERSION from 5 to 6 for the new persistent field. The TAP test cannot observe, without injection points, that revalidation follows WAL streaming rather than the initial status reply; no timing-based assertions were added. --- src/backend/replication/slot.c | 28 ++++++--- src/backend/replication/walsender.c | 96 ++++++++++++++++++++++++++++- src/include/replication/slot.h | 25 ++++++++ 3 files changed, 141 insertions(+), 8 deletions(-) diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 63ce6d27885..af4f4560e3d 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -141,7 +141,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize #define SLOT_MAGIC 0x1051CA1 /* format identifier */ -#define SLOT_VERSION 5 /* version for new files */ +#define SLOT_VERSION 6 /* version for new files */ /* Control array for replication slot management */ ReplicationSlotCtlData *ReplicationSlotCtl = NULL; @@ -731,12 +731,7 @@ retry: * invalidate the slot immediately after the check. */ if (error_if_invalid && s->data.invalidated != RS_INVAL_NONE) - ereport(ERROR, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("can no longer access replication slot \"%s\"", - NameStr(s->data.name)), - errdetail("This replication slot has been invalidated due to \"%s\".", - GetSlotInvalidationCauseName(s->data.invalidated))); + ReplicationSlotInvalidationError(s); /* Let everybody know we've modified this slot */ ConditionVariableBroadcast(&s->active_cv); @@ -761,6 +756,25 @@ retry: } } +/* + * Report an error when an invalidated replication slot cannot be used. + * + * Keep this separate from ReplicationSlotAcquire() so callers that implement + * their own invalidation policy can use the same error report. + */ +void +ReplicationSlotInvalidationError(ReplicationSlot *slot) +{ + Assert(slot->data.invalidated != RS_INVAL_NONE); + + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("can no longer access replication slot \"%s\"", + NameStr(slot->data.name)), + errdetail("This replication slot has been invalidated due to \"%s\".", + GetSlotInvalidationCauseName(slot->data.invalidated))); +} + /* * Release the replication slot that this backend considers to own. * diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index c65dd324325..eb1f80988b5 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -189,6 +189,16 @@ static XLogRecPtr sendTimeLineValidUpto = InvalidXLogRecPtr; */ static XLogRecPtr sentPtr = InvalidXLogRecPtr; +/* + * WAL position the standby must confirm before this physical streaming + * session may revalidate its slot. Set once per session by + * XLogSendPhysical(): to the end of the first WAL data message queued, or + * to the current send position if the standby is already caught up. + * Invalid until then, so an ACK processed earlier cannot revalidate the + * slot. Kept fixed for the rest of the session. + */ +static XLogRecPtr revalidationTargetPtr = InvalidXLogRecPtr; + /* Buffers for constructing outgoing messages and processing reply messages. */ static StringInfoData output_message; static StringInfoData reply_message; @@ -863,6 +873,8 @@ StartReplication(StartReplicationCmd *cmd) XLogRecPtr FlushPtr; TimeLineID FlushTLI; + revalidationTargetPtr = InvalidXLogRecPtr; + /* create xlogreader for physical replication */ xlogreader = XLogReaderAllocate(wal_segment_size, NULL, @@ -887,12 +899,29 @@ StartReplication(StartReplicationCmd *cmd) if (cmd->slotname) { - ReplicationSlotAcquire(cmd->slotname, true, true); + ReplicationSlotAcquire(cmd->slotname, true, false); if (SlotIsLogical(MyReplicationSlot)) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot use a logical replication slot for physical replication"))); + /* + * Check if the slot is invalidated. Physical slots with + * auto_revalidate can proceed -- they will be revalidated once the + * standby confirms receipt of WAL streamed by this session. All + * other invalidated slots must error out as before. + */ + if (!SlotIsValid(MyReplicationSlot)) + { + if (SlotCanBeRevalidated(MyReplicationSlot)) + ereport(WARNING, + errmsg("replication slot \"%s\" is invalidated due to \"%s\", will attempt revalidation", + NameStr(MyReplicationSlot->data.name), + GetSlotInvalidationCauseName(MyReplicationSlot->data.invalidated))); + else + ReplicationSlotInvalidationError(MyReplicationSlot); + } + /* * We don't need to verify the slot's restart_lsn here; instead we * rely on the caller requesting the starting point to use. If the @@ -2512,6 +2541,7 @@ static void PhysicalConfirmReceivedLocation(XLogRecPtr lsn) { bool changed = false; + bool revalidated = false; ReplicationSlot *slot = MyReplicationSlot; Assert(XLogRecPtrIsValid(lsn)); @@ -2521,6 +2551,26 @@ PhysicalConfirmReceivedLocation(XLogRecPtr lsn) changed = true; slot->data.restart_lsn = lsn; } + + /* + * If the slot is invalidated and eligible for auto-revalidation, clear + * the invalidation only when the standby confirms receipt of all WAL up + * to revalidationTargetPtr: the end of the first WAL data message this + * session queued, or the current send position if the standby was + * already caught up. The target is invalid until XLogSendPhysical() + * establishes it, so an ACK processed earlier cannot revalidate the + * slot. Both restart_lsn and invalidated must be updated under the + * same spinlock so ReplicationSlotsComputeRequiredLSN() sees a + * consistent pair. + */ + if (SlotCanBeRevalidated(slot) && + XLogRecPtrIsValid(revalidationTargetPtr) && + lsn >= revalidationTargetPtr) + { + slot->data.invalidated = RS_INVAL_NONE; + changed = true; + revalidated = true; + } SpinLockRelease(&slot->mutex); if (changed) @@ -2530,6 +2580,21 @@ PhysicalConfirmReceivedLocation(XLogRecPtr lsn) PhysicalWakeupLogicalWalSnd(); } + /* + * Persist the revalidation to disk immediately so the cleared state + * survives a crash. Normal restart_lsn updates are not saved here + * (the comment below explains why), but a revalidation is a significant + * one-time state change worth persisting right away. + */ + if (revalidated) + { + revalidationTargetPtr = InvalidXLogRecPtr; + ReplicationSlotSave(); + ereport(LOG, + errmsg("physical replication slot \"%s\" has been revalidated", + NameStr(slot->data.name))); + } + /* * One could argue that the slot should be saved to disk now, but that'd * be energy wasted - the worst thing lost information could cause here is @@ -3528,6 +3593,25 @@ XLogSendPhysical(void) Assert(sentPtr <= SendRqstPtr); if (SendRqstPtr <= sentPtr) { + /* + * The standby already has all the WAL we could send, so there is + * nothing left for this session to prove by streaming. If the slot + * awaits revalidation, use the current send position as the target. + * The qualifying ACK may have been processed already (e.g. the + * initial status reply), and a caught-up standby may never send + * another one, so check the confirmed flush position right away. + */ + if (MyReplicationSlot != NULL && + SlotCanBeRevalidated(MyReplicationSlot) && + !XLogRecPtrIsValid(revalidationTargetPtr)) + { + revalidationTargetPtr = sentPtr; + + if (XLogRecPtrIsValid(MyWalSnd->flush) && + MyWalSnd->flush >= revalidationTargetPtr) + PhysicalConfirmReceivedLocation(MyWalSnd->flush); + } + WalSndCaughtUp = true; return; } @@ -3643,6 +3727,16 @@ retry: pq_putmessage_noblock(PqMsg_CopyData, output_message.data, output_message.len); + /* + * If the slot is awaiting revalidation, remember the end of the first + * WAL data message we queued. Keep this first target fixed; later + * messages must not move it to a later sentPtr. + */ + if (MyReplicationSlot != NULL && + SlotCanBeRevalidated(MyReplicationSlot) && + !XLogRecPtrIsValid(revalidationTargetPtr)) + revalidationTargetPtr = endptr; + sentPtr = endptr; /* Update shared memory status */ diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 9b29444cbca..4b93538ba63 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -159,6 +159,13 @@ typedef struct ReplicationSlotPersistentData * for logical slots on the primary server. */ bool failover; + + /* + * If true, an invalidated physical slot may be automatically revalidated + * once the standby reconnects and confirms WAL receipt (flush ACK). + * Only applicable to physical slots; ignored for logical slots. + */ + bool auto_revalidate; } ReplicationSlotPersistentData; /* @@ -286,6 +293,23 @@ typedef struct ReplicationSlot #define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid) #define SlotIsLogical(slot) ((slot)->data.database != InvalidOid) +#define SlotIsValid(slot) ((slot)->data.invalidated == RS_INVAL_NONE) + +/* + * Can this slot be automatically revalidated? + * + * Only physical slots with auto_revalidate enabled and invalidated by + * an explicitly supported reason are eligible. New invalidation reasons + * must be added here to become revalidatable. + */ +static inline bool +SlotCanBeRevalidated(ReplicationSlot *s) +{ + return SlotIsPhysical(s) && + s->data.auto_revalidate && + (s->data.invalidated == RS_INVAL_WAL_REMOVED || + s->data.invalidated == RS_INVAL_IDLE_TIMEOUT); +} /* * Shared memory control area for all of replication slots. @@ -341,6 +365,7 @@ extern void ReplicationSlotAlter(const char *name, const bool *failover, extern void ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid); +pg_noreturn extern void ReplicationSlotInvalidationError(ReplicationSlot *slot); extern void ReplicationSlotRelease(void); extern void ReplicationSlotCleanup(bool synced_only); extern void ReplicationSlotSave(void); -- 2.50.1 (Apple Git-155)