stage8: integrate R4 admission gating and CR routing

src/backend/access/heap/heapam.c

@@ -3218,9 +3218,21 @@ cluster_heap_writer_wait_failclosed(Relation relation, Buffer buffer, HeapTuple
 	 * holder (lock-only / writer).
 	 */
 	{
-		bool		tt_found = cluster_tt_status_lookup_exact(&ckey, &cres);
-		bool		tt_resolved = tt_found && cres.authoritative;
-		bool		tt_terminal = tt_resolved
+		ClusterTTStatusSourceRequest source_request;
+		ClusterTTStatusSourceResult source_result;
+		bool		tt_found;
+		bool		tt_resolved;
+		bool		tt_terminal;
+
+		memset(&source_request, 0, sizeof(source_request));
+		source_request.key = &ckey;
+		tt_found = cluster_tt_status_source_dispatch(CLUSTER_TT_SOURCE_LOOKUP, &source_request,
+												   &source_result)
+			== CLUSTER_SEMANTIC_ADMISSION_OK
+			&& source_result.bool_value;
+		cres = source_result.lookup;
+		tt_resolved = tt_found && cres.authoritative;
+		tt_terminal = tt_resolved
 			&& (cres.status == CLUSTER_TT_STATUS_COMMITTED
 				|| cres.status == CLUSTER_TT_STATUS_ABORTED
 				|| cres.status == CLUSTER_TT_STATUS_CLEANED_OUT);
@@ -6766,6 +6778,8 @@ heap_lock_tuple(Relation relation, HeapTuple tuple,
 					{
 						ClusterTTStatusKey ckey;
 						ClusterTTStatusResult cres;
+						ClusterTTStatusSourceRequest source_request;
+						ClusterTTStatusSourceResult source_result;
 						bool		tt_found;
 						bool		tt_resolved;
 						bool		tt_terminal;
@@ -6803,7 +6817,13 @@ heap_lock_tuple(Relation relation, HeapTuple tuple,
 						 * the HOLDER node); never the generic resid encoder
 						 * which would key on the local node (G1).
 						 */
-						tt_found = cluster_tt_status_lookup_exact(&ckey, &cres);
+						memset(&source_request, 0, sizeof(source_request));
+						source_request.key = &ckey;
+						tt_found = cluster_tt_status_source_dispatch(CLUSTER_TT_SOURCE_LOOKUP,
+															   &source_request, &source_result)
+							== CLUSTER_SEMANTIC_ADMISSION_OK
+							&& source_result.bool_value;
+						cres = source_result.lookup;
 						tt_resolved = tt_found && cres.authoritative;
 						tt_terminal = tt_resolved
 							&& (cres.status == CLUSTER_TT_STATUS_COMMITTED

src/backend/access/heap/heapam_visibility.c

@@ -1633,17 +1633,31 @@ cluster_remote_live_xmax_keeps_visible(Buffer buffer, HeapTupleHeader tuple, Sna
 				mx_origin = cluster_mxid_origin_slot((MultiXactId)HeapTupleHeaderGetRawXmax(tuple));
 
 			if (mx_origin < 0) {
+				ClusterMultiXactSourceResult source_result;
+
 				/* PGRAC: spec-7.1 D0 census — foreign-multi refuse leg. */
-				cluster_multixact_note_underivable_read();
+				(void)cluster_multixact_source_dispatch(
+					CLUSTER_MULTI_SOURCE_NOTE_UNDERIVABLE_READ, NULL, &source_result);
 				cluster_vis53r97_note_multi_unresolvable();
 				return -1;
 			}
 			if (mx_origin != cluster_node_id) {
-				bool mx_hit = false;
+				ClusterMultiXactSourceRequest source_request;
+				ClusterMultiXactSourceResult source_result;
+				ClusterSemanticAdmissionResult source_admission;
+				ClusterVisibilityDecision mx_decision = CLUSTER_VISIBILITY_UNKNOWN;
+
+				memset(&source_request, 0, sizeof(source_request));
+				source_request.origin_slot = (uint16)mx_origin;
+				source_request.mxid = (MultiXactId)HeapTupleHeaderGetRawXmax(tuple);
+				source_request.snapshot = snapshot;
+				source_admission = cluster_multixact_source_dispatch(
+					CLUSTER_MULTI_SOURCE_REMOTE_XMAX_RESOLVE, &source_request, &source_result);
+				if (source_admission == CLUSTER_SEMANTIC_ADMISSION_OK) {
+					mx_decision = source_result.visibility;
+				}
 
-				switch (cluster_multixact_remote_xmax_resolve(
-					(uint16)mx_origin, (MultiXactId)HeapTupleHeaderGetRawXmax(tuple), snapshot,
-					&mx_hit)) {
+				switch (mx_decision) {
 				case CLUSTER_VISIBILITY_VISIBLE:
 					return 1; /* no committed updater hides the row */
 				case CLUSTER_VISIBILITY_INVISIBLE:
@@ -2158,8 +2172,11 @@ HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot, Buffer buffer)
 					mx_origin = cluster_mxid_origin_slot((MultiXactId)raw_xmax_multi);
 
 				if (mx_origin < 0) {
+					ClusterMultiXactSourceResult source_result;
+
 					/* D3-0 floor: origin not provable -> fail closed */
-					cluster_multixact_note_underivable_read();
+					(void)cluster_multixact_source_dispatch(
+						CLUSTER_MULTI_SOURCE_NOTE_UNDERIVABLE_READ, NULL, &source_result);
 					ereport(ERROR,
 							(errcode(ERRCODE_CLUSTER_MULTIXACT_MEMBER_OVERLAY_MISS),
 							 errmsg("cluster multixact %u cannot be attributed to an origin node",
@@ -2170,10 +2187,25 @@ HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot, Buffer buffer)
 									 "cluster.multi_xmax_remote_resolve. Retry the transaction.")));
 				}
 				if (mx_origin != cluster_node_id) {
+					ClusterMultiXactSourceRequest source_request;
+					ClusterMultiXactSourceResult source_result;
+					ClusterSemanticAdmissionResult source_admission;
+					ClusterVisibilityDecision mx_decision = CLUSTER_VISIBILITY_UNKNOWN;
 					bool mx_hit = false;
 
-					switch (cluster_multixact_remote_xmax_resolve(
-						(uint16)mx_origin, (MultiXactId)raw_xmax_multi, snapshot, &mx_hit)) {
+					memset(&source_request, 0, sizeof(source_request));
+					source_request.origin_slot = (uint16)mx_origin;
+					source_request.mxid = (MultiXactId)raw_xmax_multi;
+					source_request.snapshot = snapshot;
+					source_admission = cluster_multixact_source_dispatch(
+						CLUSTER_MULTI_SOURCE_REMOTE_XMAX_RESOLVE, &source_request,
+						&source_result);
+					if (source_admission == CLUSTER_SEMANTIC_ADMISSION_OK) {
+						mx_decision = source_result.visibility;
+						mx_hit = source_result.overlay_hit;
+					}
+
+					switch (mx_decision) {
 					case CLUSTER_VISIBILITY_VISIBLE:
 						return true;
 					case CLUSTER_VISIBILITY_INVISIBLE:

src/backend/access/transam/multixact.c

@@ -896,6 +896,9 @@ MultiXactIdCreateFromMembers(int nmembers, MultiXactMember *members)
 		&& nmembers <= CLUSTER_MULTIXACT_HINT_MAX_MEMBERS) {
 		ClusterMultiXactMember c_members[CLUSTER_MULTIXACT_HINT_MAX_MEMBERS];
 		ClusterMultiXactKey c_key;
+		ClusterMultiXactSourceRequest multi_request;
+		ClusterMultiXactSourceResult multi_result;
+		ClusterTTStatusHintSourceRequest hint_request;
 		bool all_local = true;
 		int i;
 
@@ -926,8 +929,21 @@ MultiXactIdCreateFromMembers(int nmembers, MultiXactMember *members)
 			c_key.multixact_id = multi;
 			c_key.cluster_epoch = (uint32)cluster_epoch_get_current();
 
-			if (cluster_multixact_member_overlay_install(&c_key, (uint16)nmembers, c_members))
-				cluster_tt_status_hint_emit_multixact_overlay(&c_key, (uint16)nmembers, c_members);
+			memset(&multi_request, 0, sizeof(multi_request));
+			multi_request.key = &c_key;
+			multi_request.member_count = (uint16)nmembers;
+			multi_request.members = c_members;
+			if (cluster_multixact_source_dispatch(CLUSTER_MULTI_SOURCE_OVERLAY_INSTALL,
+											 &multi_request, &multi_result)
+					== CLUSTER_SEMANTIC_ADMISSION_OK
+				&& multi_result.bool_value) {
+				memset(&hint_request, 0, sizeof(hint_request));
+				hint_request.multi_key = &c_key;
+				hint_request.member_count = (uint16)nmembers;
+				hint_request.members = c_members;
+				(void)cluster_tt_status_hint_source_dispatch(
+					CLUSTER_TT_HINT_SOURCE_EMIT_MULTIXACT_OVERLAY, &hint_request);
+			}
 		}
 	}
 #endif
@@ -1200,7 +1216,10 @@ GetNewMultiXactId(int nmembers, MultiXactOffset *offset)
 		 */
 		if (cluster_mxid_halfspace_exceeded(result, floor_mxid)
 			|| cluster_cr_injection_armed("cluster-mxid-halfspace-hard-limit", NULL)) {
-			cluster_multixact_note_halfspace_refuse();
+			ClusterMultiXactSourceResult source_result;
+
+			(void)cluster_multixact_source_dispatch(CLUSTER_MULTI_SOURCE_NOTE_HALFSPACE_REFUSE,
+											 NULL, &source_result);
 			ereport(
 				ERROR,
 				(errcode(ERRCODE_CLUSTER_MXID_HALFSPACE_LIMIT),
@@ -1329,7 +1348,10 @@ GetNewMultiXactId(int nmembers, MultiXactOffset *offset)
 				result = floor_mxid;
 			result = cluster_mxid_next_striped(result, mxid_stripe_slot);
 			if (cluster_mxid_halfspace_exceeded(result, floor_mxid)) {
-				cluster_multixact_note_halfspace_refuse();
+				ClusterMultiXactSourceResult source_result;
+
+				(void)cluster_multixact_source_dispatch(
+					CLUSTER_MULTI_SOURCE_NOTE_HALFSPACE_REFUSE, NULL, &source_result);
 				ereport(
 					ERROR,
 					(errcode(ERRCODE_CLUSTER_MXID_HALFSPACE_LIMIT),

src/backend/cluster/cluster_cr.c

@@ -1540,19 +1540,27 @@ cr_construct_from_copy(char *dst_page, SCN read_scn, RelFileLocator cur_locator,
 		NodeId head_origin = uba_origin_node_id(chains[0].undo_segment_head);
 
 		if (cluster_cr_coordinator_classify_origin(head_origin) == CR_COORD_ORIGIN_RUNTIME_REMOTE) {
-			bool partial = false;
+			ClusterR4SourceCrRequest source_request;
+			ClusterR4SourceCrResult source_result;
 			BufferTag tag;
 
 			InitBufferTag(&tag, &cur_locator, cur_fork, cur_block);
-			if (!cluster_gcs_block_cr_fetch_and_wait(tag, read_scn, (int32)head_origin, dst_page,
-													 &partial)) {
+			memset(&source_request, 0, sizeof(source_request));
+			source_request.tag = tag;
+			source_request.read_scn = read_scn;
+			source_request.origin_node = (int32)head_origin;
+			source_request.dst_page = dst_page;
+			if (cluster_r4_source_cr_dispatch(CLUSTER_R4_SOURCE_CR_FETCH, &source_request,
+											  &source_result)
+					!= CLUSTER_SEMANTIC_ADMISSION_OK
+				|| !source_result.fetched) {
 				if (CRShared != NULL)
 					pg_atomic_fetch_add_u64(&CRShared->cr_remote_failed_count, 1);
 				/* Unchanged spec-5.57 refusal: 53R9G + coordinator counters. */
 				cr_coordinator_refuse_runtime_remote((int)head_origin);
 			}
 
-			if (!partial) {
+			if (!source_result.partial) {
 				/* FULL: dst holds the origin-finished CR page (it already ran
 				 * prune + walk + durable resolve there). */
 				if (CRShared != NULL) {
@@ -2416,6 +2424,9 @@ cluster_cr_resolve_xmax_commit_scn(const char *cr_page, uint8 itl_idx, Transacti
 			if (cluster_itl_get_tt_ref(page, itl_idx, &ref)) {
 				ClusterTTStatusKey key;
 				ClusterTTStatusResult result;
+				ClusterTTStatusSourceRequest source_request;
+				ClusterTTStatusSourceResult source_result;
+				bool found;
 
 				memset(&key, 0, sizeof(key));
 				key.origin_node_id = ref.origin_node_id;
@@ -2424,7 +2435,14 @@ cluster_cr_resolve_xmax_commit_scn(const char *cr_page, uint8 itl_idx, Transacti
 				key.cluster_epoch = ref.cluster_epoch;
 				key.local_xid = cr_xmax;
 
-				if (cluster_tt_status_lookup_exact(&key, &result) && result.authoritative
+				memset(&source_request, 0, sizeof(source_request));
+				source_request.key = &key;
+				found = cluster_tt_status_source_dispatch(CLUSTER_TT_SOURCE_LOOKUP, &source_request,
+														 &source_result)
+						== CLUSTER_SEMANTIC_ADMISSION_OK
+					&& source_result.bool_value;
+				result = source_result.lookup;
+				if (found && result.authoritative
 					&& (result.status == CLUSTER_TT_STATUS_COMMITTED
 						|| result.status == CLUSTER_TT_STATUS_CLEANED_OUT)
 					&& SCN_VALID(result.commit_scn)) {

src/backend/cluster/cluster_cr_server.c

@@ -150,6 +150,8 @@ cluster_cr_build_on_holder(const BufferTag *tag, SCN read_scn, char dst[BLCKSZ],
 	ClusterSemanticAdmissionToken admission;
 	ClusterSemanticAdmissionResult admission_result;
 	ClusterBufmgrGcsCopyRefusal refusal = CLUSTER_BUFMGR_GCS_COPY_REFUSAL_NONE;
+	ClusterCrBuildReason reason = CLUSTER_CR_BUILD_NONE;
+	ClusterCrBuildResult result = CLUSTER_CR_BUILD_FAIL_CLOSED;
 	PGAlignedBlock current_copy;
 	XLogRecPtr page_lsn = InvalidXLogRecPtr;
 	SCN page_scn = InvalidScn;
@@ -174,61 +176,73 @@ cluster_cr_build_on_holder(const BufferTag *tag, SCN read_scn, char dst[BLCKSZ],
 		return CLUSTER_CR_BUILD_RETRYABLE;
 	}
 
-	if (tag == NULL || !SCN_VALID(read_scn)) {
-		*reason_out = CLUSTER_CR_BUILD_PROTOCOL;
-		cluster_semantic_activation_leave(&admission);
-		return CLUSTER_CR_BUILD_FAIL_CLOSED;
-	}
+	PG_TRY();
+	{
+		if (tag == NULL || !SCN_VALID(read_scn)) {
+			reason = CLUSTER_CR_BUILD_PROTOCOL;
+			goto admitted_done;
+		}
 
-	if (!cluster_bufmgr_copy_block_for_r4_cr(*tag, InvalidScn, &page_lsn, &page_scn,
+		if (!cluster_bufmgr_copy_block_for_r4_cr(*tag, InvalidScn, &page_lsn, &page_scn,
 											 current_copy.data, &refusal)) {
-		cluster_semantic_activation_leave(&admission);
-		if (refusal == CLUSTER_BUFMGR_GCS_COPY_REFUSAL_INVALID_ARGUMENT) {
-			*reason_out = CLUSTER_CR_BUILD_PROTOCOL;
-			return CLUSTER_CR_BUILD_FAIL_CLOSED;
-		}
-		switch (refusal) {
+			if (refusal == CLUSTER_BUFMGR_GCS_COPY_REFUSAL_INVALID_ARGUMENT) {
+				reason = CLUSTER_CR_BUILD_PROTOCOL;
+				goto admitted_done;
+			}
+			switch (refusal) {
 			case CLUSTER_BUFMGR_GCS_COPY_REFUSAL_NOT_RESIDENT:
 			case CLUSTER_BUFMGR_GCS_COPY_REFUSAL_CURRENT_INVALID:
 			case CLUSTER_BUFMGR_GCS_COPY_REFUSAL_CONTENT_LOCK_FIRST:
 			case CLUSTER_BUFMGR_GCS_COPY_REFUSAL_CONTENT_LOCK_SECOND:
 			case CLUSTER_BUFMGR_GCS_COPY_REFUSAL_OWNERSHIP_REVOKE_BUSY:
-				*reason_out = CLUSTER_CR_BUILD_HOLDER_MOVED;
-				return CLUSTER_CR_BUILD_RETRYABLE;
+				reason = CLUSTER_CR_BUILD_HOLDER_MOVED;
+				result = CLUSTER_CR_BUILD_RETRYABLE;
+				goto admitted_done;
 			default:
-				*reason_out = CLUSTER_CR_BUILD_PROTOCOL;
-				return CLUSTER_CR_BUILD_FAIL_CLOSED;
+				reason = CLUSTER_CR_BUILD_PROTOCOL;
+				goto admitted_done;
+			}
 		}
-	}
 
-	PG_TRY();
-	{
-		cluster_cr_construct_page_for_server(current_copy.data, read_scn, *tag, dst, &partial);
-		constructed = true;
+		PG_TRY();
+		{
+			cluster_cr_construct_page_for_server(current_copy.data, read_scn, *tag, dst, &partial);
+			constructed = true;
+		}
+		PG_CATCH();
+		{
+			constructed = false;
+			FlushErrorState();
+		}
+		PG_END_TRY();
+
+		if (!cluster_semantic_activation_recheck(&admission)) {
+			memset(dst, 0, BLCKSZ);
+			reason = CLUSTER_CR_BUILD_RF_DEFERRED;
+			result = CLUSTER_CR_BUILD_RETRYABLE;
+			goto admitted_done;
+		}
+
+		if (!constructed || partial) {
+			memset(dst, 0, BLCKSZ);
+			reason = CLUSTER_CR_BUILD_BAD_UNDO;
+			goto admitted_done;
+		}
+
+		reason = CLUSTER_CR_BUILD_NONE;
+		result = CLUSTER_CR_BUILD_FULL;
+
+admitted_done:
+		;
 	}
-	PG_CATCH();
+	PG_FINALLY();
 	{
-		constructed = false;
-		FlushErrorState();
-	}
-	PG_END_TRY();
-
-	if (!cluster_semantic_activation_recheck(&admission)) {
-		memset(dst, 0, BLCKSZ);
-		*reason_out = CLUSTER_CR_BUILD_RF_DEFERRED;
 		cluster_semantic_activation_leave(&admission);
-		return CLUSTER_CR_BUILD_RETRYABLE;
-	}
-	cluster_semantic_activation_leave(&admission);
-
-	if (!constructed || partial) {
-		memset(dst, 0, BLCKSZ);
-		*reason_out = CLUSTER_CR_BUILD_BAD_UNDO;
-		return CLUSTER_CR_BUILD_FAIL_CLOSED;
 	}
+	PG_END_TRY();
 
-	*reason_out = CLUSTER_CR_BUILD_NONE;
-	return CLUSTER_CR_BUILD_FULL;
+	*reason_out = reason;
+	return result;
 }
 
 static ClusterCrServerShared *CrServerShared = NULL;
@@ -333,6 +347,23 @@ cluster_lms_cr_submit(const GcsBlockForwardPayload *fwd)
 	return false; /* all slots busy — fail closed, requester retries/refuses */
 }
 
+/*
+ * cluster_lms_cr_submit_r4 — typed R4 FORWARD96 holder-submit boundary.
+ *
+ * D3 deliberately does not reinterpret the 96-byte route proof as the
+ * legacy 64-byte payload: that would discard the proof.  D4 owns the future
+ * stable-copy and slot-proof positive integration, so this pre-D4 boundary
+ * remains fail closed without allocating a slot or mutating shared state.
+ */
+bool
+cluster_lms_cr_submit_r4(const ClusterR4CrForwardPayload *forward)
+{
+	if (forward == NULL)
+		return false;
+
+	return false;
+}
+
 /*
  * cluster_lms_undo_fetch_submit — CONTROL-plane park (spec-6.12i D-i1).
  *