lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(worker-rejoin-resync): reconciliation round lifecycle (task 3.1)

Browse source 
New src/controller/resync.c/resync.h: resync_arm_round(cluster_t*,
worker_t*) enumerates every non-terminal job assigned to a host,
populates worker_t.resync (round_token, query job-ID set, deadline),
sets XCE_WORKER_RECONCILING, and schedules a non-blocking delaymgr
one-shot timeout. resync_resolve_round() clears the round; a fired
timeout resolves it too, via resync_round_is_current()'s by-value
round-token check -- delay.c/delay.h expose no true cancel/dequeue
primitive, so a stale timer from an already-superseded round becomes a
safe no-op instead of touching a newer round's live state (verified
safe against both misresolving the wrong round and use-after-free: the
timeout closure never holds a pointer into resync_state_t itself).

5 new tests: query-set correctness, fresh-token-per-arm, the
RECONCILING flag's set/clear lifecycle, and a direct (non-blocking)
timeout-fire proof via an XCE_TESTING-gated seam confirmed compiled out
of the production binary. Full suite: 1232 tests, 0 failures.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
Rongsong Shen 2026-09-01 12:53:09 +00:00
parent ea0658ecf4
commit 442fcf4544
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
5 changed files with 927 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

439
src/controller/resync.c Normal file
View file

@ -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 <errno.h>
#include <rtti.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#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