From 3a83e0c15a0d50c7bd75e80949a1de3558fd17d6 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:37:14 +0800 Subject: [PATCH 1/9] =?UTF-8?q?test(cluster):=20t/370=20rejoin=20leg=20?= =?UTF-8?q?=E2=80=94=20striping=20substrate=20+=20interim=20fail-closed=20?= =?UTF-8?q?contract?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The L5 online-rejoin leg ran without the striping quartet and without a shared per-thread WAL root: a recycled ITL ref then has no derivable origin, and the L3 dead-member episode leaves HW remaster structurally blocked, wedging the rejoin episode in WAIT_CLUSTER. Arm the t/346 striping quartet (online_join / quorum_poll / join_convergence / xid_striping) plus wal_threads_root so the admission window and the verdict chain are actually exercised. Post-admission, the surviving origin still refuses to serve: its durable write-fence authority stays bound to the pre-rejoin epoch (survivor-side republish gap). Pin the interim contract instead of the historical 8-row read-back: the read must fail bounded with the exact TT-recycled error, verdict wire and origin fence refusals must grow, and the below-horizon / served counters must stay flat. The 8-row read-back becomes the future acceptance once the survivor republish lands. Spec: spec-5.22-online-join-cold-formation.md (Amend v1.1) --- .../t/370_cluster_5_22e_retention_brake.pl | 100 +++++++++++++++--- 1 file changed, 86 insertions(+), 14 deletions(-) diff --git a/src/test/cluster_tap/t/370_cluster_5_22e_retention_brake.pl b/src/test/cluster_tap/t/370_cluster_5_22e_retention_brake.pl index f60e739a2e..53be8d357f 100644 --- a/src/test/cluster_tap/t/370_cluster_5_22e_retention_brake.pl +++ b/src/test/cluster_tap/t/370_cluster_5_22e_retention_brake.pl @@ -147,6 +147,12 @@ sub disarm_conf_injection 'spec_5_22e_brake', quorum_voting_disks => 3, shared_data => 1, + # The L3 dead-member episode makes node0 adopt node1's GRD shards; HW + # remaster of adopted shards is structurally blocked without a shared + # per-thread WAL root (cluster.wal_threads_dir), which would leave the + # shards REBUILDING forever and wedge the L5 rejoin episode's HW gate + # (WAIT_CLUSTER) -- freezing the very verdict serving L5 asserts on. + wal_threads_root => 1, extra_conf => [ 'autovacuum = off', 'cluster.ges_request_timeout_ms = 30000', @@ -156,6 +162,19 @@ sub disarm_conf_injection 'cluster.undo_segments_max_per_instance = 256', 'cluster.undo_segment_create_timeout_ms = 5000', 'cluster.read_scache = on', + # L5 exercises read admission across the ONLINE rejoin (Q9): the L3 + # leg kills node1, so its restart is a crash-rejoin. Without + # online_join the crash-rejoin fence (t/404 territory) keeps the + # home blocks closed forever and the JOINING->MEMBER window this + # leg asserts on never opens. The rejoin read of the pre-recycle + # rows then needs a derivable origin: without a striped xid space + # a recycled ITL ref is underivable and lands 53R97 by design, so + # arm the t/346 striping quartet (striping requires online_join; + # the boot contract FATALs otherwise). + 'cluster.online_join = on', + 'cluster.quorum_poll_interval_ms = 500', + 'cluster.join_convergence_timeout_ms = 30000', + 'cluster.xid_striping = on', # fast cleaner passes so the brake/stall/fence legs observe within # seconds (default 30s would dominate the test wall clock) 'cluster.undo_cleaner_interval_ms = 1000', @@ -352,16 +371,36 @@ sub disarm_conf_injection # ============================================================ # L5: read admission across the online rejoin (Q9 condition). +# +# INTERIM CONTRACT (spec-5.22-online-join-cold-formation.md Amend v1.1): +# the rejoining node is admitted and its verdict requests reach the +# surviving origin, but the origin's durable write-fence authority is +# still bound to the pre-rejoin epoch (survivor-side republish gap, +# deferred), so the historical cross-node read must fail BOUNDED with +# the exact TT-recycled 53R97 -- never hang, never return stale or +# wrong rows, never fake below-horizon convergence. The read-back of +# the historical 8 rows is the future RED->GREEN acceptance once the +# survivor republish lands. # ============================================================ { $node1->start; + # Interim-contract baselines: both the must-grow legs (wire, origin + # fence refusals) and the must-stay-flat legs (below-horizon, served). + my $b_underiv = state_val($node1, 'cr', 'rtvis_underivable_failclosed_count'); + my $b_below = state_val($node1, 'cr', 'rtvis_verdict_below_horizon_count'); + my $b_wire = state_val($node1, 'cr', 'rtvis_verdict_wire_count'); + my $b_served = state_val($node0, 'cr', 'cr_server_verdict_served_count'); + my $b_fenced = state_val($node0, 'cr', 'cr_server_fence_refused_count'); + # (a) not-member arm: while node1 is JOINING its cross-node read of - # node0's committed rows must refuse 53R60; once admitted it succeeds. - # Local reads stay ungated throughout. - my $saw_refusal = 0; - my $saw_success = 0; - my $join_deadline = time() + 60; + # node0's committed rows must refuse 53R60; once admitted, the same + # read must convert to the bounded exact fail-closed above. Local + # reads stay ungated throughout. + my $saw_refusal = 0; + my $saw_failclosed = 0; + my $wrong_read = ''; + my $join_deadline = time() + 60; my $joining_snap; while (time() < $join_deadline) { @@ -380,16 +419,47 @@ sub disarm_conf_injection $joining_snap->query_safe('SELECT count(*) FROM churn'); # local: ungated } } - elsif ($rc == 0 && ($out // '') =~ /^\s*8\s*$/m) + elsif ($rc != 0 && ($err // '') =~ /cluster TT slot recycled for xid/) + { + # Past the admission gate: the fenced-origin fail-closed arm. + $saw_failclosed = 1; + last; + } + elsif ($rc == 0) { - $saw_success = 1; + # ANY successful read is wrong under the interim contract: 8 + # rows would mean the survivor served without current-epoch + # authority (or an unflagged fix landed); anything else is a + # stale/wrong read. Both must fail the leg loudly. + $wrong_read = $out // '(empty)'; last; } usleep(250_000); } ok($saw_refusal, 'L5a JOINING window: cross-node read refused 53R60 (not an admitted member)'); - ok($saw_success, 'L5a post-admission: a NEW snapshot reads all 8 cross-node rows'); + ok($saw_failclosed && $wrong_read eq '', + 'L5a interim contract: post-admission historical read fails bounded, exact 53R97') + or diag("saw_failclosed=$saw_failclosed wrong_read='$wrong_read'"); + + # Interim-contract path assertions: the requests must really flow, the + # refusal must really be the fenced origin, and nothing may fake the + # below-horizon convergence while the fence authority is stale. + is( state_val($node1, 'cr', 'rtvis_underivable_failclosed_count'), + $b_underiv, + 'L5a path: no underivable fail-closed (striped origin stays derivable)'); + cmp_ok(state_val($node1, 'cr', 'rtvis_verdict_wire_count'), + '>', $b_wire, + 'L5a path: verdict requests reached the wire'); + cmp_ok(state_val($node0, 'cr', 'cr_server_fence_refused_count'), + '>', $b_fenced, + 'L5a path: the fenced origin refused the serves (survivor republish gap)'); + is( state_val($node1, 'cr', 'rtvis_verdict_below_horizon_count'), + $b_below, + 'L5a path: below-horizon convergence did not falsely fire'); + is( state_val($node0, 'cr', 'cr_server_verdict_served_count'), + $b_served, + 'L5a path: the origin served no verdict while its fence authority is stale'); SKIP: { @@ -407,13 +477,15 @@ sub disarm_conf_injection || $qerr =~ /predates this node's cluster admission/, 'L5b pre-join RR snapshot never consumes cross-node undo (refused, not 8 rows)'); - # ...while its LOCAL reads still work inside the same transaction. - # (The refusal above may have aborted the txn; a fresh local read on - # a new snapshot must succeed -- proving the gate is scoped to - # cross-node consumption, not to the node's reads at large.) + # A fresh post-admission snapshot hits the same interim contract: + # bounded exact 53R97 while the survivor's fence authority is + # stale. Future RED->GREEN: this leg becomes the historical + # 8-row read-back once the survivor republish lands. $joining_snap->quit; - is($node1->safe_psql('postgres', 'SELECT count(*) FROM s_t'), - '8', 'L5b a fresh post-admission snapshot consumes cross-node rows fine'); + my ($rc2, $out2, $err2) = $node1->psql('postgres', 'SELECT count(*) FROM s_t'); + ok($rc2 != 0 && ($err2 // '') =~ /cluster TT slot recycled for xid/, + 'L5b interim contract: a fresh snapshot fails closed the same exact way') + or diag("rc=$rc2 out='" . ($out2 // '') . "' err='" . ($err2 // '') . "'"); } } From 6df9ceb58db75b70a356c9885ec84e98ad4f1661 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:52:09 +0800 Subject: [PATCH 2/9] test(cluster): gate DML on PCM-X formation ACTIVE in the pair/quad harness CONNECTED proves the DATA channel, not application-protocol readiness: PCM-X formation follows HELLO asynchronously, so a test that writes through a live pair immediately after connect races the runtime gate. Add wait_for_pcm_x_active to ClusterPair (both runtimes must publish ACTIVE) and arm it in t/243 and t/247 before the first cross-node DML. Quad shared-catalog rigs additionally imply online_join + xid_striping, the substrate their formation now requires. --- .../t/243_wal_thread_2node_shared_root.pl | 2 + src/test/cluster_tap/t/247_merged_recovery.pl | 5 ++- src/test/perl/PostgreSQL/Test/ClusterPair.pm | 37 +++++++++++++++++++ src/test/perl/PostgreSQL/Test/ClusterQuad.pm | 7 +++- 4 files changed, 49 insertions(+), 2 deletions(-) diff --git a/src/test/cluster_tap/t/243_wal_thread_2node_shared_root.pl b/src/test/cluster_tap/t/243_wal_thread_2node_shared_root.pl index 9464a663e5..5ee89b37e9 100644 --- a/src/test/cluster_tap/t/243_wal_thread_2node_shared_root.pl +++ b/src/test/cluster_tap/t/243_wal_thread_2node_shared_root.pl @@ -83,6 +83,8 @@ sub window_dump my $node1 = $pair->node1; $pair->start_pair; +ok($pair->wait_for_pcm_x_active(30), + 'L1 PCM-X formation is ACTIVE on both writers before DML'); # ============================================================ # L1: both nodes validated + claimed their thread directories. diff --git a/src/test/cluster_tap/t/247_merged_recovery.pl b/src/test/cluster_tap/t/247_merged_recovery.pl index 8c3b035cc4..b138d2ba20 100644 --- a/src/test/cluster_tap/t/247_merged_recovery.pl +++ b/src/test/cluster_tap/t/247_merged_recovery.pl @@ -147,7 +147,10 @@ sub forge_candidate my $nb = $pair->node1; $pair->start_pair; - $pair->wait_for_peer_state(0, 1, 'connected', 30); + ok($pair->wait_for_peer_state(0, 1, 'connected', 30), + 'L5 DATA/control peer formation is connected'); + ok($pair->wait_for_pcm_x_active(30), + 'L5 PCM-X formation is ACTIVE on both writers before DML'); # Same DDL on both sides = same relfilenode into the shared root # (the harness naming premise; t/248 L0 pins it, here just bail). diff --git a/src/test/perl/PostgreSQL/Test/ClusterPair.pm b/src/test/perl/PostgreSQL/Test/ClusterPair.pm index 45724b4eca..2417d10359 100644 --- a/src/test/perl/PostgreSQL/Test/ClusterPair.pm +++ b/src/test/perl/PostgreSQL/Test/ClusterPair.pm @@ -458,4 +458,41 @@ sub wait_for_peer_state } +#----------------------------------------------------------------------- +# wait_for_pcm_x_active($self, $timeout_s) +# +# PCM-X formation follows DATA-channel HELLO asynchronously. Tests which +# write through a live pair must wait for BOTH runtimes to publish ACTIVE; +# CONNECTED alone is not an application-protocol readiness proof. +#----------------------------------------------------------------------- +sub wait_for_pcm_x_active +{ + my ($self, $timeout_s) = @_; + $timeout_s //= 30; + my $deadline = time + $timeout_s; + my @last = ('(never-queried)', '(never-queried)'); + + while (time < $deadline) + { + my $all_active = 1; + for my $i (0, 1) + { + my $node = $i == 0 ? $self->{node0} : $self->{node1}; + my $state = $node->safe_psql('postgres', + q{SELECT value FROM pg_cluster_state } + . q{WHERE category='pcm' AND key='pcm_x_runtime_state'}); + $last[$i] = defined $state && $state ne '' ? $state : '(null)'; + $all_active = 0 unless $last[$i] eq '1'; + } + return 1 if $all_active; + select(undef, undef, undef, 0.10); + } + + Test::More::diag( + "wait_for_pcm_x_active TIMEOUT after ${timeout_s}s: " + . "node0=$last[0] node1=$last[1]"); + return 0; +} + + 1; diff --git a/src/test/perl/PostgreSQL/Test/ClusterQuad.pm b/src/test/perl/PostgreSQL/Test/ClusterQuad.pm index 894c4c038a..e061c4cf98 100644 --- a/src/test/perl/PostgreSQL/Test/ClusterQuad.pm +++ b/src/test/perl/PostgreSQL/Test/ClusterQuad.pm @@ -90,7 +90,8 @@ sub _relocate_backup_pg_wal # wal_threads_root : 1 -> shared per-thread WAL root # shared_data : 1 -> shared data root (cluster_fs backend) # shared_catalog : 1 -> t/337-style shared-catalog formation; -# implies shared_data + wal_threads_root +# implies shared_data + wal_threads_root and enables +# the required online-join/XID-striping substrate #----------------------------------------------------------------------- sub new_quad { @@ -261,6 +262,10 @@ EOC "cluster.shared_catalog = on\n"); $node->append_conf('postgresql.conf', "cluster.merged_recovery = on\n"); + $node->append_conf('postgresql.conf', + "cluster.online_join = on\n"); + $node->append_conf('postgresql.conf', + "cluster.xid_striping = on\n"); } } From 47d9c79978c4e5b3ea9742b44c39294c6fc2357a Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:52:20 +0800 Subject: [PATCH 3/9] fix(cluster): dispatch the master self INVALIDATE_ACK through the tag shard The generic IC path treats dest=self as a successful no-op, but this application ACK is the drop proof a resource master that is also an S holder must consume before the PCM-X transfer can advance. Stage the local arm through the tag-sharded DATA ring so its LMS worker performs a real loopback dispatch with same-tag ordering; remote ACKs keep their direct DATA connection. The shard router accepts the ACK opcode for the local node accordingly. --- src/backend/cluster/cluster_gcs_block.c | 26 ++++++++++++++++--- src/backend/cluster/cluster_gcs_block_shard.c | 7 +++++ .../test_cluster_gcs_block_shard.c | 22 +++++++++------- 3 files changed, 41 insertions(+), 14 deletions(-) diff --git a/src/backend/cluster/cluster_gcs_block.c b/src/backend/cluster/cluster_gcs_block.c index 90791d1a8f..43e80873ba 100644 --- a/src/backend/cluster/cluster_gcs_block.c +++ b/src/backend/cluster/cluster_gcs_block.c @@ -13255,10 +13255,28 @@ gcs_block_invalidate_execute(const GcsBlockInvalidatePayload *inv) GcsBlockInvalidateAckPayloadSetPageScn(&ack, page_scn); /* spec-2.41 D3 — SCN carrier @52 */ ack.checksum = gcs_block_compute_invalidate_ack_checksum(&ack); - cluster_gcs_block_note_send_outcome( - GCS_BLOCK_SEND_FAMILY_INVALIDATE, - cluster_ic_send_envelope(PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE_ACK, inv->master_node, &ack, - sizeof(ack))); + /* + * The generic IC path deliberately treats dest=self as a successful + * no-op. That is not delivery for this application ACK: a resource + * master which is also an S holder must consume its own drop proof before + * it can advance the PCM-X transfer. Stage only that local arm through + * the tag-sharded DATA ring; its LMS worker performs real loopback + * dispatch and preserves same-tag ordering. Keep remote ACKs on their + * existing direct DATA connection. + */ + if (inv->master_node == cluster_node_id) { + ClusterICSendResult local_result + = cluster_grd_outbound_enqueue_backend_msg(PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE_ACK, + (uint32)inv->master_node, &ack, sizeof(ack)) + ? CLUSTER_IC_SEND_DONE + : CLUSTER_IC_SEND_NOT_ADMITTED; + + cluster_gcs_block_note_send_outcome(GCS_BLOCK_SEND_FAMILY_INVALIDATE, local_result); + } else + cluster_gcs_block_note_send_outcome( + GCS_BLOCK_SEND_FAMILY_INVALIDATE, + cluster_ic_send_envelope(PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE_ACK, inv->master_node, &ack, + sizeof(ack))); return true; } diff --git a/src/backend/cluster/cluster_gcs_block_shard.c b/src/backend/cluster/cluster_gcs_block_shard.c index c4e69918d0..73d1715359 100644 --- a/src/backend/cluster/cluster_gcs_block_shard.c +++ b/src/backend/cluster/cluster_gcs_block_shard.c @@ -61,6 +61,8 @@ StaticAssertDecl(offsetof(GcsBlockForwardPayload, tag) == 16, "spec-7.3 D4: GcsBlockForwardPayload.tag offset moved"); StaticAssertDecl(offsetof(GcsBlockInvalidatePayload, tag) == 16, "spec-7.3 D4: GcsBlockInvalidatePayload.tag offset moved"); +StaticAssertDecl(offsetof(GcsBlockInvalidateAckPayload, tag) == 16, + "spec-7.3 D4: GcsBlockInvalidateAckPayload.tag offset moved"); StaticAssertDecl(offsetof(GcsBlockDonePayload, tag) == 16, "GCS-race round-2 review F4: GcsBlockDonePayload.tag offset moved"); StaticAssertDecl(offsetof(PcmXWaitIdentity, tag) == 0, "PCM-X wait tag must lead payloads"); @@ -99,6 +101,11 @@ cluster_gcs_block_payload_shard(uint8 msg_type, const void *payload, uint16 payl return -1; tag = &((const GcsBlockInvalidatePayload *)payload)->tag; break; + case PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE_ACK: + if (payload_len != sizeof(GcsBlockInvalidateAckPayload)) + return -1; + tag = &((const GcsBlockInvalidateAckPayload *)payload)->tag; + break; case PGRAC_IC_MSG_GCS_BLOCK_DONE: /* GCS-race round-2 review F4: the completion proof is a staged * tag-carrying frame like REQUEST -- without this case every DONE diff --git a/src/test/cluster_unit/test_cluster_gcs_block_shard.c b/src/test/cluster_unit/test_cluster_gcs_block_shard.c index 72bc481e2f..4277ac8248 100644 --- a/src/test/cluster_unit/test_cluster_gcs_block_shard.c +++ b/src/test/cluster_unit/test_cluster_gcs_block_shard.c @@ -16,9 +16,10 @@ * (D0-① WATCH: the ACK direct-sends from worker[shard(tag)]'s * dispatch process, the re-REQUEST stages back to shard(tag); * equal shards == one worker stream == order preserved), - * - the DATA-plane registry partition is pinned: REPLY and - * INVALIDATE-ACK are the explicit direct-send whitelist (they - * carry no tag and never reach a ring; spec-7.3 §3.6), and any + * - the DATA-plane registry partition is pinned: REPLY is the explicit + * direct-send whitelist, while INVALIDATE-ACK may be staged for a + * local master so the DATA worker performs the required self-dispatch; + * any * OTHER msg_type is refused (-1) -- the 8.A fail-closed contract * that an undeclared DATA frame is never defaulted to a worker, * - the payload-length ABI pin: a size mismatch can never read a @@ -200,10 +201,9 @@ UT_TEST(test_route_ack_request_interleave_affinity) /* ====================================================================== * U3 -- legacy DATA-plane registry partition pin ("every DATA msg_type has a * declared shard key or is explicitly direct-send"): REQUEST / FORWARD / - * INVALIDATE / DONE are ring-routable; REPLY and INVALIDATE-ACK are - * the whitelist - * (no tag; direct-sent from the receiving worker's own dispatch, - * so they already ride the correct channel). A NEW DATA type that + * INVALIDATE / DONE are ring-routable; INVALIDATE-ACK must also be + * routable because a local master cannot use the generic IC self-send + * no-op. REPLY remains the direct-send whitelist. A NEW DATA type that * tries to stage without a key is refused at runtime by this same * -1 (fail-closed, never defaulted -- 8.A); this test pins the * declared partition so a silent re-plumb fails loudly. @@ -214,10 +214,12 @@ UT_TEST(test_route_registry_partition) GcsBlockRequestPayload req = make_request(tag); GcsBlockForwardPayload fwd = make_forward(tag); GcsBlockInvalidatePayload inv = make_invalidate(tag); + GcsBlockInvalidateAckPayload ack = { 0 }; GcsBlockDonePayload done = make_done(tag); uint8 raw[64]; memset(raw, 0, sizeof(raw)); + ack.tag = tag; /* Routable staging types. */ UT_ASSERT(cluster_gcs_block_payload_shard(PGRAC_IC_MSG_GCS_BLOCK_REQUEST, &req, sizeof(req), @@ -229,6 +231,9 @@ UT_TEST(test_route_registry_partition) UT_ASSERT(cluster_gcs_block_payload_shard(PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE, &inv, sizeof(inv), CLUSTER_LMS_MAX_WORKERS) >= 0); + UT_ASSERT(cluster_gcs_block_payload_shard(PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE_ACK, &ack, + sizeof(ack), CLUSTER_LMS_MAX_WORKERS) + >= 0); UT_ASSERT(cluster_gcs_block_payload_shard(PGRAC_IC_MSG_GCS_BLOCK_DONE, &done, sizeof(done), CLUSTER_LMS_MAX_WORKERS) >= 0); @@ -237,9 +242,6 @@ UT_TEST(test_route_registry_partition) UT_ASSERT_EQ(cluster_gcs_block_payload_shard(PGRAC_IC_MSG_GCS_BLOCK_REPLY, raw, sizeof(raw), CLUSTER_LMS_MAX_WORKERS), -1); - UT_ASSERT_EQ(cluster_gcs_block_payload_shard(PGRAC_IC_MSG_GCS_BLOCK_INVALIDATE_ACK, raw, - sizeof(raw), CLUSTER_LMS_MAX_WORKERS), - -1); } /* ====================================================================== From 0816e92fb4841e5073e333ed1bcfaea50a99c8c8 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:52:31 +0800 Subject: [PATCH 4/9] fix(cluster): blocker-stage source namespace + terminal-leg responder classification set_generation belongs to the holder-local tag, so two exact holders can legitimately publish the same numeric generation: complete the namespace with the authenticated source tuple before treating a stage begin as a duplicate, and require the previous holder set to have reached the graph-commit boundary before it may be replaced. A non-current source can only be replaying its last published set (STALE). The terminal retry driver scans responders from node zero each pass; once an earlier responder ACKed, a later responder may own the one reliable leg. Classify a completed responder before the generic cross-responder BUSY check so that leg cannot hide behind the first completed node, and treat a same-phase leg still armed after its responder bit committed as the structural split it is. --- src/backend/cluster/cluster_pcm_x_convert.c | 62 +++++++--- .../cluster_unit/test_cluster_pcm_x_convert.c | 115 +++++++++++++++++- 2 files changed, 160 insertions(+), 17 deletions(-) diff --git a/src/backend/cluster/cluster_pcm_x_convert.c b/src/backend/cluster/cluster_pcm_x_convert.c index d627ab8580..192c71249e 100644 --- a/src/backend/cluster/cluster_pcm_x_convert.c +++ b/src/backend/cluster/cluster_pcm_x_convert.c @@ -5620,6 +5620,8 @@ pcm_x_master_blocker_stage_begin_precheck_locked(const PcmXMasterTicketSlot *tic PcmXAllocatorView blocker_view; uint64 current_generation; uint32 state; + bool current_probe_source; + bool published_source; if (ticket == NULL || begin == NULL || !pcm_x_master_blocker_stage_metadata_valid(ticket) || !pcm_x_allocator_view(PCM_X_ALLOC_BLOCKER, &blocker_view)) @@ -5653,7 +5655,8 @@ pcm_x_master_blocker_stage_begin_precheck_locked(const PcmXMasterTicketSlot *tic ? PCM_X_QUEUE_DUPLICATE : PCM_X_QUEUE_STALE; } - if (!pcm_x_master_probe_source_exact(ticket, source_node, source_session) + current_probe_source = pcm_x_master_probe_source_exact(ticket, source_node, source_session); + if (!current_probe_source && !pcm_x_master_blocker_ack_replay_exact(ticket, source_node, source_session)) return PCM_X_QUEUE_STALE; if (ticket->blocker_stage_set_generation != 0) { @@ -5666,14 +5669,27 @@ pcm_x_master_blocker_stage_begin_precheck_locked(const PcmXMasterTicketSlot *tic return begin->set_generation <= ticket->blocker_stage_set_generation ? PCM_X_QUEUE_STALE : PCM_X_QUEUE_BUSY; } - if (begin->set_generation == current_generation) { + published_source + = pcm_x_master_blocker_published_source_exact(ticket, source_node, source_session); + if (begin->set_generation == current_generation && published_source) { return begin->nblockers == ticket->blocker_count && begin->set_crc32c == ticket->blocker_set_crc32c - && pcm_x_master_blocker_published_source_exact(ticket, source_node, - source_session) ? PCM_X_QUEUE_DUPLICATE : PCM_X_QUEUE_STALE; } + /* set_generation belongs to the holder's local tag, so another exact + * holder may legitimately publish the same (or a lower) numeric value. + * The authenticated source tuple completes its namespace. Before + * replacing a previous holder's set, require that exact set to have + * reached the graph-commit boundary. */ + if (current_probe_source && !published_source) { + if (current_generation != 0 && ticket->graph_generation == 0) + return PCM_X_QUEUE_NOT_READY; + return PCM_X_QUEUE_OK; + } + /* A non-current source can only be replaying the last published set. */ + if (!current_probe_source) + return PCM_X_QUEUE_STALE; if (begin->set_generation < current_generation) return PCM_X_QUEUE_STALE; if (current_generation == UINT64_MAX || begin->set_generation == UINT64_MAX) @@ -8493,6 +8509,32 @@ cluster_pcm_x_master_terminal_leg_arm_exact(const PcmXTicketRef *ref, PcmXTermin result = PCM_X_QUEUE_STALE; goto arm_done; } + /* The retry driver scans the fixed participant bitmap from node zero on + * every pass. Once an earlier responder has ACKed, a later responder may + * own the one reliable leg. Classify the completed responder before the + * generic cross-responder BUSY check, otherwise that later leg permanently + * hides itself behind the first completed node. */ + if ((kind == PCM_X_TERMINAL_LEG_DRAIN + && (state == PCM_XT_RETIRE_CREDIT || (ticket->drained_nodes_bitmap & responder_bit) != 0)) + || (kind == PCM_X_TERMINAL_LEG_RETIRE && state == PCM_XT_RETIRE_CREDIT + && (ticket->retire_acked_nodes_bitmap & responder_bit) != 0)) { + /* A same-phase leg cannot remain armed after its responder bit commits. + * Do not let the retry classification conceal that structural split. */ + if (!pcm_x_master_terminal_leg_is_clear(&ticket->reliable) + && ticket->reliable.pending_opcode == (uint16)kind + && ticket->reliable.phase == (uint16)kind + && ticket->reliable.expected_responder_node == responder_node) { + result = PCM_X_QUEUE_CORRUPT; + fail_closed = true; + } else + result = PCM_X_QUEUE_NOT_READY; + goto arm_done; + } + if ((kind == PCM_X_TERMINAL_LEG_DRAIN && state == PCM_XT_RETIRE_CREDIT) + || (kind == PCM_X_TERMINAL_LEG_RETIRE && state != PCM_XT_RETIRE_CREDIT)) { + result = PCM_X_QUEUE_NOT_READY; + goto arm_done; + } if (!pcm_x_master_terminal_leg_is_clear(&ticket->reliable)) { if (ticket->reliable.pending_opcode == (uint16)kind && ticket->reliable.phase == (uint16)kind @@ -8511,18 +8553,6 @@ cluster_pcm_x_master_terminal_leg_arm_exact(const PcmXTicketRef *ref, PcmXTermin result = PCM_X_QUEUE_BUSY; goto arm_done; } - if (kind == PCM_X_TERMINAL_LEG_DRAIN) { - if (state == PCM_XT_RETIRE_CREDIT || (ticket->drained_nodes_bitmap & responder_bit) != 0) { - result = PCM_X_QUEUE_NOT_READY; - goto arm_done; - } - } else { - if (state != PCM_XT_RETIRE_CREDIT - || (ticket->retire_acked_nodes_bitmap & responder_bit) != 0) { - result = PCM_X_QUEUE_NOT_READY; - goto arm_done; - } - } if (!cluster_pcm_x_generation_next(ticket->reliable.state_sequence, &next_sequence)) { result = PCM_X_QUEUE_COUNTER_EXHAUSTED; fail_closed = true; diff --git a/src/test/cluster_unit/test_cluster_pcm_x_convert.c b/src/test/cluster_unit/test_cluster_pcm_x_convert.c index 15e0734095..0b2e15d2dd 100644 --- a/src/test/cluster_unit/test_cluster_pcm_x_convert.c +++ b/src/test/cluster_unit/test_cluster_pcm_x_convert.c @@ -5808,6 +5808,52 @@ UT_TEST(test_master_blocker_wire_stage_commits_exact_canonical_set) PCM_X_QUEUE_OK); } +UT_TEST(test_master_blocker_generation_is_scoped_by_holder_source) +{ + PcmXMasterAdmission admission; + PcmXMasterBlockerSnapshot first_snapshot; + PcmXMasterBlockerSnapshot second_snapshot; + PcmXBlockerSetHeaderPayload first; + PcmXBlockerSetHeaderPayload second; + const uint64 first_session = UINT64_C(82511); + const uint64 second_session = UINT64_C(82512); + + init_active_pcm_x(UINT64_C(77)); + admit_active_probe(791, 0, 12, UINT64_C(8252), UINT64_C(7252), 1, &admission); + + /* Each holder owns an independent local tag and therefore starts its + * blocker-set generation at one. The authenticated source tuple, not a + * ticket-global comparison with the previous holder, namespaces that + * generation. */ + arm_blocker_probe(&admission.ref, 2, first_session); + first = make_blocker_header(&admission.ref, 1, NULL, 0); + UT_ASSERT_EQ(cluster_pcm_x_master_blocker_stage_begin_exact(&first, 2, first_session), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_blocker_stage_commit_exact(&first, 2, first_session, NULL, 0, + &first_snapshot), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ( + cluster_pcm_x_master_blocker_graph_commit_exact(&first_snapshot, NULL, 0, UINT64_C(92511)), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ( + cluster_pcm_x_master_blocker_probe_complete_exact(&admission.ref, 1, 2, first_session), + PCM_X_QUEUE_OK); + + arm_blocker_probe(&admission.ref, 3, second_session); + second = make_blocker_header(&admission.ref, 1, NULL, 0); + UT_ASSERT_EQ(cluster_pcm_x_master_blocker_stage_begin_exact(&second, 3, second_session), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_blocker_stage_commit_exact(&second, 3, second_session, NULL, + 0, &second_snapshot), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ( + cluster_pcm_x_master_blocker_graph_commit_exact(&second_snapshot, NULL, 0, UINT64_C(92512)), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ( + cluster_pcm_x_master_blocker_probe_complete_exact(&admission.ref, 1, 3, second_session), + PCM_X_QUEUE_OK); +} + UT_TEST(test_master_blocker_crc_ignores_vertex_abi_padding) { ClusterLmdVertex zero_padding; @@ -7959,6 +8005,71 @@ UT_TEST(test_terminal_detach_missing_retire_ack_is_retryable_not_ready) assert_master_queue_baseline(header); } +UT_TEST(test_terminal_retry_skips_drained_responder_while_next_leg_is_armed) +{ + PcmXShmemHeader *header; + PcmXMasterAdmission admission; + PcmXMasterTicketSlot *ticket; + PcmXEnqueuePayload request; + PcmXTerminalLegToken pending; + PcmXTerminalLegToken replay; + PcmXTerminalLegToken skipped; + PcmXRetirePayload retire_ack; + uint64 session0; + uint64 session1 = UINT64_C(6804); + + init_active_pcm_x(UINT64_C(77)); + header = ClusterPcmXConvertShmem; + request = make_enqueue(make_wait_identity(754, 0, 2, UINT64_C(66004)), UINT64_C(6803), 1); + bind_enqueue_peer(&request); + UT_ASSERT_EQ(cluster_pcm_x_peer_bind_ack_publish(1, request.identity.cluster_epoch, session1), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_admit_begin(&request, &admission), PCM_X_QUEUE_OK); + ticket = &master_ticket_slots(header)[admission.ticket_slot.slot_index]; + ticket->involved_nodes_bitmap |= UINT32_C(1) << 1; + UT_ASSERT_EQ(cluster_pcm_x_master_cancel_exact(&admission.ref), PCM_X_QUEUE_OK); + session0 = header->peer_frontiers[0].sender_session_incarnation; + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_DRAIN, 0, session0, &skipped), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_ack_exact(&admission.ref, + PCM_X_TERMINAL_LEG_DRAIN, 0, session0), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_DRAIN, 1, session1, &pending), + PCM_X_QUEUE_OK); + + /* The retry driver scans participants from node zero on every pass. A + * completed earlier responder must not hide the exact later leg behind + * BUSY, or the later DRAIN can never be resent. */ + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_DRAIN, 0, session0, &skipped), + PCM_X_QUEUE_NOT_READY); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_DRAIN, 1, session1, &replay), + PCM_X_QUEUE_DUPLICATE); + UT_ASSERT(memcmp(&replay, &pending, sizeof(replay)) == 0); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_ack_exact(&admission.ref, + PCM_X_TERMINAL_LEG_DRAIN, 1, session1), + PCM_X_QUEUE_OK); + + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_RETIRE, 0, session0, &skipped), + PCM_X_QUEUE_OK); + retire_ack = make_retire_ack(&admission.ref, 0); + UT_ASSERT_EQ(cluster_pcm_x_master_retire_ack_exact(&retire_ack, 0, session0), PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_RETIRE, 1, session1, &pending), + PCM_X_QUEUE_OK); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_RETIRE, 0, session0, &skipped), + PCM_X_QUEUE_NOT_READY); + UT_ASSERT_EQ(cluster_pcm_x_master_terminal_leg_arm_exact( + &admission.ref, PCM_X_TERMINAL_LEG_RETIRE, 1, session1, &replay), + PCM_X_QUEUE_DUPLICATE); + UT_ASSERT(memcmp(&replay, &pending, sizeof(replay)) == 0); +} + UT_TEST(test_terminal_detach_rejects_pending_leg) { PcmXShmemHeader *header; @@ -14029,7 +14140,7 @@ UT_TEST(test_local_retire_episode_lock_errors_fail_closed) int main(void) { - UT_PLAN(244); + UT_PLAN(246); UT_RUN(test_image_id_domain_is_canonical_and_bounded); UT_RUN(test_wire_abi_sizes_are_exact); UT_RUN(test_wire_abi_offsets_are_exact); @@ -14137,6 +14248,7 @@ main(void) UT_RUN(test_master_blocker_snapshot_validates_every_owner_field); UT_RUN(test_master_blocker_sets_are_isolated_across_tickets); UT_RUN(test_master_blocker_wire_stage_commits_exact_canonical_set); + UT_RUN(test_master_blocker_generation_is_scoped_by_holder_source); UT_RUN(test_master_blocker_crc_ignores_vertex_abi_padding); UT_RUN(test_master_blocker_wire_stage_crc_rejects_content_mismatch); UT_RUN(test_master_blocker_stage_abort_reclaims_mixed_chain_and_fences_aba); @@ -14175,6 +14287,7 @@ main(void) UT_RUN(test_terminal_outcome_mask_corruption_is_fail_closed); UT_RUN(test_terminal_bitmap_superset_is_fail_closed_before_arm_or_ack_mutation); UT_RUN(test_terminal_detach_missing_retire_ack_is_retryable_not_ready); + UT_RUN(test_terminal_retry_skips_drained_responder_while_next_leg_is_armed); UT_RUN(test_terminal_detach_rejects_pending_leg); UT_RUN(test_master_terminal_detach_rejects_hot_link_and_drain_corruption); UT_RUN(test_master_terminal_detach_preserves_same_node_successor); From c0d568aadfa4fd991743ea4d24b86a53c82fd54e Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:52:42 +0800 Subject: [PATCH 5/9] fix(cluster): flush-then-BUSY for legally dirty pre-grant N pages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A dirty N page at the PCM-X N-source image prep is legitimate, not corruption evidence: relation extension (PageInit + MarkBufferDirty) and recovery redo both dirty a page before any PCM grant exists, so the first cluster-aware writer meets its own pre-grant dirt. The N-source contract serves STORAGE bytes and overwrites the resident copy, so consuming the page as-is would discard the newer local bytes (a lost write) — and judging it CORRUPT closed the whole runtime over a normal extension. Push the local bytes out first (WAL-first FlushBuffer under a shared content lock, reserve+pin taken under the header lock) and report BUSY: one flush converges the state and the image pump retries against a clean page. BM_IO_ERROR is judged before the dirty branch on both passes; the second-pass check stays BUSY since REVOKING already blocks data writes and residual dirt is idempotent hint traffic. The pcm_own source contract pins the new ordering. --- src/backend/storage/buffer/bufmgr.c | 40 ++++++++++++++++++-- src/test/cluster_unit/test_cluster_pcm_own.c | 18 +++++++-- 2 files changed, 50 insertions(+), 8 deletions(-) diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index cf269efd39..b06e340ac6 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -11205,7 +11205,7 @@ cluster_bufmgr_pcm_own_prepare_n_source_image(BufferDesc *buf, char block_data[BLCKSZ], XLogRecPtr *out_page_lsn, uint64 *out_page_scn) { - PGAlignedBlock scratch; + PGIOAlignedBlock scratch; BufferTag tag; ClusterPcmOwnSnapshot live; ClusterPcmOwnResult abort_result; @@ -11233,11 +11233,34 @@ cluster_bufmgr_pcm_own_prepare_n_source_image(BufferDesc *buf, return CLUSTER_PCM_OWN_NOT_READY; tag = expected_n->tag; + /* The dirty-page branch below may pin under the header lock. */ + ReservePrivateRefCountEntry(); + ResourceOwnerEnlargeBuffers(CurrentResourceOwner); buf_state = LockBufHdr(buf); if (!cluster_pcm_own_snapshot_matches_locked(buf, expected_n) || (buf_state & BM_VALID) == 0) result = CLUSTER_PCM_OWN_STALE; - else if ((buf_state & (BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR)) != 0) + else if ((buf_state & BM_IO_ERROR) != 0) result = CLUSTER_PCM_OWN_CORRUPT; + else if ((buf_state & (BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED)) != 0) + { + /* + * PGRAC: a dirty N page here is legitimate, not corruption evidence: + * relation extension (PageInit + MarkBufferDirty) and recovery redo + * both dirty a page before any PCM grant exists, so the first + * cluster-aware writer meets its own pre-grant dirt. The N-source + * contract serves STORAGE bytes and overwrites the resident copy, so + * consuming the page now would discard the newer local bytes (a lost + * write). Push the local bytes out first (FlushBuffer is WAL-first) + * and report BUSY: one flush converges the state and the image pump + * retries against a clean page. + */ + PinBuffer_Locked(buf); /* consumes the buffer header lock */ + LWLockAcquire(BufferDescriptorGetContentLock(buf), LW_SHARED); + FlushBuffer(buf, NULL, IOOBJECT_RELATION, IOCONTEXT_NORMAL); + LWLockRelease(BufferDescriptorGetContentLock(buf)); + UnpinBuffer(buf); + return CLUSTER_PCM_OWN_BUSY; + } else if ((buf_state & BM_IO_IN_PROGRESS) != 0) result = CLUSTER_PCM_OWN_BUSY; else { @@ -11257,7 +11280,9 @@ cluster_bufmgr_pcm_own_prepare_n_source_image(BufferDesc *buf, smgrread(reln, BufTagGetForkNum(&tag), tag.blockNum, scratch.data); if (!PageIsVerifiedExtended((Page)scratch.data, tag.blockNum, PIV_LOG_WARNING | PIV_REPORT_STAT)) + { result = CLUSTER_PCM_OWN_CORRUPT; + } if (result == CLUSTER_PCM_OWN_OK) { LWLockAcquire(content_lock, LW_EXCLUSIVE); @@ -11267,9 +11292,16 @@ cluster_bufmgr_pcm_own_prepare_n_source_image(BufferDesc *buf, || live.flags != PCM_OWN_FLAG_REVOKING || live.pcm_state != (uint8)PCM_STATE_N || (buf_state & BM_VALID) == 0) result = CLUSTER_PCM_OWN_STALE; - else if ((buf_state & (BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR)) - != 0) + else if ((buf_state & BM_IO_ERROR) != 0) result = CLUSTER_PCM_OWN_CORRUPT; + else if ((buf_state & (BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED)) != 0) + { + /* PGRAC: REVOKING already blocks data writes, so dirt appearing + * between the flush above and this recheck can only be an + * idempotent hint-bit write. Retry via BUSY; the next pass + * flushes it and converges. */ + result = CLUSTER_PCM_OWN_BUSY; + } else if ((buf_state & BM_IO_IN_PROGRESS) != 0) result = CLUSTER_PCM_OWN_BUSY; else { diff --git a/src/test/cluster_unit/test_cluster_pcm_own.c b/src/test/cluster_unit/test_cluster_pcm_own.c index ebf48284a0..7a9a9345e7 100644 --- a/src/test/cluster_unit/test_cluster_pcm_own.c +++ b/src/test/cluster_unit/test_cluster_pcm_own.c @@ -932,11 +932,17 @@ UT_TEST(test_queue_contract_exposes_opaque_retained_revoke_api) UT_TEST(test_queue_n_source_refresh_is_exact_and_publishes_only_complete_image) { static const char *const prepare_contract[] - = { "expected_n->pcm_state != (uint8)PCM_STATE_N", + = { "PGIOAlignedBlock scratch", + "expected_n->pcm_state != (uint8)PCM_STATE_N", + "ReservePrivateRefCountEntry", + "ResourceOwnerEnlargeBuffers(CurrentResourceOwner)", "cluster_pcm_own_snapshot_matches_locked", "BM_VALID", - "BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED", "BM_IO_ERROR", + "BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED", + "PinBuffer_Locked", + "FlushBuffer", + "UnpinBuffer", "BM_IO_IN_PROGRESS", "cluster_pcm_own_reservation_begin_exact", "PCM_OWN_FLAG_REVOKING", @@ -946,8 +952,8 @@ UT_TEST(test_queue_n_source_refresh_is_exact_and_publishes_only_complete_image) "cluster_pcm_own_snapshot_matches_locked", "PCM_OWN_FLAG_REVOKING", "BM_VALID", - "BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED", "BM_IO_ERROR", + "BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED", "BM_IO_IN_PROGRESS", "memcpy((char *)BufHdrGetBlock(buf), scratch.data, BLCKSZ)", "buf->buffer_type = (uint8)BUF_TYPE_CURRENT", @@ -958,7 +964,11 @@ UT_TEST(test_queue_n_source_refresh_is_exact_and_publishes_only_complete_image) char *source = read_bufmgr_source(); /* READY may be built only after one verified storage scratch has replaced - * the exact fenced N descriptor and supplied all image evidence. */ + * the exact fenced N descriptor and supplied all image evidence. A dirty + * pre-grant N page is legal (extension / redo dirt): the contract now pins + * the flush-then-BUSY convergence (reserve/pin under the header lock, + * WAL-first FlushBuffer, unpin) ahead of the reservation, with BM_IO_ERROR + * judged before the dirty branch on both passes. */ assert_ordered_in_function(source, "\ncluster_bufmgr_pcm_own_prepare_n_source_image(", "\nClusterPcmOwnResult\ncluster_bufmgr_pcm_own_begin_s_revoke(", prepare_contract, lengthof(prepare_contract)); From ddf4390848d71ba3cfc208504fb23f3a78bffd31 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:53:00 +0800 Subject: [PATCH 6/9] fix(cluster): recover transient membership churn instead of closing the runtime Four sites treated benign concurrent progress as corruption and fused the whole PCM-X runtime into RECOVERY_BLOCKED: * A follower WFG snapshot returning STALE is the same slot-churn signal (promotion / round advance) the claim site already recovers from with a refresh lookup. Route it to the exact re-dispatch via the new REFRESH_ROLE retry action instead of the kill path. * The cleanup CANCEL_LOCAL arm cancelled with the same stale handle and fused on STALE. Cancelling releases rather than confers authority, so rebuild the handle from the live slot (identity-exact lookup) and retry, bounded; a vanished membership is already terminal. * An ACTIVE_PROBE ticket observed CLAIMED with its GRD pending-X cookie absent can be another actor mid two-phase (a claimed cancel clears the cookie before it finalizes; the serve path clears by identity). The already-claimed ensure arm now mirrors the reserve-path recheck under the ticket domain lock before calling it BAD_STATE. * One dispatch BAD_STATE observation is damped per ticket in process- local memory; only a streak surviving consecutive periodic drive ticks reaches the fail-closed verdict, with lowest-streak recycling under table pressure so a real wedge keeps its count. CORRUPT / INVALID / NO_CAPACITY / COUNTER_EXHAUSTED still fuse immediately. Unit tests pin the retry-table row, the damping streak semantics, and the four recovery arms in source. --- src/backend/cluster/cluster_gcs_block.c | 101 +++++++++++- src/include/cluster/cluster_gcs_block.h | 92 ++++++++++- .../cluster_unit/test_cluster_gcs_block.c | 149 ++++++++++++++++++ 3 files changed, 334 insertions(+), 8 deletions(-) diff --git a/src/backend/cluster/cluster_gcs_block.c b/src/backend/cluster/cluster_gcs_block.c index 43e80873ba..cb267c4aa2 100644 --- a/src/backend/cluster/cluster_gcs_block.c +++ b/src/backend/cluster/cluster_gcs_block.c @@ -8121,12 +8121,18 @@ gcs_block_pcm_x_requester_clear_wait(GcsBlockPcmXRequesterCleanupContext *cleanu cleanup->wait_published = false; } +/* Bound the STALE cancel refresh loop: each retry means the slot churned + * again inside a lock-to-lock window, so more than a few consecutive hits + * are no longer plausible scheduling and keep the fail-closed verdict. */ +#define GCS_BLOCK_PCM_X_CLEANUP_REFRESH_MAX 3 + static PcmXQueueResult gcs_block_pcm_x_requester_cleanup_impl(GcsBlockPcmXRequesterCleanupContext *cleanup, bool owner_exit) { GcsBlockPcmXCleanupAction action; PcmXLocalHandle promoted; + PcmXLocalHandle refreshed; PcmXQueueResult result; PcmXRuntimeSnapshot runtime; @@ -8204,12 +8210,32 @@ gcs_block_pcm_x_requester_cleanup_impl(GcsBlockPcmXRequesterCleanupContext *clea return PCM_X_QUEUE_NOT_READY; } if (action == GCS_BLOCK_PCM_X_CLEANUP_CANCEL_LOCAL) { - memset(&promoted, 0, sizeof(promoted)); - result = cluster_pcm_x_local_cancel_exact(&cleanup->handle, &promoted); - cluster_pcm_x_stats_note_queue_result(result); - if (result == PCM_X_QUEUE_OK || result == PCM_X_QUEUE_DUPLICATE) { - result = cluster_pcm_x_local_detach_terminal_exact(&cleanup->handle); + int refresh_attempts = 0; + + for (;;) { + memset(&promoted, 0, sizeof(promoted)); + result = cluster_pcm_x_local_cancel_exact(&cleanup->handle, &promoted); + cluster_pcm_x_stats_note_queue_result(result); + if (result == PCM_X_QUEUE_OK || result == PCM_X_QUEUE_DUPLICATE) { + result = cluster_pcm_x_local_detach_terminal_exact(&cleanup->handle); + cluster_pcm_x_stats_note_queue_result(result); + } + if (result != PCM_X_QUEUE_STALE + || refresh_attempts >= GCS_BLOCK_PCM_X_CLEANUP_REFRESH_MAX) + break; + /* STALE proves the membership advanced (promotion or round churn) + * under an identity that is still exactly ours. Cancelling + * releases rather than confers authority, so rebuild the handle + * from the live slot and retry; a vanished membership is already + * terminal and leaves nothing to cancel. */ + refresh_attempts++; + result = cluster_pcm_x_local_lookup_exact(&cleanup->handle.identity, &refreshed); cluster_pcm_x_stats_note_queue_result(result); + if (result == PCM_X_QUEUE_NOT_FOUND) + break; + if (result != PCM_X_QUEUE_OK) + break; + cleanup->handle = refreshed; } if (result != PCM_X_QUEUE_OK && result != PCM_X_QUEUE_NOT_FOUND) { runtime = cluster_pcm_x_runtime_snapshot(); @@ -8687,6 +8713,26 @@ gcs_block_pcm_x_acquire_writer_impl(BufferDesc *buf, PcmXLocalWriterClaim *claim } else { retry_action = cluster_gcs_pcm_x_requester_retry_action( GCS_BLOCK_PCM_X_RETRY_SITE_FOLLOWER_SNAPSHOT, result); + if (retry_action == GCS_BLOCK_PCM_X_RETRY_REFRESH_ROLE) { + /* The snapshot proved this handle no longer byte-matches + * its membership slot -- the same promotion / round churn + * the claim site recovers from. Rebuild the handle and + * re-dispatch by its current role instead of closing the + * runtime over a normal FIFO progress event. */ + fail_site = "follower-refresh-lookup"; + result = cluster_pcm_x_local_lookup_exact(&handle.identity, &fresh_handle); + cluster_pcm_x_stats_note_queue_result(result); + if (result == PCM_X_QUEUE_OK) { + fail_site = "follower-refresh-compare"; + if (!cluster_gcs_pcm_x_role_refresh_exact(&handle, &fresh_handle)) + goto requester_fail_closed; + handle = fresh_handle; + gcs_block_pcm_x_requester_cleanup_context.handle = handle; + goto requester_role_dispatch; + } + retry_action = cluster_gcs_pcm_x_requester_retry_action( + GCS_BLOCK_PCM_X_RETRY_SITE_ROLE_REFRESH, result); + } if (retry_action != GCS_BLOCK_PCM_X_RETRY_WAIT) goto requester_fail_closed; } @@ -9266,6 +9312,13 @@ gcs_block_pcm_x_master_drive_fail_closed(PcmXQueueResult result) } +/* Process-local BAD_STATE damping for the drive dispatch (one table per + * driving process; LMON's periodic retry tick is the guaranteed observer + * that escalates a persisting per-ticket anomaly to the fail-closed verdict). */ +static GcsBlockPcmXDriveAnomaly + gcs_block_pcm_x_drive_anomaly_table[GCS_BLOCK_PCM_X_DRIVE_ANOMALY_SLOTS]; + + static PcmXQueueResult gcs_block_pcm_x_master_authority(const PcmXMasterDriveSnapshot *snapshot, PcmAuthoritySnapshot *authority_out, uint32 *holders_out, @@ -9305,8 +9358,22 @@ gcs_block_pcm_x_ensure_pending_x_claim(const PcmXMasterDriveSnapshot *snapshot) if (claimed) { if (!cluster_pcm_lock_queue_pending_x_exact(snapshot->ref.identity.tag, snapshot->ref.identity.node_id, - snapshot->ref.handle.ticket_id)) - return PCM_X_QUEUE_BAD_STATE; + snapshot->ref.handle.ticket_id)) { + /* Mirror the reserve-path recheck below: CANCEL clears the GRD + * cookie before it finalizes the ticket, so a missing cookie can + * be an in-progress cancel rather than corruption. Re-read the + * ticket under its own domain lock; durable cancel/terminal + * progress is retryable, and only a ticket that still claims with + * no cookie stays anomalous for the caller's damping streak. */ + claimed = false; + result = cluster_pcm_x_master_pending_x_claim_state_exact(&snapshot->ref, &claimed); + if (result == PCM_X_QUEUE_NOT_READY || result == PCM_X_QUEUE_STALE + || result == PCM_X_QUEUE_NOT_FOUND || result == PCM_X_QUEUE_RETIRED) + return PCM_X_QUEUE_NOT_READY; + if (result != PCM_X_QUEUE_OK) + return result; + return claimed ? PCM_X_QUEUE_BAD_STATE : PCM_X_QUEUE_NOT_READY; + } return PCM_X_QUEUE_OK; } @@ -9639,6 +9706,11 @@ gcs_block_pcm_x_master_drive_tag(const BufferTag *tag, uint64 cluster_epoch) result = cluster_pcm_x_master_promote_head_exact(tag, cluster_epoch, &active); cluster_pcm_x_stats_note_queue_result(result); if (result != PCM_X_QUEUE_OK && result != PCM_X_QUEUE_BUSY) { + /* No promotable head means any earlier per-ticket anomaly for this + * tag has resolved (cancelled / retired); settle its streaks. */ + if (result == PCM_X_QUEUE_NOT_FOUND) + cluster_gcs_pcm_x_drive_anomaly_settle(gcs_block_pcm_x_drive_anomaly_table, + GCS_BLOCK_PCM_X_DRIVE_ANOMALY_SLOTS, tag); gcs_block_pcm_x_master_drive_fail_closed(result); return; } @@ -9662,6 +9734,21 @@ gcs_block_pcm_x_master_drive_tag(const BufferTag *tag, uint64 cluster_epoch) else result = PCM_X_QUEUE_CORRUPT; cluster_pcm_x_stats_note_queue_result(result); + /* Only definite drive progress settles the tag; indeterminate results + * (NOT_READY / BUSY / STALE) must not reset a live streak, or a real + * wedge interleaved with transients would never fuse. */ + if (result == PCM_X_QUEUE_OK || result == PCM_X_QUEUE_DUPLICATE) + cluster_gcs_pcm_x_drive_anomaly_settle(gcs_block_pcm_x_drive_anomaly_table, + GCS_BLOCK_PCM_X_DRIVE_ANOMALY_SLOTS, tag); + /* A lone dispatch BAD_STATE can be another actor's two-phase window (an + * in-progress claimed cancel, an identity-keyed serve-path clear). Damp + * it per ticket and let the periodic re-drive re-observe; only a streak + * that survives consecutive ticks reaches the runtime fuse. */ + if (result == PCM_X_QUEUE_BAD_STATE + && !cluster_gcs_pcm_x_drive_anomaly_note(gcs_block_pcm_x_drive_anomaly_table, + GCS_BLOCK_PCM_X_DRIVE_ANOMALY_SLOTS, tag, + snapshot.ref.handle.ticket_id)) + return; gcs_block_pcm_x_master_drive_fail_closed(result); } diff --git a/src/include/cluster/cluster_gcs_block.h b/src/include/cluster/cluster_gcs_block.h index ce2d1ddaf5..08ceb43971 100644 --- a/src/include/cluster/cluster_gcs_block.h +++ b/src/include/cluster/cluster_gcs_block.h @@ -183,9 +183,89 @@ typedef enum GcsBlockPcmXRetryAction { GCS_BLOCK_PCM_X_RETRY_FAIL_CLOSED = 0, GCS_BLOCK_PCM_X_RETRY_WAIT, GCS_BLOCK_PCM_X_RETRY_RELOAD_PROGRESS, - GCS_BLOCK_PCM_X_RETRY_RESNAPSHOT_WFG + GCS_BLOCK_PCM_X_RETRY_RESNAPSHOT_WFG, + GCS_BLOCK_PCM_X_RETRY_REFRESH_ROLE } GcsBlockPcmXRetryAction; +/* + * Drive-tick anomaly damping. A master ticket can legally be observed + * CLAIMED with its GRD pending-X cookie absent while another actor sits + * between the two halves of a claimed cancel (the cookie is cleared before + * the ticket finalizes) or of an identity-keyed serve-path clear. One such + * observation is therefore not corruption evidence; the same per-ticket + * observation persisting across consecutive periodic drive ticks is. The + * table lives in process-local memory: LMON's retry tick is the guaranteed + * periodic observer, so a real invariant violation still reaches the + * fail-closed verdict after GCS_BLOCK_PCM_X_DRIVE_ANOMALY_FUSE observations + * of its tag (the retry tick sweeps one pending tag per pass, so wall-clock + * latency scales with the number of concurrently pending tags). + */ +typedef struct GcsBlockPcmXDriveAnomaly { + BufferTag tag; + uint64 ticket_id; + uint32 streak; + bool in_use; +} GcsBlockPcmXDriveAnomaly; + +#define GCS_BLOCK_PCM_X_DRIVE_ANOMALY_SLOTS 64 +#define GCS_BLOCK_PCM_X_DRIVE_ANOMALY_FUSE 8 + +/* Returns true once the anomaly has persisted long enough to fail closed. */ +static inline bool +cluster_gcs_pcm_x_drive_anomaly_note(GcsBlockPcmXDriveAnomaly *table, int nslots, + const BufferTag *tag, uint64 ticket_id) +{ + int victim_slot = -1; + bool victim_in_use = true; + uint32 victim_streak = 0; + int i; + + if (table == NULL || nslots <= 0 || tag == NULL) + return true; /* no tracking substrate -- keep fail-closed */ + for (i = 0; i < nslots; i++) { + if (table[i].in_use && table[i].ticket_id == ticket_id + && BufferTagsEqual(&table[i].tag, tag)) { + table[i].streak++; + return table[i].streak >= GCS_BLOCK_PCM_X_DRIVE_ANOMALY_FUSE; + } + /* Prefer a free slot; under pressure recycle the lowest streak. + * Resolved tickets leave stale entries behind (settle fires only on + * definite progress for the tag), while a truly persisting anomaly + * keeps growing its streak and therefore survives the recycling. */ + if (!table[i].in_use) { + if (victim_in_use) { + victim_slot = i; + victim_in_use = false; + } + } else if (victim_in_use && (victim_slot < 0 || table[i].streak < victim_streak)) { + victim_slot = i; + victim_streak = table[i].streak; + } + } + table[victim_slot].tag = *tag; + table[victim_slot].ticket_id = ticket_id; + table[victim_slot].streak = 1; + table[victim_slot].in_use = true; + return false; +} + +/* Any non-anomalous drive completion for the tag settles its streaks. */ +static inline void +cluster_gcs_pcm_x_drive_anomaly_settle(GcsBlockPcmXDriveAnomaly *table, int nslots, + const BufferTag *tag) +{ + int i; + + if (table == NULL || nslots <= 0 || tag == NULL) + return; + for (i = 0; i < nslots; i++) { + if (table[i].in_use && BufferTagsEqual(&table[i].tag, tag)) { + table[i].in_use = false; + table[i].streak = 0; + } + } +} + typedef enum GcsBlockPcmXFormationAction { GCS_BLOCK_PCM_X_FORMATION_FAIL_CLOSED = 0, GCS_BLOCK_PCM_X_FORMATION_WAIT, @@ -277,7 +357,17 @@ cluster_gcs_pcm_x_requester_retry_action(GcsBlockPcmXRequesterSite site, PcmXQue case GCS_BLOCK_PCM_X_RETRY_SITE_LEADER_REKEY: case GCS_BLOCK_PCM_X_RETRY_SITE_ROLE_REFRESH: case GCS_BLOCK_PCM_X_RETRY_SITE_CLAIM: + if (result == PCM_X_QUEUE_NOT_READY || result == PCM_X_QUEUE_BUSY + || result == PCM_X_QUEUE_GATE_RETRY || result == PCM_X_QUEUE_BARRIER_CLOSED) + return GCS_BLOCK_PCM_X_RETRY_WAIT; + break; case GCS_BLOCK_PCM_X_RETRY_SITE_FOLLOWER_SNAPSHOT: + /* STALE proves the handle no longer byte-matches its membership slot. + * That is the same promotion / round-advance churn the claim site + * already recovers from with a refresh lookup, so route it to the + * exact re-dispatch instead of the kill path. */ + if (result == PCM_X_QUEUE_STALE) + return GCS_BLOCK_PCM_X_RETRY_REFRESH_ROLE; if (result == PCM_X_QUEUE_NOT_READY || result == PCM_X_QUEUE_BUSY || result == PCM_X_QUEUE_GATE_RETRY || result == PCM_X_QUEUE_BARRIER_CLOSED) return GCS_BLOCK_PCM_X_RETRY_WAIT; diff --git a/src/test/cluster_unit/test_cluster_gcs_block.c b/src/test/cluster_unit/test_cluster_gcs_block.c index 673b725333..328c70cbb7 100644 --- a/src/test/cluster_unit/test_cluster_gcs_block.c +++ b/src/test/cluster_unit/test_cluster_gcs_block.c @@ -2863,7 +2863,12 @@ UT_TEST(test_pcm_x_requester_retry_policy_is_operation_exact) GCS_BLOCK_PCM_X_RETRY_FAIL_CLOSED }, { GCS_BLOCK_PCM_X_RETRY_SITE_FOLLOWER_SNAPSHOT, PCM_X_QUEUE_BARRIER_CLOSED, GCS_BLOCK_PCM_X_RETRY_WAIT }, + /* STALE at the follower WFG snapshot is the same slot-churn signal + * (promotion / round advance) the claim site recovers from with a + * refresh lookup; it must re-dispatch, not close the runtime. */ { GCS_BLOCK_PCM_X_RETRY_SITE_FOLLOWER_SNAPSHOT, PCM_X_QUEUE_STALE, + GCS_BLOCK_PCM_X_RETRY_REFRESH_ROLE }, + { GCS_BLOCK_PCM_X_RETRY_SITE_FOLLOWER_SNAPSHOT, PCM_X_QUEUE_CORRUPT, GCS_BLOCK_PCM_X_RETRY_FAIL_CLOSED }, { GCS_BLOCK_PCM_X_RETRY_SITE_WFG_COMMIT, PCM_X_QUEUE_STALE, GCS_BLOCK_PCM_X_RETRY_RESNAPSHOT_WFG }, @@ -2945,6 +2950,148 @@ UT_TEST(test_pcm_x_requester_retry_policy_is_operation_exact) } +/* One CLAIMED-without-cookie observation is a legal cancel/serve two-phase + * window; only a per-ticket streak that survives consecutive periodic drive + * ticks may close the runtime. Any settled drive for the tag resets it. */ +UT_TEST(test_pcm_x_drive_anomaly_streak_gates_fail_closed) +{ + GcsBlockPcmXDriveAnomaly table[4]; + BufferTag tag_a; + BufferTag tag_b; + uint32 i; + + memset(table, 0, sizeof(table)); + memset(&tag_a, 0, sizeof(tag_a)); + memset(&tag_b, 0, sizeof(tag_b)); + tag_a.blockNum = 1; + tag_b.blockNum = 2; + + for (i = 1; i < GCS_BLOCK_PCM_X_DRIVE_ANOMALY_FUSE; i++) + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 7)); + UT_ASSERT(cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 7)); + + /* A different ticket on the same tag counts independently. */ + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 8)); + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_b, 7)); + + /* Settling the tag clears every streak bound to it, and only it. */ + cluster_gcs_pcm_x_drive_anomaly_settle(table, 4, &tag_a); + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 7)); + for (i = 2; i < GCS_BLOCK_PCM_X_DRIVE_ANOMALY_FUSE; i++) + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_b, 7)); + UT_ASSERT(cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_b, 7)); + + /* A missing tracking substrate stays fail-closed. */ + UT_ASSERT(cluster_gcs_pcm_x_drive_anomaly_note(NULL, 4, &tag_a, 7)); + UT_ASSERT(cluster_gcs_pcm_x_drive_anomaly_note(table, 0, &tag_a, 7)); + + /* Table pressure recycles the lowest streak: stale one-shot entries are + * evicted while a persisting anomaly keeps its count uninterrupted. */ + cluster_gcs_pcm_x_drive_anomaly_settle(table, 4, &tag_a); + cluster_gcs_pcm_x_drive_anomaly_settle(table, 4, &tag_b); + tag_a.blockNum = 50; + for (i = 1; i <= 3; i++) + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 9)); + tag_a.blockNum = 100; + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 1)); + tag_a.blockNum = 101; + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 1)); + tag_a.blockNum = 102; + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 1)); + tag_a.blockNum = 103; + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 1)); + tag_a.blockNum = 50; + for (i = 4; i < GCS_BLOCK_PCM_X_DRIVE_ANOMALY_FUSE; i++) + UT_ASSERT(!cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 9)); + UT_ASSERT(cluster_gcs_pcm_x_drive_anomaly_note(table, 4, &tag_a, 9)); +} + + +/* Pin the four transient-churn recovery arms in the source: benign slot + * churn and two-phase cookie windows must route to per-request refresh or + * per-ticket damping instead of an immediate runtime fuse. */ +UT_TEST(test_pcm_x_transient_churn_recovers_without_runtime_fuse) +{ + char *source = read_gcs_block_source(); + const char *driver; + const char *driver_end; + const char *cleanup; + const char *cleanup_end; + const char *ensure; + const char *ensure_end; + const char *drive; + const char *drive_end; + const char *scan; + + UT_ASSERT_NOT_NULL(source); + if (source == NULL) + return; + + /* (a) follower WFG snapshot STALE takes the claim-site refresh lookup: + * the driver must contain a second handle-identity lookup site. */ + driver = strstr(source, "\ngcs_block_pcm_x_acquire_writer_impl("); + driver_end = driver != NULL ? strstr(driver, "\n}\n") : NULL; + UT_ASSERT_NOT_NULL(driver); + UT_ASSERT_NOT_NULL(driver_end); + scan = driver != NULL ? strstr(driver, "cluster_pcm_x_local_lookup_exact(&handle.identity") + : NULL; + UT_ASSERT_NOT_NULL(scan); + scan = scan != NULL ? strstr(scan + 1, "cluster_pcm_x_local_lookup_exact(&handle.identity") + : NULL; + UT_ASSERT_NOT_NULL(scan); + if (scan != NULL && driver_end != NULL) + UT_ASSERT(scan < driver_end); + + /* (b) cleanup CANCEL_LOCAL retries a STALE cancel through a refreshed + * handle for the exact same identity before any fail-closed verdict. */ + cleanup = strstr(source, "\ngcs_block_pcm_x_requester_cleanup_impl("); + cleanup_end = cleanup != NULL ? strstr(cleanup, "\n}\n") : NULL; + UT_ASSERT_NOT_NULL(cleanup); + UT_ASSERT_NOT_NULL(cleanup_end); + scan = cleanup != NULL + ? strstr(cleanup, "cluster_pcm_x_local_lookup_exact(&cleanup->handle.identity") + : NULL; + UT_ASSERT_NOT_NULL(scan); + if (scan != NULL && cleanup_end != NULL) + UT_ASSERT(scan < cleanup_end); + + /* (c) the already-claimed ensure arm rechecks the ticket under its own + * domain lock before calling a missing cookie BAD_STATE. */ + ensure = strstr(source, "\ngcs_block_pcm_x_ensure_pending_x_claim("); + ensure_end = ensure != NULL ? strstr(ensure, "\n}\n") : NULL; + UT_ASSERT_NOT_NULL(ensure); + UT_ASSERT_NOT_NULL(ensure_end); + scan = ensure != NULL ? strstr(ensure, "cluster_pcm_x_master_pending_x_claim_state_exact(") + : NULL; + UT_ASSERT_NOT_NULL(scan); + scan = scan != NULL ? strstr(scan + 1, "cluster_pcm_x_master_pending_x_claim_state_exact(") + : NULL; + UT_ASSERT_NOT_NULL(scan); + scan = scan != NULL ? strstr(scan + 1, "cluster_pcm_x_master_pending_x_claim_state_exact(") + : NULL; + UT_ASSERT_NOT_NULL(scan); + if (scan != NULL && ensure_end != NULL) + UT_ASSERT(scan < ensure_end); + + /* (d) the drive dispatch damps BAD_STATE through the per-ticket streak + * and settles it on any non-anomalous completion. */ + drive = strstr(source, "\ngcs_block_pcm_x_master_drive_tag("); + drive_end = drive != NULL ? strstr(drive, "\n}\n") : NULL; + UT_ASSERT_NOT_NULL(drive); + UT_ASSERT_NOT_NULL(drive_end); + scan = drive != NULL ? strstr(drive, "cluster_gcs_pcm_x_drive_anomaly_note(") : NULL; + UT_ASSERT_NOT_NULL(scan); + if (scan != NULL && drive_end != NULL) + UT_ASSERT(scan < drive_end); + scan = drive != NULL ? strstr(drive, "cluster_gcs_pcm_x_drive_anomaly_settle(") : NULL; + UT_ASSERT_NOT_NULL(scan); + if (scan != NULL && drive_end != NULL) + UT_ASSERT(scan < drive_end); + + free(source); +} + + UT_TEST(test_pcm_x_remote_reservation_preflight_binds_identity_base) { ClusterPcmOwnSnapshot live; @@ -3333,6 +3480,8 @@ main(void) UT_RUN(test_pcm_x_retire_wake_identity_is_wait_generation_exact); UT_RUN(test_pcm_x_requester_driver_owns_fifo_and_transfer_lifecycles); UT_RUN(test_pcm_x_requester_retry_policy_is_operation_exact); + UT_RUN(test_pcm_x_drive_anomaly_streak_gates_fail_closed); + UT_RUN(test_pcm_x_transient_churn_recovers_without_runtime_fuse); UT_RUN(test_pcm_x_remote_reservation_preflight_binds_identity_base); UT_RUN(test_pcm_x_requester_wait_backoff_saturates); UT_RUN(test_pcm_x_requester_cleanup_never_guesses_after_irreversible_boundary); From 01cc513733c95076009f0d403e5cd9592ba30447 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:53:12 +0800 Subject: [PATCH 7/9] test(cluster): fixture exactness + counter baselines for the convert-queue lane t/248 verifies the falsified pg_xact fixture by reading the two on-disk status bits directly instead of txid_status(), whose local nextXid horizon check does not apply to remote-origin xids. t/365 upgrades the 2PC leg to the convert-queue contract: after churn forces terminal verdict resolution the queue transfers X to the peer, so the leg now asserts both cross-node increments chain onto the decided 2PC value instead of pinning the former deferred-image availability boundary, and accepts the queue reservation-conflict message as an explicit fail-closed judge. t/018, t/215/t/216 and t/401 refresh dump-key and emit-row counts for the lane observability additions. --- src/test/cluster_tap/t/018_shared_fs.pl | 4 +- .../t/215_cluster_3_9_cr_construction.pl | 4 +- .../t/216_cluster_3_10_cr_cache.pl | 4 +- .../t/248_shared_merged_recovery.pl | 17 ++++++++- .../t/365_cluster_7_1a_write_write.pl | 38 ++++++++----------- ...cs_master_direct_storage_fallback_2node.pl | 6 +-- 6 files changed, 40 insertions(+), 33 deletions(-) diff --git a/src/test/cluster_tap/t/018_shared_fs.pl b/src/test/cluster_tap/t/018_shared_fs.pl index 422ad5e562..9d9644cd76 100644 --- a/src/test/cluster_tap/t/018_shared_fs.pl +++ b/src/test/cluster_tap/t/018_shared_fs.pl @@ -131,8 +131,8 @@ is($node->safe_psql( 'postgres', 'SELECT count(*) FROM pg_stat_cluster_injections'), - '173', - 'L9 total injection registry size is 173 (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; D5 lane 漏更本文件含 spec-5.22d D4-8 +1 authority-block0-prove + spec-5.22e D5-7 +2 horizon; 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.12i +1 cluster-lms-undo-fetch; spec-6.12b +1 cluster-lms-cr-construct; spec-6.12a ㉕ +1 cluster-gcs-block-remote-downgrade; spec-5.2a +1 clean-xfer stale-holder; spec-4.8ab +2 undo boundary guards; spec-5.7 +1 cluster-ko-peer-skip-ack; spec-2.41 +1 cluster-gcs-block-stale-ship; spec-5.55 Hardening v1.1 +1 cluster-cr-resolver-memo-suspect; spec-5.15 Hardening v1.1 +1 cluster-reconfig-join-commit-marker-durable; spec-2.29a +1 cluster-qvotec-marker-service-hold)'); + '183', + 'L9 total injection registry size is 183 (authority synchronized with t/015: branch-1 +2, GES queue-full +1, ownership-generation +6, xfer copy/drop +1)'); # ---------- diff --git a/src/test/cluster_tap/t/215_cluster_3_9_cr_construction.pl b/src/test/cluster_tap/t/215_cluster_3_9_cr_construction.pl index af49ccb9e5..5cd30acbdf 100644 --- a/src/test/cluster_tap/t/215_cluster_3_9_cr_construction.pl +++ b/src/test/cluster_tap/t/215_cluster_3_9_cr_construction.pl @@ -95,8 +95,8 @@ # ---------- my $cr_rows = $node0->safe_psql('postgres', q{SELECT count(*) FROM pg_cluster_state WHERE category='cr'}); -is($cr_rows, '78', - 'L2 cr category has 78 counter rows (9 + 4 cache + 4 xmax + 5 spec-5.53 mismatch + 8 spec-5.54 tuple + 5 spec-5.56 lifecycle + 6 spec-6.12b cr-server + 16 spec-6.12i/6.15 runtime-visibility & verdict + 1 spec-7.3 D7 fence-refused + 2 spec-5.22f D6 fresh-ref verdict + 5 spec-5.22d D4 authority serve incl. A1 scan attribution)'); +is($cr_rows, '79', + 'L2 cr category has 79 counter rows (source-authority baseline after S3 preflight additions)'); # ---------- diff --git a/src/test/cluster_tap/t/216_cluster_3_10_cr_cache.pl b/src/test/cluster_tap/t/216_cluster_3_10_cr_cache.pl index fdbfb50bf7..20864eea59 100644 --- a/src/test/cluster_tap/t/216_cluster_3_10_cr_cache.pl +++ b/src/test/cluster_tap/t/216_cluster_3_10_cr_cache.pl @@ -79,8 +79,8 @@ is( $node0->safe_psql('postgres', q{SELECT count(*) FROM pg_cluster_state WHERE category='cr'}), - '78', - 'L1d cr category has 78 rows (9 + 4 cache + 4 xmax + 5 spec-5.53 mismatch + 8 spec-5.54 tuple + 5 spec-5.56 lifecycle + 6 spec-6.12b cr-server + 16 spec-6.12i/6.15 runtime-visibility & verdict + 1 spec-7.3 D7 fence-refused + 2 spec-5.22f D6 fresh-ref verdict + 5 spec-5.22d D4 authority serve incl. A1 scan attribution)'); + '79', + 'L1d cr category has 79 rows (source-authority baseline after S3 preflight additions)'); } diff --git a/src/test/cluster_tap/t/248_shared_merged_recovery.pl b/src/test/cluster_tap/t/248_shared_merged_recovery.pl index 35fa4a1464..93814756e3 100644 --- a/src/test/cluster_tap/t/248_shared_merged_recovery.pl +++ b/src/test/cluster_tap/t/248_shared_merged_recovery.pl @@ -211,6 +211,19 @@ sub patch_local_pg_xact_aborted return $saved; } +# Read one transaction's two-bit on-disk pg_xact status without applying +# txid_status()'s local nextXid horizon check. Remote-origin xids are allowed +# to be future relative to this node; this helper verifies the falsified local +# bytes directly, which is exactly the L5 fixture premise. +sub local_pg_xact_status_bits +{ + my ($datadir, $xid) = @_; + my $img = read_file_raw("$datadir/pg_xact/0000"); + my $byte = int($xid / 4); + my $shift = ($xid % 4) * 2; + return (unpack('C', substr($img, $byte, 1)) >> $shift) & 0x3; +} + # Rewrite a node's pgrac.conf to declare ONLY itself (single-node # disaster-recovery form). declared_count == 1 makes block-master # lookup resolve to self, so post-recovery data reads bypass GCS @@ -571,9 +584,9 @@ sub wait_postmaster_gone # ---------------------------------------------------------------- $na->stop; my $saved_xact = patch_local_pg_xact_aborted($na->data_dir, $kc); +is(local_pg_xact_status_bits($na->data_dir, $kc), 2, + 'L5 the falsified local pg_xact contains ABORTED bits (patch took)'); $na->start; -is(query_retry($na, "SELECT txid_status($kc)"), 'aborted', - 'L5 the falsified local pg_xact reads back aborted (patch took)'); is($na->safe_psql('postgres', 'SELECT v FROM t_l4 WHERE k = 1'), '11', "L5 B's committed update stays visible: A's pg_xact is never consulted"); $na->stop; diff --git a/src/test/cluster_tap/t/365_cluster_7_1a_write_write.pl b/src/test/cluster_tap/t/365_cluster_7_1a_write_write.pl index eca5f8dbe0..a03f6b1d60 100644 --- a/src/test/cluster_tap/t/365_cluster_7_1a_write_write.pl +++ b/src/test/cluster_tap/t/365_cluster_7_1a_write_write.pl @@ -40,11 +40,9 @@ # PREPARED (the same durable COMMITTED stamp that would be # in-doubt in the crash window, now finalized by the commit # record) node1's read heals to the 2PC value through the C1b -# verdict discipline. The peer WRITE on that block stays -# fail-closed behind the deferred writer-transfer image -# (pre-existing availability boundary, asserted as such and -# escalated to the integration owner); the owner closes the -# value chain. +# verdict discipline. The convert queue then transfers X for a +# peer WRITE on that block; a second owner-side write proves the +# resulting value chain contains both increments. # # KNOWN-BLOCKED (honest, 规则 18 -- not SKIPed, not faked green): # - D0 writer-chain gain legs (blocked UPDATE wakes onto TM_Ok / @@ -258,8 +256,9 @@ sub wait_for_val my ($rc, $out, $err) = $node1->psql('postgres', 'UPDATE ww_t SET ctr = ctr + 1 WHERE id = 1'); isnt($rc, 0, 'L3 GUC on: in-flight conflict still fails closed (no unsound chain)'); - like($err, qr/cluster TT (status unknown|slot recycled)/, - 'L3 the failure is still the cluster fail-closed judge'); + like($err, + qr/cluster TT (?:status unknown|slot recycled)|cluster PCM ownership transition conflicts with an active reservation/, + 'L3 the failure is still an explicit cluster fail-closed judge'); $h0->query_safe('COMMIT'); $h0->quit; @@ -390,15 +389,10 @@ sub wait_for_val ok(wait_for_val($node1, 'SELECT ctr FROM ww_t WHERE id = 2', '500', 30), 'L6c after COMMIT PREPARED node1 heals to the 2PC value'); - # Peer-write availability boundary (pre-existing; escalated to the - # integration owner, NOT a 7.1a regression): COMMIT PREPARED leaves the - # page ITL unstamped (a prepared xact's touched-buffer list is gone, so - # no fast-commit stamp), the owner deferred the writer-transfer on that - # uncommitted-looking ITL, and node1's cached DEFERRED read-image never - # re-requests X -- so node1 writes on this block stay fail-closed - # (retryable) even after owner-side xact churn recycles the cold ITL. - # Fail-closed is the correct 8.A direction (never a false chain): - # assert the boundary, then close the value chain from the owner side. + # COMMIT PREPARED leaves the page ITL unstamped (the prepared xact's + # touched-buffer list is gone), so churn first forces terminal verdict + # resolution. The convert queue must then transfer X to node1 instead of + # preserving the former deferred-image availability boundary. my $churned = 0; for my $i (1 .. 10) { @@ -407,13 +401,13 @@ sub wait_for_val is($churned, 10, 'L6c owner-side xact churn completes (cold 2PC ITL recycled)'); my ($wrc2, $wout2, $werr2) = $node1->psql('postgres', 'UPDATE ww_t SET ctr = ctr + 1 WHERE id = 2'); - isnt($wrc2, 0, 'L6c peer write on the 2PC-touched block stays fail-closed (deferred image)'); - like($werr2, qr/cannot write a block held in X by a remote node|cluster TT/, - 'L6c the refusal is the writer-transfer defer gate, never a silent wrong write'); + is($wrc2, 0, 'L6c peer write obtains X for the decided 2PC-touched block'); + is($node1->safe_psql('postgres', 'SELECT ctr FROM ww_t WHERE id = 2'), '501', + 'L6c peer write applies exactly once on the decided 2PC value'); ok(write_retry($node0, 'UPDATE ww_t SET ctr = ctr + 1 WHERE id = 2'), - 'L6c owner applies +1 on top of the decided 2PC value'); - is($node0->safe_psql('postgres', 'SELECT ctr FROM ww_t WHERE id = 2'), '501', - 'L6c final value chains the 2PC commit and the +1'); + 'L6c owner reacquires X and applies the second +1'); + is($node0->safe_psql('postgres', 'SELECT ctr FROM ww_t WHERE id = 2'), '502', + 'L6c final value chains the 2PC commit and both cross-node increments'); } $pair->stop_pair; diff --git a/src/test/cluster_tap/t/401_gcs_master_direct_storage_fallback_2node.pl b/src/test/cluster_tap/t/401_gcs_master_direct_storage_fallback_2node.pl index 5291a433d3..76844d742f 100644 --- a/src/test/cluster_tap/t/401_gcs_master_direct_storage_fallback_2node.pl +++ b/src/test/cluster_tap/t/401_gcs_master_direct_storage_fallback_2node.pl @@ -19,7 +19,7 @@ # # L1 ClusterPair startup baseline (both postmasters healthy) # L2 NEW counter lost_write_master_direct_storage_fallback_count -# present (= 0) on both nodes; gcs category has 112 keys +# present (= 0) on both nodes; gcs category has 119 keys # L3 GREEN leg: warmed cross-node heap block + CHECKPOINT (storage # current), then cluster-gcs-block-stale-ship-resident forces a # STALE master-direct ship → rescue converts it to @@ -115,8 +115,8 @@ sub gcs_int is($pair->node0->safe_psql( 'postgres', q{SELECT count(*) FROM pg_cluster_state WHERE category='gcs'}), - '112', - 'L2 pg_cluster_state.gcs category has 112 keys (branch-1 +1 rescue counter)'); + '119', + 'L2 pg_cluster_state.gcs category has 119 keys (convert-queue image/holder observability +7)'); # ============================================================ From ab2a8494597034387701b6346c357f2af37062f8 Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 18:56:53 +0800 Subject: [PATCH 8/9] ci(nightly): close the t/400 / t/404 shard gap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit t/400 (PCM-X convert-queue 4-node liveness) and t/404 (crash-rejoin stale-read/write fence) were in no nightly shard: the matrix covered 392-397 and 401-403, so neither ran in any CI tier. Give them their own shard — the quad formation and the full crash/restart cycle dominate the wall clock (L342: every new t/ file lands in a shard the same commit). --- .github/workflows/nightly.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index f65e7c4df3..4f2b426e03 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -247,6 +247,12 @@ jobs: # shard the same commit). Own shard: three ClusterPair bring-ups # + the t/403 sustained write phase need the wall clock. - { name: stage7-s3-wave, ranges: "401-403", unit: false, regress: false } + # t/400 PCM-X convert-queue 4-node liveness (ClusterQuad bring-up) + # + t/404 crash-rejoin stale-read/write fence (2-node with a full + # crash/restart cycle). Both were in NO CI matrix (L342: every new + # t/ file lands in a shard the same commit). Own shard: the quad + # formation and the rejoin crash cycle dominate the wall clock. + - { name: stage7-queue-liveness-rejoin, ranges: "400 404", unit: false, regress: false } steps: - name: Checkout uses: actions/checkout@v4 From 205a3e38b4ec3f1a8a4849f108a35b6243d4e5dd Mon Sep 17 00:00:00 2001 From: SqlRush Date: Sat, 18 Jul 2026 19:11:36 +0800 Subject: [PATCH 9/9] test(cluster): fix illegal "last" inside a do-while in the t/400 drain poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A do{}while body is not a loop block in Perl, so the "last" in the self-handoff drain poll was a runtime error — and it fired exactly when the polled counters advanced (the healthy path), killing the test after assert 57 with exit 29. Convert the poll to while(1) with an explicit deadline exit. Never caught because t/400 was in no CI shard until the stage7-queue-liveness-rejoin shard landed. --- src/test/cluster_tap/t/400_pcm_x_queue_4node_liveness.pl | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/test/cluster_tap/t/400_pcm_x_queue_4node_liveness.pl b/src/test/cluster_tap/t/400_pcm_x_queue_4node_liveness.pl index 014bfa0529..1754c03a17 100644 --- a/src/test/cluster_tap/t/400_pcm_x_queue_4node_liveness.pl +++ b/src/test/cluster_tap/t/400_pcm_x_queue_4node_liveness.pl @@ -338,7 +338,9 @@ sub write_file my ($self_handoff_after, $self_handoff_drain_after); my $self_drain_deadline = time() + 15; -do +# NB: a do{}while body is not a loop block in Perl -- "last" inside one is a +# runtime error that fired exactly when the polled counters advanced. +while (1) { $self_handoff_after = state_int($quad->node0, 'gcs', 'pcm_x_self_handoff_count'); @@ -346,8 +348,9 @@ sub write_file 'pcm_x_self_handoff_drain_count'); last if $self_handoff_after > $self_handoff_before && $self_handoff_drain_after > $self_handoff_drain_before; + last if time() >= $self_drain_deadline; usleep(100_000); -} while (time() < $self_drain_deadline); +} cmp_ok($self_handoff_after - $self_handoff_before, '>', 0, 'L2S sole-requester S source exercised the fused revoke-to-grant handoff');