Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .github/workflows/nightly.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
127 changes: 116 additions & 11 deletions src/backend/cluster/cluster_gcs_block.c
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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;
}

Expand Down Expand Up @@ -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;
}
Expand All @@ -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);
}

Expand Down Expand Up @@ -13255,10 +13342,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;
}

Expand Down
7 changes: 7 additions & 0 deletions src/backend/cluster/cluster_gcs_block_shard.c
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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
Expand Down
62 changes: 46 additions & 16 deletions src/backend/cluster/cluster_pcm_x_convert.c
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down Expand Up @@ -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) {
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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;
Expand Down
40 changes: 36 additions & 4 deletions src/backend/storage/buffer/bufmgr.c
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 {
Expand All @@ -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);
Expand All @@ -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 {
Expand Down
Loading
Loading