Skip to content

fix(cluster): GCS block-ship race trio — dedup swallow on retryable deny, invalidate-slot instant fail, duplicate-reply overwrite #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sqlrush merged 1 commit into from
Jul 12, 2026
Merged

fix(cluster): GCS block-ship race trio — dedup swallow on retryable deny, invalidate-slot instant fail, duplicate-reply overwrite #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,12 @@ jobs:
# legs pin/stall/self-heal on real cleaner cadence and need the
# wall clock.
- { name: stage5-undo-retention, ranges: "369-370 386 389", unit: false, regress: false }
# t/390 GCS block-ship race convergence (3-node triple): PENDING_X
# dedup re-evaluation (no reply-timeout burn), invalidate-slot
# bounded claim wait, duplicate-reply first-reply-wins. Own shard:
# triple bring-up + injection legs need the wall clock (L342: every
# new t/ file lands in a shard the same commit).
- { name: stage7-gcs-block-race, ranges: "390-390", unit: false, regress: false }
steps:
- name: Checkout
uses: actions/checkout@v4
Expand Down
173 changes: 157 additions & 16 deletions src/backend/cluster/cluster_gcs_block.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1712,6 +1712,16 @@ cluster_gcs_send_block_request_and_wait(BufferDesc *buf, PcmLockTransition trans
LWLockRelease(&blk->lock.lock);
}

/*
* Lock-free consume invariant (S3 RC-A): every delivery path
* (wire reply handler, direct-land completion, direct fail-slot)
* refuses to touch reply_header/reply_block_data once
* reply_received is set, so from the have_reply observation above
* until this backend rearms the slot the reply fields are
* immutable and may be read without blk->lock. A duplicate
* reply (dedup CACHED_REPLY resend / re-forward) is dropped at
* delivery, counted in stale_reply_drop_count.
*/
final_status = slot->reply_header.status;
final_page_lsn = (XLogRecPtr)slot->reply_header.page_lsn;
/* spec-2.35 HC105: capture forward source so DENIED_MASTER_
Expand Down Expand Up @@ -3910,6 +3920,13 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
CLUSTER_INJECTION_POINT("cluster-gcs-block-starvation-force-denied");
if (cluster_injection_should_skip("cluster-gcs-block-starvation-force-denied")) {
pg_atomic_fetch_add_u64(&ClusterGcsBlock->starvation_denied_pending_x_count, 1);
/* A retry-driven denial must not leave the IN_FLIGHT dedup entry
* behind: the requester's backoff retry reuses the same
* (request_id, epoch) key, so a leftover entry swallows the retry
* as IN_FLIGHT_DUPLICATE until the TTL sweep — each swallowed
* round burns a full cluster.gcs_reply_timeout_ms and the
* convergence the deny promises never re-evaluates. */
cluster_gcs_block_dedup_remove(dedup_worker_id, &key);
gcs_block_send_reply(req->sender_node, req, GCS_BLOCK_REPLY_DENIED_PENDING_X,
InvalidXLogRecPtr, NULL);
return;
Expand All @@ -3918,6 +3935,9 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
pending_x = cluster_pcm_lock_query_pending_x_requester(req->tag);
if (pending_x >= 0 && pending_x != req->sender_node) {
pg_atomic_fetch_add_u64(&ClusterGcsBlock->starvation_denied_pending_x_count, 1);
/* Same dedup-entry release as the inject branch above: the deny
* contract is "back off and retry, the retry re-evaluates". */
cluster_gcs_block_dedup_remove(dedup_worker_id, &key);
gcs_block_send_reply(req->sender_node, req, GCS_BLOCK_REPLY_DENIED_PENDING_X,
InvalidXLogRecPtr, NULL);
return;
Expand Down Expand Up @@ -4236,6 +4256,12 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
}
if (GcsBlockRequestPayloadIsDirectLandArmed(req)) {
cluster_pcm_lock_clear_pending_x(req->tag);
/* The direct-land deny asks the requester to retry with
* direct-land suppressed — same (request_id, epoch) key,
* so drop the in-flight dedup entry or the retry is
* swallowed as IN_FLIGHT_DUPLICATE (see the PENDING_X
* sites; S3 RC-B). */
cluster_gcs_block_dedup_remove(dedup_worker_id, &key);
(void)gcs_block_deny_direct_armed_forward_request(req);
return;
}
Expand Down Expand Up @@ -4416,6 +4442,16 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
}
gcs_block_broadcast_invalidate_nowait(req, holders_bm);
cluster_pcm_lock_clear_pending_x(req->tag);
/* Release the IN_FLIGHT dedup entry BEFORE the deny goes out:
* the convergence this reply promises ("a following retry
* finds no remote holder left and grants") only works if the
* retry is re-evaluated. The retry reuses the same
* (request_id, epoch) dedup key, so a leftover in-flight
* entry silently swallows it (IN_FLIGHT_DUPLICATE) until the
* TTL sweep — every swallowed round burns a full
* cluster.gcs_reply_timeout_ms at the requester and the
* S3-observed 53R90 retransmit-exhaustion storm follows. */
cluster_gcs_block_dedup_remove(dedup_worker_id, &key);
gcs_block_send_reply(req->sender_node, req, GCS_BLOCK_REPLY_DENIED_PENDING_X,
InvalidXLogRecPtr, NULL);
return;
Expand Down Expand Up @@ -4573,8 +4609,13 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
InvalidXLogRecPtr, NULL);
return;
}
if (gcs_block_deny_direct_armed_forward_request(req))
if (gcs_block_deny_direct_armed_forward_request(req)) {
/* Retry comes back with the same (request_id, epoch)
* key and direct-land suppressed — release the in-flight
* dedup entry so it is re-evaluated (S3 RC-B). */
cluster_gcs_block_dedup_remove(dedup_worker_id, &key);
return;
}

memset(&fwd, 0, sizeof(fwd));
fwd.request_id = req->request_id;
Expand Down Expand Up @@ -4646,8 +4687,13 @@ cluster_gcs_handle_block_request_envelope(const ClusterICEnvelope *env, const vo
InvalidXLogRecPtr, NULL);
return;
}
if (gcs_block_deny_direct_armed_forward_request(req))
if (gcs_block_deny_direct_armed_forward_request(req)) {
/* Retry comes back with the same (request_id, epoch) key and
* direct-land suppressed — release the in-flight dedup entry
* so it is re-evaluated (S3 RC-B). */
cluster_gcs_block_dedup_remove(dedup_worker_id, &key);
return;
}

/* Build and send GCS_BLOCK_FORWARD to holder. */
memset(&fwd, 0, sizeof(fwd));
Expand Down Expand Up @@ -4813,6 +4859,26 @@ build_and_send_reply: {
has_block_payload ? block_payload : NULL,
send_sf_dep ? &sf_dep_vec : NULL, send_sf_dep);

/*
* Duplicate-reply injection (S3 RC-A test surface). When armed with
* SKIP, ship the just-installed cached copy AHEAD of the normal send so
* the requester receives the same reply twice back-to-back — the
* deterministic stand-in for the protocol-normal duplicate (dedup
* CACHED_REPLY resend racing the original under retransmit). The
* requester-side first-reply-wins guard must drop the second delivery
* (stale_reply_drop_count++) instead of overwriting the slot mid-consume.
*/
CLUSTER_INJECTION_POINT("cluster-gcs-block-duplicate-grant-reply");
if (cluster_injection_should_skip("cluster-gcs-block-duplicate-grant-reply")) {
GcsBlockDedupEntry dup_entry;

memset(&dup_entry, 0, sizeof(dup_entry));
if (cluster_gcs_block_dedup_lookup_or_register(dedup_worker_id, &key, req->tag,
req->transition_id, &dup_entry)
== GCS_BLOCK_DEDUP_CACHED_REPLY)
gcs_block_resend_cached_reply(req->sender_node, &dup_entry);
}

