fix(cluster): close four-node PCM-X liveness campaign

src/backend/access/heap/heapam_visibility.c

@@ -828,37 +828,42 @@ cluster_satisfies_update_fork(HeapTuple htup, Buffer buffer, TM_Result *res)
 	 * striping-off/below-floor collision with a remote inserter's xid; the
 	 * pre-round-6 early return handed that collision to the PG-native cmin
 	 * path -> "attempted to update invisible tuple".  Only the no-remote-
-	 * evidence route treats the raw match as our own insert.
+	 * evidence route treats the raw match as our own insert.  VACUUM freeze is
+	 * stronger: it proves xmin committed even after that historical ITL slot is
+	 * recycled, so only the exact FROZEN bit pair skips xmin resolution.
 	 */
-	cluster_visibility_resolve_tuple(buffer, tuple, raw_xmin, CLUSTER_VIS_XMIN, &r);
-	switch (cluster_vis_evidence_route(r.evidence, TransactionIdIsCurrentTransactionId(raw_xmin))) {
-	case CLUSTER_VIS_ROUTE_FAILCLOSED_UNKNOWN:
-		ereport(ERROR, (errcode(ERRCODE_CLUSTER_TT_STATUS_UNKNOWN),
-						errmsg("cluster TT slot recycled for xmin %u", raw_xmin),
-						errhint("ITL slot no longer maps to this xid; retry.")));
-		break;
-	case CLUSTER_VIS_ROUTE_NATIVE_SELF:
-		/* Our own insert: PG-native handles cmin / self-modified correctly. */
-		return false;
-	case CLUSTER_VIS_ROUTE_REMOTE_VERDICT:
-		switch (cluster_vis_update_xmin_verdict(r.status)) {
-		case CVV_INVISIBLE:
-			*res = TM_Invisible;
-			return true;
-		case CVV_FAILCLOSED_UNKNOWN:
+	if (cluster_vis_xmin_needs_resolution(tuple->t_infomask)) {
+		cluster_visibility_resolve_tuple(buffer, tuple, raw_xmin, CLUSTER_VIS_XMIN, &r);
+		switch (
+			cluster_vis_evidence_route(r.evidence, TransactionIdIsCurrentTransactionId(raw_xmin))) {
+		case CLUSTER_VIS_ROUTE_FAILCLOSED_UNKNOWN:
 			ereport(ERROR, (errcode(ERRCODE_CLUSTER_TT_STATUS_UNKNOWN),
-							errmsg("cluster TT status unknown for xmin %u", raw_xmin),
-							errhint("Remote commit_scn not yet propagated; retry or abort.")));
+							errmsg("cluster TT slot recycled for xmin %u", raw_xmin),
+							errhint("ITL slot no longer maps to this xid; retry.")));
 			break;
-		case CVV_VISIBLE:
-			xmin_remote_visible = true;
+		case CLUSTER_VIS_ROUTE_NATIVE_SELF:
+			/* Our own insert: PG-native handles cmin / self-modified correctly. */
+			return false;
+		case CLUSTER_VIS_ROUTE_REMOTE_VERDICT:
+			switch (cluster_vis_update_xmin_verdict(r.status)) {
+			case CVV_INVISIBLE:
+				*res = TM_Invisible;
+				return true;
+			case CVV_FAILCLOSED_UNKNOWN:
+				ereport(ERROR, (errcode(ERRCODE_CLUSTER_TT_STATUS_UNKNOWN),
+								errmsg("cluster TT status unknown for xmin %u", raw_xmin),
+								errhint("Remote commit_scn not yet propagated; retry or abort.")));
+				break;
+			case CVV_VISIBLE:
+				xmin_remote_visible = true;
+				break;
+			default:
+				break;
+			}
 			break;
-		default:
+		case CLUSTER_VIS_ROUTE_NATIVE:
 			break;
 		}
-		break;
-	case CLUSTER_VIS_ROUTE_NATIVE:
-		break;
 	}
 	/* xmin NONE/LOCAL: leave xmin to PG unless we must intercept the xmax. */
 
