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

feat(worker-rejoin-resync): task 4.2 - worker answers job-state query with extras

Browse source 
Classify each queried job ID as running/finished-ok/finished-fail/
unrecognized (Req 3.1-3.3, 3.6), report unqueried running/recently-
finished jobs as extras, and send both via task 1.4's paginated wire
helper (Req 8.2, 8.3). Composes correctly with the already-implemented
controller-side consumer (resync_apply_report).

Adds xcecd_finished_job_retention_snapshot() to xcecd_worker_utils, a
small additive companion needed to enumerate all retained jobs for the
extras set (task 4.1 only exposed single-jobid lookup).

Folds the still-open OPCode_jobStateQuery receive-dispatch wiring gap
into task 4.3's scope in tasks.md, since 4.3 already needs the real
receive path to implement its post-join gate.

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 16:35:28 +00:00
parent e31012e8a7
commit cfe58c3dcf
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
7 changed files with 1084 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

11
.kiro/specs/worker-rejoin-resync/tasks.md
View file

@ -87,7 +87,7 @@
- _Requirements: 3.4, 3.5, 3.7_
- _Boundary: worker retention_
- [ ] 4.2 Answer the controller's job-state query, including extra unqueried jobs
- [x] 4.2 Answer the controller's job-state query, including extra unqueried jobs
- For every job ID in the controller's query, report exactly one outcome: still running (from the live running list), finished successfully or with failure and its true finish time (from the new retention), or unrecognized if found in neither — every queried ID gets an answer, none are silently skipped, and an indeterminate case is reported as unrecognized rather than guessed
- After answering the query, also report any job the worker is currently running or has recently finished that was not part of the query
- Send both the query answers and the extra jobs in paginated reports, correct regardless of how many jobs are involved
@ -95,12 +95,14 @@
- _Requirements: 3.1, 3.2, 3.3, 3.6, 8.2, 8.3_
- _Boundary: xcecd_resync.c_
- [ ] 4.3 Delay resuming normal post-join operation until the report has fully sent
- [ ] 4.3 Wire the worker's receipt of a job-state query and delay resuming normal post-join operation until the report has fully sent
- Also close a gap task 4.2 correctly left out of its own boundary: nothing today wires the worker's actual receipt of `OPCode_jobStateQuery` (`src/worker/src/controller_connector.c`'s `myreq_handler` switch has no case for it, confirmed by inspection — only `OPCode_startjob`/`OPCode_runtasks`/`OPCode_jobcontrol`/`OPCode_fetchfile` exist there today) to task 4.2's `xcecd_resync_answer_query()`. Add that dispatch case: reassemble an incoming query that itself arrives across multiple pages (Requirement 8.1/8.3) into the jobids array `xcecd_resync_answer_query()` expects, then call it
- Set a local gate the moment a job-state query is received; keep it set across every page of the report; clear it only once the full report, including any extra-job pages, has finished sending
- Make whatever the worker does next after rejoining (resuming its own periodic background reporting) check this gate first, so that activity never begins or interleaves while the report is still being sent
- Observable completion: the worker's post-join activity is observed to start only after the last report page has been sent, never before, across a report spanning multiple pages
- _Requirements: 3.8_
- Observable completion: the worker's post-join activity is observed to start only after the last report page has been sent, never before, across a report spanning multiple pages; a job-state query arriving from the controller (single- or multi-page) reaches `xcecd_resync_answer_query()` and produces a real reply, not just a directly-called-in-tests function
- _Requirements: 3.8, 8.1, 8.3_
- _Boundary: xcecd_resync.c_
- _Depends: 4.2_
- [ ] 5. (P) Core: exclude a reconciling host from scheduler dispatch candidacy
- Extend the scheduler's existing candidate-host filter so a host currently undergoing job-state reconciliation is excluded from dispatch candidacy the same way an already-unavailable host is
@ -160,3 +162,4 @@
- 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.
- **Task 3.2 needed 3 review rounds — a genuine cautionary tale about a boolean short-circuit hiding a second, adjacent bug.** Round 1 caught the unqueried-extra-"running" branch gating on terminal-vs-non-terminal instead of Req 4.8's real trigger (assignment status), silently leaving a job stuck PEND when its `WAL_JOB_DISPATCH` fact never reached durable commit before a failover — the feature's own core motivating scenario. The FIRST fix (`!terminal && assigned_to_w's_host`) solved that but introduced a NEW, distinct Req 4.4 violation via the same `&&`'s short-circuit: a TERMINAL job (DONE/EXIT) falls into the same `else`/assign branch and gets clobbered back to RUN — caught by round 2, and confirmed the implementer's own new test had been asserting the bug as correct behavior (a green suite is not proof of correctness when the test itself encodes the wrong expectation). The SECOND fix (a genuine three-way `if`/`else if`/`else`: terminal→noop, non-terminal-and-already-assigned-anywhere→noop, non-terminal-and-unassigned→assign) passed round 3's from-scratch re-derivation plus an explicit hunt for job==NULL reachability. **Lesson for future tasks in this codebase**: when a guard combines two orthogonal conditions with `&&`/`||` and only ONE of the two is corrected after a review finding, re-derive the FULL truth table before re-submitting — a targeted patch to the specific counterexample a reviewer names can leave an adjacent, differently-shaped counterexample in the same boolean untouched. bug-1747 (finish-time gap, split fixed) and bug-1748 (this guard defect) both logged.
- Task 4.1: found and fixed a real-but-non-blocking gap the implementer's own doc comment claimed didn't exist: `finished_job_retention_prune()` was only invoked from `xcecd_finished_job_lookup()`, never from `xcecd_retain_finished_job()`, so with `xcecd_finished_job_lookup()` having zero production callers today (task 4.2 not yet implemented) the retention list had no active pruning driver and would grow one entry per finished job for the daemon's entire uptime — even after 4.2 lands, pruning would only fire on this worker's own rejoin, so a long-uptime well-connected worker could still accumulate substantial entries between rejoins. Fixed by also pruning on retain (using `finish_time_ms`, the value already captured at genuine exit-detection time, as the prune's `now_ms`). Also confirms task 4.1 has **no runtime delivery path yet for `workerResyncWindowMs`** — `xcecd_finished_job_retention_set_window_ms()` is a real, fully-wired seam but nothing calls it in production; task 6.1's scope was already expanded (see its entry above) to plug this in via `worker_push_config`'s existing `WorkerConfig` struct.
- Task 4.2: implemented `xcecd_resync_classify_job`/`xcecd_resync_build_report`/`xcecd_resync_answer_query` (classify RUNNING/FINISHED_OK/FINISHED_FAIL/UNRECOGNIZED, compute extras, paginate via task 1.4's `xce_job_state_report_req()`) as directly-callable, fully-tested functions (20/20 tests) — necessarily added one small, purely-additive companion function to task 4.1's API, `xcecd_finished_job_retention_snapshot()`, since 4.1 only exposed single-jobid lookup and 4.2 needs to enumerate ALL retained jobs for the extras set. Confirmed by inspection that nothing wires the worker's actual receipt of `OPCode_jobStateQuery` to call into this code yet (`controller_connector.c`'s `myreq_handler` has no case for it) — folded that wiring explicitly into task 4.3's scope (see its entry above), since 4.3 already needs the real receive path to implement its post-join gate correctly and is the same file/boundary.

96
src/worker/include/xcecd_resync.h Normal file
View file

@ -0,0 +1,96 @@
#ifndef XCECD_RESYNC_H
#define XCECD_RESYNC_H
#include <stdint.h>
#include "netio.h"
#include "request.h"
#include "xcecd_worker_utils.h"
/*
* worker-rejoin-resync task 4.2 (Requirements 3.1, 3.2, 3.3, 3.6, 8.2, 8.3):
* answer the controller's job-state query from the worker's own live
* running list (xcecd_worker_t, xcecd_worker_utils.c) and the new
* finished-job retention (xcecd_worker_utils.c, task 4.1), then report any
* additional job the worker is currently running or has recently finished
* that was NOT part of the query (Requirement 3.6's "extras").
*
* Boundary/scope: this module implements the pure classify/build/send
* logic only.
* - It does NOT wire OPCode_jobStateQuery's receipt into the worker's
* dispatch loop (controller_connector.c's myreq_handler switch) --
* that opcode-dispatch wiring is not claimed by any task in
* tasks.md's current text (task 6.1 is controller-side send +
* config-window delivery; task 6.2 is controller-side report
* ingestion). Whatever task performs that wiring must also
* reassemble a query that itself arrives across multiple incoming
* pages (Req 8.1) into one flat jobids array before calling
* xcecd_resync_answer_query() below -- this module takes an
* already-assembled list, it does not itself track query pages.
* - It does NOT implement the Requirement 3.8 "resync_pending" gate
* (task 4.3's boundary, same file) that delays the worker's own
* post-join operation until the report has fully sent.
*/
/* Classify one job ID against the worker's live running list, falling
* back to the finished-job retention, per Requirements 3.1/3.3:
* JobStateStatus_running (found on the live list with worker_status ==
* WORKER_RUNNING -- i.e. not yet exited), JobStateStatus_finishedOk/
* finishedFail (found in retention, per its exit_status), or
* JobStateStatus_unrecognized (found in neither -- an indeterminate case
* is reported as unrecognized, never guessed, matching Requirement 3.3
* verbatim). `out` is fully populated except `queried`, which the caller
* sets afterward (1 for a queried job ID, 0 for an extra) -- this
* function only classifies, it does not know which set its caller is
* building. now_ms is the caller-captured wall-clock time, passed
* through verbatim to xcecd_finished_job_lookup(). */
void xcecd_resync_classify_job(xcecd_worker_t* worker_list, uint64_t jobid,
uint64_t now_ms, xce_job_state_entry_t* out);
/*
* Build the full set of report entries for a job-state query: one
* classified entry per ID in `jobids`, in the same order given, with
* queried=1 (Requirement 3.2 -- every queried ID gets exactly one
* answer, none silently skipped), followed by one entry with queried=0
* for every job the worker is currently running or has a live retained
* finished outcome for that was NOT present in `jobids` (Requirement
* 3.6).
*
* On success (0), *out_entries is a heap-allocated array of
* *out_nentries heap-allocated xce_job_state_entry_t* pointers --
* borrow-compatible with xce_job_state_report_t.entries, the exact shape
* xce_job_state_report_req() (task 1.4) expects. A query with no jobids
* and no extras yields *out_entries == NULL, *out_nentries == 0 (a valid,
* empty result -- xcecd_resync_answer_query() still sends one empty
* report page for it, matching xce_job_state_report_req()'s own
* "an empty page is still a complete exchange" contract).
*
* Free the result with xcecd_resync_free_entries(). Returns -EINVAL if
* out_entries/out_nentries is NULL, -ENOMEM on allocation failure (with
* all partial state already released).
*/
int xcecd_resync_build_report(xcecd_worker_t* worker_list,
const uint64_t* jobids, int njobids,
uint64_t now_ms,
xce_job_state_entry_t*** out_entries,
int* out_nentries);
/* Frees an entries array produced by xcecd_resync_build_report(). Safe to
* call with entries == NULL / nentries == 0. */
void xcecd_resync_free_entries(xce_job_state_entry_t** entries, int nentries);
/*
* Answer an already-reassembled job-state query end-to-end: classify
* every queried ID, append the extras (xcecd_resync_build_report()
* above), and send the whole result to the controller as one or more
* paginated OPCode_jobStateReport messages via xce_job_state_report_req()
* (task 1.4's wire helper -- Requirements 8.2, 8.3). `pktlimit` is passed
* straight through to xce_job_state_report_req() (<=0 sends everything in
* a single page). Returns 0 on success, or the first negative rc either
* xcecd_resync_build_report() or xce_job_state_report_req() returns.
*/
int xcecd_resync_answer_query(xcecd_worker_t* worker_list, endpoint_t* ep,
xce_peerinfo_t* from, const uint64_t* jobids,
int njobids, uint64_t now_ms, int pktlimit);
#endif /* XCECD_RESYNC_H */

18
src/worker/include/xcecd_worker_utils.h
View file

@ -123,6 +123,24 @@ void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms);
/* Returns the currently effective retention window in milliseconds. */
uint64_t xcecd_finished_job_retention_get_window_ms(void);
/* worker-rejoin-resync task 4.2 (Requirement 3.6): copies every currently
* retained, non-expired (as of now_ms) finished-job entry into a freshly
* heap-allocated array of xcecd_finished_job_t VALUES (not linked -- each
* copy's `next` is zeroed), so a caller outside this file (xcecd_resync.c)
* can find which recently-finished jobs are not part of a controller
* query and report them as extras. Task 4.1's own public API only
* supported single-jobid lookup (xcecd_finished_job_lookup below), which
* cannot answer "what is currently retained" -- this is the minimal
* additive companion this task needs; it does not change any existing
* function's behavior. Lazily prunes expired entries first, exactly like
* xcecd_finished_job_lookup(). On success (0) *out is the array and
* *out_n its length (0 with *out left NULL when nothing is retained);
* caller frees *out with free(). Returns -EINVAL if out/out_n is NULL,
* -ENOMEM on allocation failure. */
int xcecd_finished_job_retention_snapshot(uint64_t now_ms,
xcecd_finished_job_t** out,
int* out_n);
/* Test-only: discard all retained entries and reset the window to the
* compiled-in default, so tests don't leak state across TEST_* functions. */
void xcecd_finished_job_retention_reset(void);

