diff --git a/src/backend/cluster/cluster_debug.c b/src/backend/cluster/cluster_debug.c index b9f85d0471..2ebd5e03b9 100644 --- a/src/backend/cluster/cluster_debug.c +++ b/src/backend/cluster/cluster_debug.c @@ -1272,6 +1272,10 @@ dump_grd(ReturnSetInfo *rsinfo) fmt_int32((int32)cluster_grd_outbound_reply_dirty_depth())); emit_row(rsinfo, "grd", "grd_outbound_cleanup_dirty_depth", fmt_int32((int32)cluster_grd_outbound_cleanup_dirty_depth())); + emit_row(rsinfo, "grd", "grd_outbound_cleanup_retry_warn50_count", + fmt_int64((int64)cluster_grd_outbound_cleanup_retry_warn50_count())); + emit_row(rsinfo, "grd", "grd_outbound_cleanup_retry_warn90_count", + fmt_int64((int64)cluster_grd_outbound_cleanup_retry_warn90_count())); emit_row(rsinfo, "grd", "grd_work_queue_depth", fmt_int32((int32)cluster_grd_work_queue_depth())); emit_row(rsinfo, "grd", "grd_pending_count", fmt_int64((int64)cluster_grd_pending_count())); diff --git a/src/backend/cluster/cluster_ges.c b/src/backend/cluster/cluster_ges.c index aa298b77ad..0df0f3ae26 100644 --- a/src/backend/cluster/cluster_ges.c +++ b/src/backend/cluster/cluster_ges.c @@ -1625,10 +1625,12 @@ cluster_ges_timeout_master_reject_count(void) #define GES_ORPHAN_TOMBSTONE_TTL_MS 30000 /* - * spec-5.16 — on a bounded GES timeout, abandon the reply-wait entry as a tombstone - * (so a late orphan GRANT is auto-released by the reply handler) instead of deleting - * it. If a GRANT raced just ahead of this timeout, release the unwanted grant here - * (backend context). `req` carries the holder tuple the master will match. + * P0#3 — on a bounded GES timeout, first dequeue a blocking REQUEST from the + * remote master by its exact holder identity + the wait_seq carried on the + * original wire request. Then abandon the reply-wait entry as a tombstone (so + * a late orphan GRANT is auto-released by the reply handler) instead of deleting + * it. If a GRANT raced just ahead of this timeout, release the unwanted grant + * here (backend context). `req` carries the holder tuple the master will match. */ static void ges_abandon_wait_or_release(const GesReplyWaitKey *key, const GesRequestPayload *req, int32 master, @@ -1669,6 +1671,29 @@ ges_abandon_wait_or_release(const GesReplyWaitKey *key, const GesRequestPayload return; } + /* + * A blocking REQUEST installs both a GRD waiter and a WFG edge on the + * remote master. A requester-side timeout used to leave both behind until + * an unrelated revalidation/sweep. Reuse the deadlock-victim authority: + * CANCEL_WAIT is wait_seq-exact, reliable-staged, and idempotent. Use the + * sequence sampled into the ORIGINAL request, never a second MyProc read + * (the local wait-state may already have advanced while timeout cleanup is + * running). REQUEST_NOWAIT never enqueues and therefore must not cancel. + */ + if (send_opcode == (uint32)GES_REQ_OPCODE_REQUEST) { + ClusterResId resid; + ClusterGrdHolderId waiter; + + memcpy(&resid, req->resid, sizeof(resid)); + memset(&waiter, 0, sizeof(waiter)); + waiter.node_id = (int32)req->holder_node_id; + waiter.procno = req->holder_procno; + waiter.cluster_epoch = ges_request_holder_epoch(req); + waiter.request_id = ges_request_holder_request_id(req); + cluster_ges_send_cancel_wait(master, &resid, &waiter, req->wait_seq, 0, + GES_CANCEL_WAIT_KIND_REQUEST); + } + tombstone = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), GES_ORPHAN_TOMBSTONE_TTL_MS); if (cluster_ges_reply_wait_mark_abandoned(key, tombstone)) { GesRequestPayload rel = *req; diff --git a/src/backend/cluster/cluster_grd_outbound.c b/src/backend/cluster/cluster_grd_outbound.c index b51344dd64..ff2f5b9507 100644 --- a/src/backend/cluster/cluster_grd_outbound.c +++ b/src/backend/cluster/cluster_grd_outbound.c @@ -16,7 +16,8 @@ * - All enqueue/dequeue under cluster_grd_outbound_lock (LWLock). * - Counter bumps are atomic outside any other lock. * - No palloc / malloc / ereport ERROR / wait inside enqueue. - * - dirty-list overflow → drop oldest (NOT crash;NOT ERROR). + * - reply dirty overflow → drop oldest + counter; correctness cleanup + * overflow → explicit fail-closed PANIC (never overwrite silently). * * Spec: spec-2.16-cross-node-grant-convert-mvp.md (DRAFT v0.1) * @@ -100,11 +101,19 @@ typedef struct ClusterGrdOutboundShared { uint32 cleanup_dirty_tail; uint32 cleanup_dirty_count; ClusterGrdOutboundSlot cleanup_dirty[PGRAC_GES_CLEANUP_DIRTY_BUDGET]; + + /* Lifetime LOG-once state + exported threshold-crossing counters. */ + uint8 cleanup_retry_warned_mask; + uint64 cleanup_retry_warn50_count; + uint64 cleanup_retry_warn90_count; } ClusterGrdOutboundShared; static ClusterGrdOutboundShared *cluster_grd_outbound_state = NULL; static LWLock *cluster_grd_outbound_lock = NULL; +#define CLEANUP_RETRY_WARN50_BIT 0x01 +#define CLEANUP_RETRY_WARN90_BIT 0x02 + /* ============================================================ * Shmem lifecycle. @@ -215,22 +224,28 @@ reply_dirty_push(uint32 dest_node_id, const void *payload, uint16 payload_len) cluster_lmon_duty_mark_dirty(CLUSTER_LMON_DUTY_GRD_OUTBOUND); /* spec-7.2 D1 */ } -static void +static bool cleanup_dirty_push(uint8 msg_type, uint8 origin, uint32 dest_node_id, const void *payload, - uint16 payload_len) + uint16 payload_len, uint8 *new_warnings, uint32 *depth_after) { ClusterGrdOutboundSlot *slot; + uint8 warnings = 0; + + if (new_warnings != NULL) + *new_warnings = 0; + if (depth_after != NULL) + *depth_after = cluster_grd_outbound_state->cleanup_dirty_count; if (payload_len > PGRAC_GES_OUTBOUND_PAYLOAD_MAX) - return; + return false; - /* Bounded same as reply_dirty. Cleanup release is best-effort: - * if dirty-list also full, drop (LockReleaseAll cannot wait). */ - if (cluster_grd_outbound_state->cleanup_dirty_count >= PGRAC_GES_CLEANUP_DIRTY_BUDGET) { - cluster_grd_outbound_state->cleanup_dirty_tail - = (cluster_grd_outbound_state->cleanup_dirty_tail + 1) % PGRAC_GES_CLEANUP_DIRTY_BUDGET; - cluster_grd_outbound_state->cleanup_dirty_count--; - } + /* + * P0#3: cleanup frames carry correctness state (exact CANCEL_WAIT / RELEASE). + * Never overwrite the oldest entry. The void producer APIs turn false into + * an explicit fail-closed PANIC after releasing the outbound LWLock. + */ + if (cluster_grd_outbound_state->cleanup_dirty_count >= PGRAC_GES_CLEANUP_DIRTY_BUDGET) + return false; slot = &cluster_grd_outbound_state ->cleanup_dirty[cluster_grd_outbound_state->cleanup_dirty_head]; @@ -244,32 +259,85 @@ cleanup_dirty_push(uint8 msg_type, uint8 origin, uint32 dest_node_id, const void cluster_grd_outbound_state->cleanup_dirty_head = (cluster_grd_outbound_state->cleanup_dirty_head + 1) % PGRAC_GES_CLEANUP_DIRTY_BUDGET; cluster_grd_outbound_state->cleanup_dirty_count++; + if (cluster_grd_outbound_state->cleanup_dirty_count >= PGRAC_GES_CLEANUP_DIRTY_WARN50_DEPTH + && (cluster_grd_outbound_state->cleanup_retry_warned_mask & CLEANUP_RETRY_WARN50_BIT) + == 0) { + cluster_grd_outbound_state->cleanup_retry_warned_mask |= CLEANUP_RETRY_WARN50_BIT; + cluster_grd_outbound_state->cleanup_retry_warn50_count++; + warnings |= CLEANUP_RETRY_WARN50_BIT; + } + if (cluster_grd_outbound_state->cleanup_dirty_count >= PGRAC_GES_CLEANUP_DIRTY_WARN90_DEPTH + && (cluster_grd_outbound_state->cleanup_retry_warned_mask & CLEANUP_RETRY_WARN90_BIT) + == 0) { + cluster_grd_outbound_state->cleanup_retry_warned_mask |= CLEANUP_RETRY_WARN90_BIT; + cluster_grd_outbound_state->cleanup_retry_warn90_count++; + warnings |= CLEANUP_RETRY_WARN90_BIT; + } + if (new_warnings != NULL) + *new_warnings = warnings; + if (depth_after != NULL) + *depth_after = cluster_grd_outbound_state->cleanup_dirty_count; cluster_grd_inc_ges_cleanup_deferred(); cluster_lmon_duty_mark_dirty(CLUSTER_LMON_DUTY_GRD_OUTBOUND); /* spec-7.2 D1 */ + return true; } -static void -requeue_slot(const ClusterGrdOutboundSlot *slot) +static bool +requeue_slot(const ClusterGrdOutboundSlot *slot, uint8 *new_warnings, uint32 *depth_after) { Assert(slot != NULL); switch ((ClusterGrdOutboundOrigin)slot->origin) { case CLUSTER_GRD_OUTBOUND_LMON_REPLY: reply_dirty_push(slot->dest_node_id, slot->payload, slot->payload_len); - break; + return true; case CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE: case CLUSTER_GRD_OUTBOUND_LMD_CANCEL: - cleanup_dirty_push(slot->msg_type, slot->origin, slot->dest_node_id, slot->payload, - slot->payload_len); - break; + case CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE: + return cleanup_dirty_push(slot->msg_type, slot->origin, slot->dest_node_id, slot->payload, + slot->payload_len, new_warnings, depth_after); case CLUSTER_GRD_OUTBOUND_BACKEND_REQUEST: default: - (void)ring_push(slot->msg_type, slot->origin, slot->dest_node_id, slot->payload, - slot->payload_len); - break; + return ring_push(slot->msg_type, slot->origin, slot->dest_node_id, slot->payload, + slot->payload_len); } } +static void +cleanup_retry_log_pressure(uint8 new_warnings, uint32 depth) +{ + if ((new_warnings & CLEANUP_RETRY_WARN50_BIT) != 0) + ereport(LOG, (errmsg_internal("cluster GES reliable cleanup retry queue reached 50%% " + "(depth=%u capacity=%u max_backends=%d lmon_interval_ms=%d); " + "warning is emitted once per postmaster lifetime", + depth, PGRAC_GES_CLEANUP_DIRTY_BUDGET, MaxBackends, + cluster_lmon_main_loop_interval))); + if ((new_warnings & CLEANUP_RETRY_WARN90_BIT) != 0) + ereport(LOG, (errmsg_internal("cluster GES reliable cleanup retry queue reached 90%% " + "(depth=%u capacity=%u max_backends=%d lmon_interval_ms=%d); " + "exhaustion will PANIC fail closed, warning is emitted once " + "per postmaster lifetime", + depth, PGRAC_GES_CLEANUP_DIRTY_BUDGET, MaxBackends, + cluster_lmon_main_loop_interval))); +} + +static inline bool +cleanup_origin_requires_retry(uint8 origin) +{ + return origin == CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE + || origin == CLUSTER_GRD_OUTBOUND_LMD_CANCEL + || origin == CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE; +} + +static void +cleanup_retry_exhausted(uint8 origin, uint32 dest_node_id) +{ + ereport(PANIC, + (errmsg_internal("cluster GES reliable cleanup retry queue exhausted " + "(capacity=%u origin=%u dest=%u); refusing to lose cleanup state", + PGRAC_GES_CLEANUP_DIRTY_BUDGET, (uint32)origin, dest_node_id))); +} + /* ============================================================ * 3 producer enqueue paths. @@ -361,18 +429,27 @@ void cluster_grd_outbound_enqueue_cleanup_release(uint32 dest_node_id, const void *payload, uint16 payload_len) { + bool queued; + uint8 new_warnings = 0; + uint32 depth_after = 0; + Assert(cluster_grd_outbound_state != NULL); LWLockAcquire(cluster_grd_outbound_lock, LW_EXCLUSIVE); /* CLEANUP_RELEASE shares main ring with backend pool. If full → - * cleanup dirty-list (LockReleaseAll cannot wait). */ - if (!ring_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE, dest_node_id, - payload, payload_len)) - cleanup_dirty_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE, - dest_node_id, payload, payload_len); + * fixed reliable retry list (LockReleaseAll cannot wait). */ + queued = ring_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE, dest_node_id, + payload, payload_len); + if (!queued) + queued + = cleanup_dirty_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE, + dest_node_id, payload, payload_len, &new_warnings, &depth_after); LWLockRelease(cluster_grd_outbound_lock); + cleanup_retry_log_pressure(new_warnings, depth_after); + if (!queued) + cleanup_retry_exhausted(CLUSTER_GRD_OUTBOUND_CLEANUP_RELEASE, dest_node_id); cluster_lmon_wakeup(); } @@ -382,7 +459,7 @@ cluster_grd_outbound_enqueue_cleanup_release(uint32 dest_node_id, const void *pa * Reliability contract per spec-2.24 §1.4 example 2 — cancel forward * MUST NOT use the silent-fail backend_request path (loss → deadlock * not resolved). Mirror CLEANUP_RELEASE semantics: share main ring; - * on full → cleanup dirty-list nofail. cluster_grd_outbound origin + * on full → reliable cleanup retry list (no overwrite). cluster_grd_outbound origin * CLUSTER_GRD_OUTBOUND_LMD_CANCEL identifies the producer in pg_stat * rollups. */ @@ -390,16 +467,25 @@ void cluster_grd_outbound_enqueue_lmd_cancel(uint32 dest_node_id, const void *payload, uint16 payload_len) { + bool queued; + uint8 new_warnings = 0; + uint32 depth_after = 0; + Assert(cluster_grd_outbound_state != NULL); LWLockAcquire(cluster_grd_outbound_lock, LW_EXCLUSIVE); - if (!ring_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMD_CANCEL, dest_node_id, payload, - payload_len)) - cleanup_dirty_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMD_CANCEL, dest_node_id, - payload, payload_len); + queued = ring_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMD_CANCEL, dest_node_id, + payload, payload_len); + if (!queued) + queued + = cleanup_dirty_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMD_CANCEL, + dest_node_id, payload, payload_len, &new_warnings, &depth_after); LWLockRelease(cluster_grd_outbound_lock); + cleanup_retry_log_pressure(new_warnings, depth_after); + if (!queued) + cleanup_retry_exhausted(CLUSTER_GRD_OUTBOUND_LMD_CANCEL, dest_node_id); cluster_lmon_wakeup(); } @@ -411,7 +497,7 @@ cluster_grd_outbound_enqueue_lmd_cancel(uint32 dest_node_id, const void *payload * never replies → LMS collector slot retries on poll, but if every probe * silently drops the requester eventually hits the retry budget and * returns 53R83 erroneously). Mirror CLEANUP_RELEASE / LMD_CANCEL - * semantics: ring full → cleanup dirty-list nofail per L141 family + * semantics: ring full → reliable cleanup retry list per L141 family * pattern (the CV-broadcast wake is owned by the LMS collector slot, * not this enqueue path — outbound ring is LMON-polled). * @@ -423,16 +509,25 @@ void cluster_grd_outbound_enqueue_lms_native_probe(uint32 dest_node_id, const void *payload, uint16 payload_len) { + bool queued; + uint8 new_warnings = 0; + uint32 depth_after = 0; + Assert(cluster_grd_outbound_state != NULL); LWLockAcquire(cluster_grd_outbound_lock, LW_EXCLUSIVE); - if (!ring_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE, dest_node_id, - payload, payload_len)) - cleanup_dirty_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE, - dest_node_id, payload, payload_len); + queued = ring_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE, + dest_node_id, payload, payload_len); + if (!queued) + queued + = cleanup_dirty_push(PGRAC_IC_MSG_GES_REQUEST, CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE, + dest_node_id, payload, payload_len, &new_warnings, &depth_after); LWLockRelease(cluster_grd_outbound_lock); + cleanup_retry_log_pressure(new_warnings, depth_after); + if (!queued) + cleanup_retry_exhausted(CLUSTER_GRD_OUTBOUND_LMS_NATIVE_PROBE, dest_node_id); cluster_lmon_wakeup(); } @@ -504,6 +599,22 @@ cluster_grd_outbound_drain_dirty_lists(void) return drained; } +static void +requeue_after_send_refusal(const ClusterGrdOutboundSlot *slot) +{ + bool queued; + uint8 new_warnings = 0; + uint32 depth_after = 0; + + LWLockAcquire(cluster_grd_outbound_lock, LW_EXCLUSIVE); + queued = requeue_slot(slot, &new_warnings, &depth_after); + LWLockRelease(cluster_grd_outbound_lock); + cleanup_retry_log_pressure(new_warnings, depth_after); + + if (!queued && cleanup_origin_requires_retry(slot->origin)) + cleanup_retry_exhausted(slot->origin, slot->dest_node_id); +} + int cluster_grd_outbound_lmon_drain_send(void) { @@ -530,15 +641,15 @@ cluster_grd_outbound_lmon_drain_send(void) * backend REQUEST staging path, gcs.c owner-plane rule). scanned * bounds the batch so requeued frames cannot spin it forever. */ - while (sent < 64 && scanned < 64 && cluster_grd_outbound_dequeue(&slot)) { + while (sent < PGRAC_GES_OUTBOUND_LMON_DRAIN_BATCH + && scanned < PGRAC_GES_OUTBOUND_LMON_DRAIN_BATCH + && cluster_grd_outbound_dequeue(&slot)) { ClusterICSendResult rc; scanned++; if (slot.dest_node_id < CLUSTER_MAX_NODES && peer_blocked[slot.dest_node_id]) { - LWLockAcquire(cluster_grd_outbound_lock, LW_EXCLUSIVE); - requeue_slot(&slot); - LWLockRelease(cluster_grd_outbound_lock); + requeue_after_send_refusal(&slot); continue; } @@ -558,12 +669,17 @@ cluster_grd_outbound_lmon_drain_send(void) case CLUSTER_IC_SEND_NOT_ADMITTED: if (slot.dest_node_id < CLUSTER_MAX_NODES) peer_blocked[slot.dest_node_id] = true; - LWLockAcquire(cluster_grd_outbound_lock, LW_EXCLUSIVE); - requeue_slot(&slot); - LWLockRelease(cluster_grd_outbound_lock); + requeue_after_send_refusal(&slot); break; case CLUSTER_IC_SEND_HARD_ERROR: - /* peer-down style failures are retried by higher layers. */ + /* Backend requests/replies have higher-layer retry. Exact cleanup + * frames do not: retain them in the fixed retry list until transport + * admission (or explicit fail-closed exhaustion). */ + if (cleanup_origin_requires_retry(slot.origin)) { + if (slot.dest_node_id < CLUSTER_MAX_NODES) + peer_blocked[slot.dest_node_id] = true; + requeue_after_send_refusal(&slot); + } break; } } @@ -611,3 +727,29 @@ cluster_grd_outbound_cleanup_dirty_depth(void) LWLockRelease(cluster_grd_outbound_lock); return depth; } + +uint64 +cluster_grd_outbound_cleanup_retry_warn50_count(void) +{ + uint64 count; + + if (cluster_grd_outbound_state == NULL || cluster_grd_outbound_lock == NULL) + return 0; + LWLockAcquire(cluster_grd_outbound_lock, LW_SHARED); + count = cluster_grd_outbound_state->cleanup_retry_warn50_count; + LWLockRelease(cluster_grd_outbound_lock); + return count; +} + +uint64 +cluster_grd_outbound_cleanup_retry_warn90_count(void) +{ + uint64 count; + + if (cluster_grd_outbound_state == NULL || cluster_grd_outbound_lock == NULL) + return 0; + LWLockAcquire(cluster_grd_outbound_lock, LW_SHARED); + count = cluster_grd_outbound_state->cleanup_retry_warn90_count; + LWLockRelease(cluster_grd_outbound_lock); + return count; +} diff --git a/src/include/cluster/cluster_grd_outbound.h b/src/include/cluster/cluster_grd_outbound.h index dc1c40fe53..92881fe44e 100644 --- a/src/include/cluster/cluster_grd_outbound.h +++ b/src/include/cluster/cluster_grd_outbound.h @@ -24,16 +24,18 @@ * ges_reply_dropped_count++ (backend retry * via timeout converges) — REJECT_BUSY reply * 100% 可落地 (递归 nofail 五检查 I54). - * CLEANUP_RELEASE full: cleanup dirty-list (LMON-private, - * physically separate from reply dirty-list; - * tick body continuously drains). + * CLEANUP_RELEASE full: fixed-capacity reliable retry list + * (LMON-private, physically separate from + * reply dirty-list; never overwrite). An + * exhausted retry list fails closed explicitly. * * spec-2.16 v0.6 L1.1 nofail 五检查 (I54): * (a) shmem 预分配固定容量 (compile-time constant) * (b) bounded ring-buffer (no dynamic resize) * (c) handler path 禁 palloc / malloc / ereport ERROR / wait - * (d) full → drop oldest + counter - * (e) backend timeout retry converges via GES_REQUEST re-route + * (d) reply full → drop oldest + counter; cleanup full → fail closed + * (e) cleanup retries until transport admission; backend timeout retry + * converges via GES_REQUEST re-route * * Step 2 (this spec) ships the ring + 5 API + reserved pool + 2 * dirty-list infrastructure. Step 3 D6 wires LMON tick body drain. @@ -91,11 +93,46 @@ typedef enum ClusterGrdOutboundOrigin { #define PGRAC_GES_OUTBOUND_RING_CAPACITY 256 #define PGRAC_GES_OUTBOUND_LMON_REPLY_RESERVED_BUDGET 64 #define PGRAC_GES_OUTBOUND_CLEANUP_BUDGET 64 +#define PGRAC_GES_OUTBOUND_LMON_DRAIN_BATCH 64 /* dirty-list bounded sizes — separately预分配 reply vs cleanup paths * (I54 + I46 — physically separated; do not share buffer). */ #define PGRAC_GES_REPLY_DIRTY_BUDGET 64 -#define PGRAC_GES_CLEANUP_DIRTY_BUDGET 64 +/* + * P0#3: REQUEST timeout cleanup is correctness state, not telemetry. The + * sizing model is a finite burst budget, not a proof that exhaustion is + * unreachable: + * + * timeout slots = 2 * MaxBackends * timeout_abort_waves_per_LMON_stall + * total slots = timeout slots + backend-exit RELEASEs + LMD/native frames + * + * Here timeout_abort_waves_per_LMON_stall is the per-backend timeout/abort + * rate multiplied by the interval in which LMON makes no successful drain. + * + * A backend has at most one synchronous reply wait at a time. One timed-out + * blocking request emits one exact CANCEL_WAIT plus, only when GRANT raced the + * timeout, one RELEASE: at most two frames per backend per abort wave. Thus + * 1024 holds one full two-frame wave for 512 backends. The LMON sender drains + * at most PGRAC_GES_OUTBOUND_LMON_DRAIN_BATCH=64 frames per iteration, so the + * retry budget is 16 full drain batches. Producers wake LMON, while + * cluster.lmon_main_loop_interval is the missed-wakeup backstop. + * + * Backend-exit sweeps can additionally emit one RELEASE per remote holder; + * LMD cancel/ack and native-probe traffic also share this pool; and a stalled + * LMON permits more than one abort wave. Those terms are workload/runtime + * bounded, not compile-time bounded. Therefore 1024 is deliberately guarded + * by 50%/90% lifetime warnings and an explicit PANIC at exhaustion; it must + * never be described as unreachable or silently overwrite the oldest exact + * cleanup. + */ +#define PGRAC_GES_CLEANUP_DIRTY_BUDGET 1024 +#define PGRAC_GES_CLEANUP_DIRTY_WARN50_DEPTH (PGRAC_GES_CLEANUP_DIRTY_BUDGET / 2) +#define PGRAC_GES_CLEANUP_DIRTY_WARN90_DEPTH ((PGRAC_GES_CLEANUP_DIRTY_BUDGET * 9 + 9) / 10) + +StaticAssertDecl(PGRAC_GES_CLEANUP_DIRTY_WARN50_DEPTH < PGRAC_GES_CLEANUP_DIRTY_WARN90_DEPTH, + "cleanup retry warning thresholds must be ordered"); +StaticAssertDecl(PGRAC_GES_CLEANUP_DIRTY_WARN90_DEPTH < PGRAC_GES_CLEANUP_DIRTY_BUDGET, + "cleanup retry 90 percent warning must precede exhaustion"); /* * Max payload bytes per ring slot. This MUST be >= the largest wire payload @@ -145,8 +182,9 @@ extern void cluster_grd_outbound_shmem_register(void); * Return: true if slot inserted into ring or dirty-list (i.e. will * eventually be sent); false ONLY on caller contract * violation (invalid msg_type / origin / payload_len - * > PAYLOAD_MAX); never false due to "full" — full path - * walks reserved → dirty-list → drop oldest per origin. + * > PAYLOAD_MAX). Cleanup producers are void/nofail: ring + * saturation enters the fixed retry list; retry-list exhaustion + * fails closed instead of losing cleanup state. * * enqueue_backend_request: backend producer. Ring full → * returns false (caller waits latch + @@ -156,9 +194,8 @@ extern void cluster_grd_outbound_shmem_register(void); * ges_reply_dropped_count++ (NEVER returns * false; REJECT_BUSY 100% 可落地 contract). * enqueue_cleanup_release: LockReleaseAll / abort producer. - * Ring full → cleanup dirty-list + - * ges_cleanup_deferred_count++ (NEVER - * returns false). + * Ring full → reliable cleanup retry list + + * ges_cleanup_deferred_count++; no overwrite. */ extern bool cluster_grd_outbound_enqueue_backend_request(uint32 dest_node_id, const void *payload, uint16 payload_len); @@ -210,6 +247,8 @@ extern int cluster_grd_outbound_lmon_drain_send(void); extern uint32 cluster_grd_outbound_ring_depth(void); /* in-ring slot count */ extern uint32 cluster_grd_outbound_reply_dirty_depth(void); extern uint32 cluster_grd_outbound_cleanup_dirty_depth(void); +extern uint64 cluster_grd_outbound_cleanup_retry_warn50_count(void); +extern uint64 cluster_grd_outbound_cleanup_retry_warn90_count(void); #endif /* !FRONTEND */ diff --git a/src/test/cluster_tap/t/104_grd_entry_shmem_alloc_smoke.pl b/src/test/cluster_tap/t/104_grd_entry_shmem_alloc_smoke.pl index 6e17211e5a..d43f4bbf29 100644 --- a/src/test/cluster_tap/t/104_grd_entry_shmem_alloc_smoke.pl +++ b/src/test/cluster_tap/t/104_grd_entry_shmem_alloc_smoke.pl @@ -132,11 +132,23 @@ # convert verdict counters (granted_inplace / enqueued / illegal), # spec-6.3a adds 4 entry lifecycle reclaim counters, and # spec-5.10 D7 adds 4 starvation-fairness counters (boost / barrier_enqueued -# / barrier_publish_fail / max_skip_observed). +# / barrier_publish_fail / max_skip_observed), and P0#3 reliable cleanup adds +# 2 lifetime retry-pressure warning counters. is($node->safe_psql('postgres', q{SELECT count(*)::int FROM pg_cluster_state WHERE category='grd'}), - '51', - 'L4b dump_grd category="grd" emits 51 rows (adds spec-6.3a entry lifecycle counters)'); + '53', + 'L4b dump_grd category="grd" emits 53 rows (adds reliable cleanup pressure counters)'); + +is($node->safe_psql('postgres', + q{SELECT (SELECT value FROM pg_cluster_state + WHERE category='grd' + AND key='grd_outbound_cleanup_retry_warn50_count') + || '|' || + (SELECT value FROM pg_cluster_state + WHERE category='grd' + AND key='grd_outbound_cleanup_retry_warn90_count')}), + '0|0', + 'L4b2 reliable cleanup 50/90 percent warning counters exported at zero baseline'); # All 3 NEW atomic counter baseline 0 (本 spec 0 caller invokes # cluster_grd_entry_lookup_or_create). diff --git a/src/test/cluster_unit/Makefile b/src/test/cluster_unit/Makefile index 4e996ac980..e6c0f05bb0 100644 --- a/src/test/cluster_unit/Makefile +++ b/src/test/cluster_unit/Makefile @@ -44,7 +44,7 @@ TESTS = test_cluster_basic test_cluster_version test_cluster_backend_types \ test_cluster_startup_phase test_cluster_lmon test_cluster_lck test_cluster_diag test_cluster_stats test_cluster_cssd test_cluster_qvotec test_cluster_voting_disk_io test_cluster_quorum_decision \ test_cluster_xlog test_cluster_tt_slot test_cluster_undo_segment \ test_cluster_epoch test_cluster_fence test_cluster_reconfig test_cluster_marker_async \ - test_cluster_ges test_cluster_grd test_cluster_grd_starvation test_cluster_lmd test_cluster_lmd_graph test_cluster_lmd_wait_state test_cluster_cancel_token test_cluster_lmd_probe_collector test_cluster_lock_acquire test_cluster_advisory \ + test_cluster_ges test_cluster_grd_outbound test_cluster_grd test_cluster_grd_starvation test_cluster_lmd test_cluster_lmd_graph test_cluster_lmd_wait_state test_cluster_cancel_token test_cluster_lmd_probe_collector test_cluster_lock_acquire test_cluster_advisory \ test_cluster_tt_slot_allocator test_cluster_itl_reader_real_triple \ test_cluster_itl_cleanout test_cluster_visibility_inject test_cluster_itl_cleanout_perf \ test_cluster_heap_lock_tuple test_cluster_perf_gates \ @@ -152,6 +152,7 @@ CLUSTER_SCN_O = $(top_builddir)/src/backend/cluster/cluster_scn.o CLUSTER_ADG_O = $(top_builddir)/src/backend/cluster/cluster_adg.o CLUSTER_LNS_O = $(top_builddir)/src/backend/cluster/cluster_lns.o CLUSTER_GES_O = $(top_builddir)/src/backend/cluster/cluster_ges.o +CLUSTER_GRD_OUTBOUND_O = $(top_builddir)/src/backend/cluster/cluster_grd_outbound.o CLUSTER_GRD_O = $(top_builddir)/src/backend/cluster/cluster_grd.o CLUSTER_LMD_O = $(top_builddir)/src/backend/cluster/cluster_lmd.o CLUSTER_LMD_GRAPH_O = $(top_builddir)/src/backend/cluster/cluster_lmd_graph.o @@ -228,7 +229,7 @@ test_cluster_backup: test_cluster_backup.c unit_test.h $(CLUSTER_VERSION_O) \ # separate rules because they also link additional cluster_*.o # objects (the test files stub the PG backend symbols those # objects reference). -SIMPLE_TESTS = $(filter-out test_cluster_ic_tier1_partial test_cluster_lms_outbound test_cluster_guc test_cluster_shmem test_cluster_signal test_cluster_views test_cluster_gviews test_cluster_ic test_cluster_conf test_cluster_ic_mock test_cluster_inject test_cluster_pgstat test_cluster_debug test_cluster_shared_fs test_cluster_shared_fs_sharedfs test_cluster_shared_fs_block_device test_cluster_smgr test_cluster_startup_phase test_cluster_lmon test_cluster_lck test_cluster_diag test_cluster_stats test_cluster_cssd test_cluster_qvotec test_cluster_voting_disk_io test_cluster_quorum_decision test_cluster_scn test_cluster_scn_frontier test_cluster_adg test_cluster_epoch test_cluster_fence test_cluster_reconfig test_cluster_ges test_cluster_grd test_cluster_grd_starvation test_cluster_lmd test_cluster_lmd_graph test_cluster_lmd_wait_state test_cluster_cancel_token test_cluster_lmd_probe_collector test_cluster_lock_acquire test_cluster_advisory test_cluster_terminal_authority test_cluster_retention test_cluster_visibility_variants test_cluster_writer_chain test_cluster_tt_2pc test_cluster_stage3_acceptance test_cluster_undo_buf test_cluster_block_apply test_cluster_thread_apply test_cluster_thread_replay test_cluster_thread_driver test_cluster_thread_orchestrator test_cluster_write_fence test_cluster_stage4_acceptance test_cluster_stage5_integrated_acceptance test_cluster_stage5_beta_acceptance test_cluster_ges_mode test_cluster_sequence test_cluster_shared_catalog test_cluster_hw test_cluster_dl test_cluster_extend_gate test_cluster_ir test_cluster_ts test_cluster_ko test_cluster_hw_snapshot test_cluster_cf_authority test_cluster_cf_storage test_cluster_cf_enqueue test_cluster_cf_phase2 test_cluster_cf_stats test_cluster_hang test_cluster_hang_resolve test_cluster_cr_server_policy test_cluster_touched_peers test_cluster_clean_leave test_cluster_membership test_cluster_node_remove test_cluster_resolver_cache test_cluster_backup test_cluster_hang_acceptance test_cluster_gcs_reqid test_cluster_runtime_visibility test_cluster_xid_stripe test_cluster_mxid_stripe test_cluster_bufmgr_pcm_hook test_cluster_cr test_cluster_cr_admit test_cluster_cr_admit_stat test_cluster_cr_cache test_cluster_cr_coordinator test_cluster_cr_key test_cluster_cr_lifecycle test_cluster_cr_pool test_cluster_cr_tuple test_cluster_cr_tuple_stat test_cluster_gcs_block test_cluster_gcs_block_2way test_cluster_gcs_block_3way test_cluster_gcs_block_lost_write test_cluster_gcs_block_retransmit test_cluster_gcs_block_dedup_reclaim test_cluster_gcs_block_dedup_htab test_cluster_gcs_dispatch test_cluster_ges_handoff test_cluster_heap_lock_tuple test_cluster_hw_lease test_cluster_ic_envelope test_cluster_ic_router test_cluster_itl_cleanout test_cluster_itl_cleanout_perf test_cluster_itl_reader_real_triple test_cluster_itl_touch test_cluster_itl_wal test_cluster_multixact test_cluster_multixact_served test_cluster_pcm_lock test_cluster_perf_gates test_cluster_recovery_merge test_cluster_recovery_plan test_cluster_recovery_worker test_cluster_reverse_key test_cluster_sinval test_cluster_sinval_ack test_cluster_snapshot_source test_cluster_stage2_acceptance test_cluster_stage5_5_cr_acceptance test_cluster_subtrans test_cluster_tt_durable test_cluster_tt_slot_allocator test_cluster_tt_status test_cluster_tt_status_hint test_cluster_uba test_cluster_undo_format test_cluster_undo_lifecycle test_cluster_undo_record test_cluster_visibility_decide_scn test_cluster_visibility_fork test_cluster_visibility_inject test_cluster_wal_state test_cluster_wal_thread test_cluster_xnode_lever test_cluster_xnode_profile test_cluster_pi_shadow test_cluster_oid_lease test_cluster_xid_authority test_cluster_recovery_anchor test_cluster_relmap_authority test_cluster_lms_shard test_cluster_gcs_block_dedup test_cluster_gcs_block_shard test_cluster_undo_resid test_cluster_undo_authority test_cluster_undo_gcs test_cluster_undo_verdict test_cluster_vis_undo_verdict_map test_cluster_undo_horizon,$(TESTS)) +SIMPLE_TESTS = $(filter-out test_cluster_ic_tier1_partial test_cluster_lms_outbound test_cluster_guc test_cluster_shmem test_cluster_signal test_cluster_views test_cluster_gviews test_cluster_ic test_cluster_conf test_cluster_ic_mock test_cluster_inject test_cluster_pgstat test_cluster_debug test_cluster_shared_fs test_cluster_shared_fs_sharedfs test_cluster_shared_fs_block_device test_cluster_smgr test_cluster_startup_phase test_cluster_lmon test_cluster_lck test_cluster_diag test_cluster_stats test_cluster_cssd test_cluster_qvotec test_cluster_voting_disk_io test_cluster_quorum_decision test_cluster_scn test_cluster_scn_frontier test_cluster_adg test_cluster_epoch test_cluster_fence test_cluster_reconfig test_cluster_ges test_cluster_grd_outbound test_cluster_grd test_cluster_grd_starvation test_cluster_lmd test_cluster_lmd_graph test_cluster_lmd_wait_state test_cluster_cancel_token test_cluster_lmd_probe_collector test_cluster_lock_acquire test_cluster_advisory test_cluster_terminal_authority test_cluster_retention test_cluster_visibility_variants test_cluster_writer_chain test_cluster_tt_2pc test_cluster_stage3_acceptance test_cluster_undo_buf test_cluster_block_apply test_cluster_thread_apply test_cluster_thread_replay test_cluster_thread_driver test_cluster_thread_orchestrator test_cluster_write_fence test_cluster_stage4_acceptance test_cluster_stage5_integrated_acceptance test_cluster_stage5_beta_acceptance test_cluster_ges_mode test_cluster_sequence test_cluster_shared_catalog test_cluster_hw test_cluster_dl test_cluster_extend_gate test_cluster_ir test_cluster_ts test_cluster_ko test_cluster_hw_snapshot test_cluster_cf_authority test_cluster_cf_storage test_cluster_cf_enqueue test_cluster_cf_phase2 test_cluster_cf_stats test_cluster_hang test_cluster_hang_resolve test_cluster_cr_server_policy test_cluster_touched_peers test_cluster_clean_leave test_cluster_membership test_cluster_node_remove test_cluster_resolver_cache test_cluster_backup test_cluster_hang_acceptance test_cluster_gcs_reqid test_cluster_runtime_visibility test_cluster_xid_stripe test_cluster_mxid_stripe test_cluster_bufmgr_pcm_hook test_cluster_cr test_cluster_cr_admit test_cluster_cr_admit_stat test_cluster_cr_cache test_cluster_cr_coordinator test_cluster_cr_key test_cluster_cr_lifecycle test_cluster_cr_pool test_cluster_cr_tuple test_cluster_cr_tuple_stat test_cluster_gcs_block test_cluster_gcs_block_2way test_cluster_gcs_block_3way test_cluster_gcs_block_lost_write test_cluster_gcs_block_retransmit test_cluster_gcs_block_dedup_reclaim test_cluster_gcs_block_dedup_htab test_cluster_gcs_dispatch test_cluster_ges_handoff test_cluster_heap_lock_tuple test_cluster_hw_lease test_cluster_ic_envelope test_cluster_ic_router test_cluster_itl_cleanout test_cluster_itl_cleanout_perf test_cluster_itl_reader_real_triple test_cluster_itl_touch test_cluster_itl_wal test_cluster_multixact test_cluster_multixact_served test_cluster_pcm_lock test_cluster_perf_gates test_cluster_recovery_merge test_cluster_recovery_plan test_cluster_recovery_worker test_cluster_reverse_key test_cluster_sinval test_cluster_sinval_ack test_cluster_snapshot_source test_cluster_stage2_acceptance test_cluster_stage5_5_cr_acceptance test_cluster_subtrans test_cluster_tt_durable test_cluster_tt_slot_allocator test_cluster_tt_status test_cluster_tt_status_hint test_cluster_uba test_cluster_undo_format test_cluster_undo_lifecycle test_cluster_undo_record test_cluster_visibility_decide_scn test_cluster_visibility_fork test_cluster_visibility_inject test_cluster_wal_state test_cluster_wal_thread test_cluster_xnode_lever test_cluster_xnode_profile test_cluster_pi_shadow test_cluster_oid_lease test_cluster_xid_authority test_cluster_recovery_anchor test_cluster_relmap_authority test_cluster_lms_shard test_cluster_gcs_block_dedup test_cluster_gcs_block_shard test_cluster_undo_resid test_cluster_undo_authority test_cluster_undo_gcs test_cluster_undo_verdict test_cluster_vis_undo_verdict_map test_cluster_undo_horizon,$(TESTS)) # spec-2.4 D16: test_cluster_epoch links cluster_epoch.o standalone. # cluster_epoch.c references ShmemInitStruct + cluster_shmem_register_region @@ -1213,6 +1214,14 @@ test_cluster_ges: test_cluster_ges.c unit_test.h \ $(top_builddir)/src/common/libpgcommon_srv.a \ $(top_builddir)/src/port/libpgport_srv.a -o $@ +# P0#3: fixed-capacity, no-overwrite GES cleanup deferred-retry queue. +test_cluster_grd_outbound: test_cluster_grd_outbound.c unit_test.h \ + $(CLUSTER_VERSION_O) $(CLUSTER_GRD_OUTBOUND_O) + $(CC) $(CFLAGS) $(CPPFLAGS) $< \ + $(CLUSTER_VERSION_O) $(CLUSTER_GRD_OUTBOUND_O) \ + $(top_builddir)/src/common/libpgcommon_srv.a \ + $(top_builddir)/src/port/libpgport_srv.a -o $@ + # spec-2.14 D11: test_cluster_grd links cluster_grd.o standalone. # cluster_grd.c references ShmemInitStruct + cluster_shmem_register_region # + cluster_conf_lookup_node + pg_atomic_*; test stubs all of these diff --git a/src/test/cluster_unit/test_cluster_debug.c b/src/test/cluster_unit/test_cluster_debug.c index 73f80b2ba6..9eeb828508 100644 --- a/src/test/cluster_unit/test_cluster_debug.c +++ b/src/test/cluster_unit/test_cluster_debug.c @@ -4029,6 +4029,16 @@ cluster_grd_outbound_cleanup_dirty_depth(void) { return 0; } +uint64 +cluster_grd_outbound_cleanup_retry_warn50_count(void) +{ + return 0; +} +uint64 +cluster_grd_outbound_cleanup_retry_warn90_count(void) +{ + return 0; +} uint32 cluster_grd_work_queue_depth(void) { diff --git a/src/test/cluster_unit/test_cluster_ges.c b/src/test/cluster_unit/test_cluster_ges.c index c9d104dbd4..9204b99180 100644 --- a/src/test/cluster_unit/test_cluster_ges.c +++ b/src/test/cluster_unit/test_cluster_ges.c @@ -53,6 +53,7 @@ #include "access/transam.h" /* spec-5.8 D1c — InvalidTransactionId for the GetTopTransactionIdIfAny stub */ #include "cluster/cluster_ges.h" +#include "cluster/cluster_ges_reply_wait.h" #include "cluster/cluster_touched_peers.h" /* spec-5.14 D2 stamp stub */ #include "cluster/cluster_grd.h" #include "cluster/cluster_grd_outbound.h" @@ -326,10 +327,12 @@ cluster_grd_entry_rebind_or_insert_holder(const ClusterResId *resid pg_attribute return CLUSTER_GRD_ENTRY_OK; } +static int32 stub_remote_master = -1; + int32 cluster_grd_lookup_master(const struct ClusterResId *resid pg_attribute_unused()) { - return cluster_node_id; + return stub_remote_master >= 0 ? stub_remote_master : cluster_node_id; } void @@ -355,6 +358,11 @@ static uint64 stub_lmd_cancel_enqueue_count = 0; static uint32 stub_lmd_cancel_last_source = 0; static uint64 stub_bast_ack = 0; static uint64 stub_deadlock_probe_drop = 0; +static uint64 stub_backend_request_enqueue_count = 0; +static GesRequestPayload stub_backend_request_last; +static uint64 stub_cancel_wait_enqueue_count = 0; +static uint32 stub_cancel_wait_last_dest = 0; +static GesCancelWaitPayload stub_cancel_wait_last; void cluster_grd_inc_ges_work_queue_full(void) @@ -484,10 +492,12 @@ cluster_grd_outbound_enqueue_cleanup_release(uint32 d pg_attribute_unused(), * coverage lives in TAP 108/109 where real backend wiring runs. */ bool -cluster_grd_outbound_enqueue_backend_request(uint32 d pg_attribute_unused(), - const void *p pg_attribute_unused(), - uint16 l pg_attribute_unused()) +cluster_grd_outbound_enqueue_backend_request(uint32 d pg_attribute_unused(), const void *p, + uint16 l) { + stub_backend_request_enqueue_count++; + if (p != NULL && l == sizeof(GesRequestPayload)) + memcpy(&stub_backend_request_last, p, sizeof(stub_backend_request_last)); return true; } @@ -520,10 +530,15 @@ pfree(void *p) /* spec-2.24 D14 stub audit. */ void -cluster_grd_outbound_enqueue_lmd_cancel(uint32 d pg_attribute_unused(), - const void *p pg_attribute_unused(), - uint16 l pg_attribute_unused()) -{} +cluster_grd_outbound_enqueue_lmd_cancel(uint32 d, const void *p, uint16 l) +{ + if (p != NULL && l == sizeof(GesCancelWaitPayload) + && ((const GesCancelWaitPayload *)p)->opcode == GES_REQ_OPCODE_CANCEL_WAIT) { + stub_cancel_wait_enqueue_count++; + stub_cancel_wait_last_dest = d; + memcpy(&stub_cancel_wait_last, p, sizeof(stub_cancel_wait_last)); + } +} bool cluster_lmd_cancel_queue_enqueue(uint32 s, const void *p pg_attribute_unused(), uint16 l pg_attribute_unused()) @@ -714,40 +729,42 @@ cluster_grd_outbound_enqueue_lms_native_probe(uint32 dest, const void *p pg_attr } /* cluster_ges_reply_wait API stubs (spec-2.23 D1). */ -struct GesReplyWaitKey; -struct GesReplyWaitEntry; -struct GesReplyWaitEntry * -cluster_ges_reply_wait_insert(const struct GesReplyWaitKey *k pg_attribute_unused(), - int64 deadline pg_attribute_unused()) +static bool stub_reply_wait_insert_enabled = false; +static GesReplyWaitEntry stub_reply_wait_entry; + +GesReplyWaitEntry * +cluster_ges_reply_wait_insert(const GesReplyWaitKey *k pg_attribute_unused(), + TimestampTz deadline pg_attribute_unused()) { - return NULL; + if (!stub_reply_wait_insert_enabled) + return NULL; + memset(&stub_reply_wait_entry, 0, sizeof(stub_reply_wait_entry)); + return &stub_reply_wait_entry; } -struct GesReplyWaitEntry * -cluster_ges_reply_wait_lookup(const struct GesReplyWaitKey *k pg_attribute_unused()) +GesReplyWaitEntry * +cluster_ges_reply_wait_lookup(const GesReplyWaitKey *k pg_attribute_unused()) { return NULL; } void -cluster_ges_reply_wait_wake(struct GesReplyWaitEntry *e pg_attribute_unused(), +cluster_ges_reply_wait_wake(GesReplyWaitEntry *e pg_attribute_unused(), uint32 opcode pg_attribute_unused(), uint32 reason pg_attribute_unused()) {} void -cluster_ges_reply_wait_delete(const struct GesReplyWaitKey *k pg_attribute_unused()) +cluster_ges_reply_wait_delete(const GesReplyWaitKey *k pg_attribute_unused()) {} -/* spec-5.16 orphan-grant stubs (deliver / mark_abandoned). int/int64 returns mirror - * the minimal-stub style above (reply_wait.h not included here); cluster_ges.o links - * by symbol name and the inert bodies keep the standalone harness coverage unchanged. */ -int -cluster_ges_reply_wait_deliver(const struct GesReplyWaitKey *k pg_attribute_unused(), +/* spec-5.16 orphan-grant stubs (deliver / mark_abandoned). */ +GesReplyDeliverResult +cluster_ges_reply_wait_deliver(const GesReplyWaitKey *k pg_attribute_unused(), uint32 opcode pg_attribute_unused(), uint32 reason pg_attribute_unused()) { - return 0; /* GES_REPLY_DELIVER_NO_WAITER */ + return GES_REPLY_DELIVER_NO_WAITER; } bool -cluster_ges_reply_wait_mark_abandoned(const struct GesReplyWaitKey *k pg_attribute_unused(), - int64 tombstone_deadline pg_attribute_unused()) +cluster_ges_reply_wait_mark_abandoned(const GesReplyWaitKey *k pg_attribute_unused(), + TimestampTz tombstone_deadline pg_attribute_unused()) { return false; } @@ -910,10 +927,15 @@ DoLockModesConflict(int a pg_attribute_unused(), int b pg_attribute_unused()) return false; } -int64 +static bool stub_clock_advances = false; +static TimestampTz stub_now = 0; + +TimestampTz GetCurrentTimestamp(void) { - return 0; + if (stub_clock_advances) + stub_now += 1000; + return stub_now; } void *MyProc; @@ -1263,10 +1285,59 @@ UT_TEST(test_ges_native_lock_probe_reply_dispatch) UT_ASSERT_EQ(stub_native_probe_recv_calls, pre_recv + 1); } +/* + * P0#3 regression: a finite remote REQUEST timeout must remove the exact + * master-side waiter. The deadlock-victim path already sends CANCEL_WAIT; + * the ordinary timeout path historically left the waiter + WFG edge behind. + */ +UT_TEST(test_ges_request_timeout_sends_wait_seq_exact_cancel_wait) +{ + ClusterResId resid; + ClusterGrdHolderId holder; + uint32 result; + + memset(&resid, 0x5A, sizeof(resid)); + memset(&holder, 0, sizeof(holder)); + holder.node_id = 0; + holder.procno = 17; + holder.cluster_epoch = 0; + holder.request_id = UINT64CONST(0x1122334455667788); + + stub_remote_master = 7; + stub_reply_wait_insert_enabled = true; + stub_clock_advances = true; + stub_now = 0; + stub_backend_request_enqueue_count = 0; + stub_cancel_wait_enqueue_count = 0; + memset(&stub_backend_request_last, 0, sizeof(stub_backend_request_last)); + memset(&stub_cancel_wait_last, 0, sizeof(stub_cancel_wait_last)); + + result = cluster_ges_send_request_and_wait(&resid, AccessExclusiveLock, &holder, + holder.request_id, 1, 0); + + UT_ASSERT_EQ(result, (uint32)GES_REJECT_REASON_TIMEOUT); + UT_ASSERT_EQ(stub_backend_request_enqueue_count, (uint64)1); + UT_ASSERT_EQ(stub_cancel_wait_enqueue_count, (uint64)1); + UT_ASSERT_EQ(stub_cancel_wait_last_dest, (uint32)7); + UT_ASSERT_EQ(stub_cancel_wait_last.opcode, (uint32)GES_REQ_OPCODE_CANCEL_WAIT); + UT_ASSERT_EQ(stub_cancel_wait_last.kind, (uint32)GES_CANCEL_WAIT_KIND_REQUEST); + UT_ASSERT_EQ(stub_cancel_wait_last.waiter_node_id, holder.node_id); + UT_ASSERT_EQ(stub_cancel_wait_last.waiter_procno, holder.procno); + UT_ASSERT_EQ(stub_cancel_wait_last.waiter_cluster_epoch, holder.cluster_epoch); + UT_ASSERT_EQ(stub_cancel_wait_last.waiter_request_id, holder.request_id); + UT_ASSERT_EQ(stub_cancel_wait_last.wait_seq, stub_backend_request_last.wait_seq); + UT_ASSERT(memcmp(stub_cancel_wait_last.resid, &resid, sizeof(resid)) == 0); + + stub_remote_master = -1; + stub_reply_wait_insert_enabled = false; + stub_clock_advances = false; + stub_now = 0; +} + int main(int argc pg_attribute_unused(), char *argv[] pg_attribute_unused()) { - UT_PLAN(16); + UT_PLAN(17); UT_RUN(test_ges_request_handler_linkable); UT_RUN(test_ges_reply_handler_linkable); @@ -1284,6 +1355,7 @@ main(int argc pg_attribute_unused(), char *argv[] pg_attribute_unused()) UT_RUN(test_ges_native_lock_probe_opcode_enum_extension); UT_RUN(test_ges_native_lock_probe_request_dispatch); UT_RUN(test_ges_native_lock_probe_reply_dispatch); + UT_RUN(test_ges_request_timeout_sends_wait_seq_exact_cancel_wait); UT_DONE(); return ut_failed_count == 0 ? 0 : 1; diff --git a/src/test/cluster_unit/test_cluster_grd_outbound.c b/src/test/cluster_unit/test_cluster_grd_outbound.c new file mode 100644 index 0000000000..9fde6f3d04 --- /dev/null +++ b/src/test/cluster_unit/test_cluster_grd_outbound.c @@ -0,0 +1,345 @@ +/*------------------------------------------------------------------------- + * + * test_cluster_grd_outbound.c + * Standalone regression tests for reliable GES cleanup staging. + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * Portions Copyright (c) 2026, pgrac contributors + * + * Author: SqlRush + * + * IDENTIFICATION + * src/test/cluster_unit/test_cluster_grd_outbound.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include +#include + +#include "cluster/cluster_gcs_block.h" +#include "cluster/cluster_ges.h" +#include "cluster/cluster_grd.h" +#include "cluster/cluster_grd_outbound.h" +#include "cluster/cluster_ic.h" +#include "cluster/cluster_ic_router.h" +#include "cluster/cluster_lmon.h" +#include "cluster/cluster_lms.h" +#include "cluster/cluster_shmem.h" +#include "miscadmin.h" +#include "storage/lwlock.h" + +#undef printf + +#include "unit_test.h" + +UT_DEFINE_GLOBALS(); + +ProcessingMode Mode = NormalProcessing; +int cluster_lms_workers = 1; +int cluster_lmon_main_loop_interval = 1000; +int MaxBackends = 200; + +void +ExceptionalCondition(const char *conditionName, const char *fileName, int lineNumber) +{ + printf("# Assert failed: %s at %s:%d\n", conditionName, fileName, lineNumber); + abort(); +} + +static uint64 ut_log_count; + +bool +errstart(int elevel, const char *domain pg_attribute_unused()) +{ + if (elevel == LOG) + ut_log_count++; + return false; +} + +bool +errstart_cold(int elevel pg_attribute_unused(), const char *domain pg_attribute_unused()) +{ + return false; +} + +void +errfinish(const char *filename pg_attribute_unused(), int lineno pg_attribute_unused(), + const char *funcname pg_attribute_unused()) +{} + +int +errmsg_internal(const char *fmt pg_attribute_unused(), ...) +{ + return 0; +} + +static const ClusterShmemRegion *ut_region; +static LWLockPadded ut_lock; + +void * +ShmemInitStruct(const char *name pg_attribute_unused(), Size size, bool *found) +{ + void *p = malloc(size); + + UT_ASSERT(p != NULL); + memset(p, 0, size); + *found = false; + return p; +} + +void +cluster_shmem_register_region(const ClusterShmemRegion *region) +{ + ut_region = region; +} + +LWLockPadded * +GetNamedLWLockTranche(const char *tranche_name pg_attribute_unused()) +{ + return &ut_lock; +} + +bool +LWLockAcquire(LWLock *lock pg_attribute_unused(), LWLockMode mode pg_attribute_unused()) +{ + return true; +} + +void +LWLockRelease(LWLock *lock pg_attribute_unused()) +{} + +static uint64 ut_cleanup_deferred; +static uint64 ut_reply_deferred; +static uint64 ut_reply_dropped; +static uint64 ut_lmon_wakeup; + +void +cluster_grd_inc_ges_cleanup_deferred(void) +{ + ut_cleanup_deferred++; +} + +void +cluster_grd_inc_ges_reply_deferred(void) +{ + ut_reply_deferred++; +} + +void +cluster_grd_inc_ges_reply_dropped(void) +{ + ut_reply_dropped++; +} + +void +cluster_lmon_duty_mark_dirty(ClusterLmonDuty duty pg_attribute_unused()) +{} + +void +cluster_lmon_wakeup(void) +{ + ut_lmon_wakeup++; +} + +const ClusterICMsgTypeInfo * +cluster_ic_get_msg_type_info(uint8 msg_type pg_attribute_unused()) +{ + return NULL; +} + +int +cluster_gcs_block_payload_shard(uint8 msg_type pg_attribute_unused(), + const void *payload pg_attribute_unused(), + uint16 payload_len pg_attribute_unused(), + int nworkers pg_attribute_unused()) +{ + return -1; +} + +bool +cluster_lms_outbound_enqueue(int worker_id pg_attribute_unused(), + uint8 msg_type pg_attribute_unused(), + uint32 dest_node_id pg_attribute_unused(), + const void *payload pg_attribute_unused(), + uint16 payload_len pg_attribute_unused()) +{ + return false; +} + +void +cluster_gcs_block_lmon_prepare_outbound_request(GcsBlockRequestPayload *req pg_attribute_unused(), + int32 dest_node pg_attribute_unused()) +{} + +static ClusterICSendResult ut_send_result = CLUSTER_IC_SEND_DONE; +static uint64 ut_send_count; +static uint64 ut_release_seen[2048]; +static int ut_release_seen_count; + +ClusterICSendResult +cluster_ic_send_envelope(uint8 msg_type pg_attribute_unused(), + int32 dest_node_id pg_attribute_unused(), const void *payload, + uint32 payload_len) +{ + ut_send_count++; + if (payload != NULL && payload_len == sizeof(GesRequestPayload) + && ((const GesRequestPayload *)payload)->opcode == GES_REQ_OPCODE_RELEASE + && ut_release_seen_count < (int)lengthof(ut_release_seen)) { + const GesRequestPayload *rel = (const GesRequestPayload *)payload; + + ut_release_seen[ut_release_seen_count++] + = ((uint64)rel->holder_request_id_lo) | (((uint64)rel->holder_request_id_hi) << 32); + } + return ut_send_result; +} + +static void +ut_reset_state(void) +{ + UT_ASSERT(ut_region != NULL); + ut_region->init_fn(); + ut_send_result = CLUSTER_IC_SEND_DONE; + ut_send_count = 0; + ut_release_seen_count = 0; + ut_log_count = 0; + memset(ut_release_seen, 0, sizeof(ut_release_seen)); +} + +static GesRequestPayload +ut_release(uint64 request_id) +{ + GesRequestPayload rel; + + memset(&rel, 0, sizeof(rel)); + rel.opcode = GES_REQ_OPCODE_RELEASE; + rel.holder_node_id = 0; + rel.holder_procno = 17; + rel.holder_request_id_lo = (uint32)(request_id & UINT64CONST(0xffffffff)); + rel.holder_request_id_hi = (uint32)(request_id >> 32); + return rel; +} + +static void +ut_fill_main_ring(void) +{ + uint8 payload = 0xA5; + int i; + + for (i = 0; i < PGRAC_GES_OUTBOUND_RING_CAPACITY; i++) + cluster_grd_outbound_enqueue_lmon_reply(1, &payload, sizeof(payload)); + UT_ASSERT_EQ(cluster_grd_outbound_ring_depth(), (uint32)PGRAC_GES_OUTBOUND_RING_CAPACITY); +} + +UT_TEST(test_cleanup_retry_queue_never_overwrites_oldest) +{ + int i; + + ut_reset_state(); + ut_fill_main_ring(); + + for (i = 1; i <= 65; i++) { + GesRequestPayload rel = ut_release((uint64)i); + + cluster_grd_outbound_enqueue_cleanup_release(1, &rel, sizeof(rel)); + } + + /* Pre-fix cleanup_dirty overwrote request_id=1 at the old 64-slot limit. */ + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_dirty_depth(), (uint32)65); + + /* Drain the filler, then the deferred releases. */ + while (cluster_grd_outbound_ring_depth() > 0 || cluster_grd_outbound_cleanup_dirty_depth() > 0) + (void)cluster_grd_outbound_lmon_drain_send(); + + UT_ASSERT_EQ(ut_release_seen_count, 65); + for (i = 0; i < 65; i++) + UT_ASSERT_EQ(ut_release_seen[i], (uint64)(i + 1)); +} + +UT_TEST(test_cleanup_hard_error_is_deferred_for_retry) +{ + GesRequestPayload rel; + + ut_reset_state(); + rel = ut_release(UINT64CONST(0xABCDEF)); + cluster_grd_outbound_enqueue_cleanup_release(3, &rel, sizeof(rel)); + + ut_send_result = CLUSTER_IC_SEND_HARD_ERROR; + (void)cluster_grd_outbound_lmon_drain_send(); + UT_ASSERT_EQ(cluster_grd_outbound_ring_depth() + cluster_grd_outbound_cleanup_dirty_depth(), + (uint32)1); + + ut_send_result = CLUSTER_IC_SEND_DONE; + (void)cluster_grd_outbound_lmon_drain_send(); + UT_ASSERT_EQ(cluster_grd_outbound_ring_depth() + cluster_grd_outbound_cleanup_dirty_depth(), + (uint32)0); + UT_ASSERT_EQ(ut_release_seen_count, 2); + UT_ASSERT_EQ(ut_release_seen[0], UINT64CONST(0xABCDEF)); + UT_ASSERT_EQ(ut_release_seen[1], UINT64CONST(0xABCDEF)); +} + +UT_TEST(test_cleanup_retry_pressure_logs_once_per_postmaster_lifetime) +{ + int i; + + ut_reset_state(); + ut_fill_main_ring(); + + for (i = 1; i < PGRAC_GES_CLEANUP_DIRTY_WARN50_DEPTH; i++) { + GesRequestPayload rel = ut_release((uint64)i); + + cluster_grd_outbound_enqueue_cleanup_release(1, &rel, sizeof(rel)); + } + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn50_count(), UINT64CONST(0)); + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn90_count(), UINT64CONST(0)); + UT_ASSERT_EQ(ut_log_count, UINT64CONST(0)); + + { + GesRequestPayload rel = ut_release((uint64)PGRAC_GES_CLEANUP_DIRTY_WARN50_DEPTH); + + cluster_grd_outbound_enqueue_cleanup_release(1, &rel, sizeof(rel)); + } + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn50_count(), UINT64CONST(1)); + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn90_count(), UINT64CONST(0)); + UT_ASSERT_EQ(ut_log_count, UINT64CONST(1)); + + for (i = PGRAC_GES_CLEANUP_DIRTY_WARN50_DEPTH + 1; + i <= PGRAC_GES_CLEANUP_DIRTY_WARN90_DEPTH + 1; i++) { + GesRequestPayload rel = ut_release((uint64)i); + + cluster_grd_outbound_enqueue_cleanup_release(1, &rel, sizeof(rel)); + } + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn50_count(), UINT64CONST(1)); + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn90_count(), UINT64CONST(1)); + UT_ASSERT_EQ(ut_log_count, UINT64CONST(2)); + + /* Draining below both thresholds does not re-arm lifetime LOG-once. */ + while (cluster_grd_outbound_ring_depth() > 0 || cluster_grd_outbound_cleanup_dirty_depth() > 0) + (void)cluster_grd_outbound_lmon_drain_send(); + ut_fill_main_ring(); + for (i = 1; i <= PGRAC_GES_CLEANUP_DIRTY_WARN90_DEPTH; i++) { + GesRequestPayload rel = ut_release((uint64)i); + + cluster_grd_outbound_enqueue_cleanup_release(1, &rel, sizeof(rel)); + } + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn50_count(), UINT64CONST(1)); + UT_ASSERT_EQ(cluster_grd_outbound_cleanup_retry_warn90_count(), UINT64CONST(1)); + UT_ASSERT_EQ(ut_log_count, UINT64CONST(2)); +} + +int +main(void) +{ + cluster_grd_outbound_shmem_register(); + UT_PLAN(3); + + UT_RUN(test_cleanup_retry_queue_never_overwrites_oldest); + UT_RUN(test_cleanup_hard_error_is_deferred_for_retry); + UT_RUN(test_cleanup_retry_pressure_logs_once_per_postmaster_lifetime); + + UT_DONE(); + return ut_failed_count == 0 ? 0 : 1; +}