@@ -2465,17 +2470,15 @@ HeapTupleSatisfiesVacuumHorizon(HeapTuple htup, Buffer buffer, TransactionId *de
 
 #ifdef USE_PGRAC_CLUSTER
 	/*
-	 * spec-3.14 D5 (hole #2): a tuple with REMOTE writer evidence must never
-	 * be judged DEAD by this node's local horizon -- overlapping xid spaces
-	 * mean local oldest_xmin can pass a still-live remote xid (false-dead).
-	 * Return HEAPTUPLE_LIVE: the RECENTLY_DEAD path re-tests dead_after
-	 * against the LOCAL oldest_xmin / vistest (heap_prune_satisfies_vacuum)
-	 * and would re-introduce the false-dead, so LIVE (unconditional keep) is
-	 * the only safe verdict.  Real reclamation waits for the cross-node
-	 * vacuum-coordination spec (spec-3.14 §10).  spec wrote RECENTLY_DEAD;
-	 * LIVE is the implementation-time correction (dead_after re-test hazard).
+	 * P0-28 / spec-3.14 §10: a peer may retain this exact TID across a
+	 * PCM-X wait.  ITL evidence is not a durable prune-horizon token: its slot
+	 * may be reused before that peer reacquires the page, after which native
+	 * xmin/CLOG checks can misclassify and recycle the target line pointer.
+	 * Keep every shared-storage tuple until a real cluster horizon exists.
 	 */
-	if (cluster_storage_mode_enabled() && cluster_tuple_has_remote_evidence(buffer, tuple)) {
+	if (cluster_vis_prune_must_defer(cluster_storage_mode_enabled()
+									  && BufferIsValid(buffer) && !BufferIsLocal(buffer),
+								  false)) {
 		cluster_vis_bump_prune_remote_keep_count();
 		return HEAPTUPLE_LIVE;
 	}

src/backend/access/heap/hio.c

@@ -66,6 +66,30 @@ cluster_hio_lease_target_block(Relation relation)
 }
 #endif
 
+/* P0-20: acquire an ordered pair without ever sleeping on the second PCM-X
+ * conversion while the first content lock is held.  BARRIER_CLOSED means the
+ * first page was frozen before this nested edge became visible.  Release our
+ * first lock, resolve the second conversion alone, then retry in the original
+ * block order; pins remain owned by the caller throughout. */
+static void
+cluster_hio_lock_buffer_pair(Buffer first, Buffer second)
+{
+	Assert(BufferIsValid(first));
+	Assert(BufferIsValid(second));
+	Assert(first != second);
+
+	for (;;)
+	{
+		LockBuffer(first, BUFFER_LOCK_EXCLUSIVE);
+		if (ClusterLockBufferExclusiveBarrierAware(second))
+			return;
+
+		LockBuffer(first, BUFFER_LOCK_UNLOCK);
+		LockBuffer(second, BUFFER_LOCK_EXCLUSIVE);
+		LockBuffer(second, BUFFER_LOCK_UNLOCK);
+	}
+}
+
 /*
  * RelationPutHeapTuple - place tuple at specified page
  *
@@ -233,10 +257,12 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 		if (need_to_pin_buffer2)
 			visibilitymap_pin(relation, block2, vmbuffer2);
 
-		/* Relock buffers. */
-		LockBuffer(buffer1, BUFFER_LOCK_EXCLUSIVE);
+		/* Relock buffers.  A cross-page relock keeps the same block order,
+		 * but must unwind a frozen-barrier refusal from the second page. */
 		if (buffer2 != InvalidBuffer && buffer2 != buffer1)
-			LockBuffer(buffer2, BUFFER_LOCK_EXCLUSIVE);
+			cluster_hio_lock_buffer_pair(buffer1, buffer2);
+		else
+			LockBuffer(buffer1, BUFFER_LOCK_EXCLUSIVE);
 
 		/*
 		 * If there are two buffers involved and we pinned just one of them,
@@ -716,17 +742,15 @@ RelationGetBufferForTuple(Relation relation, Size len,
 			buffer = ReadBuffer(relation, targetBlock);
 			if (PageIsAllVisible(BufferGetPage(buffer)))
 				visibilitymap_pin(relation, targetBlock, vmbuffer);
-			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
-			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+			cluster_hio_lock_buffer_pair(otherBuffer, buffer);
 		}
 		else
 		{
 			/* lock target buffer first */
 			buffer = ReadBuffer(relation, targetBlock);
 			if (PageIsAllVisible(BufferGetPage(buffer)))
 				visibilitymap_pin(relation, targetBlock, vmbuffer);
-			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
-			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
+			cluster_hio_lock_buffer_pair(buffer, otherBuffer);
 		}
 
 		/*
@@ -900,8 +924,9 @@ RelationGetBufferForTuple(Relation relation, Size len,
 	{
 		/* released lock on target buffer above */
 		if (otherBuffer != InvalidBuffer)
-			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
-		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+			cluster_hio_lock_buffer_pair(otherBuffer, buffer);
+		else
+			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
 		recheckVmPins = true;
 	}
 	else if (otherBuffer != InvalidBuffer)
