Rongsong Shen
9f655c61c7
Add OPCode_jobStateReport dispatch in myreq_handler, reassembling a report across multiple incoming pages before handing it to task 3.2's resync_apply_report() exactly once - proven via a test-only call counter, not just end-state equivalence (a double-apply could pass an end-state check if idempotent). Standby-redirect investigation (Req 8.1/8.3): confirmed ha_dispatch_gate already redirects every rejected opcode uniformly, so no opcode-specific code was needed for that. OPCode_jobStateQuery has no controller-side receive path at all (only sent, never received) so needed no gate change either. That same investigation found a different, more consequential gap: fixes bug-1753, ha_dispatch_gate's quorum-loss fact-admission allowlist (OPCode_notify/update/setpreason/clrpreason) was missing OPCode_jobStateReport, so a report arriving during a transient quorum-loss window - this feature's own core failover scenario - would be rejected outright with no retry (worker sends fire-and-forget). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
982 lines
41 KiB
C
982 lines
41 KiB
C
/* 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 "controller.capnp.h" /* JobStateStatus_* -- worker-rejoin-resync
|
|
* task 3.2 */
|
|
#include "delay.h"
|
|
#include "persist.h" /* persist_notify_job_updated_already_persisted --
|
|
* worker-rejoin-resync task 3.2 */
|
|
#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;
|
|
}
|
|
|
|
/* ── Report application (task 3.2, Req 4.1-4.5, 6.1-6.3) ─────────────────
|
|
*
|
|
* Maps each JobStateEntry in a worker's job-state report to the correct
|
|
* idempotent job.c fact-recording call, then resolves the round once every
|
|
* queried job has a reconciled outcome. See resync.h's doc comment on
|
|
* resync_apply_report() for the full per-outcome routing contract, and
|
|
* design.md's "System Flows" sequence diagram (the `loop each reported
|
|
* entry` block) for the design this implements.
|
|
*/
|
|
|
|
/* Req 6.2: removes `jobid` from w->resync's outstanding query set, if
|
|
* present -- swap-with-last-and-shrink (the array is small, bounded by one
|
|
* host's non-terminal job count per resync_collect_query_jobids' own
|
|
* reasoning, so O(n) removal is not a concern). A no-op when w->resync is
|
|
* NULL or `jobid` is not present -- safe to call unconditionally and
|
|
* repeatedly (an already-cleared id, or one that was never queried, e.g. an
|
|
* unqueried extra). Caller holds w's write lock. */
|
|
static void resync_clear_outstanding(worker_t* w, uint64_t jobid)
|
|
{
|
|
resync_state_t* rs = w->resync;
|
|
int i;
|
|
|
|
if (rs == NULL)
|
|
return;
|
|
|
|
for (i = 0; i < rs->n_queried_jobids; i++) {
|
|
if (rs->queried_jobids[i] == jobid) {
|
|
rs->queried_jobids[i]
|
|
= rs->queried_jobids[rs->n_queried_jobids - 1];
|
|
rs->n_queried_jobids--;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
/* Req 5.1/5.2: resolves w's round if its outstanding set has just become
|
|
* empty. Caller holds w's write lock and must have already confirmed the
|
|
* round is still current for the token in question (see
|
|
* resync_settle_entry below) -- this function itself only re-checks
|
|
* w->resync != NULL (resync_resolve_round's own precondition), not token
|
|
* identity, since the caller's own check under the SAME uninterrupted lock
|
|
* hold already establishes that. */
|
|
static void resync_maybe_resolve(cluster_t* c, worker_t* w)
|
|
{
|
|
if (w->resync != NULL && w->resync->n_queried_jobids == 0)
|
|
resync_resolve_round(c, w);
|
|
}
|
|
|
|
/* Req 4.1, 5.1, 5.2, 6.1-6.3: clears `jobid` from w's outstanding set for
|
|
* round `token` and, if the set has just become empty, resolves the round
|
|
* -- both performed under a SINGLE w write-lock acquisition, both gated on
|
|
* `token` still being w's CURRENT round. The gate matters: a concurrent Req
|
|
* 5.3 timeout fire, or a fresh resync_arm_round superseding this round on
|
|
* another thread, can happen between this report-apply loop's per-entry
|
|
* fact application (which holds no lock on `w`) and this call -- without
|
|
* re-validating here, a stale round's late bookkeeping could otherwise
|
|
* mutate a DIFFERENT (newer) round's resync_state_t. This is also the exact
|
|
* mechanism that makes re-applying an already-fully-resolved report a safe
|
|
* no-op (Req 6.2): once resync_resolve_round() has run, w->resync is NULL,
|
|
* so `token` no longer reports current and this becomes a pure no-op. */
|
|
static void resync_settle_entry(cluster_t* c, worker_t* w, uint64_t token,
|
|
uint64_t jobid)
|
|
{
|
|
lock(w, RTTI_LOCK_WRITE);
|
|
if (resync_round_is_current(w, token)) {
|
|
resync_clear_outstanding(w, jobid);
|
|
resync_maybe_resolve(c, w);
|
|
}
|
|
unlock(w);
|
|
}
|
|
|
|
/* Looks up `jobid` in c->jobmgr, reference()-ing it on success -- the same
|
|
* key_find+reference idiom on_job_start/on_job_finish/on_job_fail/
|
|
* dispatch_a_job_ (job.c) already use throughout this codebase. Returns
|
|
* NULL if not found -- never treated as an error by this file's own callers
|
|
* (see resync_apply_running_entry's doc comment for why a genuinely-
|
|
* untracked jobid is a legitimate, if narrow, possibility). Caller must
|
|
* dereference() the result when done, if non-NULL. */
|
|
static job_t* resync_lookup_job(cluster_t* c, uint64_t jobid)
|
|
{
|
|
job_t* j = NULL;
|
|
|
|
lock(c->jobmgr, RTTI_LOCK_READ);
|
|
key_find(c->jobmgr, (void*)jobid, (void**)&j);
|
|
if (j != NULL)
|
|
reference(j);
|
|
unlock(c->jobmgr);
|
|
|
|
return j;
|
|
}
|
|
|
|
/* Req 4.1, 4.4, 4.7, 4.8: routes a "running" entry. A QUERIED running job's
|
|
* own state already correctly reflects "running" -- no fact is recorded
|
|
* (design.md's sequence diagram: "else still running -> clear this jobid
|
|
* from outstanding set (no-op fact)"); the caller clears the outstanding
|
|
* set for every queried entry uniformly (resync_apply_report below), so
|
|
* this function has nothing further to do for that case.
|
|
*
|
|
* An UNQUERIED extra (Req 3.6) running job is a genuine THREE-WAY split, not
|
|
* a two-way boolean (review round 2 regression fix, following round 1's own
|
|
* review-caught fix -- the round-1 guard introduced a NEW Req 4.4 violation:
|
|
* `!terminal && assigned-to-w` short-circuits to `false` whenever the job IS
|
|
* terminal, falling through to the assign branch and unconditionally
|
|
* clobbering a DONE/EXIT job back to RUN -- confirmed live: round 1's own
|
|
* test seeded a TERMINAL j_extra and asserted it became RUN, certifying the
|
|
* bug as "correct" instead of catching it):
|
|
*
|
|
* (a) TERMINAL (resync_job_is_terminal()) -> ALWAYS a no-op, regardless of
|
|
* host. Requirement 4.4 ("shall not overwrite that state with a less
|
|
* final or older report") is unconditional -- a terminal job is never
|
|
* assign-eligible, full stop. record_job_reconcile_assign_fact() has NO
|
|
* backward-transition guard of its own (its own doc comment: "no
|
|
* reject/no-op case" is a documented CALLER precondition, never
|
|
* re-checked internally) and unconditionally sets job->job.state =
|
|
* JOB_STAT_RUN -- calling it here would clobber DONE/EXIT back to RUN.
|
|
* (b) Non-terminal AND job->job.allocs != NULL (already has SOME live
|
|
* non-terminal host assignment -- to THIS reporting worker's own host,
|
|
* or to a DIFFERENT one, it does not matter which) -> a no-op. It is
|
|
* already validly assigned; record_job_reconcile_assign_fact() would
|
|
* still be wrong to call even though its net STATE effect looks
|
|
* harmless (RUN -> RUN) -- it unconditionally resets job->job.start_time
|
|
* and re-appends a fresh WAL_JOB_DISPATCH fact, neither of which is
|
|
* warranted for a job that was never actually lost/unassigned.
|
|
* (c) Non-terminal AND job->job.allocs == NULL (genuinely UNASSIGNED) ->
|
|
* Req 4.8's real trigger condition ("no existing non-terminal record...
|
|
* assigned elsewhere" -- there IS a non-terminal record, but it carries
|
|
* no assignment at all) -- reconciled via record_job_reconcile_assign_
|
|
* fact() (task 2.2), documented (and hard-asserted, job.c) as the
|
|
* running-only case.
|
|
*
|
|
* Concrete motivating scenario for (c) (the exact case design.md's own
|
|
* "Documented scope decision" paragraph describes as this function's
|
|
* target, and record_job_reconcile_assign_fact()'s core justification for
|
|
* existing at all): a job the controller currently shows PEND (non-
|
|
* terminal, allocs == NULL, unassigned to any host) because its
|
|
* WAL_JOB_DISPATCH fact never reached durable majority commit before a
|
|
* controller failover -- even though dispatch_a_job_() already sent
|
|
* xce_startjob_req() to the worker and the worker really is running it (the
|
|
* RPC fires BEFORE dispatch_a_job_()'s own eager mutation + WAL append, so
|
|
* the two can diverge exactly at a failover boundary). Such a job correctly
|
|
* reconciles via record_job_reconcile_assign_fact() (host now matches the
|
|
* reporting worker, allocs stays NULL per Req 4.8's own documented scope
|
|
* decision).
|
|
*
|
|
* A jobid with NO record AT ALL in the job table cannot be reconciled via
|
|
* record_job_reconcile_assign_fact() either -- that function mutates an
|
|
* EXISTING job_t, it does not construct one -- so this is logged and
|
|
* skipped rather than fabricating a full job record from a bare entry;
|
|
* every job a worker can legitimately run was, by this system's own
|
|
* architecture, submitted through this controller at some point, so this
|
|
* branch is expected to be unreachable in practice, not a routine path. */
|
|
static void resync_apply_running_entry(cluster_t* c, worker_t* w,
|
|
xce_job_state_entry_t* entry)
|
|
{
|
|
job_t* job;
|
|
|
|
if (entry->queried)
|
|
return;
|
|
|
|
job = resync_lookup_job(c, entry->jobid);
|
|
if (job == NULL) {
|
|
WARN("resync: unqueried extra job %lu reported running by worker "
|
|
"%s has no record at all in the job table -- cannot reconcile "
|
|
"(no existing job_t to assign onto); skipped",
|
|
(unsigned long)entry->jobid,
|
|
(w->host.name != NULL) ? w->host.name : "?");
|
|
return;
|
|
}
|
|
|
|
lock(job, RTTI_LOCK_WRITE);
|
|
if (resync_job_is_terminal(job->job.state)) {
|
|
/* (a) Terminal is always final -- Req 4.4. Never assign-eligible,
|
|
* regardless of host. */
|
|
unlock(job);
|
|
} else if (job->job.allocs != NULL) {
|
|
/* (b) Non-terminal AND already has SOME live non-terminal host
|
|
* assignment -- a genuine no-op; do not reset start_time or
|
|
* re-append a WAL_JOB_DISPATCH fact for a job that was never
|
|
* actually lost/unassigned. */
|
|
unlock(job);
|
|
} else {
|
|
/* (c) Non-terminal AND genuinely UNASSIGNED -- Req 4.8 fires. */
|
|
int rc = record_job_reconcile_assign_fact(c, job, w, entry);
|
|
unlock(job);
|
|
if (rc == (int)JOB_TRANSITION_APPLIED)
|
|
persist_notify_job_updated_already_persisted(c, job);
|
|
}
|
|
|
|
dereference(job);
|
|
}
|
|
|
|
/* Req 4.1-4.5, 4.7: routes a "finished" entry (success or failure) through
|
|
* the SAME existing, deferred-pattern fact-recording used for an ordinary
|
|
* worker finish/fail report. Applies identically whether entry->queried is
|
|
* true or false (Req 4.7: an unqueried finished extra reconciles through
|
|
* "the same idempotent commit-then-apply path... used for queried jobs" --
|
|
* there is no separate "assign" case for a finished outcome; record_job_
|
|
* reconcile_assign_fact() is documented as running-only, see job.c's own
|
|
* doc comment on it).
|
|
*
|
|
* finishedOk: bug-1747 fix (Req 4.2) -- routed through the NEW record_job_
|
|
* finish_fact_at() (job.c), not on_job_finish()/record_job_finish_fact(),
|
|
* so the job's recorded finish_time is the WORKER-reported entry->
|
|
* finish_time_ms (Req 3.5), not reconciliation-processing wall-clock time.
|
|
* record_job_finish_fact_at() expects the caller to already hold job's
|
|
* write lock (like record_job_reconcile_pending_fact/assign_fact above),
|
|
* unlike on_job_finish() which locks internally -- this function performs
|
|
* that lookup/lock/unlock/dereference sequence itself, mirroring on_job_
|
|
* finish()'s own internal sequence exactly (job.c). A jobid not found in
|
|
* the job table is a silent no-op, mirroring on_job_finish()'s own
|
|
* "can not find job" behavior for an unknown jobid.
|
|
*
|
|
* finishedFail: still routed through the existing, UNMODIFIED on_job_fail()
|
|
* /record_job_fail_fact() -- Requirement 4.2 remains unsatisfiable for this
|
|
* outcome via any existing path in this codebase, not a reconciliation-
|
|
* specific gap: WAL_JOB_FAIL's payload (persist_submit_job_fact_ex(...,
|
|
* WAL_JOB_FAIL, jobid, 0, WAL_JOB_FAIL_STAGE_FLAG|stage)) carries no
|
|
* finish-time slot at any layer, and apply_job_fail_transition() (job.c)
|
|
* unconditionally sets job->job.finish_time = time(NULL) at APPLY time
|
|
* (not even receipt time) for EVERY fail-reporting caller in this system,
|
|
* confirmed by direct inspection -- a pre-existing, systemic gap that
|
|
* predates this feature and is out of this task's boundary to fix (see
|
|
* buglog.json bug-1747, re-scoped). The reconciled job's STATE (EXIT) is
|
|
* always correct either way; only its recorded finish_time reflects
|
|
* apply-processing time rather than the worker's true report. */
|
|
static void resync_apply_finished_entry(cluster_t* c,
|
|
xce_job_state_entry_t* entry)
|
|
{
|
|
if (entry->status == JobStateStatus_finishedOk) {
|
|
job_t* job = resync_lookup_job(c, entry->jobid);
|
|
|
|
if (job != NULL) {
|
|
xce_jobfinish_t finish
|
|
= { .jobid = entry->jobid, .exit_code = 0 };
|
|
|
|
lock(job, RTTI_LOCK_WRITE);
|
|
record_job_finish_fact_at(c, job, &finish,
|
|
entry->finish_time_ms);
|
|
unlock(job);
|
|
|
|
dereference(job);
|
|
}
|
|
} else {
|
|
job_failinfo_t fail
|
|
= { .jobid = entry->jobid,
|
|
.stage = JOB_EXEC_STAGE_MAIN,
|
|
.code = entry->exit_code,
|
|
.detail = (char*)"worker-rejoin-resync reconciliation" };
|
|
on_job_fail(c, &fail);
|
|
}
|
|
}
|
|
|
|
/* Req 4.1, 4.5, 4.6: routes an "unrecognized" entry through record_job_
|
|
* reconcile_pending_fact() (task 2.1). Per Req 3.1/3.3, a worker only ever
|
|
* reports "unrecognized" for a jobid it was explicitly asked about (an
|
|
* unqueried extra is, by construction of Req 3.6's own "currently running
|
|
* or recently finished" trigger, never reported unrecognized) -- so this
|
|
* path is queried-only in practice; the caller (resync_apply_report below)
|
|
* still clears the outstanding set uniformly for every queried entry
|
|
* regardless of which branch handled it. */
|
|
static void resync_apply_unrecognized_entry(cluster_t* c,
|
|
xce_job_state_entry_t* entry)
|
|
{
|
|
job_t* job = resync_lookup_job(c, entry->jobid);
|
|
int rc;
|
|
|
|
if (job == NULL)
|
|
return;
|
|
|
|
lock(job, RTTI_LOCK_WRITE);
|
|
rc = record_job_reconcile_pending_fact(c, job);
|
|
unlock(job);
|
|
|
|
if (rc == (int)JOB_TRANSITION_APPLIED)
|
|
persist_notify_job_updated_already_persisted(c, job);
|
|
|
|
dereference(job);
|
|
}
|
|
|
|
int resync_apply_report(cluster_t* c, worker_t* w, uint64_t round_token,
|
|
xce_job_state_report_t* report)
|
|
{
|
|
int i;
|
|
|
|
if (c == NULL || w == NULL || report == NULL)
|
|
return -EINVAL;
|
|
if (report->count > 0 && report->entries == NULL)
|
|
return -EINVAL;
|
|
|
|
/* Req 6.1, 6.3: a report for a round already superseded/discarded is
|
|
* dropped in its entirety, before anything else is touched -- checked
|
|
* under w's own write lock (matching every other w->resync mutation
|
|
* this function goes on to make, via resync_settle_entry, per entry
|
|
* below). */
|
|
lock(w, RTTI_LOCK_WRITE);
|
|
if (!resync_round_is_current(w, round_token)) {
|
|
unlock(w);
|
|
return 0;
|
|
}
|
|
unlock(w);
|
|
|
|
/* Per-entry routing never holds w's write lock while also locking a
|
|
* job's own write lock -- mirrors dispatch_a_job_'s established
|
|
* discipline (job.c: every branch releases w's lock BEFORE acquiring a
|
|
* job's) rather than this file's other three functions' "caller holds
|
|
* w's lock for the whole call" convention, which fits their simple,
|
|
* single-object, single-call-site mutations but not this function's
|
|
* fan-out into per-job job.c calls, each requiring its own job-level
|
|
* lock. w's lock is only (re)acquired, narrowly, inside
|
|
* resync_settle_entry's own outstanding-set/maybe-resolve mutation. */
|
|
for (i = 0; i < report->count; i++) {
|
|
xce_job_state_entry_t* entry = report->entries[i];
|
|
|
|
if (entry == NULL)
|
|
continue;
|
|
|
|
switch (entry->status) {
|
|
case JobStateStatus_running:
|
|
resync_apply_running_entry(c, w, entry);
|
|
break;
|
|
case JobStateStatus_finishedOk:
|
|
case JobStateStatus_finishedFail:
|
|
resync_apply_finished_entry(c, entry);
|
|
break;
|
|
case JobStateStatus_unrecognized:
|
|
default:
|
|
resync_apply_unrecognized_entry(c, entry);
|
|
break;
|
|
}
|
|
|
|
if (entry->queried)
|
|
resync_settle_entry(c, w, round_token, entry->jobid);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
/* ── Multi-page report reassembly (task 6.2, Req 8.1, 8.3) ─────────────────
|
|
*
|
|
* resync_apply_report() above expects an already-fully-assembled report --
|
|
* its own doc comment (resync.h): "multi-page wire reassembly is a later
|
|
* task's concern (6.2), out of this function's scope." The functions below
|
|
* are that later task: accumulate an incoming OPCode_jobStateReport's wire
|
|
* pages into worker_t.report_accum (xce/types.h), then call resync_apply_
|
|
* report() exactly once -- never once per page -- only once the last page
|
|
* has arrived. Mirrors xcecd_resync_handle_query_page()'s (worker-side,
|
|
* xcecd_resync.c, task 4.3) accumulate-then-apply-once shape and its
|
|
* reset-on-pos-0 discard-stale-accumulation convention, adapted from a
|
|
* single file-static accumulator (a worker daemon has exactly one
|
|
* controller connection) to a per-worker_t one (a controller serves many
|
|
* workers concurrently, so the accumulator must be keyed per reporting
|
|
* worker, guarded by that worker's own RTTI lock).
|
|
*/
|
|
|
|
#ifdef XCE_TESTING
|
|
/* Structural proof seam (task 6.2): counts actual resync_apply_report()
|
|
* invocations made BY resync_handle_report_page() below, process-wide.
|
|
* Lets a test assert "a report split across N pages results in exactly ONE
|
|
* resync_apply_report() call" directly, rather than only inferring it from
|
|
* the resulting job-table end state -- job.c's own idempotent fact guards
|
|
* would make even a buggy once-per-page implementation produce a correct-
|
|
* looking end state on a happy-path report with no round change
|
|
* interleaved between pages (the exact class of placement/count bug task
|
|
* 4.3's own round-1 review missed with a purely behavioral test -- see
|
|
* tasks.md's Implementation Notes for task 4.3). */
|
|
static _Atomic int g_resync_apply_report_call_count = 0;
|
|
|
|
int resync_apply_report_call_count_for_test(void)
|
|
{
|
|
return atomic_load_explicit(&g_resync_apply_report_call_count,
|
|
memory_order_relaxed);
|
|
}
|
|
|
|
void resync_apply_report_call_count_reset_for_test(void)
|
|
{
|
|
atomic_store_explicit(&g_resync_apply_report_call_count, 0,
|
|
memory_order_relaxed);
|
|
}
|
|
#endif
|
|
|
|
/* Frees w->report_accum (if any, casting back from the opaque void* xce/
|
|
* types.h stores it as) and NULLs it. Caller holds w's write lock.
|
|
* Idempotent -- a no-op when already NULL, safe to call unconditionally
|
|
* (mirrors resync_resolve_round's own idempotence for the sibling
|
|
* w->resync field). */
|
|
static void resync_report_accum_reset(worker_t* w)
|
|
{
|
|
resync_report_accum_t* accum = (resync_report_accum_t*)w->report_accum;
|
|
int i;
|
|
|
|
if (accum == NULL)
|
|
return;
|
|
|
|
for (i = 0; i < accum->n_total; i++)
|
|
free(accum->entries[i]);
|
|
free(accum->entries);
|
|
free(accum);
|
|
|
|
w->report_accum = NULL;
|
|
}
|
|
|
|
int resync_handle_report_page(cluster_t* c, worker_t* w,
|
|
const xce_job_state_report_t* page)
|
|
{
|
|
resync_report_accum_t* accum;
|
|
int rc;
|
|
int i;
|
|
|
|
if (c == NULL || w == NULL || page == NULL)
|
|
return -EINVAL;
|
|
|
|
if (page->pos < 0 || page->count < 0)
|
|
return -EINVAL;
|
|
|
|
lock(w, RTTI_LOCK_WRITE);
|
|
|
|
if (page->pos == 0) {
|
|
/* Fresh report starting (or restarting over a stale/incomplete
|
|
* prior accumulation -- e.g. this worker disconnected mid-report
|
|
* on an earlier round) -- always reset first, mirroring xcecd_
|
|
* resync_handle_query_page()'s identical pos==0 convention. */
|
|
resync_report_accum_reset(w);
|
|
|
|
accum = (resync_report_accum_t*)calloc(1, sizeof(*accum));
|
|
if (accum == NULL) {
|
|
unlock(w);
|
|
return -ENOMEM;
|
|
}
|
|
|
|
/* Req 6.1, 6.3: captured NOW, from whatever round is current the
|
|
* instant this report starts arriving -- not re-read later at
|
|
* apply time. See resync_report_accum_t's own doc comment
|
|
* (resync.h) for why this specific capture point is load-bearing,
|
|
* not an arbitrary choice. */
|
|
accum->round_token = (w->resync != NULL) ? w->resync->round_token : 0;
|
|
accum->n_total = page->n_total;
|
|
|
|
if (page->n_total > 0) {
|
|
accum->entries = (xce_job_state_entry_t**)calloc(
|
|
(size_t)page->n_total, sizeof(*accum->entries));
|
|
if (accum->entries == NULL) {
|
|
free(accum);
|
|
unlock(w);
|
|
return -ENOMEM;
|
|
}
|
|
}
|
|
|
|
w->report_accum = accum;
|
|
}
|
|
|
|
accum = (resync_report_accum_t*)w->report_accum;
|
|
|
|
if (page->count > 0) {
|
|
if (accum == NULL || page->entries == NULL
|
|
|| (page->pos + page->count) > accum->n_total) {
|
|
resync_report_accum_reset(w);
|
|
unlock(w);
|
|
return -EINVAL;
|
|
}
|
|
|
|
/* Deep-copy each entry (plain struct-value copy -- xce_job_state_
|
|
* entry_t has no nested heap pointers): page->entries[i] is owned
|
|
* by THIS page's own decoded xce_request_t and is freed by xce_
|
|
* free_request() the moment the dispatch handler that decoded it
|
|
* returns, long before a later page (let alone the final one) has
|
|
* even arrived -- a borrowed pointer would be a use-after-free the
|
|
* instant this function returns 0 awaiting more pages. */
|
|
for (i = 0; i < page->count; i++) {
|
|
xce_job_state_entry_t* src = page->entries[i];
|
|
xce_job_state_entry_t* copy;
|
|
|
|
if (src == NULL) {
|
|
resync_report_accum_reset(w);
|
|
unlock(w);
|
|
return -EINVAL;
|
|
}
|
|
|
|
copy = (xce_job_state_entry_t*)malloc(sizeof(*copy));
|
|
if (copy == NULL) {
|
|
resync_report_accum_reset(w);
|
|
unlock(w);
|
|
return -ENOMEM;
|
|
}
|
|
*copy = *src;
|
|
|
|
accum->entries[page->pos + i] = copy;
|
|
}
|
|
|
|
accum->n_received += page->count;
|
|
} else if (accum == NULL) {
|
|
/* An orphaned continuation page (pos != 0) with no accumulation in
|
|
* progress at all (its own preceding pos==0 page never arrived, or
|
|
* arrived and was already discarded/completed) -- nothing to
|
|
* accumulate into and nothing that will ever complete; reject
|
|
* rather than silently waiting forever. */
|
|
unlock(w);
|
|
return -EINVAL;
|
|
}
|
|
|
|
if (accum->n_received < accum->n_total) {
|
|
/* More pages still expected -- gate stays set (via w->report_accum
|
|
* remaining non-NULL), nothing applied yet. */
|
|
unlock(w);
|
|
return 0;
|
|
}
|
|
|
|
/* Last page: the report is fully assembled. Detach ownership from `w`
|
|
* and release w's lock BEFORE calling resync_apply_report() -- that
|
|
* function manages its own w write-locking internally (resync.h's own
|
|
* doc comment) and must never be called with w's lock already held, or
|
|
* its own lock(w, RTTI_LOCK_WRITE) call would deadlock against this
|
|
* one. */
|
|
w->report_accum = NULL;
|
|
unlock(w);
|
|
|
|
{
|
|
xce_job_state_report_t report;
|
|
|
|
memset(&report, 0, sizeof(report));
|
|
report.n_total = accum->n_total;
|
|
report.pos = 0;
|
|
report.count = accum->n_total;
|
|
report.entries = accum->entries;
|
|
|
|
#ifdef XCE_TESTING
|
|
atomic_fetch_add_explicit(&g_resync_apply_report_call_count, 1,
|
|
memory_order_relaxed);
|
|
#endif
|
|
rc = resync_apply_report(c, w, accum->round_token, &report);
|
|
}
|
|
|
|
for (i = 0; i < accum->n_total; i++)
|
|
free(accum->entries[i]);
|
|
free(accum->entries);
|
|
free(accum);
|
|
|
|
return (rc < 0) ? rc : 1;
|
|
}
|
|
|
|
#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
|