sqlrush · sqlrush · Sep 17, 2026 · Sep 17, 2026 · Sep 17, 2026 · Sep 17, 2026
@@ -1885,12 +1885,12 @@ cluster_heap_itl_prepare_prepared_undo(Relation relation, Buffer buffer, HeapTup
 			ClusterTxwResult wait_result;
 			const char *wait_reason;
 
-			/* The existing wait owner releases content before exact blocker
-			 * resolution. Its wake cannot validate this DML's old target. */
+			/* The wait releases content, so its wake requires fresh page/tuple
+			 * qualification, not receipt invalidation. Preserve the receipt
+			 * until the requalified pending-target checks prove a change. */
 			wait_result = cluster_heap_itl_wait_capacity_after_census(
 				buffer, buffer, buffer, xid, true, capacity_wait_deadline_us, &wait_reason);
 			*content_unlocked = true;
-			*targets_invalidated = true;
 			if (wait_result == CLUSTER_TXW_RESOLVED || wait_result == CLUSTER_TXW_RETRY)
 				return CLUSTER_HEAP_PREPARED_UNDO_RETRY_REQUIRED;
 			ereport(ERROR,

@@ -680,19 +680,26 @@ cluster_lms_outbound_resource_x_intent_pump(void)
 static bool
 lms_outbound_r4_refusal_header_valid(const GcsBlockReplyHeader *header)
 {
+	int32 forwarding_master;
 	int i;
 
 	if (header == NULL || !GcsBlockReplyStatusIsR4Refusal((GcsBlockReplyStatus)header->status)
 		|| header->request_id == 0 || header->checksum != 0 || header->sender_node < 0
 		|| header->sender_node >= CLUSTER_MAX_NODES || header->requester_backend_id <= 0
-		|| header->transition_id != (uint8)PCM_TRANS_N_TO_S
-		|| GcsBlockReplyHeaderGetForwardingMasterNode(header)
-			   != GCS_BLOCK_REPLY_NO_FORWARDING_MASTER)
+		|| header->transition_id != (uint8)PCM_TRANS_N_TO_S)
+		return false;
+	forwarding_master = GcsBlockReplyHeaderGetForwardingMasterNode(header);
+	if (forwarding_master < GCS_BLOCK_REPLY_NO_FORWARDING_MASTER
+		|| forwarding_master >= CLUSTER_MAX_NODES)
 		return false;
 	for (i = 0; i < (int)sizeof(header->reserved_0); i++)
 		if (header->reserved_0[i] != 0)
 			return false;
-	if (header->status == (uint8)GCS_BLOCK_REPLY_R4_DENIED)
+	/* A holder replies directly with the real forwarding master retained for
+	 * requester authentication. Only a master-produced retry may redirect;
+	 * neither a forwarded refusal nor a denial carries a page LSN. */
+	if (forwarding_master != GCS_BLOCK_REPLY_NO_FORWARDING_MASTER
+		|| header->status == (uint8)GCS_BLOCK_REPLY_R4_DENIED)
 		return header->page_lsn == 0;
 	/* Status 25 optionally carries WRONG_MASTER as node+1.  The encoded
 	 * value is therefore either zero or in [1, CLUSTER_MAX_NODES]. */

@@ -738,6 +738,13 @@ cluster_tx_enqueue_wait_exact(const ClusterTxLocator *locator, int effective_tim
 					final_reason = CLUSTER_TX_RESOLVE_RF_DEFERRED;
 					break;
 				}
+				/* Consume against the published wait before another resolve.
+				 * Leave through the same exact cleanup as every other exit. */
+				if (cluster_cancel_token_consume()) {
+					result = CLUSTER_TXW_DEADLOCK;
+					final_reason = CLUSTER_TX_RESOLVE_NONE;
+					break;
+				}
 				memset(&resolution, 0, sizeof(resolution));
 				current_outcome = cluster_tx_resolve_exact(
 					&target_locator, CLUSTER_TX_RESOLVE_ROW_WAIT, &resolution, &current_reason);

@@ -3567,8 +3567,25 @@ CLUSTER_LMS_OUTBOUND_O = $(top_builddir)/src/backend/cluster/cluster_lms_outboun
 cluster_lms_outbound_test.o: $(top_srcdir)/src/backend/cluster/cluster_lms_outbound.c
 	$(CC) $(CFLAGS) $(CPPFLAGS) -DCLUSTER_LMS_OUTBOUND_UNIT_TEST -c $< -o $@
 
+# Run the real refusal producers and decoder against the real outbound ring.
+# Fail extraction if an owner disappears or is duplicated; do not mirror its
+# implementation in a fixture or stub the enqueue that joins these components.
+test_cluster_r4_refusal_handoff.inc: $(top_srcdir)/src/backend/cluster/cluster_gcs_block.c Makefile
+	awk '/^typedef enum ClusterGcsBlockReplyDomain/ { emit=1; domain++ } \
+	     /^typedef struct GcsBlockR4ReplyExpectation/ { emit=1; expectation++ } \
+	     /^gcs_block_compute_checksum\(/ { print "static uint32"; emit=1; checksum++ } \
+	     /^#define R4_CR_REQUIRED_HELLO_CAPS/ { cap=1; caps++ } \
+	     cap { print; if ($$0 !~ /\\$$/) cap=0; next } \
+	     /^gcs_block_r4_refusal_status_for_build\(/ { print "static bool"; emit=1; status++ } \
+	     /^gcs_block_decode_r4_reply_payload\(/ { print "static bool"; emit=1; decode++ } \
+	     /^gcs_block_r4_publish_refusal\(/ { print "static bool"; emit=1; master++ } \
+	     /^gcs_block_r4_publish_holder_refusal\(/ { print "static bool"; emit=1; holder++ } \
+	     emit { print } /^}/ { emit=0 } \
+	     END { if (domain != 1 || expectation != 1 || checksum != 1 || caps != 1 || status != 1 || decode != 1 || master != 1 || holder != 1 || emit || cap) exit 1 }' $< > $@.tmp
+	mv $@.tmp $@
+
 test_cluster_lms_outbound: test_cluster_lms_outbound.c unit_test.h \
-		$(CLUSTER_VERSION_O) cluster_lms_outbound_test.o
+		test_cluster_r4_refusal_handoff.inc $(CLUSTER_VERSION_O) cluster_lms_outbound_test.o
 	$(CC) $(CFLAGS) $(CPPFLAGS) $< \
 		$(CLUSTER_VERSION_O) cluster_lms_outbound_test.o \
 		$(top_builddir)/src/common/libpgcommon_srv.a \

@@ -16,7 +16,7 @@
   "current_product_snapshot": {
     "algorithm": "sha256-canonical-path-blob-v1",
     "path_count": 2225,
-    "sha256": "526de2a0572d0c41dccfe05c6c01f1779646383a15a54db61f91e7cb3dde4d4a"
+    "sha256": "a33690006785ed01f7bfef085653bd6b916ebe970de64d555c3436e9d11a4e3b"
   },
   "gates": {
     "L1": {

@@ -1,6 +1,6 @@
 /*-------------------------------------------------------------------------
  * test_cluster_heap_prepare_diagnostic.c
- *    Exact production prepare refusal attribution, without changing decisions.
+ *    Exact production prepare refusals and capacity-wait requalification.
  *
  * Portions Copyright (c) 2026, pgrac contributors
  * Author: SqlRush <sqlrush@gmail.com>
@@ -157,9 +157,7 @@ check_prepare(int cause, bool applied, int expected, const char *reason)
 													&invalidated);
 #endif
 	UT_ASSERT_EQ(result, expected);
-	UT_ASSERT_EQ(invalidated,
-				 cause == 7 || cause == 9
-					 || (capacity_requested && request_lock_only && cause == 1 && !applied));
+	UT_ASSERT_EQ(invalidated, cause == 7 || cause == 9);
 	UT_ASSERT_EQ(receipt.ctrc_applied_mask, applied ? 1 : 0);
 	if (reason != NULL) {
 		UT_ASSERT(observed != NULL);
@@ -212,6 +210,72 @@ UT_TEST(unproved_or_failed_wait_is_not_a_retry_success)
 	capacity_requested = false;
 }
 
+UT_TEST(capacity_wait_preserves_receipt_and_original_budgets)
+{
+	const ClusterTxwResult wakes[] = { CLUSTER_TXW_RESOLVED, CLUSTER_TXW_RETRY };
+
+	for (unsigned i = 0; i < lengthof(wakes); i++) {
+		ClusterUndoRecordPrepareReceipt receipt = { 0 }, before;
+		const char *reason = "old cause must not leak";
+		uint64 capacity_deadline = 12345;
+		bool invalidated = true;
+
+		/* The wait must not cancel or renew this previously prepared target.
+		 * Real expired-budget requalification is covered by undo_record. */
+		receipt.absolute_deadline_us = 100;
+		receipt.reservation_sequence = 17;
+		receipt.ctrc_pending_mask = receipt.ctrc_prepared_mask = 1;
+		before = receipt;
+		fault = 1;
+		wait_result = wakes[i];
+		wait_calls = 0;
+		content_unlocked = false;
+		UT_ASSERT_EQ(cluster_heap_itl_prepare_prepared_undo(NULL, 1, NULL, 700, true, &receipt, 64,
+															&invalidated, &reason,
+															&capacity_deadline, &content_unlocked),
+					 CLUSTER_HEAP_PREPARED_UNDO_RETRY_REQUIRED);
+		UT_ASSERT(!invalidated);
+		UT_ASSERT(content_unlocked);
+		UT_ASSERT_EQ(wait_calls, 1);
+		UT_ASSERT(reason == NULL);
+		UT_ASSERT_EQ(capacity_deadline, 12345);
+		UT_ASSERT_EQ(memcmp(&receipt, &before, sizeof(receipt)), 0);
+	}
+}
+
+UT_TEST(capacity_wake_does_not_bypass_fresh_target_recheck)
+{
+	const int rechecks[] = { 0, 7, 9 };
+
+	for (unsigned i = 0; i < lengthof(rechecks); i++) {
+		ClusterUndoRecordPrepareReceipt receipt = { 0 };
+		const char *reason = NULL;
+		uint64 capacity_deadline = 12345;
+		bool invalidated = false;
+
+		receipt.ctrc_pending_mask = receipt.ctrc_prepared_mask = 1;
+		fault = 1;
+		wait_result = CLUSTER_TXW_RESOLVED;
+		UT_ASSERT_EQ(cluster_heap_itl_prepare_prepared_undo(NULL, 1, NULL, 700, true, &receipt, 64,
+															&invalidated, &reason,
+															&capacity_deadline, &content_unlocked),
+					 CLUSTER_HEAP_PREPARED_UNDO_RETRY_REQUIRED);
+		UT_ASSERT(!invalidated);
+		/* Simulate the caller's fresh page bracket. Actual pending-target
+		 * mismatch still invalidates; a wake alone never publishes READY. */
+		fault = rechecks[i];
+		UT_ASSERT_EQ(cluster_heap_itl_prepare_prepared_undo(NULL, 1, NULL, 700, true, &receipt, 64,
+															&invalidated, &reason,
+															&capacity_deadline, &content_unlocked),
+					 i == 0 ? CLUSTER_HEAP_PREPARED_UNDO_READY
+							: CLUSTER_HEAP_PREPARED_UNDO_RETRY_REQUIRED);
+		UT_ASSERT_EQ(invalidated, i != 0);
+		UT_ASSERT(!content_unlocked);
+		UT_ASSERT_EQ(receipt.ctrc_applied_mask, 0);
+		UT_ASSERT_EQ(capacity_deadline, 12345);
+	}
+}
+
 UT_TEST(refusals_have_unique_exact_causes)
 {
 	check_prepare(1, false, CLUSTER_HEAP_PREPARED_UNDO_REFUSED, "ITL_CAPACITY_REFUSED");
@@ -236,11 +300,13 @@ UT_TEST(success_and_preapply_retries_remain_unchanged)
 int
 main(void)
 {
-	UT_PLAN(4);
+	UT_PLAN(6);
 	UT_RUN(refusals_have_unique_exact_causes);
 	UT_RUN(success_and_preapply_retries_remain_unchanged);
 	UT_RUN(full_lock_capacity_waits_then_requalifies_without_apply);
 	UT_RUN(unproved_or_failed_wait_is_not_a_retry_success);
+	UT_RUN(capacity_wait_preserves_receipt_and_original_budgets);
+	UT_RUN(capacity_wake_does_not_bypass_fresh_target_recheck);
 	UT_DONE();
 	return ut_failed_count == 0 ? 0 : 1;
 }