feat(worker-rejoin-resync): task 4.1 - bounded finished-job retention on worker
Retain a job's exit status/code/true finish time for a configurable window after update_worker_status() detects its exit, before cleanup_worker_list() would otherwise discard all trace of it (Requirements 3.4, 3.5, 3.7). Hooked into the single choke point both production exit-detection paths (libuv IPC callback, dbus polling) converge through, so retention happens-before any reap regardless of which mechanism is active. Also prunes on retain (not just on lookup), closing a review finding: lookup currently has zero production callers until task 4.2 lands, so without this the store would grow unboundedly for the daemon's uptime. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
parent
087b8c1804
commit
e31012e8a7
5 changed files with 613 additions and 1 deletions
|
|
@ -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,54 @@ 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);
|
||||
|
||||
/* 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,8 +7,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)
|
||||
{
|
||||
|
|
@ -299,6 +371,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)) {
|
||||
|
|
@ -318,6 +401,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)) {
|
||||
|
|
@ -407,6 +497,105 @@ 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;
|
||||
}
|
||||
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -93,3 +93,17 @@ 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'])
|
||||
|
|
|
|||
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();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue