lockless.diff

text/x-patch

Download

Patch

Same data as JSON: GET /api/v1/attachments/:id/patch the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes. API reference →

Format: unified

File	+	−
src/backend/replication/slot.c	11	0
src/include/replication/slot.h	1	0

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 4b5b24d3759..0b8c6861edb 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -473,16 +473,17 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 	memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
 	namestrcpy(&slot->data.name, name);
 	slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
 	slot->data.persistency = persistency;
 	slot->data.two_phase = two_phase;
 	slot->data.two_phase_at = InvalidXLogRecPtr;
 	slot->data.failover = failover;
 	slot->data.synced = synced;
+	slot->wal_reserved = false;
 
 	/* and then data only present in shared memory */
 	slot->just_dirtied = false;
 	slot->dirty = false;
 	slot->effective_xmin = InvalidTransactionId;
 	slot->effective_catalog_xmin = InvalidTransactionId;
 	slot->candidate_catalog_xmin = InvalidTransactionId;
 	slot->candidate_xmin_lsn = InvalidXLogRecPtr;
@@ -1568,16 +1569,17 @@ CheckSlotPermissions(void)
  * the slot and concurrency safe.
  */
 void
 ReplicationSlotReserveWal(void)
 {
 	ReplicationSlot *slot = MyReplicationSlot;
 
 	Assert(slot != NULL);
+	Assert(slot->wal_reserved == false);
 	Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
 	Assert(!XLogRecPtrIsValid(slot->last_saved_restart_lsn));
 
 	/*
 	 * The replication slot mechanism is used to prevent removal of required
 	 * WAL. As there is no interlock between this routine and checkpoints, WAL
 	 * segments could concurrently be removed when a now stale return value of
 	 * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
@@ -1613,16 +1615,22 @@ ReplicationSlotReserveWal(void)
 		INJECTION_POINT("physical-slot-reserve-wal-get-redo", NULL);
 
 		SpinLockAcquire(&slot->mutex);
 		slot->data.restart_lsn = restart_lsn;
 		SpinLockRelease(&slot->mutex);
 
 		INJECTION_POINT("physical-slot-reserve-wal-before-compute-required-lsn", NULL);
 
+		// for physical slots only!
+		if (slot->data.restart_lsn < GetRedoRecPtr())
+			continue;
+
+		slot->wal_reserved = true;
+
 		/* prevent WAL removal as fast as possible */
 		ReplicationSlotsComputeRequiredLSN();
 
 		/*
 		 * If all required WAL is still there, great, otherwise retry. The
 		 * slot should prevent further removal of WAL, unless there's a
 		 * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
 		 * the new restart_lsn above, so normally we should never need to loop
@@ -1753,16 +1761,19 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s,
 							   TimestampTz *inactive_since, TimestampTz now)
 {
 	Assert(possible_causes != RS_INVAL_NONE);
 
 	if (possible_causes & RS_INVAL_WAL_REMOVED)
 	{
 		XLogRecPtr	restart_lsn = s->data.restart_lsn;
 
+		if (!s->wal_reserved)
+			return RS_INVAL_NONE;
+
 		if (XLogRecPtrIsValid(restart_lsn) &&
 			restart_lsn < oldestLSN)
 			return RS_INVAL_WAL_REMOVED;
 	}
 
 	if (possible_causes & RS_INVAL_HORIZON)
 	{
 		/* invalid DB oid signals a shared relation */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 09c69f83d57..1aaf3a275e5 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -168,16 +168,17 @@ typedef struct ReplicationSlot
 	bool		in_use;
 
 	/* Who is streaming out changes for this slot? 0 in unused slots. */
 	pid_t		active_pid;
 
 	/* any outstanding modifications? */
 	bool		just_dirtied;
 	bool		dirty;
+	bool		wal_reserved;
 
 	/*
 	 * For logical decoding, it's extremely important that we never remove any
 	 * data that's still needed for decoding purposes, even after a crash;
 	 * otherwise, decoding will produce wrong answers.  Ordinary streaming
 	 * replication also needs to prevent old row versions from being removed
 	 * too soon, but the worst consequence we might encounter there is
 	 * unwanted query cancellations on the standby.  Thus, for logical