218
src/worker/src/xcecd_resync.c Normal file
View file

@ -0,0 +1,218 @@
#include "xcecd_resync.h"
#include <errno.h>
#include <stdlib.h>
#include <string.h>
/* worker-rejoin-resync task 4.2 (Requirements 3.1, 3.2, 3.3, 3.6, 8.2, 8.3).
* See xcecd_resync.h for the public API and its boundary/scope note. */
static bool jobid_in_array(const uint64_t* arr, int n, uint64_t jobid)
{
int i;
for (i = 0; i < n; i++) {
if (arr[i] == jobid) {
return true;
}
}
return false;
}
void xcecd_resync_classify_job(xcecd_worker_t* worker_list, uint64_t jobid,
uint64_t now_ms, xce_job_state_entry_t* out)
{
xcecd_worker_t* w;
xcecd_finished_job_t fin;
memset(out, 0, sizeof(*out));
out->jobid = jobid;
/* Requirement 3.1/3.3: check the live running list first. */
w = find_worker_by_jobid(worker_list, jobid);
if (w != NULL && w->worker_status == WORKER_RUNNING) {
out->status = JobStateStatus_running;
return;
}
/* Not (still) running -- fall back to the bounded finished-job
* retention (task 4.1). exit_status is WORKER_DONE (success) or
* WORKER_EXITED (failure), per xcecd_worker_utils.c's own
* update_worker_status() hook -- translated 1:1, no new judgment
* call. finish_time_ms is the worker-reported real finish time
* (Requirement 3.5), not the time this function runs. */
memset(&fin, 0, sizeof(fin));
if (xcecd_finished_job_lookup(jobid, now_ms, &fin)) {
out->status = (fin.exit_status == WORKER_DONE)
? JobStateStatus_finishedOk
: JobStateStatus_finishedFail;
out->exit_code = (int32_t)fin.exit_code;
out->finish_time_ms = fin.finish_time_ms;
return;
}
/* Found in neither -- an indeterminate/unknown case is reported as
* unrecognized rather than guessed (Requirement 3.3). */
out->status = JobStateStatus_unrecognized;
}
int xcecd_resync_build_report(xcecd_worker_t* worker_list,
const uint64_t* jobids, int njobids,
uint64_t now_ms,
xce_job_state_entry_t*** out_entries,
int* out_nentries)
{
xce_job_state_entry_t* storage = NULL;
xce_job_state_entry_t** ptrs = NULL;
xcecd_worker_t* w;
xcecd_finished_job_t* snapshot = NULL;
int nsnapshot = 0;
int nrunning_extra = 0;
int nfinished_extra = 0;
int total;
int idx;
int i;
int rc;
if (out_entries == NULL || out_nentries == NULL) {
return -EINVAL;
}
*out_entries = NULL;
*out_nentries = 0;
if (njobids < 0) {
njobids = 0;
}
/* Snapshot retention once up front (task 4.2 addition to
* xcecd_worker_utils.c -- see its header comment) so the counting
* pass and the building pass below see the same set. */
rc = xcecd_finished_job_retention_snapshot(now_ms, &snapshot, &nsnapshot);
if (rc != 0) {
return rc;
}
/* Count extras first so the entries array can be sized exactly once. */
for (w = worker_list; w != NULL; w = w->next) {
if (w->worker_status == WORKER_RUNNING
&& !jobid_in_array(jobids, njobids, w->job_info.jobid)) {
nrunning_extra++;
}
}
for (i = 0; i < nsnapshot; i++) {
if (!jobid_in_array(jobids, njobids, snapshot[i].jobid)) {
nfinished_extra++;
}
}
total = njobids + nrunning_extra + nfinished_extra;
if (total == 0) {
free(snapshot);
return 0;
}
storage = (xce_job_state_entry_t*)calloc((size_t)total, sizeof(*storage));
ptrs = (xce_job_state_entry_t**)calloc((size_t)total, sizeof(*ptrs));
if (storage == NULL || ptrs == NULL) {
free(storage);
free(ptrs);
free(snapshot);
return -ENOMEM;
}
idx = 0;
/* Requirement 3.2: every queried ID gets exactly one entry, in the
* order given -- none silently skipped. */
for (i = 0; i < njobids; i++) {
xcecd_resync_classify_job(worker_list, jobids[i], now_ms,
&storage[idx]);
storage[idx].queried = 1;
ptrs[idx] = &storage[idx];
idx++;
}
/* Requirement 3.6: extras -- jobs the worker is currently running
* that were not part of the query. */
for (w = worker_list; w != NULL; w = w->next) {
if (w->worker_status == WORKER_RUNNING
&& !jobid_in_array(jobids, njobids, w->job_info.jobid)) {
storage[idx].jobid = w->job_info.jobid;
storage[idx].status = JobStateStatus_running;
storage[idx].queried = 0;
ptrs[idx] = &storage[idx];
idx++;
}
}
/* Requirement 3.6: extras -- jobs the worker has recently finished
* that were not part of the query. */
for (i = 0; i < nsnapshot; i++) {
if (!jobid_in_array(jobids, njobids, snapshot[i].jobid)) {
storage[idx].jobid = snapshot[i].jobid;
storage[idx].status = (snapshot[i].exit_status == WORKER_DONE)
? JobStateStatus_finishedOk
: JobStateStatus_finishedFail;
storage[idx].exit_code = (int32_t)snapshot[i].exit_code;
storage[idx].finish_time_ms = snapshot[i].finish_time_ms;
storage[idx].queried = 0;
ptrs[idx] = &storage[idx];
idx++;
}
}
free(snapshot);
*out_entries = ptrs;
*out_nentries = total;
return 0;
}
void xcecd_resync_free_entries(xce_job_state_entry_t** entries, int nentries)
{
if (entries == NULL) {
return;
}
/* `storage` in xcecd_resync_build_report() is one contiguous
* allocation -- entries[0] points at storage[0], so freeing it frees
* the whole block. */
if (nentries > 0) {
free(entries[0]);
}
free(entries);
}
int xcecd_resync_answer_query(xcecd_worker_t* worker_list, endpoint_t* ep,
xce_peerinfo_t* from, const uint64_t* jobids,
int njobids, uint64_t now_ms, int pktlimit)
{
xce_job_state_entry_t** entries = NULL;
int nentries = 0;
xce_job_state_report_t report;
int rc;
rc = xcecd_resync_build_report(worker_list, jobids, njobids, now_ms,
&entries, &nentries);
if (rc != 0) {
return rc;
}
memset(&report, 0, sizeof(report));
report.n_total = nentries;
report.pos = 0;
report.count = nentries;
report.entries = entries;
rc = xce_job_state_report_req(ep, from, &report, NULL, NULL, pktlimit);
xcecd_resync_free_entries(entries, nentries);
return rc;
}

