From 5113e8fb35aad5373d057daac4a0db3641c1e6e3 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Thu, 17 Sep 2026 18:29:52 +0800 Subject: [PATCH 1/4] fix(cluster): deliver holder CR refusals through outbound ring --- src/backend/cluster/cluster_lms_outbound.c | 15 +- src/test/cluster_unit/Makefile | 19 +- .../cluster_unit/test_cluster_lms_outbound.c | 219 +++++++++++++++++- 3 files changed, 246 insertions(+), 7 deletions(-) diff --git a/src/backend/cluster/cluster_lms_outbound.c b/src/backend/cluster/cluster_lms_outbound.c index 6431cc8e13..062d03d3ab 100644 --- a/src/backend/cluster/cluster_lms_outbound.c +++ b/src/backend/cluster/cluster_lms_outbound.c @@ -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]. */ diff --git a/src/test/cluster_unit/Makefile b/src/test/cluster_unit/Makefile index 3b584551b3..e9fd6c7753 100644 --- a/src/test/cluster_unit/Makefile +++ b/src/test/cluster_unit/Makefile @@ -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 \ diff --git a/src/test/cluster_unit/test_cluster_lms_outbound.c b/src/test/cluster_unit/test_cluster_lms_outbound.c index 60a3f41d7f..6ccfe6c615 100644 --- a/src/test/cluster_unit/test_cluster_lms_outbound.c +++ b/src/test/cluster_unit/test_cluster_lms_outbound.c @@ -55,9 +55,11 @@ #include "cluster/cluster_lms.h" #include "cluster/cluster_clean_leave.h" #include "cluster/cluster_pcm_x_bufmgr.h" +#include "cluster/cluster_r4_observe.h" #include "cluster/cluster_shmem.h" #include "cluster/cluster_sf_dep.h" #include "miscadmin.h" +#include "port/pg_crc32c.h" #include "storage/lwlock.h" #include "storage/shmem.h" @@ -555,6 +557,27 @@ static uint8 ut_local_dispatch_marker = 0; static int ut_direct_zero_reply_count = 0; static GcsBlockReplyHeader ut_direct_zero_reply_header; static int ut_checksum_call_count = 0; +static bool ut_r4_real_checksum = false; +static char ut_r4_reply_payload[GCS_BLOCK_REPLY_PAYLOAD_TOTAL_SIZE]; + +/* Production bodies, including their capability mask and CRC, extracted by + * the Makefile. Only observation and the final transport are fixture seams. */ +#include "test_cluster_r4_refusal_handoff.inc" + +void +cluster_r4_observe_refusal(ClusterR4RefusalStage stage, ClusterCrBuildReason reason, + const BufferTag *tag, uint64 request_id, uint64 epoch, int32 requester, + int32 master, SCN read_scn) +{ + (void)stage; + (void)reason; + (void)tag; + (void)request_id; + (void)epoch; + (void)requester; + (void)master; + (void)read_scn; +} bool cluster_ic_envelope_build(ClusterICEnvelope *out_env, uint8 msg_type, uint32 source_node_id, @@ -596,6 +619,8 @@ cluster_ic_send_envelope(uint8 msg_type, int32 dest_node_id, const void *payload uint32 i; memcpy(&ut_sent_log[ut_sent_n].reply_header, payload, sizeof(GcsBlockReplyHeader)); + if (payload_len == sizeof(ut_r4_reply_payload)) + memcpy(ut_r4_reply_payload, payload, payload_len); ut_sent_log[ut_sent_n].reply_block_zero = payload_len == GCS_BLOCK_REPLY_PAYLOAD_TOTAL_SIZE; for (i = 0; ut_sent_log[ut_sent_n].reply_block_zero && i < GCS_BLOCK_DATA_SIZE; i++) @@ -611,8 +636,9 @@ cluster_ic_send_envelope(uint8 msg_type, int32 dest_node_id, const void *payload uint32 cluster_gcs_block_compute_checksum(const char *block_data) { - (void)block_data; ut_checksum_call_count++; + if (ut_r4_real_checksum) + return gcs_block_compute_checksum(block_data); return UINT32_C(0xA55A7E11); } @@ -644,6 +670,8 @@ ut_reset_log(void) ut_local_dispatch_marker = 0; ut_direct_zero_reply_count = 0; ut_checksum_call_count = 0; + ut_r4_real_checksum = false; + memset(ut_r4_reply_payload, 0, sizeof(ut_r4_reply_payload)); memset(&ut_direct_zero_reply_header, 0, sizeof(ut_direct_zero_reply_header)); ut_cap_guard_drop_count = 0; memset(ut_peer_capabilities, 0, sizeof(ut_peer_capabilities)); @@ -1020,6 +1048,189 @@ UT_TEST(test_r4_cap_bound_zero_reply_drops_drift_before_zero_expansion) UT_ASSERT_EQ(ut_cap_guard_drop_count, 1); } +/* Cover the missing handoff: a real holder refusal carries the master identity, + * unlike a master refusal. Both enqueue and drain must accept it, and the real + * requester decoder must still bind it to that exact master/request/epoch. */ +UT_TEST(test_r4_real_refusal_producers_cross_outbound_and_requester_boundary) +{ + const int masters[] = { 0, 1, UT_PEER_X, CLUSTER_MAX_NODES - 1 }; + const ClusterCrBuildReason reasons[] + = { CLUSTER_CR_BUILD_CAPACITY, CLUSTER_CR_BUILD_HOLDER_MOVED, CLUSTER_CR_BUILD_PROTOCOL }; + int m; + int r; + + for (m = 0; m < lengthof(masters); m++) { + for (r = 0; r < lengthof(reasons); r++) { + ClusterR4CrForwardPayload forward = { 0 }; + GcsBlockR4ReplyExpectation expected = { 0 }; + ClusterICEnvelope env = { 0 }; + ClusterCrBuildResult result + = r == 2 ? CLUSTER_CR_BUILD_FAIL_CLOSED : CLUSTER_CR_BUILD_RETRYABLE; + bool accepted; + bool wrong_master; + bool wrong_request; + bool wrong_epoch; + + ut_reset_log(); + ut_r4_real_checksum = true; + ut_peer_rc[UT_PEER_X] = CLUSTER_IC_SEND_DONE; + ut_peer_capabilities[UT_PEER_X] = R4_CR_REQUIRED_HELLO_CAPS; + ut_peer_cap_generation[UT_PEER_X] = 42; + forward.base.request_id = 123; + forward.base.epoch = 9; + forward.base.master_node = masters[m]; + forward.base.original_requester_node = UT_PEER_X; + forward.base.requester_backend_id = 17; + forward.base.transition_id = PCM_TRANS_N_TO_S; + UT_ASSERT(gcs_block_r4_publish_holder_refusal(2, &forward, 42, result, reasons[r])); + UT_ASSERT_EQ(cluster_lms_outbound_depth(2), 1); + UT_ASSERT_EQ(ut_sent_n, 0); + UT_ASSERT_EQ(cluster_lms_outbound_drain_send(2), 1); + UT_ASSERT_EQ(cluster_lms_outbound_depth(2), 0); + UT_ASSERT_EQ(ut_sent_n, 1); + UT_ASSERT_EQ(ut_sent_log[0].dest, UT_PEER_X); + UT_ASSERT(ut_sent_log[0].reply_block_zero); + UT_ASSERT_EQ(ut_sent_log[0].reply_header.status, + r == 2 ? GCS_BLOCK_REPLY_R4_DENIED + : GCS_BLOCK_REPLY_R4_RETRYABLE_HOLDER_MOVED); + UT_ASSERT_EQ(GcsBlockReplyHeaderGetForwardingMasterNode(&ut_sent_log[0].reply_header), + masters[m]); + expected.request_id = forward.base.request_id; + expected.epoch = forward.base.epoch; + expected.sender_node = cluster_node_id; + expected.forwarding_master_node = masters[m]; + expected.requester_backend_id = forward.base.requester_backend_id; + expected.transition_id = PCM_TRANS_N_TO_S; + expected.reply_domain = CLUSTER_GCS_BLOCK_REPLY_DOMAIN_R4_CR; + env.msg_type = PGRAC_IC_MSG_GCS_BLOCK_REPLY; + env.source_node_id = cluster_node_id; + env.dest_node_id = UT_PEER_X; + env.payload_length = sizeof(ut_r4_reply_payload); + cluster_node_id = UT_PEER_X; + accepted = gcs_block_decode_r4_reply_payload(&env, ut_r4_reply_payload, &expected); + expected.forwarding_master_node = GCS_BLOCK_REPLY_NO_FORWARDING_MASTER; + wrong_master = gcs_block_decode_r4_reply_payload(&env, ut_r4_reply_payload, &expected); + expected.forwarding_master_node = masters[m]; + expected.request_id++; + wrong_request = gcs_block_decode_r4_reply_payload(&env, ut_r4_reply_payload, &expected); + expected.request_id--; + expected.epoch++; + wrong_epoch = gcs_block_decode_r4_reply_payload(&env, ut_r4_reply_payload, &expected); + cluster_node_id = 0; + UT_ASSERT(accepted); + UT_ASSERT(!wrong_master && !wrong_request && !wrong_epoch); + } + } +} + +UT_TEST(test_r4_real_master_refusal_preserves_redirect) +{ + ClusterR4CrRequestPayload request = { 0 }; + ClusterICEnvelope env = { 0 }; + + ut_reset_log(); + ut_peer_capabilities[UT_PEER_X] = R4_CR_REQUIRED_HELLO_CAPS; + ut_peer_cap_generation[UT_PEER_X] = 42; + request.base.request_id = 321; + request.base.epoch = 9; + request.base.requester_backend_id = 17; + request.base.transition_id = PCM_TRANS_N_TO_S; + env.source_node_id = UT_PEER_X; + UT_ASSERT(gcs_block_r4_publish_refusal(2, &env, &request, 42, CLUSTER_CR_BUILD_RETRYABLE, + CLUSTER_CR_BUILD_WRONG_MASTER, false, + CLUSTER_MAX_NODES - 1)); + UT_ASSERT_EQ(cluster_lms_outbound_drain_send(2), 1); + UT_ASSERT_EQ(ut_sent_n, 1); + UT_ASSERT_EQ(ut_sent_log[0].reply_header.page_lsn, CLUSTER_MAX_NODES); + UT_ASSERT_EQ(GcsBlockReplyHeaderGetForwardingMasterNode(&ut_sent_log[0].reply_header), + GCS_BLOCK_REPLY_NO_FORWARDING_MASTER); +} + +UT_TEST(test_r4_holder_refusal_retains_backpressure_and_rejects_reconnect) +{ + GcsBlockReplyHeader hdr = ut_r4_refusal_header(GCS_BLOCK_REPLY_R4_DENIED, 0); + + ut_reset_log(); + GcsBlockReplyHeaderSetForwardingMasterNode(&hdr, 1); + ut_peer_capabilities[UT_PEER_X] = R4_CR_REQUIRED_HELLO_CAPS; + ut_peer_cap_generation[UT_PEER_X] = 42; + UT_ASSERT(cluster_lms_outbound_enqueue_zero_block_reply_cap_bound( + 2, UT_PEER_X, &hdr, R4_CR_REQUIRED_HELLO_CAPS, 42)); + ut_peer_rc[UT_PEER_X] = CLUSTER_IC_SEND_NOT_ADMITTED; + UT_ASSERT_EQ(cluster_lms_outbound_drain_send(2), 0); + UT_ASSERT_EQ(cluster_lms_outbound_depth(2), 1); + UT_ASSERT_EQ(ut_sent_n, 1); + ut_peer_rc[UT_PEER_X] = CLUSTER_IC_SEND_DONE; + UT_ASSERT_EQ(cluster_lms_outbound_drain_send(2), 1); + UT_ASSERT_EQ(cluster_lms_outbound_depth(2), 0); + UT_ASSERT_EQ(ut_sent_n, 2); + UT_ASSERT_EQ(memcmp(&ut_sent_log[0].reply_header, &ut_sent_log[1].reply_header, sizeof(hdr)), + 0); + UT_ASSERT(cluster_lms_outbound_enqueue_zero_block_reply_cap_bound( + 2, UT_PEER_X, &hdr, R4_CR_REQUIRED_HELLO_CAPS, 42)); + ut_peer_cap_generation[UT_PEER_X] = 43; + UT_ASSERT_EQ(cluster_lms_outbound_drain_send(2), 0); + UT_ASSERT_EQ(cluster_lms_outbound_depth(2), 0); + UT_ASSERT_EQ(ut_sent_n, 2); + UT_ASSERT_EQ(ut_cap_guard_drop_count, 1); +} + +UT_TEST(test_r4_holder_refusal_rejects_malformed_identity) +{ + int mutation; + + ut_reset_log(); + for (mutation = 0; mutation < 12; mutation++) { + GcsBlockReplyHeader hdr + = ut_r4_refusal_header(GCS_BLOCK_REPLY_R4_RETRYABLE_HOLDER_MOVED, 0); + + GcsBlockReplyHeaderSetForwardingMasterNode(&hdr, 1); + switch (mutation) { + case 0: + GcsBlockReplyHeaderSetForwardingMasterNode(&hdr, -2); + break; + case 1: + GcsBlockReplyHeaderSetForwardingMasterNode(&hdr, CLUSTER_MAX_NODES); + break; + case 2: + hdr.page_lsn = 1; + break; + case 3: + hdr.status = GCS_BLOCK_REPLY_R4_DENIED; + hdr.page_lsn = 1; + break; + case 4: + hdr.request_id = 0; + break; + case 5: + hdr.sender_node = -1; + break; + case 6: + hdr.sender_node = CLUSTER_MAX_NODES; + break; + case 7: + hdr.requester_backend_id = 0; + break; + case 8: + hdr.transition_id = PCM_TRANS_N_TO_X; + break; + case 9: + hdr.checksum = 1; + break; + case 10: + hdr.reserved_0[0] = 1; + break; + case 11: + hdr.status = GCS_BLOCK_REPLY_R4_CR_FULL; + break; + } + UT_ASSERT(!cluster_lms_outbound_enqueue_zero_block_reply_cap_bound( + 2, UT_PEER_X, &hdr, R4_CR_REQUIRED_HELLO_CAPS, 42)); + } + UT_ASSERT_EQ(cluster_lms_outbound_depth(2), 0); + UT_ASSERT_EQ(ut_sent_n, 0); +} + UT_TEST(test_zero_reply_wrappers_reject_the_other_status_domain) { const uint32 cap = PGRAC_IC_HELLO_CAP_SEMANTIC_ACTIVATION_V1 | PGRAC_IC_HELLO_CAP_R4_SYNC_CR_V1; @@ -1709,7 +1920,7 @@ UT_TEST(test_normal_stop_full_and_bad_frames_remain_debt) int main(void) { - UT_PLAN(36); + UT_PLAN(40); UT_RUN(test_normal_stop_missing_outbound_is_not_empty); UT_RUN(test_ring_shmem_init); @@ -1722,6 +1933,10 @@ main(void) UT_RUN(test_direct_zero_block_reply_uses_data_owner_direct_lane); UT_RUN(test_r4_cap_bound_zero_reply_sends_only_on_exact_generation); UT_RUN(test_r4_cap_bound_zero_reply_drops_drift_before_zero_expansion); + UT_RUN(test_r4_real_refusal_producers_cross_outbound_and_requester_boundary); + UT_RUN(test_r4_real_master_refusal_preserves_redirect); + UT_RUN(test_r4_holder_refusal_retains_backpressure_and_rejects_reconnect); + UT_RUN(test_r4_holder_refusal_rejects_malformed_identity); UT_RUN(test_zero_reply_wrappers_reject_the_other_status_domain); UT_RUN(test_full_worker_ring_refuses_without_overwrite); UT_RUN(test_cap_bound_frame_drops_on_connection_generation_drift); From 1e270e1e732db5a0dcc0e3edfb27f3223a08f2a4 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Thu, 17 Sep 2026 18:39:57 +0800 Subject: [PATCH 2/4] test(cluster): rebind source census after CR refusal fix --- src/test/cluster_unit/data/r11-source-removal-census-v1.json | 2 +- src/tools/check_r11_source_removal_census.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/cluster_unit/data/r11-source-removal-census-v1.json b/src/test/cluster_unit/data/r11-source-removal-census-v1.json index a44eafbe1d..b1ca2ce5a4 100644 --- a/src/test/cluster_unit/data/r11-source-removal-census-v1.json +++ b/src/test/cluster_unit/data/r11-source-removal-census-v1.json @@ -16,7 +16,7 @@ "current_product_snapshot": { "algorithm": "sha256-canonical-path-blob-v1", "path_count": 2225, - "sha256": "526de2a0572d0c41dccfe05c6c01f1779646383a15a54db61f91e7cb3dde4d4a" + "sha256": "c651005a62b5a2630179818b1281c658d120652b56af501288e0d43dd71488aa" }, "gates": { "L1": { diff --git a/src/tools/check_r11_source_removal_census.py b/src/tools/check_r11_source_removal_census.py index 16fa259b99..09604ed87f 100644 --- a/src/tools/check_r11_source_removal_census.py +++ b/src/tools/check_r11_source_removal_census.py @@ -24,7 +24,7 @@ CURRENT_PRODUCT_SNAPSHOT = { "algorithm": "sha256-canonical-path-blob-v1", "path_count": 2225, - "sha256": "526de2a0572d0c41dccfe05c6c01f1779646383a15a54db61f91e7cb3dde4d4a", + "sha256": "c651005a62b5a2630179818b1281c658d120652b56af501288e0d43dd71488aa", } LAYERS = { From e6227334f1e9849821979c3ff19b795127c5d145 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Thu, 17 Sep 2026 19:41:44 +0800 Subject: [PATCH 3/4] fix(cluster): honor deadlock cancellation in exact TX waits --- src/backend/cluster/cluster_tx_enqueue.c | 7 +++ .../data/r11-source-removal-census-v1.json | 2 +- .../cluster_unit/test_cluster_r4_tx_enqueue.c | 51 ++++++++++++++++++- src/tools/check_r11_source_removal_census.py | 2 +- 4 files changed, 59 insertions(+), 3 deletions(-) diff --git a/src/backend/cluster/cluster_tx_enqueue.c b/src/backend/cluster/cluster_tx_enqueue.c index 3cab38a8bd..f6a19c6667 100644 --- a/src/backend/cluster/cluster_tx_enqueue.c +++ b/src/backend/cluster/cluster_tx_enqueue.c @@ -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, ¤t_reason); diff --git a/src/test/cluster_unit/data/r11-source-removal-census-v1.json b/src/test/cluster_unit/data/r11-source-removal-census-v1.json index b1ca2ce5a4..050202204f 100644 --- a/src/test/cluster_unit/data/r11-source-removal-census-v1.json +++ b/src/test/cluster_unit/data/r11-source-removal-census-v1.json @@ -16,7 +16,7 @@ "current_product_snapshot": { "algorithm": "sha256-canonical-path-blob-v1", "path_count": 2225, - "sha256": "c651005a62b5a2630179818b1281c658d120652b56af501288e0d43dd71488aa" + "sha256": "8a0167c312cb1208856076e1b968998acdb1c12435bb8472deffeba3e72247df" }, "gates": { "L1": { diff --git a/src/test/cluster_unit/test_cluster_r4_tx_enqueue.c b/src/test/cluster_unit/test_cluster_r4_tx_enqueue.c index 7806cc4fb5..e5f2bcac97 100644 --- a/src/test/cluster_unit/test_cluster_r4_tx_enqueue.c +++ b/src/test/cluster_unit/test_cluster_r4_tx_enqueue.c @@ -101,6 +101,7 @@ static int test_wait_latch_calls; static int test_set_latch_calls[TEST_NSLOTS]; static bool test_wait_latch_throws; static bool test_wait_latch_sleeps; +static bool test_wait_latch_delivers_cancel; static TransactionId test_local_xid; static bool test_legacy_tt_found; static ClusterTTStatus test_legacy_tt_status; @@ -403,6 +404,8 @@ WaitLatch(Latch *latch pg_attribute_unused(), int wakeEvents pg_attribute_unused test_wait_latch_calls++; if (test_wait_latch_throws) siglongjmp(*PG_exception_stack, 1); + if (test_wait_latch_delivers_cancel) + test_cancel_token_pending = true; if (test_wait_latch_sleeps) pg_usleep((long)Max(timeout, 2) * 1000L); return WL_TIMEOUT; @@ -573,6 +576,7 @@ reset_fixture(void) memset(test_current_mx_stats, 0, sizeof(test_current_mx_stats)); test_wait_latch_throws = false; test_wait_latch_sleeps = false; + test_wait_latch_delivers_cancel = false; test_local_xid = (TransactionId)700; test_legacy_tt_found = false; test_legacy_tt_status = CLUSTER_TT_STATUS_IN_PROGRESS; @@ -859,6 +863,49 @@ UT_TEST(test_monotonic_timeout_uses_only_timeout_counter) assert_slot_clean(); } +UT_TEST(test_exact_wait_consumes_deadlock_token_before_repoll) +{ + ClusterTxLocator locator = test_locator(); + ClusterTxResolveReason reason = CLUSTER_TX_RESOLVE_PROTOCOL; + + reset_fixture(); + script_resolve(0, CLUSTER_TX_IN_PROGRESS, CLUSTER_TX_RESOLVE_NONE); + script_resolve(1, CLUSTER_TX_COMMITTED, CLUSTER_TX_RESOLVE_NONE); + test_cancel_token_pending = true; + UT_ASSERT_EQ(cluster_tx_enqueue_wait_exact(&locator, 1000, &reason), CLUSTER_TXW_DEADLOCK); + UT_ASSERT_EQ(reason, CLUSTER_TX_RESOLVE_NONE); + UT_ASSERT(!test_cancel_token_pending); + UT_ASSERT_EQ(test_resolve_pos, 1); + UT_ASSERT_EQ(test_wait_latch_calls, 0); + UT_ASSERT_EQ(test_wait_clear_calls, 1); + UT_ASSERT_EQ(test_wfg_exact_cancel_calls, 1); + UT_ASSERT(!test_wfg_live); + UT_ASSERT_EQ(pg_atomic_read_u64(&ClusterTxw->timeout_count), 0); + assert_slot_clean(); +} + +UT_TEST(test_exact_wait_consumes_deadlock_token_after_latch_wake) +{ + ClusterTxLocator locator = test_locator(); + ClusterTxResolveReason reason = CLUSTER_TX_RESOLVE_PROTOCOL; + + reset_fixture(); + script_resolve(0, CLUSTER_TX_IN_PROGRESS, CLUSTER_TX_RESOLVE_NONE); + script_resolve(1, CLUSTER_TX_IN_PROGRESS, CLUSTER_TX_RESOLVE_NONE); + script_resolve(2, CLUSTER_TX_COMMITTED, CLUSTER_TX_RESOLVE_NONE); + test_wait_latch_delivers_cancel = true; + UT_ASSERT_EQ(cluster_tx_enqueue_wait_exact(&locator, 1000, &reason), CLUSTER_TXW_DEADLOCK); + UT_ASSERT_EQ(reason, CLUSTER_TX_RESOLVE_NONE); + UT_ASSERT(!test_cancel_token_pending); + UT_ASSERT_EQ(test_resolve_pos, 2); + UT_ASSERT_EQ(test_wait_latch_calls, 1); + UT_ASSERT_EQ(test_wait_clear_calls, 1); + UT_ASSERT_EQ(test_wfg_exact_cancel_calls, 1); + UT_ASSERT(!test_wfg_live); + UT_ASSERT_EQ(pg_atomic_read_u64(&ClusterTxw->timeout_count), 0); + assert_slot_clean(); +} + UT_TEST(test_reentrant_source_and_target_slots_are_not_overwritten) { ClusterTxLocator locator = test_locator(); @@ -1486,7 +1533,7 @@ UT_TEST(test_backend_exit_counter_underflow_fails_stop_without_freeing_slot) int main(void) { - UT_PLAN(38); + UT_PLAN(40); UT_RUN(test_exact_wait_abi_and_shmem_size_are_frozen); UT_RUN(test_fixed_false_precedes_malformed_and_shared_state); UT_RUN(test_initial_terminal_never_registers); @@ -1497,6 +1544,8 @@ main(void) UT_RUN(test_zero_epoch_is_a_valid_stable_formation); UT_RUN(test_zero_to_nonzero_epoch_drift_fails_closed); UT_RUN(test_monotonic_timeout_uses_only_timeout_counter); + UT_RUN(test_exact_wait_consumes_deadlock_token_before_repoll); + UT_RUN(test_exact_wait_consumes_deadlock_token_after_latch_wake); UT_RUN(test_reentrant_source_and_target_slots_are_not_overwritten); UT_RUN(test_wfg_capacity_refusal_runs_full_cleanup); UT_RUN(test_error_longjmp_runs_same_cleanup_funnel); diff --git a/src/tools/check_r11_source_removal_census.py b/src/tools/check_r11_source_removal_census.py index 09604ed87f..a9f17209af 100644 --- a/src/tools/check_r11_source_removal_census.py +++ b/src/tools/check_r11_source_removal_census.py @@ -24,7 +24,7 @@ CURRENT_PRODUCT_SNAPSHOT = { "algorithm": "sha256-canonical-path-blob-v1", "path_count": 2225, - "sha256": "c651005a62b5a2630179818b1281c658d120652b56af501288e0d43dd71488aa", + "sha256": "8a0167c312cb1208856076e1b968998acdb1c12435bb8472deffeba3e72247df", } LAYERS = { From 9190f9fd7a78a30213e754b201ec9085bdc52a17 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Thu, 17 Sep 2026 20:40:59 +0800 Subject: [PATCH 4/4] fix(cluster): preserve ready receipts across ITL capacity waits --- src/backend/access/heap/heapam.c | 6 +- .../data/r11-source-removal-census-v1.json | 2 +- .../test_cluster_heap_prepare_diagnostic.c | 76 +++++++++++++++++-- src/tools/check_r11_source_removal_census.py | 2 +- 4 files changed, 76 insertions(+), 10 deletions(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index b049653ab5..1be293b238 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -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, diff --git a/src/test/cluster_unit/data/r11-source-removal-census-v1.json b/src/test/cluster_unit/data/r11-source-removal-census-v1.json index 050202204f..35e35f566d 100644 --- a/src/test/cluster_unit/data/r11-source-removal-census-v1.json +++ b/src/test/cluster_unit/data/r11-source-removal-census-v1.json @@ -16,7 +16,7 @@ "current_product_snapshot": { "algorithm": "sha256-canonical-path-blob-v1", "path_count": 2225, - "sha256": "8a0167c312cb1208856076e1b968998acdb1c12435bb8472deffeba3e72247df" + "sha256": "a33690006785ed01f7bfef085653bd6b916ebe970de64d555c3436e9d11a4e3b" }, "gates": { "L1": { diff --git a/src/test/cluster_unit/test_cluster_heap_prepare_diagnostic.c b/src/test/cluster_unit/test_cluster_heap_prepare_diagnostic.c index 36ba1aba49..c860664e00 100644 --- a/src/test/cluster_unit/test_cluster_heap_prepare_diagnostic.c +++ b/src/test/cluster_unit/test_cluster_heap_prepare_diagnostic.c @@ -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 @@ -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; } diff --git a/src/tools/check_r11_source_removal_census.py b/src/tools/check_r11_source_removal_census.py index a9f17209af..1e37cceb4c 100644 --- a/src/tools/check_r11_source_removal_census.py +++ b/src/tools/check_r11_source_removal_census.py @@ -24,7 +24,7 @@ CURRENT_PRODUCT_SNAPSHOT = { "algorithm": "sha256-canonical-path-blob-v1", "path_count": 2225, - "sha256": "8a0167c312cb1208856076e1b968998acdb1c12435bb8472deffeba3e72247df", + "sha256": "a33690006785ed01f7bfef085653bd6b916ebe970de64d555c3436e9d11a4e3b", } LAYERS = {