feat(worker-rejoin-resync): task 6.2 - route job-state report to reconciliation

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
This commit is contained in:
Rongsong Shen 2026-09-01 21:06:42 +00:00
parent a3d12d37d9
commit 9f655c61c7
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
12 changed files with 1178 additions and 12 deletions

View file

@ -753,6 +753,208 @@ int resync_apply_report(cluster_t* c, worker_t* w, uint64_t round_token,
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