43
src/worker/src/xcecd_worker_utils.c
View file

@ -582,6 +582,49 @@ uint64_t xcecd_finished_job_retention_get_window_ms(void)
return g_finished_job_retention_window_ms;
}
int xcecd_finished_job_retention_snapshot(uint64_t now_ms,
xcecd_finished_job_t** out,
int* out_n)
{
xcecd_finished_job_t* e;
xcecd_finished_job_t* copy;
int n = 0;
if (out == NULL || out_n == NULL) {
return -EINVAL;
}
*out = NULL;
*out_n = 0;
finished_job_retention_prune(now_ms, g_finished_job_retention_window_ms);
for (e = g_finished_job_retention; e != NULL; e = e->next) {
n++;
}
if (n == 0) {
return 0;
}
copy = (xcecd_finished_job_t*)calloc((size_t)n, sizeof(*copy));
if (copy == NULL) {
return -ENOMEM;
}
n = 0;
for (e = g_finished_job_retention; e != NULL; e = e->next) {
copy[n] = *e;
copy[n].next = NULL;
n++;
}
*out = copy;
*out_n = n;
return 0;
}
void xcecd_finished_job_retention_reset(void)
{
xcecd_finished_job_t* e = g_finished_job_retention;

16
src/worker/testsuite/meson.build
View file

@ -107,3 +107,19 @@ test('worker finished_job_retention test', finished_job_retention_test,
suite: 'worker',
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])
# worker-rejoin-resync task 4.2 (Requirements 3.1, 3.2, 3.3, 3.6, 8.2, 8.3):
# answer the controller's job-state query (running list + task 4.1's
# finished-job retention), including extras -- white-box #include of both
# ../src/xcecd_worker_utils.c and ../src/xcecd_resync.c, same convention
# as finished_job_retention_test above.
xcecd_resync_test = executable('xcecd_resync_test',
'test_xcecd_resync.c',
include_directories: [worker_test_incdir],
dependencies: worker_test_deps,
install: false)
test('worker xcecd_resync test', xcecd_resync_test,
suite: 'worker',
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])

