Double-check replication slot's effective_xmin before use.

Paranoia.

Also move related code in create_replication_slots() to more suitable place.

Author: Antonin Houska

pg_squeeze.c

@@ -970,6 +970,15 @@ setup_decoding(Oid relid, TupleDesc tup_desc, Snapshot *snap_hist)
 	 */
 	ReplicationSlotAcquire(NameStr(repl_slot->name), true);
 
+	/*
+	 * This should not really happen, but if it did, the initial load could
+	 * miss some data.
+	 */
+	if (!TransactionIdIsValid(MyReplicationSlot->effective_xmin))
+		ereport(ERROR,
+				(errmsg("replication slot \"%s\" has invalid effective_xmin",
+						NameStr(repl_slot->name))));
+
 	/*
 	 * It's pretty unlikely for some client to have consumed data changes
 	 * (accidentally?)  before this worker could acquire the slot, but it's

worker.c

@@ -1598,13 +1598,6 @@ create_replication_slots(int nslots, MemoryContext mcxt)
 		 */
 		Assert(!ctx->fast_forward);
 
-		SpinLockAcquire(&slot->mutex);
-		Assert(TransactionIdIsValid(slot->effective_xmin) &&
-			   !TransactionIdIsValid(slot->data.xmin));
-		/* Prevent ReplicationSlotRelease() from clearing effective_xmin. */
-		slot->data.xmin = slot->effective_xmin;
-		SpinLockRelease(&slot->mutex);
-
 		/*
 		 * Bring the snapshot builder into the SNAPBUILD_CONSISTENT state so
 		 * that the worker can get its snapshot and start decoding
@@ -1654,6 +1647,14 @@ create_replication_slots(int nslots, MemoryContext mcxt)
 		 * Done for now, the worker will have to setup the context on its own.
 		 */
 		FreeDecodingContext(ctx);
+
+		/* Prevent ReplicationSlotRelease() from clearing effective_xmin. */
+		SpinLockAcquire(&slot->mutex);
+		Assert(TransactionIdIsValid(slot->effective_xmin) &&
+			   !TransactionIdIsValid(slot->data.xmin));
+		slot->data.xmin = slot->effective_xmin;
+		SpinLockRelease(&slot->mutex);
+
 		ReplicationSlotRelease();
 	}