agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Andres Freund <[email protected]>
Subject: [PATCH va65 1/9] replication slots: replace invalidated_at LSN with an enum
Date: Thu, 6 Apr 2023 20:00:07 -0700
---
src/include/replication/slot.h | 15 +++++++++++++--
src/backend/replication/slot.c | 21 ++++++++++++++++++---
src/backend/replication/slotfuncs.c | 8 +++-----
3 files changed, 34 insertions(+), 10 deletions(-)
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 8872c80cdfe..793f0701b88 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -37,6 +37,17 @@ typedef enum ReplicationSlotPersistency
RS_TEMPORARY
} ReplicationSlotPersistency;
+/*
+ * Slots can be invalidated, e.g. due to max_slot_wal_keep_size. If so, the
+ * 'invalidated' field is set to a value other than _NONE.
+ */
+typedef enum ReplicationSlotInvalidationCause
+{
+ RS_INVAL_NONE,
+ /* required WAL has been removed */
+ RS_INVAL_WAL,
+} ReplicationSlotInvalidationCause;
+
/*
* On-Disk data of a replication slot, preserved across restarts.
*/
@@ -72,8 +83,8 @@ typedef struct ReplicationSlotPersistentData
/* oldest LSN that might be required by this replication slot */
XLogRecPtr restart_lsn;
- /* restart_lsn is copied here when the slot is invalidated */
- XLogRecPtr invalidated_at;
+ /* RS_INVAL_NONE if valid, or the reason for having been invalidated */
+ ReplicationSlotInvalidationCause invalidated;
/*
* Oldest LSN that the client has acked receipt for. This is used as the
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 2293c0c6fc3..06ff3559dd1 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -855,8 +855,7 @@ ReplicationSlotsComputeRequiredXmin(bool already_locked)
SpinLockAcquire(&s->mutex);
effective_xmin = s->effective_xmin;
effective_catalog_xmin = s->effective_catalog_xmin;
- invalidated = (!XLogRecPtrIsInvalid(s->data.invalidated_at) &&
- XLogRecPtrIsInvalid(s->data.restart_lsn));
+ invalidated = s->data.invalidated != RS_INVAL_NONE;
SpinLockRelease(&s->mutex);
/* invalidated slots need not apply */
@@ -901,14 +900,20 @@ ReplicationSlotsComputeRequiredLSN(void)
{
ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
XLogRecPtr restart_lsn;
+ bool invalidated;
if (!s->in_use)
continue;
SpinLockAcquire(&s->mutex);
restart_lsn = s->data.restart_lsn;
+ invalidated = s->data.invalidated != RS_INVAL_NONE;
SpinLockRelease(&s->mutex);
+ /* invalidated slots need not apply */
+ if (invalidated)
+ continue;
+
if (restart_lsn != InvalidXLogRecPtr &&
(min_required == InvalidXLogRecPtr ||
restart_lsn < min_required))
@@ -946,6 +951,7 @@ ReplicationSlotsComputeLogicalRestartLSN(void)
{
ReplicationSlot *s;
XLogRecPtr restart_lsn;
+ bool invalidated;
s = &ReplicationSlotCtl->replication_slots[i];
@@ -960,8 +966,13 @@ ReplicationSlotsComputeLogicalRestartLSN(void)
/* read once, it's ok if it increases while we're checking */
SpinLockAcquire(&s->mutex);
restart_lsn = s->data.restart_lsn;
+ invalidated = s->data.invalidated != RS_INVAL_NONE;
SpinLockRelease(&s->mutex);
+ /* invalidated slots need not apply */
+ if (invalidated)
+ continue;
+
if (restart_lsn == InvalidXLogRecPtr)
continue;
@@ -1012,6 +1023,8 @@ ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
if (s->data.database != dboid)
continue;
+ /* NB: intentionally counting invalidated slots */
+
/* count slots with spinlock held */
SpinLockAcquire(&s->mutex);
(*nslots)++;
@@ -1069,6 +1082,8 @@ restart:
if (s->data.database != dboid)
continue;
+ /* NB: intentionally including invalidated slots */
+
/* acquire slot, so ReplicationSlotDropAcquired can be reused */
SpinLockAcquire(&s->mutex);
/* can't change while ReplicationSlotControlLock is held */
@@ -1294,7 +1309,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlot *s, XLogRecPtr oldestLSN,
{
MyReplicationSlot = s;
s->active_pid = MyProcPid;
- s->data.invalidated_at = restart_lsn;
+ s->data.invalidated = RS_INVAL_WAL;
s->data.restart_lsn = InvalidXLogRecPtr;
/* Let caller know */
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 2f3c9648241..ad3e72be5ee 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -315,12 +315,10 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
nulls[i++] = true;
/*
- * If invalidated_at is valid and restart_lsn is invalid, we know for
- * certain that the slot has been invalidated. Otherwise, test
- * availability from restart_lsn.
+ * If the slot has not been invalidated, test availability from
+ * restart_lsn.
*/
- if (XLogRecPtrIsInvalid(slot_contents.data.restart_lsn) &&
- !XLogRecPtrIsInvalid(slot_contents.data.invalidated_at))
+ if (slot_contents.data.invalidated != RS_INVAL_NONE)
walstate = WALAVAIL_REMOVED;
else
walstate = GetWALAvailability(slot_contents.data.restart_lsn);
--
2.38.0
--jiyjqgogfhmg2lin
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="va65-0002-Prevent-use-of-invalidated-logical-slot-in-Crea.patch"
view thread (2+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH va65 1/9] replication slots: replace invalidated_at LSN with an enum
In-Reply-To: <no-message-id-1857364@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox