diff --git a/.kiro/specs/worker-rejoin-resync/tasks.md b/.kiro/specs/worker-rejoin-resync/tasks.md index 52ee8378..fdd089c6 100644 --- a/.kiro/specs/worker-rejoin-resync/tasks.md +++ b/.kiro/specs/worker-rejoin-resync/tasks.md @@ -62,7 +62,7 @@ - _Depends: 2.1_ - [ ] 3. Core: controller reconciliation-round bookkeeping -- [ ] 3.1 (P) Track one reconciliation round per worker with a bounded timeout +- [x] 3.1 (P) Track one reconciliation round per worker with a bounded timeout - Given a worker host with one or more non-terminal jobs currently assigned to it, construct the exact list of those jobs' unique IDs as the round's query set - Track a round identifier that distinguishes the current reconciliation attempt from any prior one for the same worker - Schedule a bounded timeout for the round using a non-blocking trigger; when it fires before the round otherwise resolves, resolve the round rather than leaving the worker withheld from dispatch indefinitely @@ -156,3 +156,4 @@ - Task 1.5: `xceadmin config dump`'s `_format_cluster_config` (`build.py`) does not print `workerResyncWindowMs`/`workerResyncTimeoutMs` — an operator can set them via `config build` but can't verify the built value via `config dump`. Confirmed genuinely out of task 1.5's boundary (config-*build*, not config-*dump*) and not required by Requirements 3.7/5.3 as written, but a real, minor observability gap worth a small follow-up task if this becomes a live operator pain point. Not blocking; not added to this plan. - **Task 2.2 found and fixed bug-1746**, a real pre-existing defect in `persist.c` (part of bug-1678's original fix, predates this feature): `capnp_encode_joballoc()` unconditionally dereferenced a NULL `alloc` via `encode_JobAlloc()` instead of the already-existing NULL-safe `encode_JobAlloc_ptr()` wrapper — never triggered before since `dispatch_a_job_` (the only prior caller) always passes non-NULL. This directly contradicted design.md's claim that the NULL-alloc case was "tested, backward-compatible" on both apply seams — true for *decode*, never actually true for *encode* until this task exercised it. Fixed with a minimal guard producing a true zero-length buffer (not a capnp-null-pointer encoding — verified against `apply_committed_entry`'s `alloc_len > 0` decode gate, which a non-zero-length "null" encoding would have wrongly satisfied). design.md corrected in 4 places. **Any future task calling `persist_submit_job_dispatch_fact` with a NULL alloc can now trust this path — it wasn't safe before task 2.2, it is now.** - Task 2.2 (self-correction): a coordinator-applied one-line comment fix to `job.c` accidentally closed the enclosing `/* */` comment block early, leaving the comment's remaining lines as bare (uncompilable) code — caught immediately by re-running the build before marking the task complete, not left for review to find. Reinforces why fresh-evidence re-verification applies even to "trivial, obviously-safe" doc-only edits made outside the TDD subagent loop. +- Task 3.1: `resync.c`'s `resync_timeout_ms()` applies its own private default-if-unset fallback (15000ms) for `workerResyncTimeoutMs` rather than following `myconf.c`'s established `SET_DEFAULT_IF_UNSET`-style canonicalization convention used for the sibling `heartbeat_interval` field (`myconf.c:74`) — `myconf.c` is outside task 3.1's boundary, so this was correctly left as-is, but if a SECOND consumer of `workerResyncTimeoutMs` (or `workerResyncWindowMs`) is ever added, it would have to either duplicate this private fallback or the two could silently disagree. Worth canonicalizing in `myconf.c` if/when that happens — not needed for this feature's own single consumer. delaymgr timer cancellation: `delay.c`/`delay.h` expose no dequeue/cancel primitive for an already-armed one-shot task; `resync.c` handles this correctly via a by-value round-token check inside the fired callback (verified safe against both the logical "resolves the wrong round" bug and use-after-free, since the callback's closure never holds a pointer into the freed `resync_state_t`) — any future delaymgr one-shot user in this codebase facing the same "need to cancel" problem should use this same by-value-token-check pattern rather than assuming a cancel API exists. diff --git a/src/controller/include/resync.h b/src/controller/include/resync.h new file mode 100644 index 00000000..5bf7b241 --- /dev/null +++ b/src/controller/include/resync.h @@ -0,0 +1,97 @@ +#if !defined(_RESYNC_H_) +#define _RESYNC_H_ 1 + +#include "config.h" + +#include + +#include "xce/cluster_def.h" +#include "xce/types.h" + +/* worker-rejoin-resync task 3.1 (Req 1.1-1.3, 2.1-2.2, 5.3, 6.1, 6.3): + * per-worker job-state reconciliation round lifecycle -- arm on trigger, + * track queried job IDs + a round token, and resolve on either the + * Requirement 5.3 delaymgr timeout or (later, task 3.2) normal report + * completion. `resync_state_t` itself lives in xce/types.h (task 1.2), + * attached via `worker_t.resync`; this file owns building/populating it + * and its lifecycle, not its shape. See design.md "Controller: resync.c". + * + * Every function below follows the SAME locking convention job.c's + * record_job_reconcile_pending_fact/record_job_reconcile_assign_fact + * already established (design.md Components): the caller holds the + * relevant per-worker write lock (matching rejoin_worker's/apply_join_ + * state's own convention, worker.c) BEFORE calling and keeps holding it + * until the call returns -- these functions never lock/unlock `w` + * themselves. + */ + +/* Arms a fresh reconciliation round for `w`: enumerates every non-terminal + * job in `c`'s job table currently assigned to w's host (head/primary host, + * matching job.c's own existing "job's assigned host" representation -- + * job->job.allocs->allocs[0]->host, e.g. job.c's exechost lookup and + * job-control routing), builds the round's query-job-ID set, assigns a + * fresh monotonic round token (Req 6.1/6.3), sets deadline = now + + * ClusterConfig.workerResyncTimeoutMs (Req 5.3), attaches the resulting + * resync_state_t to w->resync, sets XCE_WORKER_RECONCILING on + * w->host.status, and schedules a non-blocking delaymgr one-shot timeout + * that resolves the round if nothing else does first. + * + * A worker with zero matching non-terminal jobs is a documented no-op (Req + * 1.2): w->resync stays NULL/untouched, XCE_WORKER_RECONCILING is not set, + * and no timeout is scheduled. + * + * If `w` already carries an outstanding round on entry (w->resync != NULL + * -- e.g. this rejoin interrupts one that never resolved), that prior round + * is discarded first (Req 6.1, via resync_resolve_round) before the fresh + * one is armed, so the new round's token always supersedes the old one and + * a later report carrying the old token is provably stale (Req 6.3). + * + * Precondition: caller already holds `w`'s write lock. + * + * Returns the number of jobs armed into the round (>=0; 0 for the + * documented Req 1.2 no-op case) on success, or a negative errno (no round + * is armed/left partially armed on failure). */ +extern int resync_arm_round(cluster_t* c, worker_t* w); + +/* Resolves (clears) whatever reconciliation round is currently outstanding + * for `w`, if any: clears XCE_WORKER_RECONCILING on w->host.status, frees + * and detaches w->resync. Called both by the Req 5.3 timeout firing + * (internally, via resync_arm_round's own delaymgr callback) and, later + * (task 3.2), once every queried job in the round has a reconciled + * outcome. Idempotent: a no-op when w->resync is already NULL -- safe to + * call speculatively. + * + * Precondition: caller holds `w`'s write lock. */ +extern void resync_resolve_round(cluster_t* c, worker_t* w); + +/* Req 6.3: true (nonzero) iff `token` is still the round token CURRENTLY + * outstanding for `w` -- i.e. w->resync != NULL && w->resync->round_token + * == token. False for a stale/superseded round (a newer resync_arm_round + * call replaced it) or an already-resolved one (w->resync == NULL). This is + * the discard-on-stale-report mechanism Req 6.1/6.3 require: a later + * report-apply caller (task 3.2) checks a report's carried round token + * against this before applying anything from it. + * + * Precondition: caller holds at least `w`'s read lock. */ +extern int resync_round_is_current(worker_t* w, uint64_t token); + +#ifdef XCE_TESTING +/* worker-rejoin-resync task 3.1 test seam: directly invokes the SAME + * internal timeout-fire logic resync_arm_round's own delaymgr callback runs + * (resync_on_timeout, resync.c -- file-static, not otherwise reachable from + * a test) for the given `token`, without needing a real delaymgr timer + * thread or any real wall-clock wait -- treats "the timer fired" as a + * directly-testable event, exactly like any other function call. Exercises + * the round-token staleness check exactly as a genuine late fire would: a + * no-op if `token` is no longer `w`'s current round. Unlike resync_arm_ + * round/resync_resolve_round above, this ACQUIRES `w`'s own write lock + * itself (mirroring the real delaymgr callback, which runs with no lock + * already held by any caller) -- so `w` must be a real RTTI-managed + * worker_t (e.g. via makeWorker), not a bare stack struct, or the lock call + * is undefined behavior. Compiled out of production (mirrors cluster.h's + * job_array_apply_test_* seams). */ +extern void resync_test_fire_timeout(cluster_t* c, worker_t* w, + uint64_t token); +#endif + +#endif diff --git a/src/controller/meson.build b/src/controller/meson.build index b1f1fc68..dd780df2 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -27,6 +27,7 @@ controller_src = [ 'repl.c', 'ctrl_metrics.c', 'slowlane.c', + 'resync.c', ] incdir = include_directories('.', '../include', '../lib', @@ -214,6 +215,7 @@ controller_unit_test_src = [ 'config_repl.c', 'slowlane.c', 'cluster.c', + 'resync.c', ] ssl_dep = dependency('openssl') diff --git a/src/controller/resync.c b/src/controller/resync.c new file mode 100644 index 00000000..22713a3b --- /dev/null +++ b/src/controller/resync.c @@ -0,0 +1,439 @@ +/* resync.c -- worker-rejoin-resync task 3.1 (Req 1.1-1.3, 2.1, 2.2, 5.3, + * 6.1, 6.3): per-worker job-state reconciliation round lifecycle. + * + * Owns exactly what design.md's "Controller: resync.c" section assigns it: + * arm a round (query-set construction + round token + deadline + a + * non-blocking delaymgr one-shot timeout), resolve a round (on timeout or, + * later, on normal completion -- task 3.2), and a round-token staleness + * check (Req 6.3) later callers use to discard a report that arrives for an + * already-discarded/superseded round. + * + * Explicitly NOT this file's job (later tasks): sending the query/report + * over the wire (task 6.1's worker.c/on_worker_join wiring), and mapping a + * reported job-state entry to the correct job.c fact-recording call (task + * 3.2's report-apply glue). Every function here is written to be called by + * that later wiring, not to include it. + */ +#include +#include +#include +#include +#include +#include +#include + +#include "cluster.h" +#include "delay.h" +#include "resync.h" +#include "utils.h" +#include "xce/types.h" + +/* ── Round token ───────────────────────────────────────────────────────── + * + * Req 6.1/6.3: a fresh round must be distinguishable from any prior round + * for the SAME worker (a disconnect-then-rejoin re-arming the round, or a + * timeout racing a late report). A single process-wide monotonic counter, + * shared across every worker's rounds, trivially satisfies this: every arm + * anywhere gets a value strictly greater than every arm before it, so in + * particular greater than any prior round for the specific worker being + * armed right now. This is simpler and collision-free versus a wall-clock + * timestamp (which can repeat within the same millisecond under load, or + * even go backward across an NTP step) or a per-worker counter (extra state + * to carry and initialize per worker_t). Starts at 1 so 0 stays available + * as an unambiguous "no token" sentinel if ever needed. */ +static _Atomic uint64_t g_resync_next_round_token = 1; + +static uint64_t resync_next_round_token(void) +{ + return atomic_fetch_add_explicit(&g_resync_next_round_token, 1, + memory_order_relaxed); +} + +/* ── Timeout / deadline ────────────────────────────────────────────────── + * + * Req 5.3's timeout value is ClusterConfig.workerResyncTimeoutMs + * (worker_resync_timeout_ms on the wire-decoded cluster_config_t, + * clusterconfig.h), reached via c->myconf->cluster_config -- the same + * accessor chain already used for the sibling shared-config field + * heartbeat_interval (cluster.c:1347/2270's + * c->myconf->cluster_config->heartbeat_interval). No xceadmin/schema + * default is defined for this field (src/configs/schema/xce.schema.json + * declares it with no "default" key); the fixture value used throughout + * xceadmin's own tests (src/xceadmin/tests/test_mapping.py, + * fixtures/full/configs/xce.yaml) is 15000ms, adopted here as the fallback + * for when a live cluster_config isn't available at all (myconf/ + * cluster_config unset or the field left at its zero-value) -- e.g. a + * minimal unit-test cluster_t fixture that never populates myconf, mirroring + * myconf.c's own SET_DEFAULT_IF_UNSET convention for heartbeat_interval. */ +#define RESYNC_DEFAULT_TIMEOUT_MS 15000 + +static uint64_t resync_timeout_ms(cluster_t* c) +{ + if (c != NULL && c->myconf != NULL && c->myconf->cluster_config != NULL + && c->myconf->cluster_config->worker_resync_timeout_ms > 0) + return (uint64_t)c->myconf->cluster_config->worker_resync_timeout_ms; + + return RESYNC_DEFAULT_TIMEOUT_MS; +} + +/* Same CLOCK_MONOTONIC millisecond representation resync_state_t.deadline + * documents (xce/types.h) and delay.h's own delay_task_t.nextrun_ms uses + * (delay.c's monotonic_ms(), duplicated here rather than exposed by + * delay.h/delay.c -- it is a two-line static helper, not worth a new public + * API for). */ +static uint64_t resync_monotonic_ms(void) +{ + struct timespec now; + + clock_gettime(CLOCK_MONOTONIC, &now); + return (uint64_t)now.tv_sec * 1000ULL + (uint64_t)now.tv_nsec / 1000000ULL; +} + +/* ── Query-set construction (Req 1.1, 2.1, 2.2) ─────────────────────────── */ + +/* Mirrors job.c's own private JOB_STATE_TERMINAL_BITS/job_state_is_terminal + * (job.c:41-50): a job is terminal once it carries a DONE or EXIT bit. + * job.c's predicate is `static inline` and not exported via cluster.h, and + * this task's boundary is explicitly "read job.c for reference, don't + * modify it" -- so this is an independent definition built from the SAME + * public bit constants (xce/types.h's JOB_STAT_DONE/JOB_STAT_EXIT), not a + * duplicate of a private symbol and not a change to job.c to export one. */ +static inline int resync_job_is_terminal(unsigned int state) +{ + return (state & (JOB_STAT_DONE | JOB_STAT_EXIT)) != 0; +} + +/* True iff `j`'s head/primary assigned host matches `hostname`. A job's + * host assignment lives ENTIRELY in job->job.allocs->allocs[0]->host -- + * confirmed by reading job.c's own existing readers of this same + * information (job.c's mark_job_ctrl-family job-control routing at + * job.c:1922-1936; the exechost lookup at job.c:3766-3771) and by + * record_job_reconcile_assign_fact's own doc comment (job.c:3031-3045), + * which independently confirms neither job_t nor xce_job_t carries any + * OTHER field representing a job's assigned host. allocs[0] specifically + * is the job's HEAD host, matching requirements.md's Boundary Context + * ("Multi-host jobs... the Controller dispatches and tracks a multi-host + * job as a single record owned by that job's head/primary host"). Caller + * holds `j`'s read lock. */ +static int resync_job_assigned_to_host(job_t* j, const char* hostname) +{ + xce_job_alloc_t* alloc = j->job.allocs; + + if (alloc == NULL || alloc->nhosts <= 0 || alloc->allocs == NULL + || alloc->allocs[0] == NULL || alloc->allocs[0]->host == NULL) + return 0; + + return strcmp(alloc->allocs[0]->host, hostname) == 0; +} + +/* Req 1.1, 2.1, 2.2: builds the round's query-job-ID set -- every + * non-terminal job in c->jobmgr currently assigned to w's host. Locking + * mirrors persist.c's collect_job_frames() nested discipline exactly: the + * jobmgr table's own RTTI_LOCK_READ is held for the entire walk (excludes + * concurrent insert/delete), and each candidate job's OWN RTTI_LOCK_READ is + * additionally, nestedly, taken just around reading its state/allocs -- + * already the proven-safe pattern in this codebase for a point-in-time- + * consistent jobmgr walk. Independent of `w`'s own lock (a different lock + * object entirely), so this never conflicts with the write lock the caller + * of resync_arm_round already holds on `w`. + * + * On success (return 0), *out_ids is a malloc'd uint64_t[*out_n] the caller + * owns (free() when done); *out_n may be 0 with *out_ids left NULL -- Req + * 1.2's "no non-terminal jobs for this host" case is not an error. On + * failure (negative errno), *out_ids and *out_n are left NULL/0 and nothing + * remains allocated. */ +static int resync_collect_query_jobids(cluster_t* c, worker_t* w, + uint64_t** out_ids, int* out_n) +{ + iterator_t it; + int rc; + uint64_t* ids = NULL; + int n = 0; + int cap = 0; + + *out_ids = NULL; + *out_n = 0; + + if (c == NULL || c->jobmgr == NULL || w == NULL) + return -EINVAL; + + lock(c->jobmgr, RTTI_LOCK_READ); + + rc = iter_make(&it, c->jobmgr); + if (rc < 0) { + unlock(c->jobmgr); + return 0; /* empty table -- zero matches, not an error (mirrors + * collect_job_frames'/collect_jobs' own empty-table + * handling) */ + } + + for (;;) { + job_t* j = NULL; + int nrc; + int match; + uint64_t jobid; + + nrc = iter_next(&it, (void**)&j); + if (nrc < 0 || nrc == RTTI_ITER_NO_DATA) + break; + if (j == NULL) + continue; + + lock(j, RTTI_LOCK_READ); + match = !resync_job_is_terminal(j->job.state) + && resync_job_assigned_to_host(j, w->host.name); + jobid = j->job.jobid; + unlock(j); + + if (!match) + continue; + + if (n == cap) { + int newcap = (cap == 0) ? 8 : (cap * 2); + uint64_t* grown + = (uint64_t*)realloc(ids, (size_t)newcap * sizeof(uint64_t)); + if (grown == NULL) { + free(ids); + iter_destroy(&it); + unlock(c->jobmgr); + *out_ids = NULL; + *out_n = 0; + return -ENOMEM; + } + ids = grown; + cap = newcap; + } + ids[n++] = jobid; + } + + iter_destroy(&it); + unlock(c->jobmgr); + + *out_ids = ids; + *out_n = n; + return 0; +} + +/* ── Timeout trampoline (Req 5.3) ──────────────────────────────────────── + * + * delay.c/delay.h expose no "cancel a queued task" primitive (delay_task_ms + * only ever enqueues; delay_worker's own loop only ever fires-then-frees or + * re-arms). So a round that resolves normally (task 3.2, later) BEFORE its + * timeout fires cannot make the already-queued delaymgr task disappear -- + * it will still fire, eventually, on schedule. Safety against a stale fire + * acting on an already-cleared/reused resync_state_t therefore comes from + * the round token, not from dequeuing: the closure below captures the + * round token the timeout was armed FOR, and resync_on_timeout only acts if + * resync_round_is_current() still reports that exact token as current for + * `w` at fire time. A round that already resolved (normally, or via a + * newer resync_arm_round superseding it) makes the old token stale, so the + * late fire becomes a pure no-op -- never a double-clear, never touching a + * `resync_state_t` that was already freed and possibly replaced. */ +typedef struct { + cluster_t* c; + worker_t* w; + uint64_t round_token; + uint64_t timeout_ms; +} resync_timeout_ctx_t; + +/* Runs on the delaymgr timer thread (c->delaymgr's runner_fn, cluster.c's + * run_delay_tasks, calls this directly and synchronously). Per the + * async-I/O thread-role contract (ha-async-io.md's Thread-role contract + * DESIGN INVARIANT): delaymgr must fire timers and trigger work, then + * return fast -- no connect(), no file I/O, no unbounded wait. Everything + * below is bounded, in-memory bookkeeping under `w`'s own mutex (a status + * bit clear, freeing one small heap struct) -- never a blocking call, so + * this satisfies that contract by construction, not by inspection alone. + * + * worker_t* is captured directly (not re-looked-up via c->workermgr): + * confirmed by reading cluster.c that no code path ever key_del()s an entry + * from c->workermgr once created (worker_obj_killer, its kvmgr datakiller, + * is a documented no-op) -- a worker_t, once created, lives for the + * process's lifetime, the same assumption ha.c's own delaymgr trampolines + * (e.g. ha_phase2_retry_delaymgr_trampoline capturing cluster_t* c) already + * make about their own captured pointers. */ +static int resync_on_timeout(void* arg) +{ + resync_timeout_ctx_t* ctx = (resync_timeout_ctx_t*)arg; + + if (ctx == NULL) + return -EINVAL; + + lock(ctx->w, RTTI_LOCK_WRITE); + if (resync_round_is_current(ctx->w, ctx->round_token)) { + /* Req 5.3: "accepted-as-answered, timed out, or left for the next + * rejoin to retry" -- this implementation takes the literal + * "timed out" branch: resolve the round now so the worker is never + * withheld from dispatch indefinitely (Req 5.1/5.2). A report that + * later arrives for this now-resolved round is naturally discarded + * by resync_round_is_current's own stale-token check (Req 6.3) + * wherever a later caller (task 3.2) applies it -- there is no + * separate "reattempted on a subsequent rejoin" action to take + * here; that is what a fresh resync_arm_round on the next join + * already provides. */ + WARN("resync: worker %s reconciliation round %lu timed out after " + "%lums -- resolving so dispatch is not withheld indefinitely " + "(Req 5.3)", + (ctx->w->host.name != NULL) ? ctx->w->host.name : "?", + (unsigned long)ctx->round_token, + (unsigned long)ctx->timeout_ms); + resync_resolve_round(ctx->c, ctx->w); + } + /* else: stale fire -- the round this timeout was armed for already + * resolved (normally) or was superseded by a newer resync_arm_round + * before the timer got here. No-op by construction (see the comment + * above resync_timeout_ctx_t). */ + unlock(ctx->w); + + free(ctx); + return 0; +} + +/* ── Public API ──────────────────────────────────────────────────────── */ + +int resync_arm_round(cluster_t* c, worker_t* w) +{ + uint64_t* ids = NULL; + int n = 0; + int rc; + resync_state_t* rs; + resync_timeout_ctx_t* ctx; + uint64_t timeout_ms; + + if (c == NULL || w == NULL) + return -EINVAL; + + /* Req 6.1: an already-outstanding round for this worker (this rejoin + * interrupts one that never resolved) is discarded first, so the fresh + * round's token always supersedes it -- never two live resync_state_t + * generations for the same worker_t at once. */ + if (w->resync != NULL) + resync_resolve_round(c, w); + + rc = resync_collect_query_jobids(c, w, &ids, &n); + if (rc < 0) { + ERROR("resync_arm_round: worker %s: failed to build reconciliation " + "query set: %d -- no round armed (the next rejoin will " + "attempt again)", + (w->host.name != NULL) ? w->host.name : "?", rc); + return rc; + } + + if (n == 0) { + /* Req 1.2: no non-terminal jobs for this host -- documented no-op: + * no round armed, no status bit set, no timer scheduled. */ + free(ids); + return 0; + } + + rs = (resync_state_t*)calloc(1, sizeof(*rs)); + if (rs == NULL) { + free(ids); + return -ENOMEM; + } + + rs->round_token = resync_next_round_token(); + rs->n_queried_jobids = n; + rs->queried_jobids = ids; + + timeout_ms = resync_timeout_ms(c); + rs->deadline = resync_monotonic_ms() + timeout_ms; + + w->resync = rs; + w->host.status |= XCE_WORKER_RECONCILING; + + ctx = (resync_timeout_ctx_t*)calloc(1, sizeof(*ctx)); + if (ctx == NULL) { + /* The round itself is still validly armed (query set, token, + * deadline, status bit are all already set above) -- only the + * BOUNDED-TIMEOUT half of Req 5.3 is unavailable this time. Fail + * loud and stay alive (steering: never fabricate partial success + * silently) rather than unwind an otherwise-good round over an + * allocation failure for a small bookkeeping struct; a later + * report (task 3.2) still resolves it normally. */ + ERROR("resync_arm_round: worker %s round %lu: timeout context " + "alloc failed -- round armed WITHOUT a bounded timeout (Req " + "5.3 degraded: resolves normally on report, not via timeout " + "this time)", + (w->host.name != NULL) ? w->host.name : "?", + (unsigned long)rs->round_token); + return n; + } + ctx->c = c; + ctx->w = w; + ctx->round_token = rs->round_token; + ctx->timeout_ms = timeout_ms; + + if (!c->delaymgr.running) { + /* Not-yet-started/test-fixture delaymgr -- mirrors ha.c's own + * `!c->delaymgr.running` guard (e.g. schedule_upgrade_ + * materialization_retry, ha_schedule_phase2_retry): the round + * stays armed; a test drives resync_on_timeout directly instead of + * waiting on a real timer thread that was never started. */ + free(ctx); + return n; + } + + rc = delay_task_ms(&c->delaymgr, (int)timeout_ms, 1, resync_on_timeout, + ctx); + if (rc < 0) { + WARN("resync_arm_round: worker %s round %lu: delay_task_ms failed: " + "%d -- round armed without a bounded timeout", + (w->host.name != NULL) ? w->host.name : "?", + (unsigned long)rs->round_token, rc); + free(ctx); + } + + return n; +} + +void resync_resolve_round(cluster_t* c, worker_t* w) +{ + (void)c; /* accepted to match this file's other functions' shape and + * what a later caller (task 3.2) already has in hand + * alongside `w` -- this function's own body has nothing to do + * with `c` today. */ + + if (w == NULL || w->resync == NULL) + return; + + w->host.status &= ~XCE_WORKER_RECONCILING; + + free(w->resync->queried_jobids); + free(w->resync); + w->resync = NULL; +} + +int resync_round_is_current(worker_t* w, uint64_t token) +{ + if (w == NULL || w->resync == NULL) + return 0; + + return w->resync->round_token == token; +} + +#ifdef XCE_TESTING +/* worker-rejoin-resync task 3.1 test seam (resync.h): builds a + * resync_timeout_ctx_t exactly as resync_arm_round's own delay_task_ms call + * would have (heap-allocated, since resync_on_timeout unconditionally + * free()s it -- matching that exact contract here, not a stack copy, or + * the free() below would be undefined behavior) and invokes resync_on_ + * timeout directly, so a test can treat "the timeout fired" as a plain, + * synchronous function call -- no real delaymgr thread, no wall-clock + * wait. */ +void resync_test_fire_timeout(cluster_t* c, worker_t* w, uint64_t token) +{ + resync_timeout_ctx_t* ctx + = (resync_timeout_ctx_t*)calloc(1, sizeof(*ctx)); + + if (ctx == NULL) + return; + + ctx->c = c; + ctx->w = w; + ctx->round_token = token; + ctx->timeout_ms = 0; + + resync_on_timeout(ctx); +} +#endif diff --git a/src/controller/testsuite/test_worker_resync.c b/src/controller/testsuite/test_worker_resync.c index acf56dc7..8a365f27 100644 --- a/src/controller/testsuite/test_worker_resync.c +++ b/src/controller/testsuite/test_worker_resync.c @@ -52,7 +52,11 @@ #include "jobj.h" #include "kvmgr.h" #include "persist.h" +#include "resync.h" /* worker-rejoin-resync task 3.1 */ #include "wal_types.h" +#include "wobj.h" /* worker-rejoin-resync task 3.1: makeWorker, for a real + * RTTI-managed worker_t the timeout test can genuinely + * lock() */ #include "xce/cluster_def.h" #include "xce/types.h" #include "xcealloc.h" @@ -67,6 +71,16 @@ static int g_twr_jobobj_ready; static cluster_t g_twr; static char g_twr_tmpdir[256]; +/* worker-rejoin-resync task 3.1: no-op kvmgr datakiller for g_twr.workermgr, + * mirroring production's own worker_obj_killer (cluster.c) -- a worker_t, + * once created, is never explicitly torn down for the life of the process + * (no key_del(c->workermgr, ...) call site exists anywhere in this + * codebase), so there is nothing for this to actually free. */ +static void twr_worker_obj_killer(void* arg) +{ + (void)arg; +} + static void twr_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, @@ -88,6 +102,23 @@ static void twr_setup(void) g_twr.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_twr.jobmgr); + /* worker-rejoin-resync task 3.1: a real workermgr so the round-lifecycle + * tests below can create a genuine RTTI-managed worker_t (via + * makeWorker) where that matters (the timeout test, whose internal + * resync_on_timeout callback genuinely lock()s `w`) -- key type mirrors + * production's own workermgr_params (cluster.c, KVMGR_KEY_STRING, + * keyed by hostname). */ + { + kvmgr_params_t workermgr_params + = { .key_type = KVMGR_KEY_STRING, + .nslots = 32, + .delta = 16, + .datakiller = twr_worker_obj_killer }; + + g_twr.workermgr = makeInstance(&xkvmgr_type, &workermgr_params); + TEST_ASSERT_NOT_NULL(g_twr.workermgr); + } + /* Real disk-backed persist, cluster == NULL: wal_hook_append_ * classified() (persist.c) takes the direct persist_wal_async() path -- * no raft/commit-engine machinery needed for a plain RETAIN-mode fact @@ -112,6 +143,8 @@ static void twr_teardown(void) if (g_twr.jobmgr) dereference(g_twr.jobmgr); + if (g_twr.workermgr) + dereference(g_twr.workermgr); xce_hook_fini(&g_twr.hooksys); snprintf(cmd, sizeof(cmd), "rm -rf %s", g_twr_tmpdir); @@ -671,6 +704,355 @@ static void test_reconcile_assign_survives_restart_replay(void) printf("test_reconcile_assign_survives_restart_replay: PASS\n"); } +/* ── resync.c round lifecycle tests (task 3.1) ─────────────────────────── + * + * Proves resync_arm_round()/resync_resolve_round()/resync_round_is_current() + * (resync.c) -- the per-worker job-state reconciliation round lifecycle: + * query-set construction (Req 1.1, 2.1, 2.2), a fresh monotonic round token + * per arm distinguishing this round from any prior one for the same worker + * (Req 6.1, 6.3), the XCE_WORKER_RECONCILING status bit (Req 5.1, 5.2), and + * the Req 5.3 bounded timeout resolving the round without any blocking + * call. Reuses the g_twr/twr_setup/twr_teardown fixture and twr_seed_job() + * from task 2.1/2.2's tests above -- resync_arm_round only needs a + * cluster_t with a real c->jobmgr (which g_twr already provides). + * + * No live caller exists yet (worker.c's rejoin_worker/apply_join_state + * wiring is task 6.1; the report-apply glue that would call + * resync_round_is_current for real is task 3.2) -- every test here calls + * resync_arm_round/resync_resolve_round/resync_round_is_current directly, + * exactly as a real caller (task 6.1) eventually will. + * + * resync_arm_round/resync_resolve_round/resync_round_is_current never + * lock/unlock `w` themselves (their documented precondition is that the + * CALLER already holds it) -- so a bare stack worker_t is sufficient and + * safe for those tests, exactly mirroring how task 2.2's own + * record_job_reconcile_assign_fact tests above already use a bare stack + * `worker_t w` with no locking. The ONE exception is the timeout test: + * resync_on_timeout (the internal delaymgr callback resync_test_fire_ + * timeout drives directly) runs with no caller-held lock -- because a real + * delaymgr fire has none -- so it genuinely calls lock()/unlock() on `w` + * itself, which requires a real RTTI-managed worker_t (via makeWorker + * against g_twr.workermgr), not a bare stack struct. */ + +#define TWR_JOBID_RESYNC_A 81101ULL +#define TWR_JOBID_RESYNC_B 81102ULL +#define TWR_JOBID_RESYNC_OTHER 81103ULL +#define TWR_JOBID_RESYNC_DONE 81104ULL +#define TWR_JOBID_RESYNC_NOALLOC 81105ULL +#define TWR_JOBID_RESYNC_SOLO 81106ULL +#define TWR_JOBID_RESYNC_TIMEOUT 81107ULL + +/* Same builder chain as twr_make_test_alloc() above, parameterized by host + * -- twr_make_test_alloc() itself hardcodes "twr-worker", which does not + * fit these tests' need for several DISTINCT hosts within one job table. */ +static xce_job_alloc_t* twr_make_test_alloc_for_host(uint64_t jobid, + const char* host) +{ + return make_job_alloc(jobid, 1, + make_host_allocs(1, + make_host_alloc((char*)host, 1, + make_allocs(1, + make_alloc((char*)"slots", NULL, 0, 1.0f, 0))))); +} + +/* True iff `jobid` appears anywhere in ids[0..n) -- used below to prove the + * query set contains exactly the expected members, neither missing a real + * match nor carrying an extra one. */ +static int twr_ids_contains(const uint64_t* ids, int n, uint64_t jobid) +{ + int i; + + for (i = 0; i < n; i++) { + if (ids[i] == jobid) + return 1; + } + return 0; +} + +/* A real RTTI-managed worker_t (via makeWorker against g_twr.workermgr), + * for the one test below that needs genuine lock()/unlock() to succeed on + * `w` -- mirrors worker.c's own init_worker() construction + * (`tmp.name = (char*)host; tmp.attrs = attrs; w = makeWorker(c->workermgr, + * &tmp);`). */ +static worker_t* twr_make_test_worker(const char* hostname) +{ + hostattrs_t attrs; + xce_worker_t tmp; + + memset(&attrs, 0, sizeof(attrs)); + memset(&tmp, 0, sizeof(tmp)); + tmp.name = (char*)hostname; + tmp.attrs = &attrs; + + return makeWorker(g_twr.workermgr, &tmp); +} + +/* Req 1.1, 2.1, 2.2: the query set matches EXACTLY the non-terminal jobs + * assigned to the given host -- not fewer (a genuine match omitted), not + * more (a different host's job, a terminal job on the SAME host, or a + * never-dispatched job with no host assignment at all, leaking in). */ +static void test_resync_arm_round_query_set_matches_exactly(void) +{ + job_t* ja; + job_t* jb; + job_t* jother; + job_t* jdone; + job_t* jnoalloc; + worker_t w; + int rc; + + twr_setup(); + + ja = twr_seed_job(TWR_JOBID_RESYNC_A, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(ja); + ja->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_A, "resync-host-A"); + + jb = twr_seed_job(TWR_JOBID_RESYNC_B, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(jb); + jb->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_B, "resync-host-A"); + + /* Same non-terminal state, DIFFERENT host -- must NOT appear in + * resync-host-A's query set. */ + jother = twr_seed_job(TWR_JOBID_RESYNC_OTHER, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(jother); + jother->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_OTHER, + "resync-host-B"); + + /* SAME host, but terminal (DONE) -- must NOT appear (Req 1.1's + * "non-terminal" qualifier). */ + jdone = twr_seed_job(TWR_JOBID_RESYNC_DONE, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(jdone); + jdone->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_DONE, "resync-host-A"); + + /* Never dispatched (no allocation at all) -- not assigned to ANY host, + * so must NOT appear regardless of state. */ + jnoalloc = twr_seed_job(TWR_JOBID_RESYNC_NOALLOC, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(jnoalloc); + TEST_ASSERT_NULL(jnoalloc->job.allocs); + + memset(&w, 0, sizeof(w)); + w.host.name = (char*)"resync-host-A"; + + rc = resync_arm_round(&g_twr, &w); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, rc, + "exactly the 2 non-terminal resync-host-A jobs must be armed into " + "the round"); + TEST_ASSERT_NOT_NULL(w.resync); + TEST_ASSERT_EQUAL_INT(2, w.resync->n_queried_jobids); + TEST_ASSERT_TRUE_MESSAGE( + twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, + TWR_JOBID_RESYNC_A), + "job A (non-terminal, resync-host-A) must be in the query set"); + TEST_ASSERT_TRUE_MESSAGE( + twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, + TWR_JOBID_RESYNC_B), + "job B (non-terminal, resync-host-A) must be in the query set"); + TEST_ASSERT_FALSE_MESSAGE( + twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, + TWR_JOBID_RESYNC_OTHER), + "a job assigned to a DIFFERENT host must not appear"); + TEST_ASSERT_FALSE_MESSAGE( + twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, + TWR_JOBID_RESYNC_DONE), + "a TERMINAL job on the same host must not appear (Req 1.1)"); + TEST_ASSERT_FALSE_MESSAGE( + twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, + TWR_JOBID_RESYNC_NOALLOC), + "a never-dispatched job (no host assignment at all) must not " + "appear"); + TEST_ASSERT_TRUE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), + "XCE_WORKER_RECONCILING must be set once a round is armed"); + + resync_resolve_round(&g_twr, &w); + TEST_ASSERT_NULL(w.resync); + TEST_ASSERT_FALSE(IS_WORKER_RECONCILING(w.host.status)); + + twr_teardown(); + printf("test_resync_arm_round_query_set_matches_exactly: PASS\n"); +} + +/* Req 1.2: a worker with zero non-terminal jobs assigned is a documented + * no-op -- no round armed, no status bit set. */ +static void test_resync_arm_round_no_jobs_is_noop(void) +{ + worker_t w; + int rc; + + twr_setup(); + + memset(&w, 0, sizeof(w)); + w.host.name = (char*)"resync-host-empty"; + + rc = resync_arm_round(&g_twr, &w); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "a worker with no non-terminal jobs must arm zero jobs (Req 1.2)"); + TEST_ASSERT_NULL_MESSAGE(w.resync, + "no resync_state_t must be attached for the Req 1.2 no-op case"); + TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), + "XCE_WORKER_RECONCILING must not be set for the Req 1.2 no-op case"); + + twr_teardown(); + printf("test_resync_arm_round_no_jobs_is_noop: PASS\n"); +} + +/* Req 6.1, 6.3: arming a round for a worker that already carries one + * (simulating a disconnect interrupting an outstanding round, then a + * rejoin) discards the prior round and produces a FRESH, DIFFERENT token -- + * and the original token is provably no longer current once superseded. */ +static void test_resync_arm_round_fresh_token_supersedes_prior(void) +{ + job_t* j; + worker_t w; + int rc; + uint64_t token1; + uint64_t token2; + + twr_setup(); + + j = twr_seed_job(TWR_JOBID_RESYNC_SOLO, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_SOLO, + "resync-host-solo"); + + memset(&w, 0, sizeof(w)); + w.host.name = (char*)"resync-host-solo"; + + rc = resync_arm_round(&g_twr, &w); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_NOT_NULL(w.resync); + token1 = w.resync->round_token; + TEST_ASSERT_TRUE_MESSAGE(resync_round_is_current(&w, token1), + "the round just armed must report its own token as current"); + + /* Simulated disconnect-then-rejoin: the round never resolved, but the + * worker rejoins and is armed again. */ + rc = resync_arm_round(&g_twr, &w); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_NOT_NULL(w.resync); + token2 = w.resync->round_token; + + TEST_ASSERT_NOT_EQUAL_MESSAGE(token1, token2, + "re-arming must produce a token distinct from the prior round's " + "(Req 6.1, 6.3)"); + TEST_ASSERT_FALSE_MESSAGE(resync_round_is_current(&w, token1), + "the ORIGINAL token must now report stale/superseded (Req 6.3) -- " + "the mechanism a later report-apply caller uses to discard a " + "report carrying it"); + TEST_ASSERT_TRUE_MESSAGE(resync_round_is_current(&w, token2), + "the NEW token must report current"); + + resync_resolve_round(&g_twr, &w); + twr_teardown(); + printf("test_resync_arm_round_fresh_token_supersedes_prior: PASS\n"); +} + +/* Req 5.1, 5.2: XCE_WORKER_RECONCILING is set once a round is armed and + * cleared once resync_resolve_round runs -- resync_round_is_current also + * correctly reports the resolved round as no longer current. */ +static void test_resync_reconciling_flag_set_after_arm_cleared_after_resolve( + void) +{ + job_t* j; + worker_t w; + int rc; + uint64_t token; + + twr_setup(); + + j = twr_seed_job(TWR_JOBID_RESYNC_SOLO, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_SOLO, + "resync-host-flag"); + + memset(&w, 0, sizeof(w)); + w.host.name = (char*)"resync-host-flag"; + TEST_ASSERT_FALSE(IS_WORKER_RECONCILING(w.host.status)); + + rc = resync_arm_round(&g_twr, &w); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_TRUE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), + "XCE_WORKER_RECONCILING must be set once armed (Req 5.1)"); + token = w.resync->round_token; + + resync_resolve_round(&g_twr, &w); + TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), + "XCE_WORKER_RECONCILING must clear once resolved (Req 5.2)"); + TEST_ASSERT_NULL_MESSAGE(w.resync, + "resync_resolve_round must detach/free w->resync"); + TEST_ASSERT_FALSE_MESSAGE(resync_round_is_current(&w, token), + "a resolved round's token must no longer report current"); + + /* Idempotent: resolving an already-resolved (NULL) round is a safe + * no-op, never a double-free. */ + resync_resolve_round(&g_twr, &w); + TEST_ASSERT_NULL(w.resync); + + twr_teardown(); + printf("test_resync_reconciling_flag_set_after_arm_cleared_after_" + "resolve: PASS\n"); +} + +/* Req 5.3: the timeout resolves an outstanding round -- and does so via a + * plain, synchronous function call with no sleep/wait of any kind, proving + * the resolution itself never blocks (ha-async-io.md's delaymgr + * thread-role contract). g_twr never starts a real delaymgr (matching + * every other bare-cluster_t fixture in this file), so resync_arm_round's + * own `!c->delaymgr.running` guard already took the no-real-timer branch + * when the round above was armed; resync_test_fire_timeout below drives + * the EXACT SAME internal resolution logic + * (resync_on_timeout) a real timer firing would have. */ +static void test_resync_timeout_resolves_round_without_blocking(void) +{ + job_t* j; + worker_t* w; + int rc; + uint64_t token; + + twr_setup(); + + j = twr_seed_job(TWR_JOBID_RESYNC_TIMEOUT, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_TIMEOUT, + "resync-host-timeout"); + + w = twr_make_test_worker("resync-host-timeout"); + TEST_ASSERT_NOT_NULL(w); + + rc = resync_arm_round(&g_twr, w); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_NOT_NULL(w->resync); + TEST_ASSERT_TRUE(IS_WORKER_RECONCILING(w->host.status)); + token = w->resync->round_token; + + /* resync_on_timeout (invoked via resync_test_fire_timeout) locks/ + * unlocks `w` itself -- this call must NOT be wrapped in an external + * lock(w, ...), or it would deadlock against `w`'s own non-recursive + * write lock. */ + resync_test_fire_timeout(&g_twr, w, token); + TEST_ASSERT_NULL_MESSAGE(w->resync, + "the timeout must resolve (free/detach) the round"); + TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status), + "the timeout must clear XCE_WORKER_RECONCILING so the worker is " + "not withheld from dispatch indefinitely (Req 5.1, 5.2, 5.3)"); + + /* A second, late fire for the SAME (now-stale) token must be a safe + * no-op -- never a double-free/double-clear against an already- + * resolved round. This is exactly the round-token staleness check + * (Req 6.3) protecting against a real delaymgr task that could not be + * dequeued/cancelled (delay.c exposes no cancel primitive) firing late + * against a resync_state_t that is already gone or has already been + * replaced by a newer round. */ + resync_test_fire_timeout(&g_twr, w, token); + TEST_ASSERT_NULL(w->resync); + TEST_ASSERT_FALSE(IS_WORKER_RECONCILING(w->host.status)); + + twr_teardown(); + printf("test_resync_timeout_resolves_round_without_blocking: PASS\n"); +} + /* ── Test group entry point ───────────────────────────────────────────── */ void test_worker_resync(void) @@ -681,4 +1063,9 @@ void test_worker_resync(void) RUN_TEST(test_reconcile_pending_rejects_terminal_job); RUN_TEST(test_reconcile_assign_applies_running_no_allocs_and_notifies); RUN_TEST(test_reconcile_assign_survives_restart_replay); + RUN_TEST(test_resync_arm_round_query_set_matches_exactly); + RUN_TEST(test_resync_arm_round_no_jobs_is_noop); + RUN_TEST(test_resync_arm_round_fresh_token_supersedes_prior); + RUN_TEST(test_resync_reconciling_flag_set_after_arm_cleared_after_resolve); + RUN_TEST(test_resync_timeout_resolves_round_without_blocking); }