merge: sync Rongsong's scheduler-framework (88 commits, f01ae353..c5b172c7)
Two-sided divergence from f01ae353 (2026-08-30): theirs +88 (worker
rejoin-resync reconciliation, terminal-state bug-1737/1738/1739/1764,
scheduler-sim verbosity, hermetic offline build, go mod vendor of Go deps
in apiserver -- 3375 pure-adds), ours +89 (job-finish robustness bug-1509
B1/B3, PEND-reason, worker reconnect resend, apiserver job control).
Conflict resolutions (kept both sides' fixes):
- controller_connector.c checking_connection(): combine theirs
xcecd_resync_is_pending() rejoin deferral gate with ours
resend_job_status() actual-state re-report on rejoin success.
- job.c: keep both unknown-finish registry (ours, bug-1509) and
record_job_finish_fact_at (theirs, bug-1747); keep both
mark_jobs_unknown_core family (ours) and record_job_reconcile_*
(theirs). No double-notify: on_job_finish unchanged, theirs is
resync-path-only.
- scheduler handler.c/test_handlers.c/meson: theirs schedExitJob
exit_code branch (bug-1737) + ours jobDispatchFail success-ack no-op
(bug-1512) + configpush enq_pend_update (bug-1513) both present; both
test suites and REPL/sim targets wired.
- apiserver jobs.go: keep ours normalizeJobInt64/jobArrayIdx superset +
JobIDHuman; drop theirs strict IntOrInt64 early-reject (superseded --
ours tolerates float64/json.Number, no panic path).
- jobs_test.go: keep both regression suites (ours bug-1490, theirs
bug-1742).
- .kiro/specs scheduler-sim-*/worker-rejoin-resync + .wolf/*: keep our
deletions (established convention -- drop Rongsong bookkeeping on merge).
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen
commit
5e2985542f
3469 changed files with 1466599 additions and 382 deletions
169
src/worker/include/xcecd_resync.h
Normal file
169
src/worker/include/xcecd_resync.h
Normal file
|
|
@ -0,0 +1,169 @@
|
|||
#ifndef XCECD_RESYNC_H
|
||||
#define XCECD_RESYNC_H
|
||||
|
||||
#include <stdbool.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").
|
||||
*
|
||||
* task 4.3 (Requirements 3.8, 8.1, 8.3) adds the actual wire-receipt entry
|
||||
* point, xcecd_resync_handle_query_page() below: it is what
|
||||
* controller_connector.c's myreq_handler dispatch case for
|
||||
* OPCode_jobStateQuery calls, once per incoming query page, and it owns
|
||||
* both the multi-page query reassembly (a query may itself arrive across
|
||||
* several pages, Requirement 8.1/8.3) and the Requirement 3.8
|
||||
* `resync_pending` gate (mirroring controller_connector.c's existing
|
||||
* `needs_rejoin` static-bool pattern) that delays the worker's own
|
||||
* resumption of normal post-join operation until the full report --
|
||||
* queried and extra pages alike -- has finished sending.
|
||||
* xcecd_resync_answer_query() itself remains the pure classify/build/send
|
||||
* primitive task 4.2 left it as; it takes an already-assembled jobids
|
||||
* array and knows nothing about the gate or multi-page query receipt.
|
||||
*/
|
||||
|
||||
/* 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);
|
||||
|
||||
/*
|
||||
* worker-rejoin-resync task 4.3 (Requirements 3.8, 8.1, 8.3): the real
|
||||
* OPCode_jobStateQuery receive-side entry point. Called once per incoming
|
||||
* query page from controller_connector.c's myreq_handler dispatch case,
|
||||
* with `page` the just-decoded xce_job_state_query_t for that one page
|
||||
* (its own {n_total, pos, count, jobids} slice, the same pagination idiom
|
||||
* xce_job_state_query_req() -- the sender side, task 1.4 -- uses).
|
||||
*
|
||||
* Reassembles however many pages the query itself arrives in into one flat
|
||||
* jobids array (a page with pos == 0 starts a fresh accumulation, discarding
|
||||
* any prior incomplete one -- e.g. left over from a worker-side disconnect
|
||||
* mid-query; only one reconciliation round is ever outstanding on a given
|
||||
* worker's single long-lived controller connection at a time, per
|
||||
* myreq_handler's own synchronous per-message dispatch). Once the last page
|
||||
* has arrived, calls xcecd_resync_answer_query() above with the fully
|
||||
* assembled list.
|
||||
*
|
||||
* Also owns the Requirement 3.8 `resync_pending` gate (mirroring
|
||||
* controller_connector.c's existing `needs_rejoin` static-bool pattern,
|
||||
* queried via xcecd_resync_is_pending() below): set the moment the first
|
||||
* page of a query is received (before this page is even fully validated),
|
||||
* held set across every remaining query page and every page of the report
|
||||
* this function goes on to send, and cleared only once
|
||||
* xcecd_resync_answer_query() has returned -- on success OR failure, so a
|
||||
* transport error can never leave the worker's own periodic post-join
|
||||
* activity gated forever (the controller's own bounded reconciliation
|
||||
* timeout, Requirement 5.3, is what resolves an incomplete round from its
|
||||
* side; this is purely a local, non-blocking gate on the worker's own
|
||||
* follow-on behavior -- not a new wire-level ack, not an extension of
|
||||
* xce_join()'s blocking wait, per design.md's Boundary Commitments).
|
||||
*
|
||||
* Returns 0 while still waiting for more query pages (report not sent
|
||||
* yet), 1 once this query has been fully answered (report sent, whether
|
||||
* or not the send itself succeeded), or a negative errno for a malformed
|
||||
* page (NULL page/ep/from, or a page whose pos/count fall outside its own
|
||||
* declared n_total) or allocation failure -- the gate and any in-progress
|
||||
* accumulator are both cleared/discarded in that case too, never left
|
||||
* stuck.
|
||||
*/
|
||||
int xcecd_resync_handle_query_page(xcecd_worker_t* worker_list, endpoint_t* ep,
|
||||
xce_peerinfo_t* from,
|
||||
const xce_job_state_query_t* page,
|
||||
uint64_t now_ms, int pktlimit);
|
||||
|
||||
/* Requirement 3.8: true from the moment a job-state query's first page is
|
||||
* received until the full report (queried and extra pages alike) has
|
||||
* finished sending -- see xcecd_resync_handle_query_page() above. Checked
|
||||
* by controller_connector.c's checking_connection() before it resumes the
|
||||
* needs_rejoin-triggered rejoin action -- NOT before its own liveness probe
|
||||
* (xcecli_hello()) or the needs_rejoin check itself, both of which run every
|
||||
* tick unconditionally so the worker's own reconnect-detection can never be
|
||||
* gated off (Requirement 6.1). */
|
||||
bool xcecd_resync_is_pending(void);
|
||||
|
||||
/*
|
||||
* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
|
||||
* 6.1): unconditionally clears resync_pending and discards any in-progress
|
||||
* multi-page query accumulation. xcecd_resync_handle_query_page() can only
|
||||
* clear the gate itself by being called again with a query's LAST page --
|
||||
* if the worker's connection drops after a non-final page arrived, no
|
||||
* further page will ever come, and resync_pending would otherwise stay
|
||||
* stuck true forever with nothing left to clear it.
|
||||
*
|
||||
* Call this the moment the worker itself detects it has reconnected --
|
||||
* controller_connector.c's xcecd_on_reconnect() (the client library's
|
||||
* transparent-reconnect callback, the same signal that sets needs_rejoin)
|
||||
* is that moment: the connection any in-flight query arrived on is already
|
||||
* gone, and the controller's own Requirement 6.1 Acceptance Criterion 1
|
||||
* discards that incomplete round and re-initiates a fresh query on this
|
||||
* worker's next rejoin, so nothing here is worth preserving. Safe to call
|
||||
* from that exact site: it touches only this module's own statics (a bool
|
||||
* and a small heap buffer), no I/O, no xcecli_* call. */
|
||||
void xcecd_resync_reset(void);
|
||||
|
||||
#endif /* XCECD_RESYNC_H */
|
||||
|
|
@ -38,6 +38,25 @@ struct xcecd_worker {
|
|||
struct xcecd_db_t;
|
||||
typedef struct xcecd_worker xcecd_worker_t;
|
||||
|
||||
/* worker-rejoin-resync task 4.1 (Requirements 3.4, 3.5, 3.7): a job's true
|
||||
* outcome, retained for a bounded window after exit so a later
|
||||
* reconciliation query (task 4.2, xcecd_resync.c) can still answer for it
|
||||
* even after cleanup_worker_list() has reaped the live xcecd_worker_t
|
||||
* entry that used to be the only record of it. Field list matches
|
||||
* design.md's Data Models section exactly: {jobid, exit_status, exit_code,
|
||||
* finish_time_ms} -- no retain-until field is stored per-entry; expiry is
|
||||
* computed at query time from finish_time_ms plus the currently configured
|
||||
* window (xcecd_finished_job_retention_set_window_ms()), i.e. lazy
|
||||
* expiry-on-query, not active/periodic pruning (either is permitted by
|
||||
* design.md; this is the chosen strategy -- see xcecd_worker_utils.c). */
|
||||
typedef struct xcecd_finished_job {
|
||||
uint64_t jobid;
|
||||
int exit_status; /* WORKER_DONE or WORKER_EXITED */
|
||||
int exit_code;
|
||||
uint64_t finish_time_ms; /* wall-clock (epoch) ms */
|
||||
struct xcecd_finished_job* next;
|
||||
} xcecd_finished_job_t;
|
||||
|
||||
/* TODO: New Worker states */
|
||||
typedef enum {
|
||||
SERVICE_UNKNOWN = 0,
|
||||
|
|
@ -60,6 +79,72 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
|
|||
/* Cleanup the worker in the list which finished work */
|
||||
void cleanup_worker_list(xcecd_worker_t** worker_list, sqlite3* db);
|
||||
|
||||
/* Record a job's true outcome at the moment its exit is detected --
|
||||
* called from update_worker_status()'s WORKER_DONE/WORKER_EXITED handling,
|
||||
* BEFORE cleanup_worker_list() would otherwise discard all trace of the
|
||||
* xcecd_worker_t entry (Requirement 3.4). Does not change
|
||||
* cleanup_worker_list()'s own reaping behavior -- this is a second,
|
||||
* parallel, smaller structure fed at the same trigger point, not a rework
|
||||
* of the existing one. finish_time_ms is the caller-captured real exit
|
||||
* time (wall-clock/epoch milliseconds), not the time this function itself
|
||||
* runs (Requirement 3.5) -- callers must capture it once, at genuine
|
||||
* exit-detection time, and pass it through. Replaces any existing
|
||||
* retained entry for the same jobid. */
|
||||
void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code,
|
||||
uint64_t finish_time_ms);
|
||||
|
||||
/* Look up a job's retained outcome. Returns true and fills *out when jobid
|
||||
* is present and has not yet aged past the configured retention window
|
||||
* (now_ms - finish_time_ms <= window in effect); returns false (*out left
|
||||
* untouched) when absent or expired -- this is what task 4.2's query
|
||||
* handler (xcecd_resync.c) calls for a queried job ID not found on the
|
||||
* live running list. now_ms is caller-supplied, not captured internally,
|
||||
* so callers (and tests) can query deterministically. Lazily prunes every
|
||||
* expired entry as a side effect, so the store never grows unbounded even
|
||||
* without an external periodic driver. */
|
||||
bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms,
|
||||
xcecd_finished_job_t* out);
|
||||
|
||||
/* Configure the retention window in milliseconds (Requirement 3.7,
|
||||
* ClusterConfig.workerResyncWindowMs). Passing 0 restores the compiled-in
|
||||
* default. Confirmed while implementing task 4.1: the worker daemon has no
|
||||
* runtime delivery path for this ClusterConfig field today --
|
||||
* xcecd_config_t/xcecd_context_t (xcecd.h) carry no ClusterConfig fields,
|
||||
* worker_push_config() (controller/worker.c) only ever pushes
|
||||
* hosttype/hostmodel/resource confvars via collect_confvars() (an
|
||||
* unrelated xce_config_t key/value store, not the capnp ClusterConfig
|
||||
* struct), and the worker has no local ClusterConfig file the way the
|
||||
* controller's myconf.c::load_cluster_config_from_file() does. This setter
|
||||
* is the seam a later task wires the real operator-configured value into
|
||||
* once that delivery path exists; until then the compiled-in default
|
||||
* applies. */
|
||||
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);
|
||||
|
||||
xcecd_worker_t* find_worker_by_pid(void* worker_list, int pid);
|
||||
|
||||
xcecd_worker_t* find_worker_by_jobid(xcecd_worker_t* worker_list,
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ xcecd_worker_srcs = [
|
|||
'src/static_array.c',
|
||||
'src/xcecd_cgroups.c',
|
||||
'src/xcecd_worker_utils.c',
|
||||
'src/xcecd_resync.c',
|
||||
'src/controller_connector.c',
|
||||
'src/xcecd_dbus_manager.c',
|
||||
'src/xcecd_db.c',
|
||||
|
|
@ -84,6 +85,7 @@ xcecd_srcs = [
|
|||
'src/controller_connector.c',
|
||||
'src/resource_sync.c',
|
||||
'src/xcecd_worker_utils.c',
|
||||
'src/xcecd_resync.c',
|
||||
'src/hardware_monitor.c',
|
||||
'src/node_information.c',
|
||||
'src/job_information.c'
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
#include "uv_ipc.h"
|
||||
#include "xcecd_cgroups.h"
|
||||
#include "xcecd_exdata.h"
|
||||
#include "xcecd_resync.h"
|
||||
#include "xcecd_worker_utils.h"
|
||||
#include "xcelog.h"
|
||||
#include <errno.h>
|
||||
|
|
@ -271,11 +272,26 @@ static void format_candidate_list(char* buf, size_t bufsz)
|
|||
* XCECLI_FAILOVER_CALL retry loop) -- per design.md's "Rejoin callback
|
||||
* constraint", this callback MUST NOT perform I/O or call any xcecli_*
|
||||
* function: doing so would try to re-acquire the same already-held lock and
|
||||
* self-deadlock the calling thread forever. It only sets a flag and returns
|
||||
* 0 immediately so the interrupted operation resumes normally. Consuming
|
||||
* this flag to issue the actual xcecli_join() as a separate, later call
|
||||
* (once the triggering call has returned and the lock is free) is task
|
||||
* 2.2's job.
|
||||
* self-deadlock the calling thread forever. It only sets flags/statics and
|
||||
* returns 0 immediately so the interrupted operation resumes normally.
|
||||
* Consuming needs_rejoin to issue the actual xcecli_join() as a separate,
|
||||
* later call (once the triggering call has returned and the lock is free)
|
||||
* is task 2.2's job.
|
||||
*
|
||||
* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
|
||||
* 6.1): also resets xcecd_resync.c's resync_pending gate here. This is the
|
||||
* moment the worker itself learns its connection to the controller was
|
||||
* re-established -- any job-state query it was still assembling (or report
|
||||
* it had not finished sending) belongs to the connection that just died and
|
||||
* can never complete; the controller's own Requirement 6.1 Acceptance
|
||||
* Criterion 1 already discards that incomplete round and re-initiates a
|
||||
* fresh query on this worker's next rejoin. Without this reset,
|
||||
* resync_pending can only be cleared by xcecd_resync_handle_query_page()
|
||||
* seeing a query's LAST page -- which never arrives on a dead connection --
|
||||
* leaving it stuck true forever (round 1's confirmed defect). Calling
|
||||
* xcecd_resync_reset() here is safe under the constraint above: it only
|
||||
* touches xcecd_resync.c's own statics (a bool and a small heap buffer), no
|
||||
* I/O and no xcecli_* call.
|
||||
*/
|
||||
static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
|
||||
{
|
||||
|
|
@ -283,6 +299,7 @@ static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
|
|||
(void)arg;
|
||||
|
||||
needs_rejoin = true;
|
||||
xcecd_resync_reset();
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -465,6 +482,25 @@ static int (*rejoin_call)(void) = perform_pending_rejoin;
|
|||
* index) but the old `rc == 0` check logged "rejoin ... failed" on every
|
||||
* subsequent liveness probe.
|
||||
*
|
||||
* worker-rejoin-resync task 4.3 (Requirement 3.8), narrowed per round-2
|
||||
* remediation: xcecd_resync_is_pending() gates ONLY the needs_rejoin-
|
||||
* triggered rejoin_call() below -- the "resume normal post-join operation"
|
||||
* step Requirement 3.8 actually names -- not this function's entry.
|
||||
* xcecli_hello() and the needs_rejoin check itself both still run every
|
||||
* tick unconditionally, so checking_connection() keeps the exact
|
||||
* self-healing property needs_rejoin already had before this task: a fresh
|
||||
* chance every cycle to detect and recover the worker's own connection
|
||||
* (src/lib/xcectx.c's handle_disconnect() -- "so that checking_connection
|
||||
* can detect the failure and attempt to rejoin"). A resync_pending stuck
|
||||
* true (round 1's confirmed defect: gating this function's entry made the
|
||||
* worker's sole connection-recovery mechanism permanently inert, breaking
|
||||
* Requirement 6.1's "disconnect during an incomplete reconciliation round"
|
||||
* case) can therefore never disable this probe. In practice resync_pending
|
||||
* is also cleared the instant a reconnect is detected
|
||||
* (xcecd_on_reconnect() -> xcecd_resync_reset(), above), so this check is a
|
||||
* defensive backstop against the rejoin racing an in-flight report, not the
|
||||
* mechanism that unsticks a stalled one.
|
||||
*
|
||||
* @return int Always returns 0.
|
||||
*/
|
||||
int checking_connection()
|
||||
|
|
@ -480,25 +516,31 @@ int checking_connection()
|
|||
}
|
||||
|
||||
if (needs_rejoin) {
|
||||
int rc = rejoin_call();
|
||||
if (rc >= 0) {
|
||||
needs_rejoin = false;
|
||||
XCECD_LOG(XCE_LOG_INFO,
|
||||
"checking_connection: rejoin after transparent "
|
||||
"reconnect succeeded");
|
||||
|
||||
/* bug-1509 Phase 3 (B2/B3): re-report running/finished job state
|
||||
* on every rejoin, not just at startup, so the controller
|
||||
* reconciles its view after the disconnect window (clears UNKWN
|
||||
* on a live claim; converges to a terminal state for processes
|
||||
* that are gone). */
|
||||
(void)resend_job_status(*cc_ctx->worker_list);
|
||||
if (xcecd_resync_is_pending()) {
|
||||
XCECD_LOG(XCE_LOG_DEBUG,
|
||||
"checking_connection: deferring rejoin -- job-state "
|
||||
"resync report still being sent");
|
||||
} else {
|
||||
XCECD_LOG(XCE_LOG_WARN,
|
||||
"checking_connection: rejoin after transparent "
|
||||
"reconnect failed (rc=%d); will retry on the next "
|
||||
"liveness probe",
|
||||
rc);
|
||||
int rc = rejoin_call();
|
||||
if (rc >= 0) {
|
||||
needs_rejoin = false;
|
||||
XCECD_LOG(XCE_LOG_INFO,
|
||||
"checking_connection: rejoin after transparent "
|
||||
"reconnect succeeded");
|
||||
|
||||
/* bug-1509 Phase 3 (B2/B3): re-report running/finished job
|
||||
* state on every rejoin, not just at startup, so the
|
||||
* controller reconciles its view after the disconnect window
|
||||
* (clears UNKWN on a live claim; converges to a terminal
|
||||
* state for processes that are gone). */
|
||||
(void)resend_job_status(*cc_ctx->worker_list);
|
||||
} else {
|
||||
XCECD_LOG(XCE_LOG_WARN,
|
||||
"checking_connection: rejoin after transparent "
|
||||
"reconnect failed (rc=%d); will retry on the next "
|
||||
"liveness probe",
|
||||
rc);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1578,6 +1620,121 @@ static int handle_jobcontrol(xce_request_t* request)
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* worker-rejoin-resync task 4.3 (Req 8.2/8.3 precedent): page-size cap for
|
||||
* this worker's own OPCode_jobStateReport sends, matching the pktlimit
|
||||
* value apiserver.c's own xce_push_data_req() call sites already use for
|
||||
* the same {pos,count,n_total} pagination idiom. */
|
||||
#define XCECD_RESYNC_REPORT_PKTLIMIT 100
|
||||
|
||||
/* Wall-clock (epoch) milliseconds -- same rationale and implementation as
|
||||
* xcecd_worker_utils.c's private finished_job_wallclock_now_ms() (not
|
||||
* reusable here: internal linkage, separate translation unit). Needed here
|
||||
* because xcecd_resync.c's functions all take now_ms as a caller-captured
|
||||
* parameter rather than sampling the clock themselves (task 4.1's own
|
||||
* convention -- deterministic values in tests, real wall-clock in
|
||||
* production), and this dispatch case is that production caller. */
|
||||
static uint64_t resync_wallclock_now_ms(void)
|
||||
{
|
||||
struct timespec ts;
|
||||
|
||||
if (clock_gettime(CLOCK_REALTIME, &ts) != 0) {
|
||||
return (uint64_t)time(NULL) * 1000ULL;
|
||||
}
|
||||
|
||||
return (uint64_t)ts.tv_sec * 1000ULL + (uint64_t)ts.tv_nsec / 1000000ULL;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Handles one incoming OPCode_jobStateQuery page from the controller
|
||||
* (worker-rejoin-resync task 4.3, Requirements 3.8, 8.1, 8.3).
|
||||
*
|
||||
* Delegates to xcecd_resync_handle_query_page() (xcecd_resync.c), which
|
||||
* reassembles a query that itself spans multiple incoming pages and, once
|
||||
* the last page has arrived, answers it with a paginated
|
||||
* OPCode_jobStateReport -- see that function's own header comment
|
||||
* (xcecd_resync.h) for the resync_pending gate it also owns.
|
||||
*
|
||||
* @param request Pointer to the xce_request_t structure containing the
|
||||
* request data.
|
||||
* @return int 0 on success (whether or not this page completed the query),
|
||||
* negative on error.
|
||||
*/
|
||||
static int handle_jobstatequery(xce_request_t* request)
|
||||
{
|
||||
endpoint_t* ep = xcecli_get_my_endpoint(xcecli_ctx);
|
||||
xce_peerinfo_t* from = xcecli_get_myinfo(xcecli_ctx);
|
||||
xcecd_worker_t* worker_list
|
||||
= (cc_ctx != NULL) ? (xcecd_worker_t*)*cc_ctx->worker_list : NULL;
|
||||
int rc;
|
||||
|
||||
if (ep == NULL || from == NULL) {
|
||||
XCECD_LOG(XCE_LOG_ERROR,
|
||||
"handle_jobstatequery: no active controller connection");
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
rc = xcecd_resync_handle_query_page(
|
||||
worker_list, ep, from, request->body.jobstatequerymsg,
|
||||
resync_wallclock_now_ms(), XCECD_RESYNC_REPORT_PKTLIMIT);
|
||||
if (rc < 0) {
|
||||
XCECD_LOG(XCE_LOG_ERROR,
|
||||
"handle_jobstatequery: failed to process query page "
|
||||
"(rc=%d)",
|
||||
rc);
|
||||
return rc;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Handles an incoming OPCode_push config-push message from the
|
||||
* controller (worker-rejoin-resync task 6.1, Requirement 3.7).
|
||||
*
|
||||
* Confirmed by inspection: before this task, myreq_handler's dispatch
|
||||
* switch below had NO case for OPCode_push at all -- a config push
|
||||
* (worker_push_config()/on_worker_join(), controller/worker.c) fell through
|
||||
* to the `default: break;` case and was silently discarded in its
|
||||
* entirety, even for the pre-existing clustername field.
|
||||
* xcecd_finished_job_retention_set_window_ms()'s own doc comment
|
||||
* (xcecd_worker_utils.h, task 4.1) independently confirms this same gap:
|
||||
* "the worker daemon has no runtime delivery path for this ClusterConfig
|
||||
* field today... this setter is the seam a later task wires the real
|
||||
* operator-configured value into once that delivery path exists." This
|
||||
* function is that delivery path, scoped narrowly to the ONE field this
|
||||
* task needs (workerResyncWindowMs) -- clustername/confvars remain
|
||||
* unconsumed by the worker, unchanged and out of this task's boundary; a
|
||||
* general config-receive rework is not attempted here.
|
||||
*
|
||||
* Fire-and-forget, matching worker_push_config()'s own controller-side
|
||||
* send (no reply is expected or sent back).
|
||||
*
|
||||
* @param request Pointer to the xce_request_t structure containing the
|
||||
* request data.
|
||||
* @return int 0 always -- a malformed/unexpected payload shape (wrong
|
||||
* body_which, wrong ConfigData union arm, NULL conf4worker) is
|
||||
* simply ignored, not an error to report back.
|
||||
*/
|
||||
static int handle_pushconfig(xce_request_t* request)
|
||||
{
|
||||
xce_configdata_t* configdata;
|
||||
|
||||
if (request->body_which != Request_body_configpushmsg)
|
||||
return 0;
|
||||
|
||||
configdata = request->body.configpushmsg;
|
||||
if (configdata == NULL
|
||||
|| configdata->data_which != ConfigData_data_workerconfig
|
||||
|| configdata->data.conf4worker == NULL) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
xcecd_finished_job_retention_set_window_ms(
|
||||
(uint64_t)configdata->data.conf4worker->worker_resync_window_ms);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Handles incoming requests from the controller.
|
||||
*
|
||||
|
|
@ -1603,6 +1760,12 @@ static int myreq_handler(xce_request_t* request, void* args)
|
|||
case OPCode_fetchfile:
|
||||
ret = handle_fetchfile(request);
|
||||
break;
|
||||
case OPCode_jobStateQuery:
|
||||
ret = handle_jobstatequery(request);
|
||||
break;
|
||||
case OPCode_push:
|
||||
ret = handle_pushconfig(request);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
#include "job_information.h"
|
||||
#include "node_information.h"
|
||||
#include "xcecd.h"
|
||||
#include "xcecd_resync.h"
|
||||
#include "xcecd_utils.h"
|
||||
#include "xcelog.h"
|
||||
#include <pthread.h>
|
||||
|
|
@ -58,10 +59,23 @@ static void free_resvals(xce_resval_t* vals, int count)
|
|||
|
||||
static void init_builtin_load_indices(void)
|
||||
{
|
||||
static bool initialized = false;
|
||||
if (initialized)
|
||||
return;
|
||||
|
||||
/* bug-1756 (companion, worker-rejoin-resync task 4.4): this used to
|
||||
* guard against re-running registration with its own function-local
|
||||
* `static bool initialized`, but that guard is a duplicate of -- and
|
||||
* conflicts with -- collect_load_indices()'s own `g_load_indices ==
|
||||
* NULL` check, this function's only caller. Once g_load_indices is torn
|
||||
* down by resource_sync_cleanup_load_indices() (resource_sync_destroy(),
|
||||
* e.g. a create/start/stop/destroy cycle repeated within one process,
|
||||
* as the task 4.4 tests below now do for the first time), the caller's
|
||||
* check correctly wants to re-populate the registry, but the
|
||||
* function-local `initialized` stayed stuck true from the first run and
|
||||
* silently no-op'd forever after, leaving g_load_indices permanently
|
||||
* NULL/empty for the rest of the process ("No load indices to report").
|
||||
* Removed: the caller's own guard already ensures this body runs
|
||||
* exactly once per populated-registry window, with no other call site
|
||||
* (grepped) and no concurrent caller (collect_load_indices() only runs
|
||||
* on resource_sync_thread()'s single dedicated pthread in production,
|
||||
* or single-threaded test code). */
|
||||
g_load_indices_capacity = MAX_LOAD_INDICES;
|
||||
g_load_indices = (load_index_info_t*)calloc(g_load_indices_capacity,
|
||||
sizeof(load_index_info_t));
|
||||
|
|
@ -89,8 +103,6 @@ static void init_builtin_load_indices(void)
|
|||
true);
|
||||
resource_sync_register_load_index("io", Value_u_fvalue, node_info_get_io,
|
||||
true);
|
||||
|
||||
initialized = true;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -322,6 +334,30 @@ static void* resource_sync_thread(void* arg)
|
|||
|
||||
sync->load_report_counter++;
|
||||
if (sync->load_report_counter >= LOAD_INDICES_INTERVAL_DEFAULT) {
|
||||
/* worker-rejoin-resync task 4.4 (Requirement 3.8): this periodic
|
||||
* load-indices send is worker post-join background activity,
|
||||
* same as checking_connection()'s rejoin_call() (task 4.3,
|
||||
* controller_connector.c) -- defer it while a job-state
|
||||
* reconciliation report is still in flight rather than
|
||||
* interleaving with it. Unlike the vals==NULL/valptrs==NULL
|
||||
* skip branches below (genuine local failures, where resetting
|
||||
* the counter avoids a tight failure-retry loop), this branch
|
||||
* deliberately does NOT reset load_report_counter: the deferral
|
||||
* is caused by an external condition outside this loop's
|
||||
* control, of unknown duration, so the threshold stays crossed
|
||||
* and the very next tick re-checks immediately once
|
||||
* xcecd_resync_is_pending() clears -- resuming the send with no
|
||||
* extra retry/backoff delay, rather than making a resync that
|
||||
* just finished wait out a fresh full interval on top. */
|
||||
if (xcecd_resync_is_pending()) {
|
||||
XCECD_LOG(XCE_LOG_DEBUG,
|
||||
"resource_sync_thread: deferring load-indices "
|
||||
"send -- job-state resync report still being "
|
||||
"sent");
|
||||
sleep(SYNC_INTERVAL);
|
||||
continue;
|
||||
}
|
||||
|
||||
vals = collect_load_indices(&vals_count);
|
||||
if (vals == NULL || vals_count <= 0) {
|
||||
XCECD_LOG(XCE_LOG_ERROR, "No load indices to report");
|
||||
|
|
@ -381,7 +417,17 @@ void resource_sync_destroy(resource_sync_t* sync)
|
|||
XCECD_LOG(XCE_LOG_ERROR, "Invalid resource sync module");
|
||||
return;
|
||||
}
|
||||
resource_sync_stop(sync);
|
||||
/* bug-1756: resource_sync_stop() joins sync->thread but never resets the
|
||||
* handle afterward, so an unconditional second call here double-joins
|
||||
* the same pthread_t -- UB, caught by ASAN as "Joining already joined
|
||||
* thread, aborting." xcecd.c's own shutdown sequence (xcecd_cleanup())
|
||||
* calls resource_sync_stop() then resource_sync_destroy() back-to-back,
|
||||
* hitting this exact hazard on every daemon shutdown. Guard the same
|
||||
* way connection_handler.c's network_handler_destroy() already guards
|
||||
* its own analogous stop-then-destroy call (`if (handler->running)`). */
|
||||
if (atomic_load(&sync->running)) {
|
||||
resource_sync_stop(sync);
|
||||
}
|
||||
resource_sync_cleanup_load_indices();
|
||||
XCECD_LOG(XCE_LOG_DEBUG, "Resource sync module destroyed successfully");
|
||||
FREEUP(sync);
|
||||
|
|
|
|||
349
src/worker/src/xcecd_resync.c
Normal file
349
src/worker/src/xcecd_resync.c
Normal file
|
|
@ -0,0 +1,349 @@
|
|||
#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;
|
||||
}
|
||||
|
||||
/* worker-rejoin-resync task 4.3 (Requirements 3.8, 8.1, 8.3). See
|
||||
* xcecd_resync.h for the full contract of both symbols below.
|
||||
*
|
||||
* `resync_pending` mirrors controller_connector.c's `needs_rejoin` exactly:
|
||||
* a single file-scope static bool, set by the event it gates on, read
|
||||
* through a small accessor, cleared once the gated work is done. */
|
||||
static bool resync_pending = false;
|
||||
|
||||
/* Accumulates one incoming query's pages (Requirement 8.1/8.3) between
|
||||
* calls to xcecd_resync_handle_query_page() -- reset whenever a page with
|
||||
* pos == 0 arrives, so a stale/incomplete accumulation from an aborted
|
||||
* prior query (e.g. the worker itself disconnected mid-query) is always
|
||||
* discarded by the next query's first page rather than corrupting it. */
|
||||
static uint64_t* query_accum = NULL;
|
||||
static int query_accum_ntotal = 0;
|
||||
static int query_accum_received = 0;
|
||||
|
||||
static void reset_query_accum(void)
|
||||
{
|
||||
free(query_accum);
|
||||
query_accum = NULL;
|
||||
query_accum_ntotal = 0;
|
||||
query_accum_received = 0;
|
||||
}
|
||||
|
||||
bool xcecd_resync_is_pending(void) { return resync_pending; }
|
||||
|
||||
/* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8/6.1):
|
||||
* bounds resync_pending's worst case. xcecd_resync_handle_query_page() below
|
||||
* can only clear the gate by being called again with a query's LAST page --
|
||||
* if the worker's connection to the controller drops after a non-final page
|
||||
* has already arrived, no further page will ever arrive on that now-dead
|
||||
* connection, and resync_pending would otherwise stay stuck true forever
|
||||
* with nothing left to clear it. That was round 1's confirmed defect: it
|
||||
* permanently disabled checking_connection()'s entry-guard-gated body,
|
||||
* including the worker's own liveness/reconnect-detection -- the exact
|
||||
* mechanism src/lib/xcectx.c's handle_disconnect() documents as the sole
|
||||
* recovery path ("so that checking_connection can detect the failure and
|
||||
* attempt to rejoin").
|
||||
*
|
||||
* The worker's own reconnect-detection (xcecd_on_reconnect(),
|
||||
* controller_connector.c -- fired by the client library the moment it
|
||||
* transparently reconnects) is exactly the signal that makes any such
|
||||
* in-flight query moot: the connection it arrived on is already gone, and
|
||||
* the controller's own Requirement 6.1 Acceptance Criterion 1 already
|
||||
* discards that incomplete reconciliation round and re-initiates a
|
||||
* brand-new query (starting again at pos == 0) on this worker's next
|
||||
* rejoin. There is nothing worth preserving from the old, abandoned
|
||||
* accumulation -- discard it and clear the gate immediately, the same way
|
||||
* a malformed page already does above. Called from controller_connector.c's
|
||||
* xcecd_on_reconnect() (see that function's own comment for why this call
|
||||
* site is safe under its "flag-only, no I/O, no xcecli_*" constraint: this
|
||||
* function only touches this module's own statics -- a bool and a small
|
||||
* heap buffer -- no socket/file I/O and no call back into the client
|
||||
* library). */
|
||||
void xcecd_resync_reset(void)
|
||||
{
|
||||
reset_query_accum();
|
||||
resync_pending = false;
|
||||
}
|
||||
|
||||
int xcecd_resync_handle_query_page(xcecd_worker_t* worker_list, endpoint_t* ep,
|
||||
xce_peerinfo_t* from,
|
||||
const xce_job_state_query_t* page,
|
||||
uint64_t now_ms, int pktlimit)
|
||||
{
|
||||
int rc;
|
||||
|
||||
if (page == NULL || ep == NULL || from == NULL) {
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
/* Requirement 3.8: set the moment a query is received -- the first
|
||||
* page, even before this page (or any later one) is validated, so the
|
||||
* worker's own periodic post-join activity never interleaves with any
|
||||
* part of receiving/answering this exchange. Setting it again on every
|
||||
* later page of the same query is harmless (already true). */
|
||||
resync_pending = true;
|
||||
|
||||
if (page->pos == 0) {
|
||||
reset_query_accum();
|
||||
|
||||
if (page->n_total > 0) {
|
||||
query_accum = (uint64_t*)calloc((size_t)page->n_total,
|
||||
sizeof(*query_accum));
|
||||
if (query_accum == NULL) {
|
||||
resync_pending = false;
|
||||
return -ENOMEM;
|
||||
}
|
||||
}
|
||||
|
||||
query_accum_ntotal = page->n_total;
|
||||
}
|
||||
|
||||
if (page->count > 0) {
|
||||
if (page->jobids == NULL || query_accum == NULL || page->pos < 0
|
||||
|| page->count < 0
|
||||
|| (page->pos + page->count) > query_accum_ntotal) {
|
||||
reset_query_accum();
|
||||
resync_pending = false;
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
memcpy(&query_accum[page->pos], page->jobids,
|
||||
(size_t)page->count * sizeof(*query_accum));
|
||||
query_accum_received += page->count;
|
||||
}
|
||||
|
||||
if (query_accum_received < query_accum_ntotal) {
|
||||
/* More pages of this query still expected -- gate stays set,
|
||||
* nothing sent yet. */
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Last page: the query is fully assembled -- answer it for real. */
|
||||
rc = xcecd_resync_answer_query(worker_list, ep, from, query_accum,
|
||||
query_accum_ntotal, now_ms, pktlimit);
|
||||
|
||||
reset_query_accum();
|
||||
|
||||
/* Requirement 3.8: clear only once the full report -- queried and
|
||||
* extra pages alike -- has finished sending, whether or not the send
|
||||
* itself succeeded. A transport error must not leave the worker's own
|
||||
* periodic activity gated forever; the controller's own bounded
|
||||
* reconciliation timeout (Requirement 5.3) resolves an incomplete
|
||||
* round from its side, and this gate is purely local/non-blocking. */
|
||||
resync_pending = false;
|
||||
|
||||
return (rc < 0) ? rc : 1;
|
||||
}
|
||||
|
|
@ -8,8 +8,80 @@
|
|||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
/* worker-rejoin-resync task 4.1 (Requirements 3.4, 3.5, 3.7): bounded
|
||||
* finished-job retention. See xcecd_worker_utils.h for the public API and
|
||||
* data-model doc comments. */
|
||||
|
||||
/* Placeholder default until the worker daemon gains a real delivery path
|
||||
* for ClusterConfig.workerResyncWindowMs (see
|
||||
* xcecd_finished_job_retention_set_window_ms()'s header comment for what
|
||||
* was confirmed missing). Not derived from any existing constant -- there
|
||||
* is no established codebase precedent for this specific field's default
|
||||
* (unlike workerResyncTimeoutMs's own resync.c-local 15000ms fallback,
|
||||
* task 3.1, a distinct sibling field on the controller side). */
|
||||
#define XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS 60000ULL
|
||||
|
||||
static xcecd_finished_job_t* g_finished_job_retention = NULL;
|
||||
static uint64_t g_finished_job_retention_window_ms
|
||||
= XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS;
|
||||
|
||||
static bool finished_job_entry_expired(const xcecd_finished_job_t* e,
|
||||
uint64_t now_ms, uint64_t window_ms)
|
||||
{
|
||||
if (now_ms < e->finish_time_ms) {
|
||||
/* now_ms predates the entry's own finish time (clock skew, or a
|
||||
* caller/test passing an earlier now_ms) -- treat as not yet
|
||||
* expired rather than underflow the unsigned subtraction below. */
|
||||
return false;
|
||||
}
|
||||
|
||||
return (now_ms - e->finish_time_ms) > window_ms;
|
||||
}
|
||||
|
||||
/* Drops every entry older than window_ms as of now_ms. This is the lazy
|
||||
* expiry-on-query strategy documented in xcecd_worker_utils.h -- called
|
||||
* from both xcecd_retain_finished_job() and xcecd_finished_job_lookup()
|
||||
* so the store never grows unbounded even without an external periodic
|
||||
* driver, without needing a new call site in cleanup_worker_list() or
|
||||
* anywhere else. */
|
||||
static void finished_job_retention_prune(uint64_t now_ms, uint64_t window_ms)
|
||||
{
|
||||
xcecd_finished_job_t** cur = &g_finished_job_retention;
|
||||
|
||||
while (*cur) {
|
||||
xcecd_finished_job_t* e = *cur;
|
||||
|
||||
if (finished_job_entry_expired(e, now_ms, window_ms)) {
|
||||
*cur = e->next;
|
||||
FREEUP(e);
|
||||
} else {
|
||||
cur = &e->next;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Wall-clock (epoch) milliseconds -- used to capture finish_time_ms at
|
||||
* genuine job-exit-detection time in update_worker_status() below.
|
||||
* Deliberately wall-clock, not monotonic (unlike delay.c's private
|
||||
* monotonic_ms()/resync.c's resync_monotonic_ms(), which time an
|
||||
* in-process deadline): finish_time_ms is reported to the controller and
|
||||
* recorded there as the job's authoritative finish time (Requirements 3.5,
|
||||
* 4.2), so it must be a real, cross-process-meaningful timestamp, not an
|
||||
* arbitrary-origin monotonic value. */
|
||||
static uint64_t finished_job_wallclock_now_ms(void)
|
||||
{
|
||||
struct timespec ts;
|
||||
|
||||
if (clock_gettime(CLOCK_REALTIME, &ts) != 0) {
|
||||
return (uint64_t)time(NULL) * 1000ULL;
|
||||
}
|
||||
|
||||
return (uint64_t)ts.tv_sec * 1000ULL + (uint64_t)ts.tv_nsec / 1000000ULL;
|
||||
}
|
||||
|
||||
void add_worker(xcecd_worker_t** worker_list, int worker_pid, int worker_status,
|
||||
uint64_t job_id, int ntasks, job_startinfo_t* startinfo)
|
||||
{
|
||||
|
|
@ -300,6 +372,17 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
|
|||
xcecd_job_finished(job);
|
||||
/* FIXME: add error handling when msg send failure */
|
||||
|
||||
/* worker-rejoin-resync task 4.1 (Requirements 3.4, 3.5):
|
||||
* this is the exact moment the job's exit is detected --
|
||||
* capture its true outcome into the bounded finished-job
|
||||
* retention BEFORE cleanup_worker_list() reaps this
|
||||
* xcecd_worker_t entry on its next pass. Does not change
|
||||
* cleanup_worker_list()'s own reaping behavior (unmodified,
|
||||
* still keyed on worker->worker_status below). */
|
||||
xcecd_retain_finished_job(job->job_id, WORKER_DONE,
|
||||
job->exit_code,
|
||||
finished_job_wallclock_now_ms());
|
||||
|
||||
if (db
|
||||
&& !xcecd_db_update_worker(db, worker->worker_pid,
|
||||
WORKER_DONE, JOB_DONE)) {
|
||||
|
|
@ -319,6 +402,13 @@ int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
|
|||
jobidStr);
|
||||
xcecd_job_finished(job);
|
||||
|
||||
/* worker-rejoin-resync task 4.1 (Requirements 3.4, 3.5) --
|
||||
* see the WORKER_DONE case above for the full rationale;
|
||||
* identical hook, failure outcome. */
|
||||
xcecd_retain_finished_job(job->job_id, WORKER_EXITED,
|
||||
job->exit_code,
|
||||
finished_job_wallclock_now_ms());
|
||||
|
||||
if (db
|
||||
&& !xcecd_db_update_worker(db, worker->worker_pid,
|
||||
WORKER_EXITED, JOB_EXITED)) {
|
||||
|
|
@ -408,6 +498,148 @@ void cleanup_worker_list(xcecd_worker_t** worker_list, sqlite3* db)
|
|||
}
|
||||
}
|
||||
|
||||
void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code,
|
||||
uint64_t finish_time_ms)
|
||||
{
|
||||
char jobidStr[64];
|
||||
xcecd_finished_job_t** cur = &g_finished_job_retention;
|
||||
xcecd_finished_job_t* entry = NULL;
|
||||
|
||||
/* Prune here too, not just on lookup: xcecd_finished_job_lookup() is
|
||||
* only reached when this worker is queried on its own rejoin, which on
|
||||
* a long-uptime, well-connected worker may be rare relative to how
|
||||
* often jobs finish. Pruning on every retain as well keeps the store
|
||||
* bounded by the retention window regardless of query frequency. */
|
||||
finished_job_retention_prune(finish_time_ms,
|
||||
g_finished_job_retention_window_ms);
|
||||
|
||||
/* Replace any stale entry for this jobid first, so the store holds at
|
||||
* most one entry per jobid. */
|
||||
while (*cur) {
|
||||
if ((*cur)->jobid == jobid) {
|
||||
xcecd_finished_job_t* stale = *cur;
|
||||
*cur = stale->next;
|
||||
FREEUP(stale);
|
||||
break;
|
||||
}
|
||||
cur = &(*cur)->next;
|
||||
}
|
||||
|
||||
entry = (xcecd_finished_job_t*)calloc(1, sizeof(xcecd_finished_job_t));
|
||||
if (!entry) {
|
||||
XCECD_LOG(XCE_LOG_ERROR,
|
||||
"Failed to allocate memory for finished-job retention "
|
||||
"entry");
|
||||
return;
|
||||
}
|
||||
|
||||
entry->jobid = jobid;
|
||||
entry->exit_status = exit_status;
|
||||
entry->exit_code = exit_code;
|
||||
entry->finish_time_ms = finish_time_ms;
|
||||
|
||||
entry->next = g_finished_job_retention;
|
||||
g_finished_job_retention = entry;
|
||||
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
XCECD_LOG(XCE_LOG_DEBUG,
|
||||
"Retaining finished-job outcome for job %s (exit_status=%d, "
|
||||
"exit_code=%d, finish_time_ms=%llu)",
|
||||
jobidStr, exit_status, exit_code,
|
||||
(unsigned long long)finish_time_ms);
|
||||
}
|
||||
|
||||
bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms,
|
||||
xcecd_finished_job_t* out)
|
||||
{
|
||||
xcecd_finished_job_t* e;
|
||||
|
||||
finished_job_retention_prune(now_ms, g_finished_job_retention_window_ms);
|
||||
|
||||
e = g_finished_job_retention;
|
||||
while (e) {
|
||||
if (e->jobid == jobid) {
|
||||
if (out) {
|
||||
*out = *e;
|
||||
out->next = NULL;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
e = e->next;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms)
|
||||
{
|
||||
g_finished_job_retention_window_ms
|
||||
= (window_ms > 0) ? window_ms
|
||||
: XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS;
|
||||
}
|
||||
|
||||
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;
|
||||
|
||||
while (e) {
|
||||
xcecd_finished_job_t* next = e->next;
|
||||
FREEUP(e);
|
||||
e = next;
|
||||
}
|
||||
|
||||
g_finished_job_retention = NULL;
|
||||
g_finished_job_retention_window_ms = XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS;
|
||||
}
|
||||
|
||||
xcecd_worker_t* find_worker_by_pid(void* worker_list, int pid)
|
||||
{
|
||||
XCECD_LOG(XCE_LOG_DEBUG, "Finding worker %d in worker list", pid);
|
||||
|
|
|
|||
|
|
@ -97,3 +97,33 @@ test('worker uv_ipc job_exit_severity test', uv_ipc_job_exit_severity_test,
|
|||
suite: 'worker',
|
||||
timeout: 60,
|
||||
env: ['ASAN_OPTIONS=detect_leaks=0'])
|
||||
|
||||
# worker-rejoin-resync task 4.1 (Requirements 3.4, 3.5, 3.7): bounded
|
||||
# finished-job retention -- white-box #include of xcecd_worker_utils.c,
|
||||
# same convention as resource_sync_test above.
|
||||
finished_job_retention_test = executable('finished_job_retention_test',
|
||||
'test_finished_job_retention.c',
|
||||
include_directories: [worker_test_incdir],
|
||||
dependencies: worker_test_deps,
|
||||
install: false)
|
||||
|
||||
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'])
|
||||
|
|
|
|||
|
|
@ -142,6 +142,70 @@ bool resend_job_status(xcecd_worker_t* worker_list)
|
|||
return true;
|
||||
}
|
||||
|
||||
/* worker-rejoin-resync task 4.3 (Requirements 3.8, 8.1, 8.3): link-
|
||||
* satisfying stubs for xcecd_resync.c's public API -- this file's white-box
|
||||
* target is controller_connector.c only, it does not #include
|
||||
* xcecd_resync.c (that module's own resync_pending lifecycle and
|
||||
* multi-page query reassembly are covered against the real production
|
||||
* code by test_xcecd_resync.c). g_test_resync_pending lets
|
||||
* checking_connection()'s gate-check tests below control what the gate
|
||||
* reads without a real xcecd_resync.c backing it; the call-count/last-page
|
||||
* globals let the dispatch-wiring tests below observe that myreq_handler
|
||||
* actually reaches this function, without needing a real report to be
|
||||
* built and sent. */
|
||||
static bool g_test_resync_pending = false;
|
||||
static int g_resync_handle_query_page_call_count = 0;
|
||||
static const xce_job_state_query_t* g_resync_handle_query_page_last_page = NULL;
|
||||
/* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
|
||||
* 6.1): lets the tests below observe that xcecd_on_reconnect() actually
|
||||
* calls xcecd_resync_reset() (the round-2 fix that bounds a stalled
|
||||
* multi-page query's worst case), and lets a deferred-rejoin test simulate
|
||||
* the gate clearing on its own, independent of checking_connection() ever
|
||||
* touching it directly. */
|
||||
static int g_resync_reset_call_count = 0;
|
||||
|
||||
/* worker-rejoin-resync task 6.1 (Requirement 3.7): link-satisfying spy stub
|
||||
* for xcecd_worker_utils.c's real retention-window setter -- this file's
|
||||
* white-box target is controller_connector.c only (does not #include
|
||||
* xcecd_worker_utils.c), same rationale as the xcecd_resync.c stubs above.
|
||||
* Lets the dispatch-wiring tests below observe that handle_pushconfig()
|
||||
* (myreq_handler's new OPCode_push case) actually reaches this function
|
||||
* with the decoded value, without needing a real xcecd_worker_utils.c
|
||||
* retention store backing it. */
|
||||
static int g_retention_set_call_count;
|
||||
static uint64_t g_retention_window_ms_set;
|
||||
|
||||
void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms)
|
||||
{
|
||||
g_retention_set_call_count++;
|
||||
g_retention_window_ms_set = window_ms;
|
||||
}
|
||||
|
||||
bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
|
||||
|
||||
void xcecd_resync_reset(void)
|
||||
{
|
||||
g_resync_reset_call_count++;
|
||||
g_test_resync_pending = false;
|
||||
}
|
||||
|
||||
int xcecd_resync_handle_query_page(xcecd_worker_t* worker_list, endpoint_t* ep,
|
||||
xce_peerinfo_t* from,
|
||||
const xce_job_state_query_t* page,
|
||||
uint64_t now_ms, int pktlimit)
|
||||
{
|
||||
(void)worker_list;
|
||||
(void)ep;
|
||||
(void)from;
|
||||
(void)now_ms;
|
||||
(void)pktlimit;
|
||||
|
||||
g_resync_handle_query_page_call_count++;
|
||||
g_resync_handle_query_page_last_page = page;
|
||||
|
||||
return (page != NULL) ? 1 : -EINVAL;
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Fixtures
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
|
@ -155,6 +219,17 @@ void setUp(void)
|
|||
/* Restore the real production call by default (task 2.2) -- individual
|
||||
* rejoin-retry tests below override this to a controllable double. */
|
||||
rejoin_call = perform_pending_rejoin;
|
||||
|
||||
/* worker-rejoin-resync task 4.3: default the resync gate to "not
|
||||
* pending" and reset the dispatch-wiring spy globals. */
|
||||
g_test_resync_pending = false;
|
||||
g_resync_handle_query_page_call_count = 0;
|
||||
g_resync_handle_query_page_last_page = NULL;
|
||||
g_resync_reset_call_count = 0;
|
||||
|
||||
/* worker-rejoin-resync task 6.1: reset the retention-window setter spy. */
|
||||
g_retention_set_call_count = 0;
|
||||
g_retention_window_ms_set = 0;
|
||||
}
|
||||
|
||||
void tearDown(void) { return; }
|
||||
|
|
@ -284,6 +359,35 @@ static void test_rejoin_callback_sets_flag_and_returns_immediately(void)
|
|||
printf("test_rejoin_callback_sets_flag_and_returns_immediately: PASS\n");
|
||||
}
|
||||
|
||||
/* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
|
||||
* 6.1): the reconnect callback must also reset xcecd_resync.c's gate, not
|
||||
* just set needs_rejoin -- this is what bounds a stalled multi-page
|
||||
* query's worst case (see xcecd_resync_reset()'s own doc comment). Without
|
||||
* this call, a query whose connection dropped mid-reassembly would leave
|
||||
* resync_pending stuck true forever with nothing left to clear it. */
|
||||
static void test_rejoin_callback_also_resets_resync_gate(void)
|
||||
{
|
||||
g_test_resync_pending = true;
|
||||
g_resync_reset_call_count = 0;
|
||||
|
||||
int rc = xcecd_on_reconnect((xcecli_ctx_t)0x1, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_TRUE(needs_rejoin);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, g_resync_reset_call_count,
|
||||
"xcecd_on_reconnect() must call xcecd_resync_reset() exactly once "
|
||||
"so a resync gate left stuck by a stalled multi-page query is "
|
||||
"always cleared the instant the worker detects it has "
|
||||
"reconnected");
|
||||
TEST_ASSERT_FALSE_MESSAGE(
|
||||
g_test_resync_pending,
|
||||
"the resync gate must actually be cleared, not merely have its "
|
||||
"reset function called");
|
||||
|
||||
printf("test_rejoin_callback_also_resets_resync_gate: PASS\n");
|
||||
}
|
||||
|
||||
static void test_rejoin_callback_registered_via_public_api(void)
|
||||
{
|
||||
/* Register through the real public entry point the production code
|
||||
|
|
@ -591,6 +695,421 @@ test_checking_connection_rejoin_retried_on_next_probe_after_failure(void)
|
|||
"failure: PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Task 4.3 (worker-rejoin-resync, Requirements 3.8, 8.1, 8.3) — gating the
|
||||
* needs_rejoin-triggered rejoin action on xcecd_resync.c's resync_pending
|
||||
* flag (the "resume normal post-join operation" step Requirement 3.8
|
||||
* actually names), and wiring OPCode_jobStateQuery's real receipt.
|
||||
*
|
||||
* Round-2 remediation note: round 1 gated checking_connection()'s ENTRY on
|
||||
* resync_pending, which also silently skipped xcecli_hello() and the
|
||||
* needs_rejoin check itself -- confirmed to permanently disable the
|
||||
* worker's own connection-recovery mechanism whenever resync_pending got
|
||||
* stuck true (a stalled multi-page query, e.g.). The gate now sits INSIDE
|
||||
* the needs_rejoin branch, guarding only rejoin_call(); see
|
||||
* test_checking_connection_survives_permanently_stuck_resync_pending() and
|
||||
* test_checking_connection_hello_and_needs_rejoin_precede_resync_gate_in_
|
||||
* source() further below for the regression coverage of that distinction.
|
||||
*
|
||||
* What THIS file proves is the two things that are genuinely
|
||||
* controller_connector.c's own: (1) checking_connection() actually
|
||||
* consults the gate before attempting the rejoin -- the same way it
|
||||
* already consults needs_rejoin -- and (2) myreq_handler's dispatch switch
|
||||
* actually routes OPCode_jobStateQuery to the resync module at all, the
|
||||
* exact gap task 4.2 left unclosed (confirmed by inspection: no case
|
||||
* existed for this opcode before this task). xcecd_resync.c's own
|
||||
* resync_pending set/clear lifecycle and multi-page query reassembly are
|
||||
* covered end-to-end, against the real production code and real wire
|
||||
* encode/decode, by test_xcecd_resync.c -- not duplicated here.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static void
|
||||
test_checking_connection_defers_rejoin_action_when_resync_pending(void)
|
||||
{
|
||||
controller_connector_ctx_t fake_cc = { 0 };
|
||||
fake_cc.running = true;
|
||||
cc_ctx = &fake_cc;
|
||||
xcecli_ctx = NULL;
|
||||
|
||||
needs_rejoin = true; /* would normally trigger a rejoin attempt */
|
||||
g_rejoin_call_count = 0;
|
||||
rejoin_call = fake_rejoin_always_succeeds;
|
||||
g_test_resync_pending = true;
|
||||
|
||||
checking_connection();
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_rejoin_call_count,
|
||||
"checking_connection() must not attempt a rejoin while a job-state "
|
||||
"resync report is still being sent (Req 3.8) -- the worker's "
|
||||
"resumption of normal post-join operation must not begin or "
|
||||
"interleave with it");
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
needs_rejoin,
|
||||
"a deferred tick must leave needs_rejoin untouched, not clear it "
|
||||
"as if a rejoin had been attempted");
|
||||
|
||||
printf(
|
||||
"test_checking_connection_defers_rejoin_action_when_resync_pending: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
static void test_checking_connection_resumes_once_resync_no_longer_pending(void)
|
||||
{
|
||||
controller_connector_ctx_t fake_cc = { 0 };
|
||||
fake_cc.running = true;
|
||||
cc_ctx = &fake_cc;
|
||||
xcecli_ctx = NULL;
|
||||
|
||||
needs_rejoin = true;
|
||||
g_rejoin_call_count = 0;
|
||||
rejoin_call = fake_rejoin_always_succeeds;
|
||||
g_test_resync_pending = false; /* report already finished sending */
|
||||
|
||||
checking_connection();
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, g_rejoin_call_count,
|
||||
"once the gate clears, checking_connection() must resume its "
|
||||
"normal periodic activity on its very next tick");
|
||||
TEST_ASSERT_FALSE(needs_rejoin);
|
||||
|
||||
printf("test_checking_connection_resumes_once_resync_no_longer_pending: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Round-2 remediation regression guard (Requirements 3.8, 6.1): FINDING 1
|
||||
* of the round-1 review -- gating checking_connection()'s ENTRY on
|
||||
* resync_pending permanently disabled the worker's own connection-recovery
|
||||
* mechanism whenever a multi-page job-state query's final page could never
|
||||
* arrive (e.g. the worker's own connection to the controller dropped after
|
||||
* a non-final page). resync_pending must now guard ONLY the
|
||||
* needs_rejoin-triggered rejoin action -- xcecli_hello() (the worker's
|
||||
* liveness probe / reconnect-detection trigger) and the needs_rejoin check
|
||||
* itself must keep running every single tick regardless, extending
|
||||
* test_checking_connection_never_permanently_disables_worker's exact
|
||||
* pattern (many ticks, must never get permanently stuck) into this new
|
||||
* adversarial condition: resync_pending genuinely stuck true (simulating a
|
||||
* stalled multi-page query whose final page never arrives) for the whole
|
||||
* loop, never cleared by anything inside checking_connection() itself.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static void
|
||||
test_checking_connection_survives_permanently_stuck_resync_pending(void)
|
||||
{
|
||||
controller_connector_ctx_t fake_cc = { 0 };
|
||||
fake_cc.running = true;
|
||||
cc_ctx = &fake_cc;
|
||||
/* No connection at all -> xcecli_hello(NULL, ...) fails immediately
|
||||
* with -EINVAL, same fast/deterministic setup as
|
||||
* test_checking_connection_never_permanently_disables_worker. */
|
||||
xcecli_ctx = NULL;
|
||||
|
||||
needs_rejoin = true; /* a rejoin is genuinely pending throughout */
|
||||
g_rejoin_call_count = 0;
|
||||
rejoin_call = fake_rejoin_always_succeeds;
|
||||
|
||||
/* Simulates a multi-page job-state query that stalled after a
|
||||
* non-final page (the worker's own connection dropped mid-query, so no
|
||||
* further page -- and therefore no call into
|
||||
* xcecd_resync_handle_query_page() that could ever clear the gate --
|
||||
* will ever arrive). Left true for the entire loop below: nothing in
|
||||
* this test ever clears it, matching the round-1 defect's worst case
|
||||
* exactly. */
|
||||
g_test_resync_pending = true;
|
||||
|
||||
struct timespec t0, t1;
|
||||
clock_gettime(CLOCK_MONOTONIC, &t0);
|
||||
|
||||
/* Far more iterations than the old MAX_RETRY_COUNT (5) -- see
|
||||
* test_checking_connection_never_permanently_disables_worker. Every
|
||||
* single tick must still leave the worker able to detect/recover its
|
||||
* own connection (cc_ctx->running never forced false), and the pending
|
||||
* rejoin must never be silently dropped (needs_rejoin stays set, exactly
|
||||
* because it is legitimately still deferred by the still-pending
|
||||
* resync, not because the worker gave up). */
|
||||
for (int i = 0; i < 25; i++) {
|
||||
checking_connection();
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
cc_ctx->running,
|
||||
"a permanently-stuck resync_pending must never permanently "
|
||||
"disable the worker -- checking_connection()'s liveness probe "
|
||||
"and needs_rejoin check must keep running every tick "
|
||||
"regardless (Requirement 6.1)");
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
needs_rejoin,
|
||||
"the rejoin action stays correctly deferred tick after tick "
|
||||
"while resync_pending is genuinely stuck true -- it must "
|
||||
"never be silently cleared without actually being attempted");
|
||||
}
|
||||
|
||||
clock_gettime(CLOCK_MONOTONIC, &t1);
|
||||
long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L
|
||||
+ (t1.tv_nsec - t0.tv_nsec) / 1000000L;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_rejoin_call_count,
|
||||
"the rejoin action itself must stay deferred for as long as "
|
||||
"resync_pending is genuinely (if pathologically) stuck true -- "
|
||||
"narrowing the gate must not mean it stops applying");
|
||||
|
||||
/* 25 ticks completing well under a second proves this loop never fell
|
||||
* into any blocking wait -- same bound as the sibling invariant test. */
|
||||
TEST_ASSERT_LESS_THAN_INT(1000, elapsed_ms);
|
||||
|
||||
printf(
|
||||
"test_checking_connection_survives_permanently_stuck_resync_"
|
||||
"pending: PASS (elapsed_ms=%ld)\n",
|
||||
elapsed_ms);
|
||||
}
|
||||
|
||||
/* A dummy backing buffer large enough for the real api_ctx_t
|
||||
* (client_internal.h) that xcecli_get_my_endpoint()/xcecli_get_myinfo()
|
||||
* (real functions, src/lib/client.c) cast xcecli_ctx to internally --
|
||||
* this file must not include client_internal.h itself (its api_ctx_t
|
||||
* conflicts with controller_connector.h's own unused same-named type, see
|
||||
* this file's top-of-file comment), but neither accessor dereferences
|
||||
* anything through the returned pointer in THIS translation unit, since
|
||||
* xcecd_resync_handle_query_page() is a stub here that never dereferences
|
||||
* its ep/from parameters either -- only their addresses need to be
|
||||
* computable, which pure pointer arithmetic over a large enough zeroed
|
||||
* buffer safely provides regardless of the real struct's exact layout. */
|
||||
static uint8_t g_dummy_api_ctx[8192];
|
||||
|
||||
/* Req 8.1/8.3: myreq_handler's OPCode_jobStateQuery case must route an
|
||||
* incoming query page to xcecd_resync_handle_query_page(). */
|
||||
static void test_myreq_handler_routes_jobstatequery_to_resync_module(void)
|
||||
{
|
||||
controller_connector_ctx_t fake_cc = { 0 };
|
||||
void* worker_list_head = NULL;
|
||||
fake_cc.running = true;
|
||||
fake_cc.worker_list = &worker_list_head;
|
||||
cc_ctx = &fake_cc;
|
||||
|
||||
memset(g_dummy_api_ctx, 0, sizeof(g_dummy_api_ctx));
|
||||
xcecli_ctx = (xcecli_ctx_t)g_dummy_api_ctx;
|
||||
|
||||
uint64_t jobids[2] = { 42, 43 };
|
||||
xce_job_state_query_t page = { 0 };
|
||||
page.n_total = 2;
|
||||
page.pos = 0;
|
||||
page.count = 2;
|
||||
page.jobids = jobids;
|
||||
|
||||
xce_request_t request = { 0 };
|
||||
request.opcode = OPCode_jobStateQuery;
|
||||
request.body_which = Request_body_jobstatequerymsg;
|
||||
request.body.jobstatequerymsg = &page;
|
||||
|
||||
int rc = myreq_handler(&request, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, g_resync_handle_query_page_call_count,
|
||||
"OPCode_jobStateQuery must reach xcecd_resync_handle_query_page() "
|
||||
"exactly once per incoming page -- this is the real dispatch "
|
||||
"wiring task 4.2 left out (no case existed for this opcode before "
|
||||
"this task)");
|
||||
TEST_ASSERT_EQUAL_PTR_MESSAGE(
|
||||
&page, g_resync_handle_query_page_last_page,
|
||||
"the decoded query page must be passed through unchanged");
|
||||
|
||||
xcecli_ctx = NULL;
|
||||
printf("test_myreq_handler_routes_jobstatequery_to_resync_module: PASS\n");
|
||||
}
|
||||
|
||||
/* Without an active controller connection (xcecli_ctx == NULL), the
|
||||
* dispatch case must fail fast rather than call into the resync module
|
||||
* with a NULL endpoint/peerinfo. */
|
||||
static void
|
||||
test_myreq_handler_jobstatequery_without_connection_is_rejected(void)
|
||||
{
|
||||
controller_connector_ctx_t fake_cc = { 0 };
|
||||
void* worker_list_head = NULL;
|
||||
fake_cc.running = true;
|
||||
fake_cc.worker_list = &worker_list_head;
|
||||
cc_ctx = &fake_cc;
|
||||
xcecli_ctx = NULL;
|
||||
|
||||
uint64_t jobids[1] = { 1 };
|
||||
xce_job_state_query_t page = { 0 };
|
||||
page.n_total = 1;
|
||||
page.pos = 0;
|
||||
page.count = 1;
|
||||
page.jobids = jobids;
|
||||
|
||||
xce_request_t request = { 0 };
|
||||
request.opcode = OPCode_jobStateQuery;
|
||||
request.body_which = Request_body_jobstatequerymsg;
|
||||
request.body.jobstatequerymsg = &page;
|
||||
|
||||
int rc = myreq_handler(&request, NULL);
|
||||
|
||||
TEST_ASSERT_TRUE(rc < 0);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_resync_handle_query_page_call_count,
|
||||
"without an active controller connection, the resync module must "
|
||||
"never be called");
|
||||
|
||||
printf("test_myreq_handler_jobstatequery_without_connection_is_rejected: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* worker-rejoin-resync task 6.1 (Requirement 3.7): myreq_handler's new
|
||||
* OPCode_push case (handle_pushconfig) must apply a received
|
||||
* workerResyncWindowMs value to the retention window before the push
|
||||
* completes.
|
||||
*
|
||||
* Before this task, myreq_handler had NO case for OPCode_push at all --
|
||||
* confirmed by inspection (this file's own white-box #include of the real,
|
||||
* current controller_connector.c) -- so a config push fell through to
|
||||
* `default: break;` and was silently discarded in its entirety, even for
|
||||
* the pre-existing clustername field. xcecd_finished_job_retention_set_
|
||||
* window_ms()'s own doc comment (xcecd_worker_utils.h, task 4.1)
|
||||
* independently confirms the same gap. This is the first test coverage of
|
||||
* that dispatch case existing at all.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
/* Builds a real, wire-encoded OPCode_push carrying a Conf4Worker payload
|
||||
* with the given workerResyncWindowMs value, pops it back off the queue,
|
||||
* and decodes it with the real xce_decode() -- exercising the actual
|
||||
* capnp encode/decode round trip (controller.capnp.c, generated), not just
|
||||
* a directly-constructed xce_request_t. Mirrors src/lib/testsuite/
|
||||
* test_job_state_wire.c's own real-endpoint round-trip pattern. Caller
|
||||
* must xce_free_request(out) and endpoint_fini()/free() the endpoint. */
|
||||
static endpoint_t*
|
||||
build_and_decode_pushconfig(uint32_t worker_resync_window_ms,
|
||||
xce_request_t* out)
|
||||
{
|
||||
xce_conf4worker_t conf4worker = { 0 };
|
||||
conf4worker.clustername = (char*)"test-cluster";
|
||||
conf4worker.worker_resync_window_ms = worker_resync_window_ms;
|
||||
|
||||
xce_configdata_t configdata = { 0 };
|
||||
configdata.data_which = ConfigData_data_workerconfig;
|
||||
configdata.data.conf4worker = &conf4worker;
|
||||
|
||||
char nm[] = "test-controller";
|
||||
char ky[] = "test-key";
|
||||
xce_peerinfo_t from = { .name = nm, .key = ky };
|
||||
|
||||
endpoint_t* ep = endpoint_get();
|
||||
TEST_ASSERT_NOT_NULL(ep);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, xce_pushconfig_req(ep, &from, &configdata));
|
||||
|
||||
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));
|
||||
|
||||
return ep;
|
||||
}
|
||||
|
||||
/* A config push carrying a non-default workerResyncWindowMs value must be
|
||||
* applied to the retention window -- synchronously, within myreq_handler's
|
||||
* own call, i.e. before the push "completes" from the caller's point of
|
||||
* view (Requirement 3.7's own observable completion criterion). */
|
||||
static void
|
||||
test_myreq_handler_routes_pushconfig_workerconfig_to_retention_setter(void)
|
||||
{
|
||||
xce_request_t decoded;
|
||||
endpoint_t* ep = build_and_decode_pushconfig(4242, &decoded);
|
||||
|
||||
int rc = myreq_handler(&decoded, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, g_retention_set_call_count,
|
||||
"OPCode_push carrying a Conf4Worker payload must reach "
|
||||
"xcecd_finished_job_retention_set_window_ms() exactly once -- the "
|
||||
"dispatch case task 6.1 adds (no case existed for OPCode_push "
|
||||
"before this task)");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
|
||||
4242, g_retention_window_ms_set,
|
||||
"the applied value must be the real value decoded off the wire, "
|
||||
"not a hardcoded/default one");
|
||||
|
||||
xce_free_request(&decoded);
|
||||
endpoint_fini(ep);
|
||||
free(ep);
|
||||
printf("test_myreq_handler_routes_pushconfig_workerconfig_to_retention_"
|
||||
"setter: PASS\n");
|
||||
}
|
||||
|
||||
/* A second, different value must also be applied -- confirms the setter is
|
||||
* called with the field genuinely read from each push, not a fixed
|
||||
* constant from the first call. */
|
||||
static void
|
||||
test_myreq_handler_routes_pushconfig_second_value_also_applied(void)
|
||||
{
|
||||
xce_request_t decoded;
|
||||
endpoint_t* ep = build_and_decode_pushconfig(7, &decoded);
|
||||
|
||||
int rc = myreq_handler(&decoded, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(1, g_retention_set_call_count);
|
||||
TEST_ASSERT_EQUAL_UINT64(7, g_retention_window_ms_set);
|
||||
|
||||
xce_free_request(&decoded);
|
||||
endpoint_fini(ep);
|
||||
free(ep);
|
||||
printf("test_myreq_handler_routes_pushconfig_second_value_also_applied: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* A config push for a DIFFERENT ConfigData union arm (e.g. a scheduler's
|
||||
* Conf4Sched, or the raw ClusterConfig arm) must NOT be misread as a
|
||||
* Conf4Worker payload -- the retention setter must not fire at all. */
|
||||
static void test_myreq_handler_pushconfig_ignores_non_workerconfig_data(void)
|
||||
{
|
||||
cluster_config_t clusterconfig = { 0 };
|
||||
clusterconfig.clustername = (char*)"test-cluster";
|
||||
|
||||
xce_configdata_t configdata = { 0 };
|
||||
configdata.data_which = ConfigData_data_clusterconfig;
|
||||
configdata.data.clusterconfig = &clusterconfig;
|
||||
|
||||
xce_request_t request = { 0 };
|
||||
request.opcode = OPCode_push;
|
||||
request.body_which = Request_body_configpushmsg;
|
||||
request.body.configpushmsg = &configdata;
|
||||
|
||||
int rc = myreq_handler(&request, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_retention_set_call_count,
|
||||
"a non-workerconfig ConfigData arm must never reach the retention "
|
||||
"setter");
|
||||
|
||||
printf("test_myreq_handler_pushconfig_ignores_non_workerconfig_data: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* A NULL configpushmsg (defensive: should never happen for a genuinely
|
||||
* decoded OPCode_push, but the handler must not crash) must also be a
|
||||
* silent no-op. */
|
||||
static void test_myreq_handler_pushconfig_ignores_null_payload(void)
|
||||
{
|
||||
xce_request_t request = { 0 };
|
||||
request.opcode = OPCode_push;
|
||||
request.body_which = Request_body_configpushmsg;
|
||||
request.body.configpushmsg = NULL;
|
||||
|
||||
int rc = myreq_handler(&request, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(0, g_retention_set_call_count);
|
||||
|
||||
printf("test_myreq_handler_pushconfig_ignores_null_payload: PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Task 2.3 — structural regression guard: xcecli_set_reconnect_handler()
|
||||
* must be called AFTER xcecli_start() succeeds, never before.
|
||||
|
|
@ -670,6 +1189,69 @@ test_reconnect_handler_registered_after_start_not_before_in_source(void)
|
|||
"source: PASS\n");
|
||||
}
|
||||
|
||||
/* Structural companion to
|
||||
* test_checking_connection_survives_permanently_stuck_resync_pending()
|
||||
* above: proves WHY checking_connection() can never again become
|
||||
* permanently inert the way round 1's entry-guard placement did. Same
|
||||
* source-text-scan genre as the xcecli_start()-before-
|
||||
* xcecli_set_reconnect_handler() guard just above, because the behavioral
|
||||
* test cannot, by itself, distinguish "the gate is checked first and
|
||||
* returns early" from "the gate is checked last and defers only the
|
||||
* rejoin action" when needs_rejoin/resync_pending start out already true
|
||||
* in both cases -- both produce the same rejoin_call_count/needs_rejoin
|
||||
* observations. The distinction is textual: xcecli_hello() and the
|
||||
* needs_rejoin check must appear in checking_connection()'s body strictly
|
||||
* BEFORE any xcecd_resync_is_pending() call, never gated behind it. */
|
||||
static void
|
||||
test_checking_connection_hello_and_needs_rejoin_precede_resync_gate_in_source(
|
||||
void)
|
||||
{
|
||||
char* src = read_whole_file(XCE_CONTROLLER_CONNECTOR_C_PATH);
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(
|
||||
src, "could not read controller_connector.c source for scan");
|
||||
|
||||
const char* fn_start = strstr(src, "int checking_connection()");
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(fn_start,
|
||||
"checking_connection() definition not found");
|
||||
|
||||
/* Search only from the function definition onward, so an unrelated,
|
||||
* earlier mention of any of these tokens elsewhere in the file (e.g. in
|
||||
* a doc comment) cannot produce a false pass. */
|
||||
const char* hello_call = strstr(fn_start, "xcecli_hello(xcecli_ctx");
|
||||
const char* needs_rejoin_check = strstr(fn_start, "if (needs_rejoin)");
|
||||
const char* pending_check
|
||||
= strstr(fn_start, "xcecd_resync_is_pending()");
|
||||
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(hello_call,
|
||||
"xcecli_hello(xcecli_ctx, ...) call site "
|
||||
"not found in checking_connection()");
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(
|
||||
needs_rejoin_check,
|
||||
"if (needs_rejoin) check not found in checking_connection()");
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(
|
||||
pending_check,
|
||||
"xcecd_resync_is_pending() call not found in checking_connection()");
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
hello_call < needs_rejoin_check,
|
||||
"xcecli_hello() must run unconditionally, before the needs_rejoin "
|
||||
"check -- never gated behind resync_pending (Requirement 6.1's "
|
||||
"self-healing liveness probe)");
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
needs_rejoin_check < pending_check,
|
||||
"the resync_pending gate must be checked INSIDE the needs_rejoin "
|
||||
"branch (guarding only the rejoin action itself), not before it -- "
|
||||
"an entry-guard placement (round 1's confirmed defect) would put "
|
||||
"xcecd_resync_is_pending() before both xcecli_hello() and the "
|
||||
"needs_rejoin check, permanently disabling the worker's own "
|
||||
"connection-recovery mechanism whenever resync_pending gets stuck");
|
||||
|
||||
FREEUP(src);
|
||||
printf(
|
||||
"test_checking_connection_hello_and_needs_rejoin_precede_resync_"
|
||||
"gate_in_source: PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Task 2.3 — Req 1.6/5.2: all-candidates-unreachable failure identifies
|
||||
* every attempted candidate, not a generic error.
|
||||
|
|
@ -861,6 +1443,7 @@ int main(void)
|
|||
test_build_cluster_config_single_entry_peers_matches_synthesized_shape);
|
||||
RUN_TEST(test_build_cluster_config_multi_candidate_passthrough);
|
||||
RUN_TEST(test_rejoin_callback_sets_flag_and_returns_immediately);
|
||||
RUN_TEST(test_rejoin_callback_also_resets_resync_gate);
|
||||
RUN_TEST(test_rejoin_callback_registered_via_public_api);
|
||||
RUN_TEST(test_checking_connection_never_permanently_disables_worker);
|
||||
RUN_TEST(test_checking_connection_skips_rejoin_when_not_pending);
|
||||
|
|
@ -870,8 +1453,22 @@ int main(void)
|
|||
test_checking_connection_rejoin_success_with_nonzero_index_clears_flag);
|
||||
RUN_TEST(
|
||||
test_checking_connection_rejoin_retried_on_next_probe_after_failure);
|
||||
RUN_TEST(
|
||||
test_checking_connection_defers_rejoin_action_when_resync_pending);
|
||||
RUN_TEST(test_checking_connection_resumes_once_resync_no_longer_pending);
|
||||
RUN_TEST(
|
||||
test_checking_connection_survives_permanently_stuck_resync_pending);
|
||||
RUN_TEST(test_myreq_handler_routes_jobstatequery_to_resync_module);
|
||||
RUN_TEST(test_myreq_handler_jobstatequery_without_connection_is_rejected);
|
||||
RUN_TEST(
|
||||
test_myreq_handler_routes_pushconfig_workerconfig_to_retention_setter);
|
||||
RUN_TEST(test_myreq_handler_routes_pushconfig_second_value_also_applied);
|
||||
RUN_TEST(test_myreq_handler_pushconfig_ignores_non_workerconfig_data);
|
||||
RUN_TEST(test_myreq_handler_pushconfig_ignores_null_payload);
|
||||
RUN_TEST(
|
||||
test_reconnect_handler_registered_after_start_not_before_in_source);
|
||||
RUN_TEST(
|
||||
test_checking_connection_hello_and_needs_rejoin_precede_resync_gate_in_source);
|
||||
RUN_TEST(test_format_candidate_list_names_every_candidate_in_order);
|
||||
RUN_TEST(test_xcecd_join_cluster_all_unreachable_reports_every_candidate);
|
||||
RUN_TEST(
|
||||
|
|
|
|||
341
src/worker/testsuite/test_finished_job_retention.c
Normal file
341
src/worker/testsuite/test_finished_job_retention.c
Normal file
|
|
@ -0,0 +1,341 @@
|
|||
/*
|
||||
* test_finished_job_retention.c -- worker-rejoin-resync task 4.1
|
||||
* (Requirements 3.4, 3.5, 3.7): a job's true outcome (exit status, exit
|
||||
* code, real finish time) must remain queryable for a configurable bounded
|
||||
* window after the job exits, instead of being discarded the moment
|
||||
* cleanup_worker_list() reaps the xcecd_worker_t entry on its next pass.
|
||||
*
|
||||
* White-box #include of ../src/xcecd_worker_utils.c (same convention as
|
||||
* this suite's test_resource_sync.c/test_dbus_workers_update.c) so the
|
||||
* REAL update_worker_status() -- the exact job-exit-detection hook point
|
||||
* this task adds its retention call to -- and the real retention
|
||||
* functions under test both run for real, not against reimplemented
|
||||
* stand-ins. Link-satisfying stubs below cover every OTHER external
|
||||
* symbol xcecd_worker_utils.c references (dbus_worker_clean,
|
||||
* xcecd_job_finished, xcecd_db_update_worker/delete_worker,
|
||||
* ipc_master_check_connection, block_signal/unblock_signal) -- none of
|
||||
* which this task's own retention logic depends on.
|
||||
*/
|
||||
#include <stdbool.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "../src/xcecd_worker_utils.c"
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Link-satisfying stubs -- none exercised by the assertions below; they
|
||||
* only need to let update_worker_status()'s other, unrelated branches run
|
||||
* to completion without touching real sqlite3/dbus/IPC.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
/* ipc_master_check_connection() gates update_worker_status()'s own
|
||||
* exit(1) fatal path -- must return true, or every test below would kill
|
||||
* the test process. */
|
||||
bool ipc_master_check_connection() { return true; }
|
||||
|
||||
/* block_signal()/unblock_signal() (xcecd_sig.c) -- not #include'd here
|
||||
* (only xcecd_worker_utils.c is), no-op stand-ins matching the header
|
||||
* signatures exactly. */
|
||||
void block_signal(int sig) { (void)sig; }
|
||||
void unblock_signal(int sig) { (void)sig; }
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Fixtures
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static uint64_t test_wallclock_now_ms(void)
|
||||
{
|
||||
struct timespec ts;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, clock_gettime(CLOCK_REALTIME, &ts));
|
||||
return (uint64_t)ts.tv_sec * 1000ULL + (uint64_t)ts.tv_nsec / 1000000ULL;
|
||||
}
|
||||
|
||||
static xcecd_worker_t* make_running_worker(int pid, uint64_t jobid)
|
||||
{
|
||||
xcecd_worker_t* w = (xcecd_worker_t*)calloc(1, sizeof(*w));
|
||||
w->worker_pid = pid;
|
||||
w->worker_status = WORKER_RUNNING;
|
||||
w->job_status = JOB_RUNNING;
|
||||
w->job_info.jobid = jobid;
|
||||
return w;
|
||||
}
|
||||
|
||||
void setUp(void)
|
||||
{
|
||||
xcecd_finished_job_retention_reset();
|
||||
}
|
||||
|
||||
void tearDown(void)
|
||||
{
|
||||
xcecd_finished_job_retention_reset();
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Population: a job's exit (via the real update_worker_status()
|
||||
* WORKER_DONE/WORKER_EXITED hook this task adds) must populate the
|
||||
* retention store with the correct outcome.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static void test_worker_done_populates_retention_with_correct_outcome(void)
|
||||
{
|
||||
xcecd_worker_t* worker = make_running_worker(4001, 90001);
|
||||
xcecd_job_t job = { .job_id = 0,
|
||||
.job_pid = 0,
|
||||
.job_status = JOB_DONE,
|
||||
.exit_code = 0 };
|
||||
|
||||
uint64_t before_ms = test_wallclock_now_ms();
|
||||
int rc = update_worker_status(worker, 4001, WORKER_DONE, &job, NULL);
|
||||
uint64_t after_ms = test_wallclock_now_ms();
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(XCECD_OK, rc);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
bool found = xcecd_finished_job_lookup(90001, after_ms, &out);
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
found, "job 90001's exit (WORKER_DONE) must be retained");
|
||||
TEST_ASSERT_EQUAL_UINT64(90001, out.jobid);
|
||||
TEST_ASSERT_EQUAL_INT(WORKER_DONE, out.exit_status);
|
||||
TEST_ASSERT_EQUAL_INT(0, out.exit_code);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
out.finish_time_ms >= before_ms && out.finish_time_ms <= after_ms,
|
||||
"finish_time_ms must be the real exit-detection time, not an "
|
||||
"arbitrary value -- it must fall within the [before, after] "
|
||||
"wall-clock bound taken around the update_worker_status() call, "
|
||||
"not the time this assertion happens to run");
|
||||
|
||||
free(worker);
|
||||
}
|
||||
|
||||
static void test_worker_exited_populates_retention_with_exit_code(void)
|
||||
{
|
||||
xcecd_worker_t* worker = make_running_worker(4002, 90002);
|
||||
xcecd_job_t job = { .job_id = 0,
|
||||
.job_pid = 0,
|
||||
.job_status = JOB_EXITED,
|
||||
.exit_code = 137 };
|
||||
|
||||
uint64_t before_ms = test_wallclock_now_ms();
|
||||
int rc
|
||||
= update_worker_status(worker, 4002, WORKER_EXITED, &job, NULL);
|
||||
uint64_t after_ms = test_wallclock_now_ms();
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(XCECD_OK, rc);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
bool found = xcecd_finished_job_lookup(90002, after_ms, &out);
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
found, "job 90002's exit (WORKER_EXITED) must be retained");
|
||||
TEST_ASSERT_EQUAL_UINT64(90002, out.jobid);
|
||||
TEST_ASSERT_EQUAL_INT(WORKER_EXITED, out.exit_status);
|
||||
TEST_ASSERT_EQUAL_INT(137, out.exit_code);
|
||||
TEST_ASSERT_TRUE(out.finish_time_ms >= before_ms
|
||||
&& out.finish_time_ms <= after_ms);
|
||||
|
||||
free(worker);
|
||||
}
|
||||
|
||||
/* Confirms this task's own critical constraint: cleanup_worker_list()'s
|
||||
* existing in-memory reaping behavior for the xcecd_worker_t entry is
|
||||
* completely unmodified by the retention hook -- the worker is still
|
||||
* exactly as eligible for reap (worker_status == WORKER_DONE) as before,
|
||||
* and the retention entry exists as a SEPARATE, additional record, not a
|
||||
* replacement for it. */
|
||||
static void
|
||||
test_retention_hook_does_not_change_cleanup_worker_list_eligibility(void)
|
||||
{
|
||||
xcecd_worker_t* worker = make_running_worker(4003, 90003);
|
||||
xcecd_job_t job = { .job_id = 0, .job_pid = 0,
|
||||
.job_status = JOB_DONE, .exit_code = 0 };
|
||||
|
||||
int rc = update_worker_status(worker, 4003, WORKER_DONE, &job, NULL);
|
||||
TEST_ASSERT_EQUAL_INT(XCECD_OK, rc);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(WORKER_DONE, worker->worker_status);
|
||||
|
||||
xcecd_worker_t* list = worker;
|
||||
cleanup_worker_list(&list, NULL);
|
||||
|
||||
TEST_ASSERT_NULL_MESSAGE(
|
||||
list, "cleanup_worker_list() must still reap a WORKER_DONE entry "
|
||||
"exactly as before -- the retention hook must not interfere "
|
||||
"with its existing behavior");
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
xcecd_finished_job_lookup(90003, test_wallclock_now_ms(), &out),
|
||||
"the retention entry must survive cleanup_worker_list()'s reap of "
|
||||
"the live xcecd_worker_t entry -- that is the entire point of "
|
||||
"this task");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Query: unknown jobid.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static void test_lookup_unknown_jobid_returns_not_found(void)
|
||||
{
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
|
||||
bool found = xcecd_finished_job_lookup(123456789, test_wallclock_now_ms(),
|
||||
&out);
|
||||
|
||||
TEST_ASSERT_FALSE_MESSAGE(
|
||||
found, "a jobid that was never retained must report not-found");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Expiry: the configured window boundary, driven deterministically via
|
||||
* explicit finish_time_ms/now_ms values (no sleeping in the test).
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static void test_lookup_within_window_returns_outcome(void)
|
||||
{
|
||||
const uint64_t finish_time_ms = 1000000ULL;
|
||||
const uint64_t window_ms = 5000ULL;
|
||||
|
||||
xcecd_finished_job_retention_set_window_ms(window_ms);
|
||||
xcecd_retain_finished_job(70001, WORKER_EXITED, 42, finish_time_ms);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
|
||||
/* Exactly at the boundary (elapsed == window_ms) is still within the
|
||||
* window -- inclusive, per xcecd_worker_utils.c's
|
||||
* finished_job_entry_expired() ("> window_ms" is expired, not
|
||||
* ">="). */
|
||||
bool found
|
||||
= xcecd_finished_job_lookup(70001, finish_time_ms + window_ms, &out);
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
found, "a job retained within its configured window must still be "
|
||||
"queryable, including exactly at the boundary");
|
||||
TEST_ASSERT_EQUAL_UINT64(70001, out.jobid);
|
||||
TEST_ASSERT_EQUAL_INT(WORKER_EXITED, out.exit_status);
|
||||
TEST_ASSERT_EQUAL_INT(42, out.exit_code);
|
||||
TEST_ASSERT_EQUAL_UINT64(finish_time_ms, out.finish_time_ms);
|
||||
}
|
||||
|
||||
static void test_lookup_just_outside_window_returns_not_found(void)
|
||||
{
|
||||
const uint64_t finish_time_ms = 2000000ULL;
|
||||
const uint64_t window_ms = 5000ULL;
|
||||
|
||||
xcecd_finished_job_retention_set_window_ms(window_ms);
|
||||
xcecd_retain_finished_job(70002, WORKER_DONE, 0, finish_time_ms);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
|
||||
/* One millisecond past the boundary -- must no longer be returned. */
|
||||
bool found = xcecd_finished_job_lookup(
|
||||
70002, finish_time_ms + window_ms + 1, &out);
|
||||
|
||||
TEST_ASSERT_FALSE_MESSAGE(
|
||||
found, "an entry one millisecond past its configured retention "
|
||||
"window must no longer be queryable (Observable completion: "
|
||||
"'no longer available once the window elapses')");
|
||||
}
|
||||
|
||||
static void test_lookup_after_expiry_prunes_entry_permanently(void)
|
||||
{
|
||||
const uint64_t finish_time_ms = 3000000ULL;
|
||||
const uint64_t window_ms = 1000ULL;
|
||||
|
||||
xcecd_finished_job_retention_set_window_ms(window_ms);
|
||||
xcecd_retain_finished_job(70003, WORKER_DONE, 0, finish_time_ms);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
|
||||
/* First query, well past expiry, prunes the entry as a side effect
|
||||
* (lazy expiry-on-query, see xcecd_worker_utils.c). A second query,
|
||||
* even with a now_ms that would have been within the window relative
|
||||
* to the original finish_time_ms, must still report not-found -- the
|
||||
* entry is gone, not merely "currently judged expired". */
|
||||
TEST_ASSERT_FALSE(xcecd_finished_job_lookup(
|
||||
70003, finish_time_ms + window_ms + 100, &out));
|
||||
TEST_ASSERT_FALSE(
|
||||
xcecd_finished_job_lookup(70003, finish_time_ms, &out));
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* main
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
int main(void)
|
||||
{
|
||||
xce_init_xcecd_log("/tmp/test_finished_job_retention.log", 1024 * 1024,
|
||||
2);
|
||||
|
||||
UNITY_BEGIN();
|
||||
|
||||
RUN_TEST(test_worker_done_populates_retention_with_correct_outcome);
|
||||
RUN_TEST(test_worker_exited_populates_retention_with_exit_code);
|
||||
RUN_TEST(test_retention_hook_does_not_change_cleanup_worker_list_eligibility);
|
||||
RUN_TEST(test_lookup_unknown_jobid_returns_not_found);
|
||||
RUN_TEST(test_lookup_within_window_returns_outcome);
|
||||
RUN_TEST(test_lookup_just_outside_window_returns_not_found);
|
||||
RUN_TEST(test_lookup_after_expiry_prunes_entry_permanently);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
@ -28,11 +28,13 @@
|
|||
*/
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#include "xcecd.h"
|
||||
#include "xcecd_resync.h"
|
||||
|
||||
/* Real globals this white-box TU's resource_sync.c references
|
||||
* (xcecd_ctx.virtual_name) -- not linking xcecd.c itself (its own
|
||||
|
|
@ -62,20 +64,41 @@ char* node_info_get_mem(void) { return stub_getter(); }
|
|||
char* node_info_get_tmp(void) { return stub_getter(); }
|
||||
char* node_info_get_io(void) { return stub_getter(); }
|
||||
|
||||
/* Linker stub for controller_connector.c's real implementation -- only
|
||||
* referenced by resource_sync_thread() (the background polling loop),
|
||||
* never invoked by this test (which calls collect_load_indices()
|
||||
* directly). */
|
||||
/* Linker stub for controller_connector.c's real implementation, spied on by
|
||||
* the worker-rejoin-resync task 4.4 tests below (Requirement 3.8) to observe
|
||||
* whether resource_sync_thread()'s periodic loop actually attempted a send
|
||||
* on a given tick. */
|
||||
static int g_send_load_indices_call_count = 0;
|
||||
|
||||
bool xcecd_send_load_indices(xce_resval_t** vals, int count)
|
||||
{
|
||||
(void)vals;
|
||||
(void)count;
|
||||
g_send_load_indices_call_count++;
|
||||
return true;
|
||||
}
|
||||
|
||||
/* worker-rejoin-resync task 4.4 (Requirement 3.8): link-satisfying stub for
|
||||
* xcecd_resync.c's public gate query -- this file's white-box target is
|
||||
* resource_sync.c only, it does not #include xcecd_resync.c (that module's
|
||||
* own resync_pending lifecycle is covered against the real production code
|
||||
* by test_xcecd_resync.c). g_test_resync_pending lets the tests below
|
||||
* control what resource_sync_thread()'s new gate check reads, same
|
||||
* convention as test_controller_connector.c's own g_test_resync_pending
|
||||
* double for the same real function. */
|
||||
static bool g_test_resync_pending = false;
|
||||
|
||||
bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
|
||||
|
||||
#include "../src/resource_sync.c"
|
||||
|
||||
void setUp(void) { return; }
|
||||
void setUp(void)
|
||||
{
|
||||
memset(&xcecd_ctx, 0, sizeof(xcecd_ctx));
|
||||
xcecd_ctx.virtual_name = (char*)"pc13v";
|
||||
g_test_resync_pending = false;
|
||||
g_send_load_indices_call_count = 0;
|
||||
}
|
||||
void tearDown(void) { return; }
|
||||
|
||||
static void
|
||||
|
|
@ -129,10 +152,116 @@ test_collect_load_indices_falls_back_to_gethostname_when_virtual_name_unset(
|
|||
"virtual_name_unset: PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* worker-rejoin-resync task 4.4 (Requirement 3.8): resource_sync_thread()'s
|
||||
* periodic load-indices send is worker post-join background activity
|
||||
* distinct from checking_connection()'s rejoin_call() (task 4.3, a separate
|
||||
* pthread -- controller_connector.c's connector_to_controller() vs. this
|
||||
* file's resource_sync_start()) -- it must also defer to an in-flight
|
||||
* job-state reconciliation report rather than interleaving with it.
|
||||
*
|
||||
* resource_sync_thread() is a real pthread with its own SYNC_INTERVAL=1s
|
||||
* sleep between checks; load_report_counter is seeded to
|
||||
* LOAD_INDICES_INTERVAL_DEFAULT - 1 before starting the thread so the very
|
||||
* first tick already crosses the reporting threshold, keeping these tests
|
||||
* fast instead of waiting a full 15s interval.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
static void
|
||||
test_resource_sync_thread_skips_load_indices_send_while_resync_pending(void)
|
||||
{
|
||||
resource_sync_t* sync = resource_sync_create();
|
||||
TEST_ASSERT_NOT_NULL(sync);
|
||||
|
||||
g_test_resync_pending = true;
|
||||
sync->load_report_counter = LOAD_INDICES_INTERVAL_DEFAULT - 1;
|
||||
|
||||
TEST_ASSERT_TRUE(resource_sync_start(sync));
|
||||
|
||||
/* Several ticks, resync_pending held true throughout -- proves the skip
|
||||
* is not a one-shot fluke of the first check. */
|
||||
sleep(2);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_send_load_indices_call_count,
|
||||
"a load-indices send must not fire while resync_pending is true "
|
||||
"(Requirement 3.8)");
|
||||
|
||||
resource_sync_stop(sync);
|
||||
resource_sync_destroy(sync);
|
||||
|
||||
printf("test_resource_sync_thread_skips_load_indices_send_while_resync_"
|
||||
"pending: PASS\n");
|
||||
}
|
||||
|
||||
static void
|
||||
test_resource_sync_thread_resumes_load_indices_send_once_resync_clears(void)
|
||||
{
|
||||
resource_sync_t* sync = resource_sync_create();
|
||||
TEST_ASSERT_NOT_NULL(sync);
|
||||
|
||||
g_test_resync_pending = true;
|
||||
sync->load_report_counter = LOAD_INDICES_INTERVAL_DEFAULT - 1;
|
||||
|
||||
TEST_ASSERT_TRUE(resource_sync_start(sync));
|
||||
|
||||
sleep(2);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_send_load_indices_call_count,
|
||||
"sanity: no send should have fired yet while resync_pending was "
|
||||
"true");
|
||||
|
||||
/* Simulates xcecd_resync_handle_query_page() finishing the report and
|
||||
* clearing the gate (xcecd_resync.c, mirroring task 4.3's own
|
||||
* resync_pending lifecycle). load_report_counter was never reset by
|
||||
* the skip path above, so the very next tick's check should fire a
|
||||
* send immediately -- no extra retry/backoff delay. */
|
||||
g_test_resync_pending = false;
|
||||
|
||||
sleep(2);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
g_send_load_indices_call_count >= 1,
|
||||
"the load-indices send must resume on the next periodic check once "
|
||||
"resync_pending clears (Requirement 3.8)");
|
||||
|
||||
resource_sync_stop(sync);
|
||||
resource_sync_destroy(sync);
|
||||
|
||||
printf("test_resource_sync_thread_resumes_load_indices_send_once_"
|
||||
"resync_clears: PASS\n");
|
||||
}
|
||||
|
||||
static void
|
||||
test_resource_sync_thread_sends_load_indices_normally_when_not_pending(void)
|
||||
{
|
||||
resource_sync_t* sync = resource_sync_create();
|
||||
TEST_ASSERT_NOT_NULL(sync);
|
||||
|
||||
g_test_resync_pending = false;
|
||||
sync->load_report_counter = LOAD_INDICES_INTERVAL_DEFAULT - 1;
|
||||
|
||||
TEST_ASSERT_TRUE(resource_sync_start(sync));
|
||||
|
||||
sleep(2);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
g_send_load_indices_call_count >= 1,
|
||||
"normal (non-reconciling) periodic load-indices reporting must be "
|
||||
"unchanged by the new resync_pending gate");
|
||||
|
||||
resource_sync_stop(sync);
|
||||
resource_sync_destroy(sync);
|
||||
|
||||
printf("test_resource_sync_thread_sends_load_indices_normally_when_not_"
|
||||
"pending: PASS\n");
|
||||
}
|
||||
|
||||
int main(void)
|
||||
{
|
||||
UNITY_BEGIN();
|
||||
RUN_TEST(test_collect_load_indices_tags_host_with_virtual_name_not_gethostname);
|
||||
RUN_TEST(test_collect_load_indices_falls_back_to_gethostname_when_virtual_name_unset);
|
||||
RUN_TEST(test_resource_sync_thread_skips_load_indices_send_while_resync_pending);
|
||||
RUN_TEST(test_resource_sync_thread_resumes_load_indices_send_once_resync_clears);
|
||||
RUN_TEST(test_resource_sync_thread_sends_load_indices_normally_when_not_pending);
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
1048
src/worker/testsuite/test_xcecd_resync.c
Normal file
1048
src/worker/testsuite/test_xcecd_resync.c
Normal file
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue