fix(gcs): rescue STALE master-direct ship via shared-storage fallback

src/backend/cluster/cluster_debug.c

@@ -2282,6 +2282,9 @@ dump_gcs(ReturnSetInfo *rsinfo)
 			 fmt_int64((int64)cluster_gcs_get_lost_write_invalidscn_failclosed_count()));
 	emit_row(rsinfo, "gcs", "lost_write_not_scn_tracked_skip_count",
 			 fmt_int64((int64)cluster_gcs_get_lost_write_not_scn_tracked_skip_count()));
+	/* PGRAC: branch-1 — master-direct stale ship rescued via shared storage. */
+	emit_row(rsinfo, "gcs", "lost_write_master_direct_storage_fallback_count",
+			 fmt_int64((int64)cluster_gcs_get_lost_write_master_direct_storage_fallback_count()));
 	/* PGRAC: spec-5.2 D2 — X-holder read-image ship counter. */
 	emit_row(rsinfo, "gcs", "cf_xheld_read_ship_count",
 			 fmt_int64((int64)cluster_gcs_get_cf_xheld_read_ship_count()));

src/backend/cluster/cluster_gcs_block.c

@@ -257,6 +257,12 @@ typedef struct ClusterGcsBlockShared {
 		lost_write_invalidscn_failclosed_count; /* §2.6 b2: tracked block, shipped InvalidScn */
 	pg_atomic_uint64
 		lost_write_not_scn_tracked_skip_count; /* §2.6 b1: expected InvalidScn → skip */
+	/* PGRAC: branch-1 (S3 step-2 forensics) — a STALE master-direct ship whose
+	 * shared-storage version covers the watermark is rescued to
+	 * GRANTED_STORAGE_FALLBACK instead of DENIED_LOST_WRITE (availability:
+	 * the requester reads storage instead of aborting 53R93).  The refused
+	 * twin (storage unprovable) keeps bumping lost_write_detected_count. */
+	pg_atomic_uint64 lost_write_master_direct_storage_fallback_count;
 	pg_atomic_uint64
 		redo_coverage_required_lsn_zero_count;		 /* serve-gate required_lsn==0 (cold/degrade) */
 	pg_atomic_uint64 redo_coverage_gate_block_count; /* serve-gate not-covered (block) */
@@ -544,6 +550,7 @@ cluster_gcs_block_shmem_init(void)
 		/* PGRAC: spec-2.41 D7 — 4 NEW SCN detector + redo-coverage counters init. */
 		pg_atomic_init_u64(&ClusterGcsBlock->lost_write_invalidscn_failclosed_count, 0);
 		pg_atomic_init_u64(&ClusterGcsBlock->lost_write_not_scn_tracked_skip_count, 0);
+		pg_atomic_init_u64(&ClusterGcsBlock->lost_write_master_direct_storage_fallback_count, 0);
 		pg_atomic_init_u64(&ClusterGcsBlock->redo_coverage_required_lsn_zero_count, 0);
 		pg_atomic_init_u64(&ClusterGcsBlock->redo_coverage_gate_block_count, 0);
 		/* PGRAC: spec-5.2 D2 — X-holder read-image ship counter init. */
@@ -5435,6 +5442,22 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
 		if (cluster_injection_should_skip("cluster-gcs-block-stale-ship"))
 			shipped_scn = InvalidScn;
 
+		/* branch-1 (S3 step-2 forensics) inject — simulate a RETAINED STALE
+		 * RESIDENT (a kept Past Image serving as the grant payload): force the
+		 * SHIPPED pd_block_scn one time-step below the valid watermark so the
+		 * verdict is STALE (§2.6 branch 1) while shared storage keeps its real
+		 * version.  Master-direct site ONLY — the holder-forward twin must not
+		 * fire this, so a test that greens can only have exercised THIS path.
+		 * The predecessor comes from the SCN layer (fails closed to InvalidScn
+		 * = leave the shipped value alone; the test round simply retries). */
+		CLUSTER_INJECTION_POINT("cluster-gcs-block-stale-ship-resident");
+		if (cluster_injection_should_skip("cluster-gcs-block-stale-ship-resident")) {
+			SCN forced_stale_scn = cluster_scn_time_predecessor(expected_scn);
+
+			if (SCN_VALID(forced_stale_scn))
+				shipped_scn = forced_stale_scn;
+		}
+
 		verdict = gcs_block_lost_write_verdict(expected_scn, shipped_scn);
 		if (verdict == GCS_LOST_WRITE_SKIP) {
 			/* spec-2.41 D7 observability — block not SCN-tracked (no fire). */
@@ -5473,6 +5496,103 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
 					wm_have ? wm_prov.epoch : 0, wm_have ? (uint64)wm_prov.old_scn : 0,
 					wm_have ? (uint64)wm_prov.new_scn : 0,
 					wm_have ? (int)(wm_prov.new_scn == expected_scn) : -1)));
+
+			/*
+			 * PGRAC: branch-1 (S3 step-2 forensics) — storage-fallback rescue.
+			 *
+			 *	A STALE verdict here means the master is holding a RETAINED
+			 *	stale resident (a kept Past Image) as the would-be grant
+			 *	payload.  When the shared-storage page already covers the
+			 *	authoritative watermark, the cluster-proven current version is
+			 *	durably readable: convert the reply to
+			 *	GRANTED_STORAGE_FALLBACK (ship no image; page_lsn carries the
+			 *	watermark — the same contract as the state=N grant above) so
+			 *	the requester proves/refreshes its copy through
+			 *	cluster_gcs_block_fallback_verify_refresh instead of aborting
+			 *	53R93 on every hit.  The verdict re-check uses the same
+			 *	gcs_block_lost_write_verdict SCN order the detector itself
+			 *	trusts (spec-2.41 §2.6).
+			 *
+			 *	ANOMALY (shipped InvalidScn on a tracked tag) is NOT rescued:
+			 *	an unstamped resident says the master's own view is broken —
+			 *	stay fail-closed.  A failed/unverifiable storage read keeps
+			 *	the fail-closed DENIED too (Rule 8.A).
+			 */
+			if (verdict == GCS_LOST_WRITE_FAIL_STALE) {
+				SCN storage_scn = InvalidScn;
+				bool storage_read_ok = false;
+				MemoryContext probe_cxt = CurrentMemoryContext;
+
+				/*
+				 * PGRAC: the storage probe is the only disk I/O on this
+				 * self-check path and smgrread can ereport(ERROR) (short
+				 * read on a concurrent truncate/drop, real I/O failure).
+				 * An uncaught throw here would leak the ship image — a
+				 * live_sge borrow is a raw pin outside ResourceOwner
+				 * tracking — and drop the reply after produce_reply already
+				 * applied the PCM transition, wedging the requester.  Catch
+				 * locally and fall through to the fail-closed DENIED arm,
+				 * which releases the image and still replies (Rule 8.A).
+				 */
+				PG_TRY();
+				{
+					storage_read_ok
+						= cluster_bufmgr_read_storage_scn_for_gcs(req->tag, &storage_scn);
+				}
+				PG_CATCH();
+				{
+					ErrorData *edata;
+
+					MemoryContextSwitchTo(probe_cxt);
+					edata = CopyErrorData();
+					FlushErrorState();
+					ereport(LOG, (errmsg_internal(
+									 "cluster_gcs_block: master-direct storage-fallback probe "
+									 "failed; keeping fail-closed DENIED_LOST_WRITE: %s",
+									 edata->message != NULL ? edata->message : "(no message)")));
+					FreeErrorData(edata);
+					storage_read_ok = false;
+					storage_scn = InvalidScn;
+				}
+				PG_END_TRY();
+
+				/* Test hook: pretend storage is unverifiable so the rescue
+				 * refuses and the original fail-closed DENIED ships. */
+				CLUSTER_INJECTION_POINT("cluster-gcs-block-master-direct-fallback-storage-stale");
+				if (cluster_injection_should_skip(
+						"cluster-gcs-block-master-direct-fallback-storage-stale"))
+					storage_scn = InvalidScn;
+
+				if (storage_read_ok
+					&& gcs_block_lost_write_verdict(expected_scn, storage_scn)
+						   == GCS_LOST_WRITE_PASS) {
+					ereport(
+						LOG,
+						(errmsg_internal(
+							"cluster_gcs_block: master-direct stale ship rescued to "
+							"GRANTED_STORAGE_FALLBACK: tag spc=%u db=%u rel=%u block=%u "
+							"fork=%d expected pi_watermark_scn=" UINT64_FORMAT
+							" shipped pd_block_scn=" UINT64_FORMAT
+							" storage pd_block_scn=" UINT64_FORMAT
+							" requester=%d request_id=" UINT64_FORMAT " epoch=" UINT64_FORMAT,
+							req->tag.spcOid, req->tag.dbOid, req->tag.relNumber, req->tag.blockNum,
+							(int)req->tag.forkNum, (uint64)expected_scn, (uint64)shipped_scn,
+							(uint64)storage_scn, req->sender_node, req->request_id, req->epoch)));
+					status = GCS_BLOCK_REPLY_GRANTED_STORAGE_FALLBACK;
+					page_lsn = (XLogRecPtr)expected_scn;
+					gcs_block_release_ship_image(block_payload_release_cb,
+												 block_payload_release_arg);
+					block_payload = NULL;
+					block_payload_lkey = 0;
+					block_payload_release_cb = NULL;
+					block_payload_release_arg = NULL;
+					if (ClusterGcsBlock != NULL)
+						pg_atomic_fetch_add_u64(
+							&ClusterGcsBlock->lost_write_master_direct_storage_fallback_count, 1);
+					goto build_and_send_reply;
+				}
+			}
+
 			status = GCS_BLOCK_REPLY_DENIED_LOST_WRITE;
 			page_lsn = InvalidXLogRecPtr;
 			gcs_block_release_ship_image(block_payload_release_cb, block_payload_release_arg);
@@ -8770,6 +8890,15 @@ cluster_gcs_get_lost_write_avoid_count(void)
 	return ClusterGcsBlock ? pg_atomic_read_u64(&ClusterGcsBlock->lost_write_avoid_count) : 0;
 }
 
+/* PGRAC: branch-1 master-direct storage-fallback rescue accessor. */
+uint64
+cluster_gcs_get_lost_write_master_direct_storage_fallback_count(void)
+{
+	return ClusterGcsBlock ? pg_atomic_read_u64(
+								 &ClusterGcsBlock->lost_write_master_direct_storage_fallback_count)
+						   : 0;
+}
+
 /* PGRAC: spec-2.41 D7 — SCN detector + redo-coverage observability accessors. */
 uint64
 cluster_gcs_get_lost_write_invalidscn_failclosed_count(void)

src/backend/cluster/cluster_inject.c

@@ -661,6 +661,26 @@ static ClusterInjectPoint cluster_injection_points[] = {
 	 *	  (deferred behavioral trigger), activated here for spec-2.41 §6 D.
 	 */
 	{ .name = "cluster-gcs-block-stale-ship" },
+	/*
+	 * branch-1 master-direct storage-fallback (S3 step-2 forensics verdict:
+	 * a master-retained Past Image served as the master-direct grant payload).
+	 *
+	 *	cluster-gcs-block-stale-ship-resident:
+	 *	  Fires ONLY at the master-direct self-check.  SKIP forces the
+	 *	  SHIPPED pd_block_scn to (valid watermark - 1) — §2.6 branch 1
+	 *	  STALE with a real shared-storage version left intact — so the
+	 *	  ship converts to GRANTED_STORAGE_FALLBACK when storage covers the
+	 *	  watermark (lost_write_master_direct_storage_fallback_count grows)
+	 *	  instead of failing the requester with 53R93.
+	 *
+	 *	cluster-gcs-block-master-direct-fallback-storage-stale:
+	 *	  Fires inside the fallback rescue probe.  SKIP forces the storage
+	 *	  pd_block_scn read to InvalidScn so the rescue is REFUSED and the
+	 *	  original fail-closed DENIED_LOST_WRITE ships — proves the rescue
+	 *	  never converts when shared storage cannot be proven current.
+	 */
+	{ .name = "cluster-gcs-block-stale-ship-resident" },
+	{ .name = "cluster-gcs-block-master-direct-fallback-storage-stale" },
 	/*
 	 * spec-2.38 D14 — SI Broadcaster fault injection.
 	 *

src/backend/storage/buffer/bufmgr.c

@@ -6993,6 +6993,36 @@ cluster_bufmgr_probe_block_for_gcs(BufferTag tag)
 	return valid;
 }
 
+/*
+ * Read the shared-storage version for a GCS lost-write rescue probe.  This is
+ * deliberately independent of the local buffer table: a BM_VALID mirror is
+ * residency only and must not be consulted as the storage-version carrier (the
+ * resident copy is exactly what the caller has just proven stale).  The
+ * physical read happens without any PCM/GRD lock held by the caller.
+ *
+ * Returns false (with *out_page_scn = InvalidScn) when the page fails
+ * verification; the caller must then keep its fail-closed verdict.
+ */
+bool
+cluster_bufmgr_read_storage_scn_for_gcs(BufferTag tag, SCN *out_page_scn)
+{
+	PGAlignedBlock scratch;
+	SMgrRelation reln;
+
+	if (out_page_scn != NULL)
+		*out_page_scn = InvalidScn;
+
+	reln = smgropen(BufTagGetRelFileLocator(&tag), InvalidBackendId);
+	smgrread(reln, BufTagGetForkNum(&tag), tag.blockNum, scratch.data);
+	if (!PageIsVerifiedExtended((Page) scratch.data, tag.blockNum,
+								PIV_LOG_WARNING | PIV_REPORT_STAT))
+		return false;
+
+	if (out_page_scn != NULL)
+		*out_page_scn = ((PageHeader) scratch.data)->pd_block_scn;
+	return true;
+}
+
 /*
  * Copy the 8KB block bytes for `tag` into *dst, flushing WAL up to the
  * page's LSN before reading the bytes (HC82 I-WAL-before-ship).  Sets

src/include/cluster/cluster_gcs_block.h

@@ -1827,6 +1827,7 @@ StaticAssertDecl(GCS_BLOCK_DATA_SIZE == BLCKSZ,
 #include "access/xlogdefs.h"		  /* XLogRecPtr */
 #include "cluster/cluster_pcm_lock.h" /* PcmLockMode for invalidate helper */
 extern bool cluster_bufmgr_probe_block_for_gcs(BufferTag tag);
+extern bool cluster_bufmgr_read_storage_scn_for_gcs(BufferTag tag, SCN *out_page_scn);
 extern bool cluster_bufmgr_copy_block_for_gcs(BufferTag tag, XLogRecPtr *out_page_lsn, char *dst);
 extern bool cluster_bufmgr_borrow_block_for_gcs_live_sge(BufferTag tag, XLogRecPtr *out_page_lsn,
 														 void **out_page_addr,
@@ -2334,6 +2335,8 @@ extern uint64 cluster_gcs_get_lost_write_detected_count(void);
 extern uint64 cluster_gcs_get_lost_write_avoid_count(void);
 /* PGRAC: spec-2.41 D7 — SCN detector + redo-coverage observability accessors. */
 extern uint64 cluster_gcs_get_lost_write_invalidscn_failclosed_count(void);
+/* PGRAC: branch-1 master-direct storage-fallback rescue accessor. */
+extern uint64 cluster_gcs_get_lost_write_master_direct_storage_fallback_count(void);
 extern uint64 cluster_gcs_get_lost_write_not_scn_tracked_skip_count(void);
 extern uint64 cluster_gcs_get_redo_coverage_required_lsn_zero_count(void);
 extern uint64 cluster_gcs_get_redo_coverage_gate_block_count(void);