686
src/worker/testsuite/test_xcecd_resync.c Normal file
View file

@ -0,0 +1,686 @@
/*
* test_xcecd_resync.c -- worker-rejoin-resync task 4.2 (Requirements 3.1,
* 3.2, 3.3, 3.6, 8.2, 8.3): answer the controller's job-state query from
* the worker's own live running list and the new finished-job retention
* (task 4.1), including extra jobs the worker knows about that were not
* part of the query, sent as one or more paginated OPCode_jobStateReport
* messages.
*
* White-box #include of both ../src/xcecd_worker_utils.c (task 4.1's
* retention + running-list lookups that xcecd_resync.c's classify logic
* calls into for real, not against reimplemented stand-ins) and
* ../src/xcecd_resync.c (this task's own production code under test) --
* same convention as test_finished_job_retention.c and
* test_controller_connector.c. Link-satisfying stubs for
* xcecd_worker_utils.c's OTHER external dependencies are copied verbatim
* from test_finished_job_retention.c's own header comment/stub set; none
* of them are exercised by the assertions below.
*
* Wire-level assertions (paginated OPCode_jobStateReport sends) use
* test_job_state_wire.c's own no-live-socket technique: a real (heap)
* endpoint_t via endpoint_get() with ep->iomgr == NULL, so
* xce_job_state_report_req() can be exercised end-to-end -- real
* xce_encode() on send, real xce_decode() on "receive" -- without a
* socket. Each queued page is popped with endpoint_deq()/ITPTR_TO_PACKET()
* and decoded with xce_decode(), which takes ownership of (frees) the
* packet.
*/
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "../src/xcecd_resync.c"
#include "../src/xcecd_worker_utils.c"
/* ════════════════════════════════════════════════════════════════════════
* Link-satisfying stubs -- none exercised by the assertions below; they
* only need to let xcecd_worker_utils.c's other, unrelated code paths
* link and (where reached at all) run to completion without touching
* real sqlite3/dbus/IPC. Copied from test_finished_job_retention.c.
* ════════════════════════════════════════════════════════════════════════ */
bool xcecd_reply_startjob(xcecd_job_t* job)
{
(void)job;
return true;
}
int xcecd_job_stopped(uint64_t job_id)
{
(void)job_id;
return 0;
}
int xcecd_job_resumed(uint64_t job_id)
{
(void)job_id;
return 0;
}
int xcecd_job_finished(xcecd_job_t* job)
{
(void)job;
return 0;
}
bool xcecd_db_update_worker(sqlite3* db, int worker_pid, int worker_status,
int job_status)
{
(void)db;
(void)worker_pid;
(void)worker_status;
(void)job_status;
return true;
}
bool xcecd_db_delete_worker(sqlite3* db, int worker_pid)
{
(void)db;
(void)worker_pid;
return true;
}
bool dbus_worker_clean(xcecd_worker_t* worker, xcecd_job_t* job)
{
(void)worker;
(void)job;
return true;
}
bool ipc_master_check_connection() { return true; }
void block_signal(int sig) { (void)sig; }
void unblock_signal(int sig) { (void)sig; }
/* ════════════════════════════════════════════════════════════════════════
* Fixtures
* ════════════════════════════════════════════════════════════════════════ */
static xcecd_worker_t* make_worker(int pid, uint64_t jobid, int worker_status)
{
xcecd_worker_t* w = (xcecd_worker_t*)calloc(1, sizeof(*w));
w->worker_pid = pid;
w->worker_status = worker_status;
w->job_status
= (worker_status == WORKER_RUNNING) ? JOB_RUNNING : JOB_EXITED;
w->job_info.jobid = jobid;
return w;
}
static void free_worker_list(xcecd_worker_t* list)
{
xcecd_worker_t* w = list;
while (w) {
xcecd_worker_t* next = w->next;
free(w);
w = next;
}
}
static void link_worker(xcecd_worker_t** list, xcecd_worker_t* w)
{
if (*list == NULL) {
*list = w;
return;
}
xcecd_worker_t* cur = *list;
while (cur->next) {
cur = cur->next;
}
cur->next = w;
w->prev = cur;
}
static endpoint_t* new_test_endpoint(void)
{
endpoint_t* ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
return ep;
}
static void pop_and_decode(endpoint_t* ep, xce_request_t* out)
{
data_item_t* it = endpoint_deq(ep);
TEST_ASSERT_NOT_NULL(it);
io_packet_t* pkt = ITPTR_TO_PACKET(it);
memset(out, 0, sizeof(*out));
TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, out));
}
static void assert_queue_empty(endpoint_t* ep)
{
TEST_ASSERT_NULL(endpoint_deq(ep));
}
void setUp(void) { xcecd_finished_job_retention_reset(); }
void tearDown(void) { xcecd_finished_job_retention_reset(); }
/* ════════════════════════════════════════════════════════════════════════
* xcecd_resync_classify_job: per-ID classification (Requirements 3.1, 3.3)
* ════════════════════════════════════════════════════════════════════════ */
static void test_classify_running_job(void)
{
xcecd_worker_t* w = make_worker(1, 111, WORKER_RUNNING);
xce_job_state_entry_t out;
xcecd_resync_classify_job(w, 111, 5000, &out);
TEST_ASSERT_EQUAL_UINT64(111, out.jobid);
TEST_ASSERT_EQUAL_INT(JobStateStatus_running, out.status);
free(w);
}
static void test_classify_finished_ok_job(void)
{
xcecd_retain_finished_job(222, WORKER_DONE, 0, 90000);
xce_job_state_entry_t out;
xcecd_resync_classify_job(NULL, 222, 95000, &out);
TEST_ASSERT_EQUAL_UINT64(222, out.jobid);
TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedOk, out.status);
TEST_ASSERT_EQUAL_UINT64(90000, out.finish_time_ms);
}
static void test_classify_finished_fail_job(void)
{
xcecd_retain_finished_job(333, WORKER_EXITED, 137, 90500);
xce_job_state_entry_t out;
xcecd_resync_classify_job(NULL, 333, 95000, &out);
TEST_ASSERT_EQUAL_UINT64(333, out.jobid);
TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedFail, out.status);
TEST_ASSERT_EQUAL_INT32(137, out.exit_code);
TEST_ASSERT_EQUAL_UINT64(90500, out.finish_time_ms);
}
static void test_classify_unrecognized_job(void)
{
xce_job_state_entry_t out;
xcecd_resync_classify_job(NULL, 444, 95000, &out);
TEST_ASSERT_EQUAL_UINT64(444, out.jobid);
TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, out.status);
}
/* A job present in the worker list but already terminal (worker_status
* WORKER_DONE/EXITED, lingering before cleanup_worker_list() reaps it)
* must NOT be reported as running -- it must fall through to retention
* (populated by the same update_worker_status() hook at the same
* transition, task 4.1), matching design.md's "check the running list
* first; if not running, check retention" order exactly. */
static void test_classify_terminal_worker_entry_falls_through_to_retention(void)
{
xcecd_worker_t* w = make_worker(5, 555, WORKER_DONE);
xcecd_retain_finished_job(555, WORKER_DONE, 0, 80000);
xce_job_state_entry_t out;
xcecd_resync_classify_job(w, 555, 85000, &out);
TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedOk, out.status);
free(w);
}
/* Requirement 3.3: a job whose process state is inconclusive (here,
* WORKER_UNKNOWN -- the one worker_status update_worker_status() never
* feeds into retention) must be reported unrecognized, not guessed. */
static void test_classify_worker_unknown_status_is_unrecognized(void)
{
xcecd_worker_t* w = make_worker(6, 666, WORKER_UNKNOWN);
xce_job_state_entry_t out;
xcecd_resync_classify_job(w, 666, 85000, &out);
TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, out.status);
free(w);
}
/* Expiry boundary: a retained job older than the configured window must
* classify as unrecognized, not finished (task 4.1's own boundary
* semantics, consumed here through the classify layer). */
static void test_classify_expired_retention_is_unrecognized(void)
{
xcecd_finished_job_retention_set_window_ms(1000);
xcecd_retain_finished_job(777, WORKER_DONE, 0, 100000);
xce_job_state_entry_t out;
xcecd_resync_classify_job(NULL, 777, 100000 + 1000 + 1, &out);
TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, out.status);
}
/* ════════════════════════════════════════════════════════════════════════
* xcecd_resync_build_report: query answers + extras (Requirements 3.2, 3.6)
* ════════════════════════════════════════════════════════════════════════ */
static xce_job_state_entry_t* find_entry(xce_job_state_entry_t** entries, int n,
uint64_t jobid)
{
int i;
for (i = 0; i < n; i++) {
if (entries[i]->jobid == jobid) {
return entries[i];
}
}
return NULL;
}
/* Every queried ID gets exactly one entry, none silently skipped, mixing
* all four outcome kinds (Requirement 3.2, and the task's own Observable
* Completion: "a query covering a mix of running, finished, and unknown
* job IDs produces one correctly-classified entry per ID"). */
static void test_build_report_answers_every_queried_id_exactly_once(void)
{
xcecd_worker_t* list = NULL;
link_worker(&list, make_worker(1, 1001, WORKER_RUNNING));
xcecd_retain_finished_job(1002, WORKER_DONE, 0, 50000);
xcecd_retain_finished_job(1003, WORKER_EXITED, 9, 50500);
/* 1004 is queried but the worker has no record of it at all. */
uint64_t jobids[4] = { 1001, 1002, 1003, 1004 };
xce_job_state_entry_t** entries = NULL;
int n = 0;
int rc = xcecd_resync_build_report(list, jobids, 4, 60000, &entries, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(4, n);
xce_job_state_entry_t* e1001 = find_entry(entries, n, 1001);
xce_job_state_entry_t* e1002 = find_entry(entries, n, 1002);
xce_job_state_entry_t* e1003 = find_entry(entries, n, 1003);
xce_job_state_entry_t* e1004 = find_entry(entries, n, 1004);
TEST_ASSERT_NOT_NULL(e1001);
TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e1001->status);
TEST_ASSERT_EQUAL_INT(1, e1001->queried);
TEST_ASSERT_NOT_NULL(e1002);
TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedOk, e1002->status);
TEST_ASSERT_EQUAL_UINT64(50000, e1002->finish_time_ms);
TEST_ASSERT_EQUAL_INT(1, e1002->queried);
TEST_ASSERT_NOT_NULL(e1003);
TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedFail, e1003->status);
TEST_ASSERT_EQUAL_INT32(9, e1003->exit_code);
TEST_ASSERT_EQUAL_INT(1, e1003->queried);
TEST_ASSERT_NOT_NULL(e1004);
TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, e1004->status);
TEST_ASSERT_EQUAL_INT(1, e1004->queried);
xcecd_resync_free_entries(entries, n);
free_worker_list(list);
}
/* Requirement 3.6: a job the worker is running that was NOT part of the
* query is reported as an extra (queried=0). */
static void test_build_report_includes_running_extra(void)
{
xcecd_worker_t* list = NULL;
link_worker(&list, make_worker(1, 2001, WORKER_RUNNING));
link_worker(&list, make_worker(2, 2002, WORKER_RUNNING)); /* extra */
uint64_t jobids[1] = { 2001 };
xce_job_state_entry_t** entries = NULL;
int n = 0;
int rc = xcecd_resync_build_report(list, jobids, 1, 1000, &entries, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, n);
xce_job_state_entry_t* queried = find_entry(entries, n, 2001);
xce_job_state_entry_t* extra = find_entry(entries, n, 2002);
TEST_ASSERT_NOT_NULL(queried);
TEST_ASSERT_EQUAL_INT(1, queried->queried);
TEST_ASSERT_NOT_NULL(extra);
TEST_ASSERT_EQUAL_INT(0, extra->queried);
TEST_ASSERT_EQUAL_INT(JobStateStatus_running, extra->status);
xcecd_resync_free_entries(entries, n);
free_worker_list(list);
}
/* Requirement 3.6: a job the worker recently finished that was NOT part
* of the query is reported as an extra, with its real finish time. */
static void test_build_report_includes_finished_extra(void)
{
xcecd_retain_finished_job(3001, WORKER_EXITED, 5, 70000);
xce_job_state_entry_t** entries = NULL;
int n = 0;
int rc = xcecd_resync_build_report(NULL, NULL, 0, 75000, &entries, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, n);
TEST_ASSERT_EQUAL_UINT64(3001, entries[0]->jobid);
TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedFail, entries[0]->status);
TEST_ASSERT_EQUAL_INT32(5, entries[0]->exit_code);
TEST_ASSERT_EQUAL_UINT64(70000, entries[0]->finish_time_ms);
TEST_ASSERT_EQUAL_INT(0, entries[0]->queried);
xcecd_resync_free_entries(entries, n);
}
/* A job that IS part of the query must never also appear a second time
* as an extra -- the queried answer is the only entry for it. */
static void test_build_report_queried_job_not_duplicated_as_extra(void)
{
xcecd_worker_t* list = NULL;
link_worker(&list, make_worker(1, 4001, WORKER_RUNNING));
xcecd_retain_finished_job(4002, WORKER_DONE, 0, 10000);
uint64_t jobids[2] = { 4001, 4002 };
xce_job_state_entry_t** entries = NULL;
int n = 0;
int rc = xcecd_resync_build_report(list, jobids, 2, 20000, &entries, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, n);
int count4001 = 0, count4002 = 0;
int i;
for (i = 0; i < n; i++) {
if (entries[i]->jobid == 4001)
count4001++;
if (entries[i]->jobid == 4002)
count4002++;
}
TEST_ASSERT_EQUAL_INT(1, count4001);
TEST_ASSERT_EQUAL_INT(1, count4002);
xcecd_resync_free_entries(entries, n);
free_worker_list(list);
}
/* Empty query, nothing running or retained -> a valid, empty (zero-entry)
* result, not an error. */
static void test_build_report_empty_query_no_state_yields_empty_result(void)
{
xce_job_state_entry_t** entries
= (xce_job_state_entry_t**)0x1; /* sentinel */
int n = -1;
int rc = xcecd_resync_build_report(NULL, NULL, 0, 1000, &entries, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, n);
TEST_ASSERT_NULL(entries);
}
static void test_build_report_null_out_params_rejected(void)
{
uint64_t jobids[1] = { 1 };
xce_job_state_entry_t** entries = NULL;
int n = 0;
TEST_ASSERT_EQUAL_INT(
-EINVAL, xcecd_resync_build_report(NULL, jobids, 1, 1000, NULL, &n));
TEST_ASSERT_EQUAL_INT(-EINVAL, xcecd_resync_build_report(
NULL, jobids, 1, 1000, &entries, NULL));
}
/* ════════════════════════════════════════════════════════════════════════
* xcecd_resync_answer_query: end-to-end paginated send (Requirements 8.2,
* 8.3)
* ════════════════════════════════════════════════════════════════════════ */
static void test_answer_query_single_page(void)
{
xcecd_worker_t* list = NULL;
link_worker(&list, make_worker(1, 5001, WORKER_RUNNING));
xcecd_retain_finished_job(5002, WORKER_DONE, 0, 30000);
uint64_t jobids[2] = { 5001, 5002 };
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
int rc = xcecd_resync_answer_query(list, ep, &from, jobids, 2, 40000, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode);
TEST_ASSERT_EQUAL_INT(Request_body_jobstatereportmsg, decoded.body_which);
TEST_ASSERT_EQUAL_INT(2, decoded.body.jobstatereportmsg->n_total);
TEST_ASSERT_EQUAL_INT(2, decoded.body.jobstatereportmsg->count);
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
free_worker_list(list);
}
/* Requirement 8.3: nothing truncated/dropped across multiple pages, for
* a mix of queried answers and extras. */
static void test_answer_query_multipage_covers_every_job(void)
{
xcecd_worker_t* list = NULL;
int i;
/* 3 queried running jobs. */
uint64_t jobids[3] = { 6001, 6002, 6003 };
for (i = 0; i < 3; i++) {
link_worker(&list, make_worker(100 + i, jobids[i], WORKER_RUNNING));
}
/* 2 extra running jobs, not queried. */
link_worker(&list, make_worker(200, 6004, WORKER_RUNNING));
link_worker(&list, make_worker(201, 6005, WORKER_RUNNING));
/* 2 extra finished jobs, not queried. */
xcecd_retain_finished_job(6006, WORKER_DONE, 0, 90000);
xcecd_retain_finished_job(6007, WORKER_EXITED, 3, 90500);
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
/* 7 total entries, pktlimit=3 -> pages of 3, 3, 1. */
int rc = xcecd_resync_answer_query(list, ep, &from, jobids, 3, 95000, 3);
TEST_ASSERT_EQUAL_INT(0, rc);
uint64_t rebuilt[7];
int queried_flags[7];
int rebuilt_n = 0;
int page;
for (page = 0; page < 3; page++) {
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode);
TEST_ASSERT_EQUAL_INT(7, decoded.body.jobstatereportmsg->n_total);
int c = decoded.body.jobstatereportmsg->count;
int i2;
for (i2 = 0; i2 < c; i2++) {
xce_job_state_entry_t* got
= decoded.body.jobstatereportmsg->entries[i2];
rebuilt[rebuilt_n] = got->jobid;
queried_flags[rebuilt_n] = got->queried;
rebuilt_n++;
}
xce_free_request(&decoded);
}
assert_queue_empty(ep);
TEST_ASSERT_EQUAL_INT(7, rebuilt_n);
/* Every job from 6001..6007 must appear exactly once across all
* pages, with the correct queried flag. */
uint64_t expected[7] = { 6001, 6002, 6003, 6004, 6005, 6006, 6007 };
int expected_q[7] = { 1, 1, 1, 0, 0, 0, 0 };
int j;
for (j = 0; j < 7; j++) {
int found = 0;
int k;
for (k = 0; k < 7; k++) {
if (rebuilt[k] == expected[j]) {
TEST_ASSERT_EQUAL_INT(expected_q[j], queried_flags[k]);
found = 1;
}
}
TEST_ASSERT_TRUE_MESSAGE(found, "every job must appear exactly once "
"across all report pages, none "
"dropped");
}
endpoint_fini(ep);
free(ep);
free_worker_list(list);
}
/* An empty query against a worker with no running/retained jobs still
* sends exactly one (empty) report page -- not skipped. */
static void test_answer_query_empty_still_sends_one_page(void)
{
uint64_t* jobids = NULL;
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
int rc = xcecd_resync_answer_query(NULL, ep, &from, jobids, 0, 1000, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode);
TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->n_total);
TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->count);
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
}
/* ════════════════════════════════════════════════════════════════════════
* xcecd_finished_job_retention_snapshot: task 4.2's companion addition to
* xcecd_worker_utils.c/h (needed to enumerate retention for extras, since
* task 4.1's own public API was jobid-keyed lookup only).
* ════════════════════════════════════════════════════════════════════════ */
static void test_retention_snapshot_returns_all_unexpired_entries(void)
{
xcecd_retain_finished_job(9001, WORKER_DONE, 0, 10000);
xcecd_retain_finished_job(9002, WORKER_EXITED, 1, 10500);
xcecd_finished_job_t* snap = NULL;
int n = 0;
int rc = xcecd_finished_job_retention_snapshot(20000, &snap, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, n);
TEST_ASSERT_NOT_NULL(snap);
free(snap);
}
static void test_retention_snapshot_excludes_expired_entries(void)
{
xcecd_finished_job_retention_set_window_ms(1000);
xcecd_retain_finished_job(9101, WORKER_DONE, 0, 10000);
xcecd_finished_job_t* snap = NULL;
int n = 0;
int rc = xcecd_finished_job_retention_snapshot(10000 + 1000 + 1, &snap, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, n);
TEST_ASSERT_NULL(snap);
}
static void test_retention_snapshot_empty_store(void)
{
xcecd_finished_job_t* snap = (xcecd_finished_job_t*)0x1;
int n = -1;
int rc = xcecd_finished_job_retention_snapshot(1000, &snap, &n);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, n);
TEST_ASSERT_NULL(snap);
}
static void test_retention_snapshot_null_out_params_rejected(void)
{
xcecd_finished_job_t* snap = NULL;
int n = 0;
TEST_ASSERT_EQUAL_INT(
-EINVAL, xcecd_finished_job_retention_snapshot(1000, NULL, &n));
TEST_ASSERT_EQUAL_INT(
-EINVAL, xcecd_finished_job_retention_snapshot(1000, &snap, NULL));
}
/* ════════════════════════════════════════════════════════════════════════
* main
* ════════════════════════════════════════════════════════════════════════ */
int main(void)
{
xce_init_xcecd_log("/tmp/test_xcecd_resync.log", 1024 * 1024, 2);
UNITY_BEGIN();
RUN_TEST(test_classify_running_job);
RUN_TEST(test_classify_finished_ok_job);
RUN_TEST(test_classify_finished_fail_job);
RUN_TEST(test_classify_unrecognized_job);
RUN_TEST(test_classify_terminal_worker_entry_falls_through_to_retention);
RUN_TEST(test_classify_worker_unknown_status_is_unrecognized);
RUN_TEST(test_classify_expired_retention_is_unrecognized);
RUN_TEST(test_build_report_answers_every_queried_id_exactly_once);
RUN_TEST(test_build_report_includes_running_extra);
RUN_TEST(test_build_report_includes_finished_extra);
RUN_TEST(test_build_report_queried_job_not_duplicated_as_extra);
RUN_TEST(test_build_report_empty_query_no_state_yields_empty_result);
RUN_TEST(test_build_report_null_out_params_rejected);
RUN_TEST(test_answer_query_single_page);
RUN_TEST(test_answer_query_multipage_covers_every_job);
RUN_TEST(test_answer_query_empty_still_sends_one_page);
RUN_TEST(test_retention_snapshot_returns_all_unexpired_entries);
RUN_TEST(test_retention_snapshot_excludes_expired_entries);
RUN_TEST(test_retention_snapshot_empty_store);
RUN_TEST(test_retention_snapshot_null_out_params_rejected);
return UNITY_END();
}