@@ -926,8 +951,7 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		{
 			unlockedTargetBuffer = true;
 			LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
-			LockBuffer(otherBuffer, BUFFER_LOCK_EXCLUSIVE);
-			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+			cluster_hio_lock_buffer_pair(otherBuffer, buffer);
 		}
 		recheckVmPins = true;
 	}

src/backend/access/heap/visibilitymap.c

@@ -255,6 +255,32 @@ visibilitymap_pin(Relation rel, BlockNumber heapBlk, Buffer *vmbuf)
 	*vmbuf = vm_readbuf(rel, mapBlock, true);
 }
 
+/*
+ * Repin the exact visibility-map block in a recently observed descriptor.
+ *
+ * Unlike visibilitymap_pin(), this never performs a mapping lookup, storage
+ * read, or relation extension.  A heap mutation can therefore release its VM
+ * pin before a potentially blocking cluster PCM acquire, take heap content
+ * authority, and use this helper without introducing I/O below that lock.  A
+ * false result means the descriptor was invalidated or retagged; the caller
+ * must release heap content authority and retry through visibilitymap_pin().
+ */
+bool
+visibilitymap_pin_recent(Relation rel, BlockNumber heapBlk, Buffer recent_buffer,
+						 Buffer *vmbuf)
+{
+	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
+
+	if (rel == NULL || vmbuf == NULL || BufferIsValid(*vmbuf)
+		|| !BufferIsValid(recent_buffer))
+		return false;
+	if (!ReadRecentBuffer(RelationGetSmgr(rel)->smgr_rlocator.locator,
+						  VISIBILITYMAP_FORKNUM, mapBlock, recent_buffer))
+		return false;
+	*vmbuf = recent_buffer;
+	return true;
+}
+
 /*
  *	visibilitymap_pin_ok - do we already have the correct page pinned?
  *

src/backend/cluster/cluster_cr_server.c

@@ -1049,7 +1049,7 @@ cr_serve_slot(ClusterLmsCrSlot *slot)
 		CLUSTER_INJECTION_POINT("cluster-lms-cr-construct");
 		if (!cluster_injection_should_skip("cluster-lms-cr-construct")
 			&& cluster_crossnode_cr_data_plane
-			&& cluster_bufmgr_copy_block_for_gcs(slot->tag, &page_lsn, cur_copy.data)) {
+			&& cluster_bufmgr_copy_block_for_gcs(slot->tag, &page_lsn, cur_copy.data, NULL)) {
 			PG_TRY();
 			{
 				cluster_cr_construct_page_for_server(cur_copy.data, slot->read_scn, slot->tag,

src/backend/cluster/cluster_debug.c

@@ -1451,6 +1451,8 @@ dump_lms(ReturnSetInfo *rsinfo)
 			 fmt_int64((int64)cluster_lms_obs_get_outbound_not_admitted(-1)));
 	emit_row(rsinfo, "lms", "lms_outbound_requeue_drop_count",
 			 fmt_int64((int64)cluster_lms_obs_get_outbound_requeue_drop(-1)));
+	emit_row(rsinfo, "lms", "lms_outbound_cap_guard_drop_count",
+			 fmt_int64((int64)cluster_lms_obs_get_outbound_cap_guard_drop(-1)));
 	{
 		int w, hb;
 
@@ -1474,6 +1476,9 @@ dump_lms(ReturnSetInfo *rsinfo)
 			snprintf(wkey, sizeof(wkey), "lms_outbound_requeue_drop_count_w%d", w);
 			emit_row(rsinfo, "lms", wkey,
 					 fmt_int64((int64)cluster_lms_obs_get_outbound_requeue_drop(w)));
+			snprintf(wkey, sizeof(wkey), "lms_outbound_cap_guard_drop_count_w%d", w);
+			emit_row(rsinfo, "lms", wkey,
+					 fmt_int64((int64)cluster_lms_obs_get_outbound_cap_guard_drop(w)));
 		}
 
 		/* Inline-serve duration histogram: aggregate 16 rows + per live
@@ -1995,15 +2000,60 @@ dump_pcm(ReturnSetInfo *rsinfo)
 	emit_row(rsinfo, "pcm", "wm_prov_insert_fail_count",
 			 fmt_int64((int64)cluster_pcm_get_wm_prov_insert_fail_count()));
 
-	/* PCM-X queue core: 30 stable keys in the existing pcm category. */
+	/* PCM-X queue core: 30 stable integer keys in the existing pcm category,
+	 * plus one text-valued fail-closed provenance key (excluded from the
+	 * integer-only snapshot tooling in the TAP suite). */
 	{
 		PcmXStatsSnapshot stats = { 0 };
 		PcmXRuntimeSnapshot runtime = cluster_pcm_x_runtime_snapshot();
+		char fail_closed_site[PCM_X_FAIL_CLOSED_SITE_LEN];
 
 		(void)cluster_pcm_x_stats_snapshot(&stats);
 		emit_row(rsinfo, "pcm", "pcm_x_runtime_state", fmt_int32((int32)runtime.state));
 		emit_row(rsinfo, "pcm", "pcm_x_runtime_generation",
 				 fmt_int64((int64)runtime.gate_generation));
+		if (!cluster_pcm_x_runtime_fail_closed_site(fail_closed_site, sizeof(fail_closed_site)))
+			strlcpy(fail_closed_site, "(none)", sizeof(fail_closed_site));
+		emit_row(rsinfo, "pcm", "pcm_x_runtime_fail_closed_site", fail_closed_site);
+
+		/* Text-valued per-slot diagnostic rows; present only while master tag
+		 * or ticket slots are occupied, so the quiescent key count above is
+		 * unchanged. */
+		{
+			Size cursor = 0;
+			Size index = 0;
+			char line[768];
+			char key[64];
+
+			while (cluster_pcm_x_master_tag_debug_next(&cursor, &index, line, sizeof(line))) {
+				snprintf(key, sizeof(key), "pcm_x_tag_%zu", index);
+				emit_row(rsinfo, "pcm", key, line);
+			}
+			cursor = 0;
+			while (cluster_pcm_x_master_ticket_debug_next(&cursor, &index, line, sizeof(line))) {
+				snprintf(key, sizeof(key), "pcm_x_ticket_%zu", index);
+				emit_row(rsinfo, "pcm", key, line);
+			}
+			cursor = 0;
+			while (cluster_pcm_x_local_tag_debug_next(&cursor, &index, line, sizeof(line))) {
+				snprintf(key, sizeof(key), "pcm_x_ltag_%zu", index);
+				emit_row(rsinfo, "pcm", key, line);
+			}
+			{
+				uint32 note_op = 0;
+				uint32 note_result = 0;
+				uint32 note_count = 0;
+				uint64 note_ticket = 0;
+
+				if (cluster_pcm_x_terminal_note_read(&note_op, &note_result, &note_ticket,
+													 &note_count)) {
+					snprintf(line, sizeof(line),
+							 "op=%u result=%u ticket=" UINT64_FORMAT " count=%u", note_op,
+							 note_result, note_ticket, note_count);
+					emit_row(rsinfo, "pcm", "pcm_x_terminal_last_note", line);
+				}
+			}
+		}
 		emit_row(rsinfo, "pcm", "pcm_x_queue_enqueue_count", fmt_int64((int64)stats.enqueue_count));
 		emit_row(rsinfo, "pcm", "pcm_x_queue_admit_count", fmt_int64((int64)stats.admit_count));
 		emit_row(rsinfo, "pcm", "pcm_x_queue_confirm_count", fmt_int64((int64)stats.confirm_count));
@@ -2043,6 +2093,8 @@ dump_pcm(ReturnSetInfo *rsinfo)
 		emit_row(rsinfo, "pcm", "pcm_x_own_busy_count", fmt_int64((int64)stats.own_busy_count));
 		emit_row(rsinfo, "pcm", "pcm_x_own_corrupt_count",
 				 fmt_int64((int64)stats.own_corrupt_count));
+		emit_row(rsinfo, "pcm", "pcm_x_queue_barrier_unwind_count",
+				 fmt_int64((int64)stats.barrier_unwind_count));
 	}
 }
 
@@ -2354,6 +2406,8 @@ dump_gcs(ReturnSetInfo *rsinfo)
 			 fmt_int64((int64)cluster_gcs_get_pi_watermark_advance_count()));
 	emit_row(rsinfo, "gcs", "pi_watermark_retire_count",
 			 fmt_int64((int64)cluster_gcs_get_pi_watermark_retire_count()));
+	emit_row(rsinfo, "gcs", "pi_durable_note_apply_count",
+			 fmt_int64((int64)cluster_gcs_get_pi_durable_note_apply_count()));
 	emit_row(rsinfo, "gcs", "lost_write_detected_count",
 			 fmt_int64((int64)cluster_gcs_get_lost_write_detected_count()));
 	emit_row(rsinfo, "gcs", "lost_write_avoid_count",