fix(cluster): close S3 preflight coherence gaps

src/backend/cluster/cluster_gcs_block.c

@@ -3370,8 +3370,15 @@ cluster_gcs_pcm_x_fetch_image_and_install(BufferDesc *buf, const PcmXLocalHandle
 			reply = slot->reply_header;
 			memcpy(reply_block, slot->reply_block_data, sizeof(reply_block));
 			LWLockRelease(&blk->lock.lock);
-			if (reply.status == (uint8)GCS_BLOCK_REPLY_DENIED_MASTER_NOT_HOLDER) {
-				pg_atomic_fetch_add_u64(&ClusterGcsBlock->block_master_not_holder_count, 1);
+			/* Both denials describe transient source authority, not malformed
+			 * image bytes.  Re-arm the same generation-exact request within the
+			 * existing bounded retry budget; only a putative READ_IMAGE reply is
+			 * eligible for byte/checksum validation below.
+			 */
+			if (reply.status == (uint8)GCS_BLOCK_REPLY_DENIED_MASTER_NOT_HOLDER
+				|| 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);
 				continue;
 			}
 			if (!cluster_pcm_x_image_fetch_reply_exact(&reply, reply_block, &progress_now,

src/backend/storage/buffer/bufmgr.c

@@ -1195,6 +1195,8 @@ StaticAssertDecl(lengthof(cluster_bufmgr_pcm_x_writer_ledger) == LWLOCK_MAX_HELD
 				 "PCM-X writer ledger must match the process held-LWLock bound");
 
 static ClusterPcmXWriterLedgerEntry *cluster_bufmgr_pcm_x_writer_find(BufferDesc *buf);
+static void cluster_bufmgr_pcm_x_holder_report_failure(PcmXQueueResult result, BufferDesc *buf,
+													   const char *operation);
 static bool cluster_bufmgr_pcm_x_writer_claim_entry_exact(const ClusterPcmXWriterLedgerEntry *entry,
 														  BufferDesc *buf);
 static bool cluster_bufmgr_pcm_x_writer_entry_exact(const ClusterPcmXWriterLedgerEntry *entry,
@@ -1210,6 +1212,7 @@ cluster_bufmgr_pcm_x_holder_retry_wait(LWLock *content_lock, int32 buffer_id,
 								   uint32 wait_index)
 {
 	long		delay_ms;
+	PcmXQueueResult guard_result;
 
 	/*
 	 * This is a production lock-order guard.  Waiting while the page content
@@ -1220,6 +1223,15 @@ cluster_bufmgr_pcm_x_holder_retry_wait(LWLock *content_lock, int32 buffer_id,
 				(errcode(ERRCODE_DATA_CORRUPTED),
 				 errmsg("cannot wait for a cluster PCM-X holder gate while holding content authority"),
 				 errdetail("buffer=%d wait_index=%u", buffer_id, wait_index)));
+	/* A different held content lock may itself be the holder that a frozen
+	 * PROBE is draining.  Reuse the protocol's exact held-lock snapshot so a
+	 * safe lock-coupling path can still wait, but never sleep across a closed
+	 * holder barrier or a torn snapshot.
+	 */
+	guard_result = cluster_pcm_x_nested_wait_guard_before_block();
+	if (guard_result != PCM_X_QUEUE_OK)
+		cluster_bufmgr_pcm_x_holder_report_failure(
+			guard_result, GetBufferDescriptor(buffer_id), "retry wait nested guard");
 	delay_ms = cluster_pcm_x_holder_retry_delay_ms(wait_index);
 	CHECK_FOR_INTERRUPTS();
 	(void) WaitLatch(MyLatch, WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, delay_ms,
@@ -8381,15 +8393,12 @@ LockBufferForFreeSpaceMapPageInit(Buffer buffer)
  * This assumes the caller wants BUFFER_LOCK_EXCLUSIVE mode.
  *
  * PGRAC ownership-generation audit — this is NOT a PCM bypass entrance for the
- * ownership-generation wave: it acquires only the content lock and never reads
- * or mutates the (pcm_state, generation, reservation_token, flags) tuple, so
- * it cannot create a stale-generation TOCTOU (W1) nor an install/drop window
- * (W2/W3).  It deliberately does NOT register a PCM X grant: a caller that
- * then WRITES the page is backstopped by
- * cluster_bufmgr_block_write_permitted() at write time.  The post-acquire
- * reservation check below additionally refuses protocol-owned bytes without
- * adding a blocking PCM round trip, which would violate this API's don't-wait
- * contract.
+ * ownership-generation wave: it acquires only the content lock and never
+ * initiates a PCM conversion.  A shared-buffer caller may therefore succeed
+ * only when the coherent ownership tuple already names an unencumbered local
+ * X image.  N/S ownership, retained bytes, or a live transition returns false
+ * so the caller cannot write around the queue while preserving this API's
+ * don't-wait contract.
  */
 bool
 ConditionalLockBuffer(Buffer buffer)
@@ -8419,6 +8428,7 @@ ConditionalLockBuffer(Buffer buffer)
 		 */
 		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;
 		UnlockBufHdr(buf, buf_state);
 		if (blocked)

src/test/cluster_tap/t/020_shmem_registry.pl

@@ -283,8 +283,8 @@
 is($node->safe_psql(
 		'postgres',
 		'SELECT count(*) FROM pg_stat_cluster_injections'),
-   '181',
-   'L15 total injection registry size is 181 (gcs-race-round4c +2 cluster-gcs-block-fallback-refresh-stale + cluster-gcs-block-yield-notify-drop; gcs-race-fix-2 +1 cluster-gcs-block-done-drop; gcs-race-fix +1 cluster-gcs-block-duplicate-grant-reply; spec-5.22d Hardening +1 cluster-undo-authority-scan; spec-5.22e D5-7 +2; spec-5.22d D4-8 +1; spec-7.3 review P1-1 +1 cluster-lms-cr-fence-recheck; spec-7.1 D3-a +1 cluster-mxid-halfspace-hard-limit; spec-7.3 D7 +1 cluster-lms-cr-fence-refuse; spec-7.2 D6 +2 cluster-lms-data-dispatch + cluster-lms-conn-reset; spec-6.14 D5+D8 +3; spec-5.6a +1; spec-6.12e2 +1 cluster-gcs-block-bast-nudge; spec-6.15 D3 +2 cluster-xid-herding-stall + cluster-xid-window-hard-limit; spec-6.12i +1 cluster-lms-undo-fetch; spec-6.12b +1 cluster-lms-cr-construct; spec-6.12a ㉕ +1 cluster-gcs-block-remote-downgrade; full breakdown in t/015)');
+   '183',
+   'L15 total injection registry size is 183 (branch-1 +2; full breakdown in t/015)');
 
 
 # ----------

src/test/cluster_tap/t/021_block_format.pl

@@ -188,8 +188,8 @@
 is($node->safe_psql(
 		'postgres',
 		'SELECT count(*) FROM pg_stat_cluster_injections'),
-   '181',
-   'L11 pg_stat_cluster_injections is 181 (gcs-race-round4c +2 cluster-gcs-block-fallback-refresh-stale + cluster-gcs-block-yield-notify-drop; gcs-race-fix-2 +1 cluster-gcs-block-done-drop; gcs-race-fix +1 cluster-gcs-block-duplicate-grant-reply; spec-5.22d Hardening +1 cluster-undo-authority-scan; spec-5.22e D5-7 +2; spec-5.22d D4-8 +1; spec-7.3 review P1-1 +1 cluster-lms-cr-fence-recheck; spec-7.1 D3-a +1 cluster-mxid-halfspace-hard-limit; spec-7.3 D7 +1 cluster-lms-cr-fence-refuse; spec-7.2 D6 +2 lms data-dispatch/conn-reset; spec-6.14 D5+D8 +3; spec-5.6a +1; spec-6.12e2 +1 cluster-gcs-block-bast-nudge; spec-6.15 D3 +2 xid herding/hard-limit; spec-6.12 waves a/b/i +3; full breakdown in t/015)');
+   '183',
+   'L11 pg_stat_cluster_injections is 183 (branch-1 +2; full breakdown in t/015)');
 
 
 # ----------

src/test/cluster_tap/t/022_itl_slot.pl

@@ -204,8 +204,8 @@
 is($node->safe_psql(
 		'postgres',
 		'SELECT count(*) FROM pg_stat_cluster_injections'),
-   '181',
-   'L12a pg_stat_cluster_injections is 181 (gcs-race-round4c +2 cluster-gcs-block-fallback-refresh-stale + cluster-gcs-block-yield-notify-drop; gcs-race-fix-2 +1 cluster-gcs-block-done-drop; gcs-race-fix +1 cluster-gcs-block-duplicate-grant-reply; spec-5.22d Hardening +1 cluster-undo-authority-scan; spec-5.22e D5-7 +2; spec-5.22d D4-8 +1; spec-7.3 review P1-1 +1 cluster-lms-cr-fence-recheck; spec-7.1 D3-a +1 cluster-mxid-halfspace-hard-limit; spec-7.3 D7 +1 cluster-lms-cr-fence-refuse; spec-7.2 D6 +2 lms data-dispatch/conn-reset; spec-6.14 D5+D8 +3; spec-5.6a +1; spec-6.12e2 +1 cluster-gcs-block-bast-nudge; spec-6.15 D3 +2 xid herding/hard-limit; spec-6.12 waves a/b/i +3; full breakdown in t/015)');
+   '183',
+   'L12a pg_stat_cluster_injections is 183 (branch-1 +2; full breakdown in t/015)');
 
 is($node->safe_psql(
 		'postgres',

src/test/cluster_tap/t/023_buffer_descriptor.pl

@@ -157,8 +157,8 @@
 is($node->safe_psql(
 		'postgres',
 		'SELECT count(*) FROM pg_stat_cluster_injections'),
-   '181',
-   'L11 pg_stat_cluster_injections is 181 (gcs-race-round4c +2 cluster-gcs-block-fallback-refresh-stale + cluster-gcs-block-yield-notify-drop; gcs-race-fix-2 +1 cluster-gcs-block-done-drop; gcs-race-fix +1 cluster-gcs-block-duplicate-grant-reply; spec-5.22d Hardening +1 cluster-undo-authority-scan; spec-5.22e D5-7 +2; spec-5.22d D4-8 +1; spec-7.3 review P1-1 +1 cluster-lms-cr-fence-recheck; spec-7.1 D3-a +1 cluster-mxid-halfspace-hard-limit; spec-7.3 D7 +1 cluster-lms-cr-fence-refuse; spec-7.2 D6 +2 lms data-dispatch/conn-reset; spec-6.14 D5+D8 +3; spec-5.6a +1; spec-6.12e2 +1 cluster-gcs-block-bast-nudge; spec-6.15 D3 +2 xid herding/hard-limit; spec-6.12 waves a/b/i +3; full breakdown in t/015)');
+   '183',
+   'L11 pg_stat_cluster_injections is 183 (branch-1 +2; full breakdown in t/015)');
 
 
 # ----------

src/test/cluster_unit/test_cluster_gcs_block.c

@@ -2188,6 +2188,8 @@ UT_TEST(test_pcm_x_requester_fetch_revalidates_queue_and_reservation_before_inst
 	const char *install_runtime_before;
 	const char *install_publish;
 	const char *install_runtime_after;
+	const char *recovering_retry;
+	const char *reply_validation;
 	char *reply_handler;
 
 	UT_ASSERT_NOT_NULL(source);
@@ -2205,6 +2207,12 @@ UT_TEST(test_pcm_x_requester_fetch_revalidates_queue_and_reservation_before_inst
 		UT_ASSERT_NOT_NULL(strstr(fetch, "cluster_pcm_x_image_fetch_reservation_exact("));
 		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");
+		reply_validation = strstr(fetch, "cluster_pcm_x_image_fetch_reply_exact(");
+		UT_ASSERT_NOT_NULL(recovering_retry);
+		UT_ASSERT(recovering_retry == NULL || recovering_retry < fetch_end);
+		UT_ASSERT(recovering_retry == NULL || reply_validation == NULL
+				  || recovering_retry < 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,15 +1211,17 @@ 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 *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 && flags != NULL && release != NULL)
-			UT_ASSERT(content < flags && flags < release);
+		if (content != NULL && ownership != NULL && flags != NULL && release != NULL)
+			UT_ASSERT(content < ownership && ownership < flags && flags < release);
 	}
 	if (resident_stamp != NULL) {
 		const char *content = strstr(
@@ -1408,6 +1410,8 @@ UT_TEST(test_bufmgr_pcm_x_holder_gate_retry_is_bounded_outside_content_lock)
 	static const char *const wait_contract[]
 		= { "content_lock == NULL || LWLockHeldByMe(content_lock)",
 			"ereport(ERROR",
+			"cluster_pcm_x_nested_wait_guard_before_block()",
+			"cluster_bufmgr_pcm_x_holder_report_failure(",
 			"cluster_pcm_x_holder_retry_delay_ms(wait_index)",
 			"CHECK_FOR_INTERRUPTS()",
 			"WaitLatch(MyLatch",