fix(cluster): close S3 preflight runtime gate

src/backend/cluster/cluster_debug.c

@@ -2411,6 +2411,8 @@ dump_gcs(ReturnSetInfo *rsinfo)
 	/* PGRAC: spec-4.7 D6 — 8 NEW counter rows for GCS/PCM warm recovery. */
 	emit_row(rsinfo, "gcs_recovery", "block_resources_recovering",
 			 fmt_int64((int64)cluster_gcs_get_recovery_block_resources_recovering()));
+	emit_row(rsinfo, "gcs_recovery", "pcm_x_image_fetch_recovering_retry_count",
+			 fmt_int64((int64)cluster_gcs_get_pcm_x_image_fetch_recovering_retry_count()));
 	emit_row(rsinfo, "gcs_recovery", "buffers_redeclared",
 			 fmt_int64((int64)cluster_gcs_get_recovery_buffers_redeclared()));
 	emit_row(rsinfo, "gcs_recovery", "block_state_rebuilt",

src/backend/cluster/cluster_gcs_block.c

@@ -297,13 +297,16 @@ typedef struct ClusterGcsBlockShared {
 		clean_page_xfer_stale_holder_recover_count; /* RESERVED Stage 6 (was DENIED recover) */
 	pg_atomic_uint64 clean_page_xfer_third_party_denied_count; /* 3-node third-party master DENY */
 	/* PGRAC: spec-4.7 D6 — GCS/PCM warm-recovery observability (dump category
-	 * 'gcs_recovery').  8 counters per spec §2.4. */
+	 * 'gcs_recovery'). */
 	pg_atomic_uint64 recovery_block_resources_recovering; /* phase_for_tag → RECOVERING hits */
-	pg_atomic_uint64 recovery_buffers_redeclared;		  /* survivor re-declare sent (D2) */
-	pg_atomic_uint64 recovery_block_state_rebuilt;		  /* master rebuild applied (D2/D3) */
-	pg_atomic_uint64 recovery_redo_boundary_waits;		  /* redo gate: not yet covered (D5) */
-	pg_atomic_uint64 recovery_redo_boundary_reached;	  /* redo gate: covered (D5) */
-	pg_atomic_uint64 recovery_stale_block_drop; /* re-declare dropped: off-epoch/bad (D2) */
+	/* PCM-X requester received a retryable RESOURCE_RECOVERING denial while
+	 * fetching its generation-exact holder image. */
+	pg_atomic_uint64 pcm_x_image_fetch_recovering_retry_count;
+	pg_atomic_uint64 recovery_buffers_redeclared;	 /* survivor re-declare sent (D2) */
+	pg_atomic_uint64 recovery_block_state_rebuilt;	 /* master rebuild applied (D2/D3) */
+	pg_atomic_uint64 recovery_redo_boundary_waits;	 /* redo gate: not yet covered (D5) */
+	pg_atomic_uint64 recovery_redo_boundary_reached; /* redo gate: covered (D5) */
+	pg_atomic_uint64 recovery_stale_block_drop;		 /* re-declare dropped: off-epoch/bad (D2) */
 	pg_atomic_uint64 recovery_ambiguous_owner_failclosed; /* not-double-X conflict (D3) */
 	pg_atomic_uint64 recovery_before_boundary_failclosed; /* served-before-redo gate fail (D5) */
 	/* PGRAC: spec-2.36 D3 (HC116) — master broadcast invalidate slot.
@@ -600,6 +603,7 @@ cluster_gcs_block_shmem_init(void)
 		pg_atomic_init_u64(&ClusterGcsBlock->clean_page_xfer_third_party_denied_count, 0);
 		/* PGRAC: spec-4.7 D6 — 8 NEW warm-recovery counters init. */
 		pg_atomic_init_u64(&ClusterGcsBlock->recovery_block_resources_recovering, 0);
+		pg_atomic_init_u64(&ClusterGcsBlock->pcm_x_image_fetch_recovering_retry_count, 0);
 		pg_atomic_init_u64(&ClusterGcsBlock->recovery_buffers_redeclared, 0);
 		pg_atomic_init_u64(&ClusterGcsBlock->recovery_block_state_rebuilt, 0);
 		pg_atomic_init_u64(&ClusterGcsBlock->recovery_redo_boundary_waits, 0);
@@ -3379,6 +3383,9 @@ cluster_gcs_pcm_x_fetch_image_and_install(BufferDesc *buf, const PcmXLocalHandle
 				|| reply.status == (uint8)GCS_BLOCK_REPLY_DENIED_RESOURCE_RECOVERING) {
 				if (reply.status == (uint8)GCS_BLOCK_REPLY_DENIED_MASTER_NOT_HOLDER)
 					pg_atomic_fetch_add_u64(&ClusterGcsBlock->block_master_not_holder_count, 1);
+				else
+					pg_atomic_fetch_add_u64(
+						&ClusterGcsBlock->pcm_x_image_fetch_recovering_retry_count, 1);
 				continue;
 			}
 			if (!cluster_pcm_x_image_fetch_reply_exact(&reply, reply_block, &progress_now,
@@ -14480,6 +14487,14 @@ cluster_gcs_get_recovery_block_resources_recovering(void)
 			   ? pg_atomic_read_u64(&ClusterGcsBlock->recovery_block_resources_recovering)
 			   : 0;
 }
+
+uint64
+cluster_gcs_get_pcm_x_image_fetch_recovering_retry_count(void)
+{
+	return ClusterGcsBlock
+			   ? pg_atomic_read_u64(&ClusterGcsBlock->pcm_x_image_fetch_recovering_retry_count)
+			   : 0;
+}
 uint64
 cluster_gcs_get_recovery_buffers_redeclared(void)
 {

src/backend/storage/buffer/bufmgr.c

@@ -8427,9 +8427,10 @@ ConditionalLockBuffer(Buffer buffer)
 		 * GRANT_PENDING must not modify protocol-owned bytes.
 		 */
 		buf_state = LockBufHdr(buf);
-		blocked = cluster_bufmgr_pcm_x_retained_image_locked(buf, buf_state)
-			|| buf->pcm_state != (uint8) PCM_STATE_X
-			|| cluster_pcm_own_flags_get(buf->buf_id) != 0;
+		blocked = !cluster_pcm_x_conditional_lock_allowed(
+			cluster_pcm_is_active(), cluster_bufmgr_should_pcm_track(buf),
+			cluster_bufmgr_pcm_x_retained_image_locked(buf, buf_state), buf->pcm_state,
+			cluster_pcm_own_flags_get(buf->buf_id));
 		UnlockBufHdr(buf, buf_state);
 		if (blocked)
 		{

src/include/cluster/cluster_gcs_block.h

@@ -3513,6 +3513,7 @@ extern void cluster_gcs_block_send_pi_discard_invalidate(BufferTag tag, int32 ta
 
 /* PGRAC: spec-4.7 D6 — 8 warm-recovery observability accessors. */
 extern uint64 cluster_gcs_get_recovery_block_resources_recovering(void);
+extern uint64 cluster_gcs_get_pcm_x_image_fetch_recovering_retry_count(void);
 extern uint64 cluster_gcs_get_recovery_buffers_redeclared(void);
 extern uint64 cluster_gcs_get_recovery_block_state_rebuilt(void);
 extern uint64 cluster_gcs_get_recovery_redo_boundary_waits(void);

src/include/cluster/cluster_pcm_x_bufmgr.h

@@ -165,6 +165,18 @@ cluster_pcm_x_cached_cover_bypasses_queue(bool local_cache, bool requested_x, ui
 	return local_cache && requested_x && pcm_state == (uint8)PCM_STATE_X && flags == 0;
 }
 
+/* ConditionalLockBuffer cannot initiate a PCM conversion.  Preserve native
+ * PostgreSQL behavior while PCM is inactive and for relations outside the
+ * coherence domain; an active tracked page must already hold exact X.  Live
+ * transition/retained evidence remains closed regardless of runtime state. */
+static inline bool
+cluster_pcm_x_conditional_lock_allowed(bool runtime_active, bool tracked, bool retained_image,
+									   uint8 pcm_state, uint32 flags)
+{
+	return !retained_image && flags == 0
+		   && (!runtime_active || !tracked || pcm_state == (uint8)PCM_STATE_X);
+}
+
 typedef ClusterPcmOwnSnapshot ClusterPcmOwnEvictionCapture;
 
 typedef enum ClusterPcmXGrantReservationKind {

src/test/cluster_tap/t/251_gcs_pcm_warm_recovery.pl

@@ -69,7 +69,7 @@
 #      L2/L3 (sig b/c) DEFERRED to D1-D5 acceptance + cluster_unit (L239,
 #           see finding above).  SKIPed here with reason.
 #      Lobs (D6/observability, FLIPPED) gcs_recovery dump category exposes the
-#           8 warm-recovery counters under category='gcs_recovery'.
+#           warm-recovery counters under category='gcs_recovery'.
 #
 # Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
 # Portions Copyright (c) 1994, Regents of the University of California
@@ -215,19 +215,19 @@
 #     8 warm-recovery counters: block_resources_recovering / buffers_redeclared /
 #     block_state_rebuilt / redo_boundary_waits / redo_boundary_reached /
 #     stale_block_drop / ambiguous_owner_failclosed / before_boundary_failclosed).
-#     spec-2.41 D7 adds 2 redo-coverage serve-gate counters to the same category
-#     (redo_coverage_required_lsn_zero_count / redo_coverage_gate_block_count) → 10.
+#     spec-2.41 D7 adds 2 redo-coverage serve-gate counters to the same category;
+#     PCM-X adds one image-fetch RESOURCE_RECOVERING retry counter → 11.
 # ----------
 is($triple->node0->safe_psql('postgres',
 		q{SELECT count(*) FROM cluster_dump_state()
 		    WHERE category = 'gcs_recovery'}),
-	'10',
-	'Lobs (D6 flipped + spec-2.41 D7): gcs_recovery dump category exposes 10 '
+	'11',
+	'Lobs (D6 flipped + spec-2.41 D7 + PCM-X): gcs_recovery dump category exposes 11 '
 	. 'counters — 8 warm-recovery (block_resources_recovering / buffers_redeclared / '
 	. 'block_state_rebuilt / redo_boundary_waits / redo_boundary_reached / '
 	. 'stale_block_drop / ambiguous_owner_failclosed / before_boundary_failclosed) '
 	. '+ 2 redo-coverage serve-gate (redo_coverage_required_lsn_zero_count / '
-	. 'redo_coverage_gate_block_count)');
+	. 'redo_coverage_gate_block_count) + 1 PCM-X image-fetch recovering retry');
 
 
 $triple->stop_triple;

src/test/cluster_unit/test_cluster_debug.c

@@ -1650,6 +1650,11 @@ cluster_gcs_get_recovery_block_resources_recovering(void)
 	return 0;
 }
 uint64
+cluster_gcs_get_pcm_x_image_fetch_recovering_retry_count(void)
+{
+	return 0;
+}
+uint64
 cluster_gcs_get_recovery_buffers_redeclared(void)
 {
 	return 0;

src/test/cluster_unit/test_cluster_gcs_block.c

@@ -2189,6 +2189,7 @@ UT_TEST(test_pcm_x_requester_fetch_revalidates_queue_and_reservation_before_inst
 	const char *install_publish;
 	const char *install_runtime_after;
 	const char *recovering_retry;
+	const char *recovering_retry_counter;
 	const char *reply_validation;
 	char *reply_handler;
 
@@ -2208,11 +2209,16 @@ UT_TEST(test_pcm_x_requester_fetch_revalidates_queue_and_reservation_before_inst
 		UT_ASSERT_NOT_NULL(strstr(fetch, "gcs_block_pcm_x_install_reserved_image_exact("));
 		UT_ASSERT_NOT_NULL(strstr(fetch, "gcs_block_release_slot(slot)"));
 		recovering_retry = strstr(fetch, "GCS_BLOCK_REPLY_DENIED_RESOURCE_RECOVERING");
+		recovering_retry_counter = strstr(fetch, "pcm_x_image_fetch_recovering_retry_count");
 		reply_validation = strstr(fetch, "cluster_pcm_x_image_fetch_reply_exact(");
 		UT_ASSERT_NOT_NULL(recovering_retry);
+		UT_ASSERT_NOT_NULL(recovering_retry_counter);
 		UT_ASSERT(recovering_retry == NULL || recovering_retry < fetch_end);
+		UT_ASSERT(recovering_retry_counter == NULL || recovering_retry_counter < fetch_end);
 		UT_ASSERT(recovering_retry == NULL || reply_validation == NULL
 				  || recovering_retry < reply_validation);
+		UT_ASSERT(recovering_retry_counter == NULL || reply_validation == NULL
+				  || recovering_retry_counter < reply_validation);
 		backoff_branch = strstr(fetch, "if (retry_attempt > 0)");
 		backoff_wait
 			= backoff_branch != NULL ? strstr(backoff_branch, "(void)WaitLatch(MyLatch") : NULL;

src/test/cluster_unit/test_cluster_pcm_own.c

@@ -1211,17 +1211,15 @@ UT_TEST(test_retained_image_release_and_writeback_gates_are_exact)
 	if (conditional != NULL) {
 		const char *content
 			= strstr(conditional, "LWLockConditionalAcquire(BufferDescriptorGetContentLock(buf)");
-		const char *ownership = strstr(conditional, "buf->pcm_state != (uint8) PCM_STATE_X");
-		const char *flags = strstr(conditional, "cluster_pcm_own_flags_get(buf->buf_id) != 0");
+		const char *ownership = strstr(conditional, "cluster_pcm_x_conditional_lock_allowed(");
 		const char *release
 			= strstr(conditional, "LWLockRelease(BufferDescriptorGetContentLock(buf))");
 
 		UT_ASSERT_NOT_NULL(content);
 		UT_ASSERT_NOT_NULL(ownership);
-		UT_ASSERT_NOT_NULL(flags);
 		UT_ASSERT_NOT_NULL(release);
-		if (content != NULL && ownership != NULL && flags != NULL && release != NULL)
-			UT_ASSERT(content < ownership && ownership < flags && flags < release);
+		if (content != NULL && ownership != NULL && release != NULL)
+			UT_ASSERT(content < ownership && ownership < release);
 	}
 	if (resident_stamp != NULL) {
 		const char *content = strstr(
@@ -1659,6 +1657,18 @@ UT_TEST(test_current_image_shape_accepts_monotone_xcur_after_x_to_s_yield)
 	UT_ASSERT(!cluster_pcm_x_current_image_shape((uint8)PCM_STATE_S, (uint8)BUF_TYPE_XCUR, false));
 }
 
+UT_TEST(test_conditional_lock_preserves_native_off_and_enforces_tracked_x)
+{
+	UT_ASSERT(cluster_pcm_x_conditional_lock_allowed(false, true, false, (uint8)PCM_STATE_N, 0));
+	UT_ASSERT(cluster_pcm_x_conditional_lock_allowed(true, false, false, (uint8)PCM_STATE_N, 0));
+	UT_ASSERT(!cluster_pcm_x_conditional_lock_allowed(true, true, false, (uint8)PCM_STATE_N, 0));
+	UT_ASSERT(!cluster_pcm_x_conditional_lock_allowed(true, true, false, (uint8)PCM_STATE_S, 0));
+	UT_ASSERT(cluster_pcm_x_conditional_lock_allowed(true, true, false, (uint8)PCM_STATE_X, 0));
+	UT_ASSERT(!cluster_pcm_x_conditional_lock_allowed(false, false, true, (uint8)PCM_STATE_X, 0));
+	UT_ASSERT(!cluster_pcm_x_conditional_lock_allowed(false, false, false, (uint8)PCM_STATE_X,
+													  PCM_OWN_FLAG_GRANT_PENDING));
+}
+
 UT_TEST(test_queue_passive_n_mirror_is_never_gcs_ship_authority)
 {
 	static const char *const probe_contract[]
@@ -1977,7 +1987,7 @@ UT_TEST(test_lockbuffer_pcm_x_writer_ledger_is_distinct_and_brackets_content_aut
 int
 main(void)
 {
-	UT_PLAN(44);
+	UT_PLAN(45);
 	UT_RUN(test_shmem_initializes_complete_entry);
 	UT_RUN(test_begin_abort_is_exact_and_monotonic);
 	UT_RUN(test_invalid_live_flag_shapes_are_corrupt_not_busy);
@@ -2017,6 +2027,7 @@ main(void)
 	UT_RUN(test_queue_holder_snapshot_by_tag_is_mapping_and_header_exact);
 	UT_RUN(test_queue_passive_pinned_s_release_serializes_bytes_and_ownership);
 	UT_RUN(test_current_image_shape_accepts_monotone_xcur_after_x_to_s_yield);
+	UT_RUN(test_conditional_lock_preserves_native_off_and_enforces_tracked_x);
 	UT_RUN(test_queue_installed_image_publication_is_exact_and_content_locked);
 	UT_RUN(test_queue_self_source_handoff_is_single_lifecycle_and_readonly_drain);
 	UT_RUN(test_queue_passive_n_mirror_is_never_gcs_ship_authority);