diff --git a/.kiro/specs/raft-drive-async/tasks.md b/.kiro/specs/raft-drive-async/tasks.md index 66e178a..8760744 100644 --- a/.kiro/specs/raft-drive-async/tasks.md +++ b/.kiro/specs/raft-drive-async/tasks.md @@ -96,7 +96,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - _Boundary: PeerCtrlDispatch, SlowLane_ - _Depends: 3.1_ -- [ ] 3.3 (P) Offload the transport-only config-push to the slow lane +- [~] 3.3 (P) Offload the transport-only config-push to the slow lane - Move the requester-less iomgr/promotion config-push transport (and its wait) to the lane; replace the drive-thread join with a lane-internal wait and non-blocking return; leave the admin upgrade path (with a waiting requester) on its existing vtask-blocking path unchanged. - Observable: a promotion/heartbeat-driven config push no longer blocks iomgr; the admin upgrade reply path is unchanged. - _Requirements: 1.2_ @@ -148,3 +148,5 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - **Part B (production-live, default-ON, no flag)**: `trim_wal_segments` (`persist.c`) now bounds deletion by `floor = repl_trim_lsn(&p->repl, active_seg_lsn, snap_interval_s)` = `min(active_seg_lsn, every active standby acked_lsn, every valid disconnected-slot last_disconnected_lsn)`, after a `repl_expire_disconnect_slots` call to age out stale slot guards. `floor ≤ active_seg_lsn` structurally (repl_trim_lsn caps at its `checkpoint_lsn` arg), so the new `seg_start_lsn < floor` check subsumes the old `< active_seg_lsn` guard and can only PRESERVE more segments — never delete one a live standby still needs (closes the bug-795/828 trim-vs-cursor race for connected/recently-disconnected standbys; does NOT fix empty/cursor=0 join — needs entity-bearing snapshots, separate spec). Lock-safe: `trim_wal_segments` runs with `p->lock` released; `repl_trim_lsn`/`repl_expire_disconnect_slots` take only `r->lock` — no inversion. Tests: `test_trim_guard_preserves_standby_segments`, `test_trim_without_standbys_proceeds_normally` (test_e2e_checkpoint.c). - **Part A (DORMANT, `g_drive_async_snapshot_offload` DEFAULT 0/OFF)**: `ha_snapshot_send_dispatch` routes `HA_REPL_SEND_SNAPSHOT` inline (flag OFF — byte-for-byte legacy, send on io_worker) or to `slowlane_submit`+`snapshot_lane_fn` (flag ON). Send body factored into `ha_snapshot_do_send` (shared by both paths, holds the XCE_TESTING hook). Per-peer single-flight via `ha_state_t.snapshot_inflight[MAX_HA_CANDIDATES]` CAS + slowlane key `peer_idx+1`; flag cleared on all 4 terminal paths (alloc-fail, -EEXIST, hard-error, lane-fn completion) exactly like reconnect (2.1). Tests: `test_snapshot_offload_runs_on_lane`, `test_snapshot_inline_when_flag_off`, `test_snapshot_offload_dedups_per_peer` (test_ha_liveness.c). - **WHY DORMANT (flag-ON gate, owned by 3.4)**: `snapshot_lane_fn` captures a BORROWED standby endpoint (unlike reconnect, which owns its ep). With the flag ON the send runs on a lane thread while the io_worker may RDHUP-`endpoint_fini` that ep → cross-thread UAF (bug-1197 family). Flag OFF keeps send+teardown serialized on the io_worker. **3.4 is the single gate** that flips BOTH `g_drive_async_snapshot_offload` AND `g_drive_async_peerctrl_dispatch` ON, after establishing the endpoint-lifetime discipline (refcount or lane-serialized teardown) and proving liveness under pool saturation. Caught + fixed pre-commit: the offload flag was initially authored DEFAULT-ON (would have shipped the UAF) → corrected to OFF in both ha.c and the ha.h accessor doc (bug-1210). + +- **3.3 (landed DORMANT — checkbox `[~]` until 3.4 gate)**: requester-less config-push transport offload, flag `g_drive_async_configpush_offload` **DEFAULT 0/OFF** (toggle `ha_set_configpush_offload_for_test`). `ha_config_push_offloadable(c, bytes, blen, sha256, new_tag)` routes inline (flag OFF — verbatim `push_config_transport_only`, caller still owns+frees `bytes`) or to `slowlane_submit`+`configpush_lane_fn` (flag ON — DEEP-COPIES `bytes`+sha into a ctx the lane frees). Single cluster-wide single-flight via scalar `ha_state_t.configpush_inflight` CAS + lane key=1 (config push is a cluster-wide fan-out, not per-peer); flag cleared on all terminal paths (alloc-fail×2, -EEXIST, hard-error, lane completion). Send body factored into `ha_config_push_do_send` (production → `push_config_transport_only`; XCE_TESTING → `ha_configpush_probe` stub since cluster.c/transport isn't linked in test). THREE requester-less drive-thread call sites converted: `raft_cb_become_leader` (promotion, both bytes/NULL branches), the `HA_REPL_PUSH_CONFIG` heartbeat case, and the direct-CTRL_PRIMARY FSM promotion branch (~ha.c:4476). UNCHANGED: `push_config_to_standbys` admin-upgrade barrier (has a waiting requester) and `join.c`. Tests: `test_configpush_offload_runs_on_lane`, `test_configpush_inline_when_flag_off`, `test_configpush_offload_dedups` (test_ha_liveness.c). Same BORROWED-ENDPOINT/bug-1197 dormancy rationale as 3.2 (lane reads `c->ha.standby_endpoints`); **3.4 flips this ON together with snapshot + peerctrl flags**. cpp-review: no CRITICAL/HIGH/MEDIUM; controller suite EXIT=0. diff --git a/.wolf/memory.md b/.wolf/memory.md index 3f90cd4..b3099bc 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -3472,3 +3472,17 @@ | 08:42 | Edited .kiro/specs/raft-drive-async/tasks.md | inline fix | ~41 | | 08:42 | Edited .kiro/specs/raft-drive-async/tasks.md | modified completion() | ~830 | | 08:42 | raft-drive-async 3.2: snapshot SlowLane offload (Part A, DORMANT flag-OFF) + trim cursor-guard (Part B, LIVE) | ha.c, ha.h, persist.c, test_ha_liveness.c, test_e2e_checkpoint.c | 196/0 ctrl tests, 0 new leaks, cpp-review GO; caught+fixed default-ON UAF (bug-1210) | ~62k | +| 08:49 | Edited src/controller/ha.h | expanded (+14 lines) | ~364 | +| 08:49 | Edited src/controller/ha.h | modified OFF() | ~351 | +| 08:50 | Edited src/controller/ha.c | modified ha_snapshot_offload_enabled() | ~513 | +| 08:51 | Edited src/controller/ha.c | added 10 condition(s) | ~2166 | +| 08:51 | Edited src/controller/ha.c | expanded (+10 lines) | ~152 | +| 08:51 | Edited src/controller/ha.c | 14→16 lines | ~228 | +| 08:52 | Edited src/controller/ha.c | modified XCE_TESTING() | ~179 | +| 08:52 | Edited src/controller/ha.c | expanded (+7 lines) | ~176 | +| 08:52 | Edited src/controller/ha.c | 14→16 lines | ~212 | +| 08:53 | Edited src/controller/testsuite/test_ha_liveness.c | added 3 condition(s) | ~2156 | +| 08:54 | Edited src/controller/testsuite/test_ha_liveness.c | 4→8 lines | ~113 | +| 08:57 | Edited src/controller/ha.c | modified XCE_TESTING() | ~115 | +| 09:05 | Edited .kiro/specs/raft-drive-async/tasks.md | inline fix | ~51 | +| 09:05 | raft-drive-async 3.3: requester-less config-push SlowLane offload (DORMANT flag-OFF), 3 drive-thread sites + deep-copy bytes ownership | ha.c, ha.h, test_ha_liveness.c | 3 tests pass, ctrl EXIT=0, cpp-review GO no CRIT/HIGH/MED | ~58k | diff --git a/src/controller/ha.c b/src/controller/ha.c index 2279dd2..f5e0c25 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -87,6 +87,13 @@ extern int push_config_transport_only(cluster_t *c, const uint8_t *bytes, uint64_t new_tag, int *n_out); #endif +/* raft-drive-async Task 3.3: requester-less config-push offload router (defined + * below, near the snapshot helpers). Forward-declared because + * raft_cb_become_leader (earlier) is one of its two call sites. */ +static void ha_config_push_offloadable(cluster_t *c, const uint8_t *bytes, + size_t blen, const uint8_t sha256[32], + uint64_t new_tag); + /* config_repl.c is linked in BOTH production and XCE_TESTING builds. * ha-term-safety task 4.3 (remediation r1): the commit-advance hook activates a * newly-committed config marker (Finding B); ha_dispatch_gate / the pre-promotion @@ -1190,14 +1197,16 @@ static void raft_cb_become_leader(void *ctx, uint64_t term) extern int conf_read_raw(const char *, uint8_t **, size_t *); if (c->myconf && c->myconf->conffile && conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) { - push_config_transport_only(c, cfg_bytes, cfg_len, + /* Task 3.3: inline (flag OFF, default) or slow-lane (flag ON). + * The offload deep-copies cfg_bytes, so this free is still + * correct regardless of routing. */ + ha_config_push_offloadable(c, cfg_bytes, cfg_len, c->myconf->config_sha256, - c->myconf->config_version_tag, NULL); + c->myconf->config_version_tag); free(cfg_bytes); } else { - push_config_transport_only(c, NULL, 0, NULL, - c->myconf ? c->myconf->config_version_tag : 0, - NULL); + ha_config_push_offloadable(c, NULL, 0, NULL, + c->myconf ? c->myconf->config_version_tag : 0); } } #endif @@ -2783,6 +2792,190 @@ static void ha_snapshot_send_dispatch(ha_state_t *ha, cluster_t *c, /* 0 or -EAGAIN: enqueued exactly once; snapshot_lane_fn clears the flag. */ } +/* ─────────────────────────────────────────────────────────────────────── + * raft-drive-async Task 3.3 — requester-less ConfigPush slow-lane offload + * (bug-1198). + * + * The two REQUESTER-LESS config-push sites — post-election promotion + * (raft_cb_become_leader) and the heartbeat-driven HA_REPL_PUSH_CONFIG + * convergence case — call push_config_transport_only, which spawns one push + * thread per standby and then BLOCKS the caller in wait_all_with_timeout up to + * CONF_PUSH_TIMEOUT_MS. Run inline on the single io_worker that drives both of + * those sites, that wait stalls every heartbeat/vote for the whole timeout. + * This offloads it to the bounded slow lane. Config push is a CLUSTER-WIDE + * fan-out, so a single cluster-wide single-flight guard (configpush_inflight) + + * a single dedup key (key=1) suffice — at most one push in flight at a time. + * + * The admin-upgrade path (push_config_to_standbys, which has a waiting + * requester / majority barrier) is NOT routed here and stays unchanged. + * ─────────────────────────────────────────────────────────────────────── */ + +#ifdef XCE_TESTING +extern void (*ha_configpush_probe)(void); +#endif + +typedef struct { + cluster_t *c; + uint8_t *bytes; /* deep copy, owned by the ctx; NULL when blen==0 */ + size_t blen; + uint8_t sha[32]; /* valid only when has_sha != 0 */ + int has_sha; + uint64_t new_tag; +} configpush_ctx_t; + +/* ha_config_push_do_send: the actual (blocking) config-push work — shared by + * the inline path and the lane worker so both behave identically. In + * XCE_TESTING builds a non-NULL ha_configpush_probe() is invoked right around + * the send so a test can record the running thread / park the lane worker. */ +static void ha_config_push_do_send(cluster_t *c, const uint8_t *bytes, + size_t blen, const uint8_t sha256[32], + uint64_t new_tag) +{ +#ifdef XCE_TESTING + /* XCE_TESTING does not link cluster.c and uses hook shims for the real + * transport (push_config_transport_only's extern is #ifndef XCE_TESTING). + * The probe stands in for the send and records the running thread. */ + if (ha_configpush_probe != NULL) + ha_configpush_probe(); + (void)c; (void)bytes; (void)blen; (void)sha256; (void)new_tag; +#else + push_config_transport_only(c, bytes, blen, sha256, new_tag, NULL); +#endif +} + +/* + * configpush_lane_fn: slow-lane worker entry. Runs the bounded blocking + * config push off the drive thread, then ALWAYS clears the cluster-wide + * in-flight flag (success or failure) so the dedup never wedges and a later + * promotion/heartbeat can retry. Frees the deep-copied bytes and the ctx. + * + * ⚠ BORROWED-ENDPOINT LIFETIME HAZARD (bug-1197 family — why this path is + * gated OFF by default). push_config_transport_only here reads + * c->ha.standby_endpoints and writes to those standby endpoints, which are + * shared with the io_worker epoll loop, NOT owned by this lane (contrast + * reconnect_lane_fn, which owns the ep it creates). If a standby + * RDHUP-disconnects while this push runs, the io_worker can endpoint_fini/free + * those endpoints concurrently → UAF. Today (flag OFF) the push runs inline ON + * the io_worker, so teardown and push are serialized on one thread and cannot + * overlap. Flipping g_drive_async_configpush_offload ON is therefore only safe + * once Task 3.4 establishes the endpoint-lifetime discipline for the whole + * pool/lane dispatch model. Until then this code lands DORMANT and is + * exercised only by unit tests with a stable (or absent) endpoint set. + */ +static void configpush_lane_fn(void *arg) +{ + configpush_ctx_t *cx = (configpush_ctx_t *)arg; + if (!cx) return; + cluster_t *c = cx->c; + + ha_config_push_do_send(c, cx->bytes, cx->blen, + cx->has_sha ? cx->sha : NULL, cx->new_tag); + + /* Clear in-flight LAST so a concurrent promotion/heartbeat tick cannot + * enqueue a second config push until this one has fully finished. */ + atomic_store_explicit(&c->ha.configpush_inflight, 0, memory_order_release); + free(cx->bytes); + free(cx); +} + +/* + * ha_config_push_offloadable: route a requester-less config push either inline + * (flag OFF, production default — runs on the caller/drive thread, unchanged + * legacy behavior; the CALLER still owns `bytes` and frees them after we + * return) or onto the slow lane (flag ON — non-blocking; we take a deep copy of + * `bytes` so the caller's free is safe). The cluster-wide single-flight CAS on + * ha.configpush_inflight is the authoritative dedup; the lane's per-key dedup + * is a redundant backstop. Mirrors the snapshot-offload terminal-path handling + * (Task 3.2). Never self-terminates: on any failure it clears the guard and + * the next cycle retries. + */ +static void ha_config_push_offloadable(cluster_t *c, const uint8_t *bytes, + size_t blen, const uint8_t sha256[32], + uint64_t new_tag) +{ + /* Flag OFF (default) or no cluster handle (no lane) → inline. */ + if (!ha_configpush_offload_enabled() || c == NULL) { + ha_config_push_do_send(c, bytes, blen, sha256, new_tag); + return; + } + + /* Cluster-wide single-flight: CAS 0->1. Already 1 → a config push is in + * flight; enqueue nothing. */ + int expected = 0; + if (!atomic_compare_exchange_strong_explicit( + &c->ha.configpush_inflight, &expected, 1, + memory_order_acq_rel, memory_order_acquire)) { + return; + } + + configpush_ctx_t *cx = malloc(sizeof(*cx)); + if (!cx) { + atomic_store_explicit(&c->ha.configpush_inflight, 0, + memory_order_release); + return; + } + cx->c = c; + cx->new_tag = new_tag; + cx->has_sha = (sha256 != NULL) ? 1 : 0; + if (cx->has_sha) + memcpy(cx->sha, sha256, 32); + if (bytes != NULL && blen > 0) { + cx->bytes = malloc(blen); + if (!cx->bytes) { + free(cx); + atomic_store_explicit(&c->ha.configpush_inflight, 0, + memory_order_release); + return; + } + memcpy(cx->bytes, bytes, blen); + cx->blen = blen; + } else { + cx->bytes = NULL; + cx->blen = 0; + } + + uint64_t key = 1; /* single cluster-wide dedup key (one config-push stream) */ + int srv = slowlane_submit(&c->slowlane, key, configpush_lane_fn, cx); + if (srv == -EEXIST) { + /* Lane still holds the key from a prior config-push task that has + * already cleared configpush_inflight (its last action) but whose key + * the lane has not yet removed. Our CAS saw 0 and set 1, but the + * submit is refused — release the flag we set so no clear is lost, and + * let the next tick retry. */ + free(cx->bytes); + free(cx); + atomic_store_explicit(&c->ha.configpush_inflight, 0, + memory_order_release); + return; + } + if (srv < 0 && srv != -EAGAIN) { + /* Hard submit failure (alloc / lane stopping). -EAGAIN is NOT a + * failure (still queued, will run). Only on a real failure tear down + * and release the guard so the next tick retries; never self-terminate. */ + WARN("HA repl: slowlane_submit configpush failed: %d — retry next cycle", + srv); + free(cx->bytes); + free(cx); + atomic_store_explicit(&c->ha.configpush_inflight, 0, + memory_order_release); + return; + } + /* 0 or -EAGAIN: enqueued exactly once; configpush_lane_fn clears the flag. */ +} + +#ifdef XCE_TESTING +/* Test seam: directly drive ha_config_push_offloadable (the two production call + * sites are #ifndef XCE_TESTING). Returns 0 (the dispatch itself cannot fail + * fatally — failures clear the guard and retry). */ +int ha_config_push_offloadable_for_test(cluster_t *c, const uint8_t *bytes, + size_t blen, const uint8_t sha256[32], + uint64_t new_tag) +{ + ha_config_push_offloadable(c, bytes, blen, sha256, new_tag); + return 0; +} +#endif + /* * ha_dispatch_repl_action_for_standby — execute the decision-loop action for * one standby on the current heartbeat cycle (Task 3.3). @@ -2836,14 +3029,16 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, size_t cfg_len = 0; if (c->myconf && c->myconf->conffile && conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) { - push_config_transport_only(c, cfg_bytes, cfg_len, + /* Task 3.3: inline (flag OFF, default) or slow-lane (flag ON). + * The offload deep-copies cfg_bytes, so this free is still + * correct regardless of routing. */ + ha_config_push_offloadable(c, cfg_bytes, cfg_len, c->myconf->config_sha256, - c->myconf->config_version_tag, NULL); + c->myconf->config_version_tag); free(cfg_bytes); } else { - push_config_transport_only(c, NULL, 0, NULL, - c->myconf ? c->myconf->config_version_tag : 0, - NULL); + ha_config_push_offloadable(c, NULL, 0, NULL, + c->myconf ? c->myconf->config_version_tag : 0); } #endif } @@ -3960,6 +4155,42 @@ int ha_snapshot_offload_enabled(void) memory_order_acquire); } +/* raft-drive-async task 3.3: ConfigPush feature flag. + * + * DEFAULT OFF (0): the requester-less config push runs INLINE on the caller + * (io_worker/drive) thread — byte-for-byte the legacy behavior, i.e. + * push_config_transport_only's fan-out followed by its blocking + * wait_all_with_timeout(CONF_PUSH_TIMEOUT_MS). This task lands DORMANT, + * exactly like the Task 3.1/3.2 flags, because the offload borrows standby + * endpoints it does not own (bug-1197 family — push_config_transport_only + * reads c->ha.standby_endpoints): moving the push to a lane thread while the + * io_worker may RDHUP-free those endpoints is a cross-thread UAF. Today (flag + * OFF) the push and the endpoint teardown are serialized on the single + * io_worker, so they cannot overlap. + * + * Flipping ON is the job of Task 3.4, which establishes the endpoint-lifetime + * discipline for the whole pool/lane dispatch model and flips this together + * with the snapshot + peer-control dispatch flags. Until then this offload is + * exercised only by unit tests with a stable (or absent) endpoint set. + * + * ONLY the requester-less call sites (post-election promotion, heartbeat-driven + * HA_REPL_PUSH_CONFIG) route through this flag; the admin-upgrade barrier path + * (push_config_to_standbys) is unaffected. Compiled in ALL builds; the setter + * is the test seam to flip RED<->GREEN. */ +static _Atomic int g_drive_async_configpush_offload = 0; + +void ha_set_configpush_offload_for_test(int on) +{ + atomic_store_explicit(&g_drive_async_configpush_offload, on ? 1 : 0, + memory_order_release); +} + +int ha_configpush_offload_enabled(void) +{ + return atomic_load_explicit(&g_drive_async_configpush_offload, + memory_order_acquire); +} + /* raft-drive-async task 3.1: PeerCtrlDispatch feature flag. * * DEFAULT OFF (0): the registered peer-control handler runs its routing body @@ -4085,6 +4316,16 @@ int (*ha_repl_stream_wal_hook)(void *ep, const char *wal_path, */ int (*ha_repl_send_snapshot_hook)(int peer_idx) = NULL; +/* + * ha_configpush_probe: test-only observation seam for the requester-less + * config-push offload (raft-drive-async Task 3.3). When non-NULL, + * ha_config_push_do_send() invokes it right around push_config_transport_only + * (on BOTH the inline and the lane path), so a test can record the running + * thread and optionally park the lane worker in flight to prove the + * cluster-wide single-flight dedup. NULL in production (zero cost). + */ +void (*ha_configpush_probe)(void) = NULL; + #endif /* XCE_TESTING */ /* ── Election Evaluation (Req 2.1, 2.7, 2.8) ─────────────────────────────── @@ -4226,13 +4467,15 @@ apply_role: ", first_boot=%d)", atomic_load_explicit(&ha->term, memory_order_acquire), is_first_boot); - /* Push config to standbys with lower config_tag (Req 7.1) */ + /* Push config to standbys with lower config_tag (Req 7.1). + * Task 3.3: requester-less promotion push → offloadable (dormant + * flag OFF = inline, unchanged; flag ON = slow lane). */ #ifdef XCE_TESTING if (ha_push_config_hook != NULL) ha_push_config_hook(c); #else if (c != NULL) - push_config_transport_only(c, NULL, 0, NULL, 0, NULL); + ha_config_push_offloadable(c, NULL, 0, NULL, 0); #endif /* Send CtrlHelloRequest to all standbys to trigger WAL catch-up (Req 7.2) */ diff --git a/src/controller/ha.h b/src/controller/ha.h index 34182a6..90d87a5 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -285,6 +285,20 @@ typedef struct { * snapshot for that peer. Indexed by * ABSOLUTE candidate index. 0 at * init (ha_init memset). */ + /* ── New field added by raft-drive-async spec (Task 3.3) ──────── */ + _Atomic int configpush_inflight; /* CLUSTER-WIDE single-flight guard + * for the offloaded requester-less + * config push. Config push is a + * cluster-wide fan-out (one push thread + * per standby), so at most ONE is in + * flight at a time — hence a scalar, not + * a per-peer array. Set (CAS 0->1) when + * a config push is submitted to the slow + * lane, cleared by the lane worker on + * completion (success or failure). While + * set, the heartbeat decision loop / + * promotion enqueues no duplicate config + * push. 0 at init (ha_init memset). */ } ha_state_t; /* ── Public API ────────────────────────────────────────────── */ @@ -1252,6 +1266,27 @@ int ha_reconnect_offload_enabled(void); void ha_set_snapshot_offload_for_test(int on); int ha_snapshot_offload_enabled(void); +/* + * ha_set_configpush_offload_for_test / ha_configpush_offload_enabled: + * TDD kill-switch for raft-drive-async task 3.3 requester-less config-push + * slow-lane offload. + * OFF (0, PRODUCTION DEFAULT): the config push runs INLINE on the caller + * (io_worker/drive) thread — byte-for-byte the legacy path (the + * requester-less push_config_transport_only fan-out + its blocking + * wait_all_with_timeout up to CONF_PUSH_TIMEOUT_MS). The offload lands + * DORMANT because the lane task borrows standby endpoints it does not own + * (bug-1197 family) and reads c->ha.standby_endpoints; moving the push to a + * lane thread while the io_worker may RDHUP-free those endpoints is a + * cross-thread UAF. Flag-ON is therefore only safe once Task 3.4 + * establishes endpoint-lifetime discipline, and flipping it ON is owned by + * Task 3.4 (the single gate that flips the snapshot + peerctrl flags too). + * ON (1): the config push runs on a slow-lane worker, off the drive thread. + * NOTE: the admin-upgrade path (push_config_to_standbys, which has a waiting + * requester / barrier) is NOT affected by this flag. + */ +void ha_set_configpush_offload_for_test(int on); +int ha_configpush_offload_enabled(void); + /* * ha_set_peerctrl_dispatch_for_test / ha_peerctrl_dispatch_enabled: * Feature-flag accessors (raft-drive-async Task 3.1). The dispatch flag gates diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index 9255ec6..d96bc09 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -1969,6 +1969,210 @@ static void test_snapshot_offload_dedups_per_peer(void) printf("test_snapshot_offload_dedups_per_peer: PASS\n"); } +/* ─────────────────────────────────────────────────────────────────────── + * Task 3.3 — requester-less ConfigPush slow-lane offload (Req 1.1, 1.5; bug-1198). + * + * Probe: ha_configpush_probe records the thread the push body ran on and + * supports an optional gate so a test can hold the lane worker in flight while + * it submits a second config push (cluster-wide dedup proof, Req 1.5). The two + * production call sites are #ifndef XCE_TESTING, so the tests drive the router + * directly via ha_config_push_offloadable_for_test(). Mirrors the snapshot + * offload probe/gate above. + * ─────────────────────────────────────────────────────────────────────── */ +extern void (*ha_configpush_probe)(void); +extern int ha_config_push_offloadable_for_test(cluster_t *c, const uint8_t *bytes, + size_t blen, const uint8_t sha256[32], + uint64_t new_tag); + +static _Atomic int g_cp_runs = 0; +static _Atomic pthread_t g_cp_last_thread; +static _Atomic int g_cp_gate_on = 0; +static pthread_mutex_t g_cp_mu = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t g_cp_release_cv = PTHREAD_COND_INITIALIZER; +static int g_cp_parked = 0; +static int g_cp_release = 0; + +static void cp_probe(void) +{ + atomic_store_explicit(&g_cp_last_thread, pthread_self(), + memory_order_relaxed); + if (atomic_load(&g_cp_gate_on)) { + pthread_mutex_lock(&g_cp_mu); + g_cp_parked = 1; + while (!g_cp_release) + pthread_cond_wait(&g_cp_release_cv, &g_cp_mu); + pthread_mutex_unlock(&g_cp_mu); + } + atomic_fetch_add(&g_cp_runs, 1); +} + +/* Bounded poll until the gated body parks; 0 on park, -1 on timeout. */ +static int cp_wait_parked(int timeout_ms) +{ + pthread_mutex_lock(&g_cp_mu); + int parked = 0; + for (int i = 0; i < timeout_ms; i++) { + if (g_cp_parked) { parked = 1; break; } + pthread_mutex_unlock(&g_cp_mu); + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */ + nanosleep(&ts, NULL); + pthread_mutex_lock(&g_cp_mu); + } + pthread_mutex_unlock(&g_cp_mu); + return parked ? 0 : -1; +} + +static int cp_wait_runs(int want, int timeout_ms) +{ + for (int i = 0; i < timeout_ms; i++) { + if (atomic_load(&g_cp_runs) >= want) break; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */ + nanosleep(&ts, NULL); + } + return atomic_load(&g_cp_runs); +} + +static void cp_release_gate(void) +{ + pthread_mutex_lock(&g_cp_mu); + g_cp_release = 1; + pthread_cond_broadcast(&g_cp_release_cv); + pthread_mutex_unlock(&g_cp_mu); +} + +static void cp_reset_probe(void) +{ + atomic_store(&g_cp_runs, 0); + atomic_store(&g_cp_gate_on, 0); + pthread_mutex_lock(&g_cp_mu); + g_cp_parked = 0; + g_cp_release = 0; + pthread_mutex_unlock(&g_cp_mu); +} + +/* + * Task 3.3 GREEN (Req 1.1, 8.1/8.2): with the offload flag ON, a requester-less + * config push is submitted to the slow lane — the push body runs on a LANE + * worker (a DIFFERENT thread from the caller) and the dispatch returns promptly. + * This proves the config push no longer blocks the io_worker/drive thread. + * + * RED before Task 3.3 lands: the push ran inline on the caller thread, so the + * different-thread assertion fails. + */ +static void test_configpush_offload_runs_on_lane(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4)); + + cp_reset_probe(); + ha_configpush_probe = cp_probe; + ha_set_configpush_offload_for_test(1); + + pthread_t caller = pthread_self(); + int rc = ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "Req 8.1/8.2: dispatch must return promptly (0) after lane submit"); + + int runs = cp_wait_runs(1, 2000); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs, + "Req 1.1: the config push body must run on the lane exactly once"); + TEST_ASSERT_FALSE_MESSAGE( + pthread_equal(atomic_load_explicit(&g_cp_last_thread, + memory_order_relaxed), caller), + "Req 1.1: the config push must run on a LANE worker, NOT the " + "caller's (io_worker/drive) thread"); + + ha_set_configpush_offload_for_test(0); /* restore production default (OFF) */ + ha_configpush_probe = NULL; + slowlane_fini(&c.slowlane); + ha_fini(&c.ha); + printf("test_configpush_offload_runs_on_lane: PASS\n"); +} + +/* + * Task 3.3 no-regression (Req 6.x): with the offload flag OFF (the production + * default until Task 3.4 resolves the borrowed-endpoint lifetime), the config + * push runs INLINE on the calling thread — byte-for-byte the pre-task behavior. + */ +static void test_configpush_inline_when_flag_off(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + + cp_reset_probe(); + ha_configpush_probe = cp_probe; + ha_set_configpush_offload_for_test(0); /* explicit OFF */ + + pthread_t caller = pthread_self(); + int rc = ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Synchronous: the body already ran on THIS thread before returning. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_cp_runs), + "Req 6.x: with the flag OFF the config push must run inline"); + TEST_ASSERT_TRUE_MESSAGE( + pthread_equal(atomic_load_explicit(&g_cp_last_thread, + memory_order_relaxed), caller), + "Req 6.x: with the flag OFF the push must run on the caller's thread"); + + ha_configpush_probe = NULL; + ha_fini(&c.ha); + printf("test_configpush_inline_when_flag_off: PASS\n"); +} + +/* + * Task 3.3 (Req 1.5): cluster-wide single-flight. While one config push is in + * flight on the lane (parked at the gate), a second push enqueues nothing — the + * CAS on configpush_inflight fails. Proof: with the first body parked, the run + * count stays at 0; after release, exactly ONE body has run (the duplicate was + * deduped, not queued). + */ +static void test_configpush_offload_dedups(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4)); + + cp_reset_probe(); + atomic_store(&g_cp_gate_on, 1); /* lane body parks at the gate */ + ha_configpush_probe = cp_probe; + ha_set_configpush_offload_for_test(1); + + /* First config push — submitted; its body parks in the lane. */ + TEST_ASSERT_EQUAL_INT(0, ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cp_wait_parked(2000), + "the first config push body must reach the gate (running on the lane)"); + + /* Second config push while the first is in flight. */ + TEST_ASSERT_EQUAL_INT(0, ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7)); + + /* Give a generous window: the duplicate must NOT run. */ + struct timespec ts = { .tv_sec = 0, .tv_nsec = 200 * 1000000L }; /* 200 ms */ + nanosleep(&ts, NULL); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_cp_runs), + "Req 1.5: the duplicate config push must be deduped while one is in flight"); + + /* Release; exactly the first body completes. */ + cp_release_gate(); + int runs = cp_wait_runs(1, 2000); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs, + "Req 1.5: exactly one config push ran (duplicate deduped)"); + /* And no second body sneaks in after release. */ + nanosleep(&ts, NULL); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_cp_runs), + "Req 1.5: the deduped duplicate was dropped, not queued behind the first"); + + ha_set_configpush_offload_for_test(0); /* restore production default (OFF) */ + ha_configpush_probe = NULL; + slowlane_fini(&c.slowlane); + ha_fini(&c.ha); + printf("test_configpush_offload_dedups: PASS\n"); +} + /* ── Suite entry point ──────────────────────────────────────────────────── */ void test_ha_liveness(void) @@ -2010,5 +2214,9 @@ void test_ha_liveness(void) RUN_TEST(test_snapshot_offload_runs_on_lane); RUN_TEST(test_snapshot_inline_when_flag_off); RUN_TEST(test_snapshot_offload_dedups_per_peer); + /* raft-drive-async Task 3.3 — requester-less ConfigPush slow-lane offload (Req 1.1, 1.5, 6.x) */ + RUN_TEST(test_configpush_offload_runs_on_lane); + RUN_TEST(test_configpush_inline_when_flag_off); + RUN_TEST(test_configpush_offload_dedups); UNITY_END(); }