/*
* spec-2.34 D17 — drop-reply injection. When active with SKIP,
* master DOES NOT send the reply envelope (sender experiences
Expand Down Expand Up @@ -5016,7 +5082,13 @@ gcs_block_direct_fail_slot(ClusterGcsBlockBackendBlock *blk, ClusterGcsBlockOuts
slot->direct_target_lkey = 0;
slot->direct_target_prepared = false;
gcs_block_note_direct_abort();
if (authoritative_denial && hdr != NULL) {
/* First-reply-wins (S3 RC-A): if a wire reply already landed for this
* attempt, leave the slot reply fields untouched — the requester may be
* consuming them without the lock — and do not mark stale either (the
* landed reply is the outcome). Just signal. */
if (slot->reply_received) {
/* keep landed reply */
} else if (authoritative_denial && hdr != NULL) {
slot->reply_header = *hdr;
memset(slot->reply_block_data, 0, sizeof(slot->reply_block_data));
slot->reply_received = true;
Expand Down Expand Up @@ -5120,6 +5192,15 @@ cluster_gcs_block_lmon_handle_direct_land_completion(int32 peer_node, uint64 wr_
return;
}

/* First-reply-wins (S3 RC-A): a wire reply that already landed for this
* attempt owns the slot reply fields (the requester may be consuming
* them without the lock). Clean up the direct target without touching
* them; the landed reply is the outcome. */
if (slot->reply_received) {
gcs_block_direct_fail_slot(blk, slot, GCS_BLOCK_DIRECT_ABORT_TIMEOUT, false, NULL);
return;
}

cluster_bufmgr_finish_direct_land_target_for_gcs(slot->direct_target_buf, true,
(XLogRecPtr)hdr->page_lsn);
slot->direct_target_prepared = false;
Expand Down Expand Up @@ -5314,6 +5395,23 @@ cluster_gcs_handle_block_reply_envelope(const ClusterICEnvelope *env, const void
int32 fwd_master;
bool authorized = false;

/*
* First-reply-wins (S3 RC-A). Duplicate replies for the same
* armed attempt are protocol-normal (dedup CACHED_REPLY resend
* after a requester retransmit, FORWARDED re-forward), but the
* requester consumes reply_header/reply_block_data WITHOUT this
* lock once it has observed reply_received under it — the slot
* reply fields are immutable from reply_received=true until the
* owner rearms the slot. Overwriting here mid-consume tears the
* 8KB image under the CRC32C verify and surfaces a false
* DENIED_CHECKSUM_FAIL (the S3 loopback "CRC verify reject").
*/
if (slot->reply_received) {
pg_atomic_fetch_add_u64(&ClusterGcsBlock->stale_reply_drop_count, 1);
LWLockRelease(&blk->lock.lock);
return;
}

if (slot->direct_state == GCS_BLOCK_DIRECT_ARMED
|| slot->direct_state == GCS_BLOCK_DIRECT_LANDED
|| slot->direct_state == GCS_BLOCK_DIRECT_ABORTING) {
Expand Down Expand Up @@ -6007,19 +6105,58 @@ gcs_block_broadcast_invalidate_and_wait_ext(const GcsBlockRequestPayload *req, u

/* Claim and stamp the broadcast slot as one critical section. ACK
* validation reads the same identity under this lock, so a late ACK from
* an older broadcast cannot match a newly claimed slot by request_id alone. */
LWLockAcquire(&ClusterGcsBlock->invalidate_broadcast_lock.lock, LW_EXCLUSIVE);
if (pg_atomic_read_u64(&ClusterGcsBlock->invalidate_broadcast_request_id) != 0) {
LWLockRelease(&ClusterGcsBlock->invalidate_broadcast_lock.lock);
cluster_xp_abort(&xp_inv); /* PGRAC: spec-5.59 — slot busy, no sample */
return false;
* an older broadcast cannot match a newly claimed slot by request_id alone.
*
* The slot is a node-wide singleton, and every caller of this blocking
* variant runs in BACKEND context (the LMON/LMS dispatch paths use the
* nowait fan-out instead), so a busy slot must be WAITED OUT, not failed
* instantly: two concurrent local S->X upgrades — even on unrelated
* blocks — collide here, and the pre-wait behavior surfaced the loser as
* a spurious "S->X upgrade invalidate did not complete" ERROR (the
* S3-observed low-concurrency failure class). Bound the wait by the
* same ACK-collection budget; a genuine exhaustion still returns false. */
{
TimestampTz claim_deadline
= GetCurrentTimestamp() + (TimestampTz)timeout_ms * (TimestampTz)1000;
bool claimed = false;

ConditionVariablePrepareToSleep(&ClusterGcsBlock->invalidate_broadcast_cv);
for (;;) {
TimestampTz now;
long remaining_ms;

LWLockAcquire(&ClusterGcsBlock->invalidate_broadcast_lock.lock, LW_EXCLUSIVE);
if (pg_atomic_read_u64(&ClusterGcsBlock->invalidate_broadcast_request_id) == 0) {
ClusterGcsBlock->invalidate_broadcast_epoch = current_epoch;
ClusterGcsBlock->invalidate_broadcast_tag = req->tag;
pg_atomic_write_u32(&ClusterGcsBlock->invalidate_broadcast_expected_bm, holders_bm);
pg_atomic_write_u32(&ClusterGcsBlock->invalidate_broadcast_acked_bm, 0);
pg_atomic_write_u64(&ClusterGcsBlock->invalidate_broadcast_request_id,
req->request_id);
LWLockRelease(&ClusterGcsBlock->invalidate_broadcast_lock.lock);
claimed = true;
break;
}
LWLockRelease(&ClusterGcsBlock->invalidate_broadcast_lock.lock);

CHECK_FOR_INTERRUPTS();
now = GetCurrentTimestamp();
if (now >= claim_deadline)
break;
remaining_ms = (long)((claim_deadline - now) / 1000);
if (remaining_ms <= 0)
remaining_ms = 1;
(void)ConditionVariableTimedSleep(&ClusterGcsBlock->invalidate_broadcast_cv,
remaining_ms,
WAIT_EVENT_GCS_BLOCK_INVALIDATE_ACK_WAIT);
}
ConditionVariableCancelSleep();

if (!claimed) {
cluster_xp_abort(&xp_inv); /* PGRAC: spec-5.59 — slot busy, no sample */
return false;
}
}
ClusterGcsBlock->invalidate_broadcast_epoch = current_epoch;
ClusterGcsBlock->invalidate_broadcast_tag = req->tag;
pg_atomic_write_u32(&ClusterGcsBlock->invalidate_broadcast_expected_bm, holders_bm);
pg_atomic_write_u32(&ClusterGcsBlock->invalidate_broadcast_acked_bm, 0);
pg_atomic_write_u64(&ClusterGcsBlock->invalidate_broadcast_request_id, req->request_id);
LWLockRelease(&ClusterGcsBlock->invalidate_broadcast_lock.lock);

/* Build and dispatch INVALIDATE to each holder bit. */
memset(&inv, 0, sizeof(inv));
Expand Down Expand Up @@ -6067,7 +6204,10 @@ gcs_block_broadcast_invalidate_and_wait_ext(const GcsBlockRequestPayload *req, u
}
ConditionVariableCancelSleep();

/* Release the slot. */
/* Release the slot. Broadcast the CV afterwards: concurrent claimants
* sleep on the same invalidate_broadcast_cv (see the bounded claim-wait
* above), so the release must wake them or they only recheck on their
* timeout slices. */
LWLockAcquire(&ClusterGcsBlock->invalidate_broadcast_lock.lock, LW_EXCLUSIVE);
pg_atomic_write_u64(&ClusterGcsBlock->invalidate_broadcast_request_id, 0);
ClusterGcsBlock->invalidate_broadcast_epoch = 0;
Expand All @@ -6076,6 +6216,7 @@ gcs_block_broadcast_invalidate_and_wait_ext(const GcsBlockRequestPayload *req, u
pg_atomic_write_u32(&ClusterGcsBlock->invalidate_broadcast_expected_bm, 0);
pg_atomic_write_u32(&ClusterGcsBlock->invalidate_broadcast_acked_bm, 0);
LWLockRelease(&ClusterGcsBlock->invalidate_broadcast_lock.lock);
ConditionVariableBroadcast(&ClusterGcsBlock->invalidate_broadcast_cv);

cluster_xp_end(&xp_inv); /* PGRAC: spec-5.59 D2 */
return full_ack;
Expand Down
10 changes: 10 additions & 0 deletions src/backend/cluster/cluster_inject.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,16 @@ static ClusterInjectPoint cluster_injection_points[] = {
*/
{ .name = "cluster-gcs-block-drop-reply-before-send" },
{ .name = "cluster-gcs-block-force-epoch-stale-reply" },
/*
* cluster-gcs-block-duplicate-grant-reply:
* Fires inside the master build_and_send_reply tail after the dedup
* install. SKIP ships the cached copy AHEAD of the normal send so
* the requester receives the same reply twice — deterministic
* coverage for the requester-side first-reply-wins guard (a slot
* whose reply_received is already set drops the duplicate,
* stale_reply_drop_count++, never overwrites mid-consume).
*/
{ .name = "cluster-gcs-block-duplicate-grant-reply" },
/*
* spec-2.35 D15 — CF 2-way protocol fault injection.
*
Expand Down
Loading
Loading