From 3030012f518866fc2517733361df0b1a077b7714 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 12:52:07 +0000 Subject: [PATCH] feat: JOB_STAT_TRANSIENT, job_should_retain, worker_should_retain, dead_since (P0-5a, P0-8, P0-13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add terminal-state constants and GC retention predicates for the checkpoint xzarray_thaw() GC pass: - types.h: add JOB_STAT_TRANSIENT (0x400), XCE_WORKER_HIDDEN (0x00200000), QUEUE_STAT_HIDDEN (0x0100) - persist.c: implement job_is_terminal(), job_should_retain(), worker_should_retain() — placed here so the unit test binary (which links only persist.c) can access them without RTTI deps - persist.h: declare the three GC predicates; include xce/types.h - worker.c: set dead_since = time(NULL) in mark_worker_unavail() when dead_since is not already set - cluster.c: add TODO P0-8b marker on zarray_del endpoint cleanup - meson.build: register test_gc_retention.c; raise test timeout 60→120s - testsuite/test_gc_retention.c: 19 tests covering job_is_terminal, job_should_retain (TRANSIENT TTL, terminal TTL, non-terminal always retained), and worker_should_retain (HIDDEN/UNAVAIL/healthy/zero dead_since); all pass (63 total, 0 failures) --- src/controller/cluster.c | 1 + src/controller/job.c | 6 + src/controller/meson.build | 3 +- src/controller/persist.c | 55 +++++ src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_gc_retention.c | 244 +++++++++++++++++++ src/controller/worker.c | 9 + src/include/persist.h | 7 + src/include/xce/types.h | 5 +- 9 files changed, 330 insertions(+), 2 deletions(-) create mode 100644 src/controller/testsuite/test_gc_retention.c diff --git a/src/controller/cluster.c b/src/controller/cluster.c index c4b4a72c..4f1ab6b8 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -82,6 +82,7 @@ static void handle_endpoint_error(endpoint_t* ep) if (ep->idx >= 0) { /* zarray_del will finalize ep */ + /* TODO P0-8b: replace with deferred cleanup after P0-0a zarray merge is complete */ zarray_del(&(c->endpoints), ep->idx); return; } diff --git a/src/controller/job.c b/src/controller/job.c index 59ef0d67..b52bc646 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1,9 +1,11 @@ #include #include +#include #include #include "cluster.h" #include "jobj.h" +#include "persist.h" #include "preason.h" #include "xce/cluster_def.h" #include "xce/errnox.h" @@ -1040,3 +1042,7 @@ int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request) return 0; } + +/* job_is_terminal and job_should_retain live in persist.c — + * they only depend on xce_job_t fields and persist_config_t, + * and must be accessible from the unit test binary which links persist.c. */ diff --git a/src/controller/meson.build b/src/controller/meson.build index 4612d5ac..b2e3ef27 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -75,6 +75,7 @@ controller_unit_test_src = [ 'testsuite/test_persist_recovery.c', 'testsuite/test_hooks.c', 'testsuite/test_persist_hooks.c', + 'testsuite/test_gc_retention.c', 'persist.c', ] @@ -95,5 +96,5 @@ controller_unit_test = executable('controller_unit_test', test('controller test', controller_unit_test, suite: 'controller', - timeout: 60) + timeout: 120) diff --git a/src/controller/persist.c b/src/controller/persist.c index cbc64565..99768c50 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -3,8 +3,10 @@ #include "config.h" #include "hook.h" #include "xce/cluster_def.h" +#include "xce/types.h" #include +#include #include #include #include @@ -873,3 +875,56 @@ int persist_worker_hook_fun(int type, int situation, int detail, uint32_t payload = (uint32_t)detail; return persist_wal_async(p, wal_type, &payload, sizeof(payload)); } + +/* ============================================================ + * GC retention predicates (P0-5a / P0-8 / P0-13) + * + * Placed in persist.c so the unit test binary (which links only + * persist.c from the controller sources) can access them without + * pulling in RTTI-dependent job.c / worker.c. + * ============================================================ */ + +bool job_is_terminal(const job_t* j) +{ + return (j->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE | JOB_STAT_PERR)) != 0; +} + +bool job_should_retain(const job_t* j, const persist_config_t* cfg) +{ + time_t now = time(NULL); + + /* TRANSIENT jobs: auto-terminate when TTL elapses after failover. + * transient_since == 0 means the timestamp was never set — do not GC. */ + if (j->job.state & JOB_STAT_TRANSIENT) { + if (j->job.transient_since > 0 && + now > (time_t)j->job.transient_since + (time_t)cfg->transient_job_timeout_s) { + return false; + } + return true; + } + + /* Non-terminal jobs are always retained */ + if (!job_is_terminal(j)) return true; + + /* Terminal jobs: retain until finished_retain_s expires */ + return (j->job.finish_time + (time_t)cfg->finished_retain_s >= now); +} + +bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg) +{ + time_t now = time(NULL); + + /* HIDDEN workers: admin explicitly removed — clean up at next checkpoint */ + if (w->status & XCE_WORKER_HIDDEN) return false; + + /* UNAVAILABLE workers: retain until worker_retain_s expires. + * dead_since == 0 means not yet set — do not GC. */ + if (w->status & XCE_WORKER_UNAVAIL) { + if (w->dead_since > 0 && + now > w->dead_since + (time_t)cfg->worker_retain_s) { + return false; + } + } + + return true; +} diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 4d957255..8186cd1f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -12,6 +12,7 @@ extern void test_persist_checkpoint(void); extern void test_persist_recovery(void); extern void test_hooks(void); extern void test_persist_hooks(void); +extern void test_gc_retention(void); void setUp(void) { return; } @@ -93,6 +94,7 @@ int main(int argc, char* argv[]) TEST(test_persist_recovery); TEST(test_hooks); TEST(test_persist_hooks); + TEST(test_gc_retention); return UNITY_END(); } diff --git a/src/controller/testsuite/test_gc_retention.c b/src/controller/testsuite/test_gc_retention.c new file mode 100644 index 00000000..5456b42b --- /dev/null +++ b/src/controller/testsuite/test_gc_retention.c @@ -0,0 +1,244 @@ +#define _GNU_SOURCE +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" +#include "xce/types.h" + +/* ------------------------------------------------------------------ */ +/* Helpers to build minimal job_t / xce_worker_t fixtures */ +/* ------------------------------------------------------------------ */ + +static job_t make_job(unsigned int state, time_t finish_time, + uint64_t transient_since) +{ + job_t j; + memset(&j, 0, sizeof(j)); + j.job.state = state; + j.job.finish_time = finish_time; + j.job.transient_since = (uint64_t)transient_since; + return j; +} + +static xce_worker_t make_worker(uint32_t status, time_t dead_since) +{ + xce_worker_t w; + memset(&w, 0, sizeof(w)); + w.status = status; + w.dead_since = dead_since; + return w; +} + +/* ------------------------------------------------------------------ */ +/* Forward declarations for functions under test */ +/* ------------------------------------------------------------------ */ + +extern bool job_is_terminal(const job_t* j); +extern bool job_should_retain(const job_t* j, const persist_config_t* cfg); +extern bool worker_should_retain(const xce_worker_t* w, + const persist_config_t* cfg); + +/* ------------------------------------------------------------------ */ +/* job_is_terminal tests */ +/* ------------------------------------------------------------------ */ + +static void test_running_job_not_terminal(void) +{ + job_t j = make_job(JOB_STAT_RUN, 0, 0); + TEST_ASSERT_FALSE(job_is_terminal(&j)); +} + +static void test_pending_job_not_terminal(void) +{ + job_t j = make_job(JOB_STAT_PEND, 0, 0); + TEST_ASSERT_FALSE(job_is_terminal(&j)); +} + +static void test_exit_job_is_terminal(void) +{ + job_t j = make_job(JOB_STAT_EXIT, time(NULL), 0); + TEST_ASSERT_TRUE(job_is_terminal(&j)); +} + +static void test_done_job_is_terminal(void) +{ + job_t j = make_job(JOB_STAT_DONE, time(NULL), 0); + TEST_ASSERT_TRUE(job_is_terminal(&j)); +} + +static void test_perr_job_is_terminal(void) +{ + job_t j = make_job(JOB_STAT_PERR, time(NULL), 0); + TEST_ASSERT_TRUE(job_is_terminal(&j)); +} + +static void test_transient_job_not_terminal(void) +{ + /* JOB_STAT_TRANSIENT is NOT terminal — it's a special non-terminal state */ + job_t j = make_job(JOB_STAT_TRANSIENT, 0, (uint64_t)time(NULL)); + TEST_ASSERT_FALSE(job_is_terminal(&j)); +} + +/* ------------------------------------------------------------------ */ +/* job_should_retain: TRANSIENT timeout tests */ +/* ------------------------------------------------------------------ */ + +static void test_transient_job_retained_within_timeout(void) +{ + /* transient_since = now - 100s, timeout = 300s → retain */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.transient_job_timeout_s = 300; + job_t j = make_job(JOB_STAT_TRANSIENT, 0, now - 100); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); +} + +static void test_transient_job_expired(void) +{ + /* transient_since = now - 400s, timeout = 300s → do NOT retain */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.transient_job_timeout_s = 300; + job_t j = make_job(JOB_STAT_TRANSIENT, 0, now - 400); + TEST_ASSERT_FALSE(job_should_retain(&j, &cfg)); +} + +static void test_transient_job_zero_since_always_retained(void) +{ + /* transient_since == 0 means unset → never expire (operator must decide) */ + persist_config_t cfg = persist_config_defaults(); + cfg.transient_job_timeout_s = 300; + job_t j = make_job(JOB_STAT_TRANSIENT, 0, 0); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); +} + +/* ------------------------------------------------------------------ */ +/* job_should_retain: terminal job TTL tests */ +/* ------------------------------------------------------------------ */ + +static void test_finished_job_retained_within_ttl(void) +{ + /* finish_time = now - 1800s, retain = 3600s → retain */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.finished_retain_s = 3600; + job_t j = make_job(JOB_STAT_EXIT, now - 1800, 0); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); +} + +static void test_finished_job_expired(void) +{ + /* finish_time = now - 7200s, retain = 3600s → do NOT retain */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.finished_retain_s = 3600; + job_t j = make_job(JOB_STAT_EXIT, now - 7200, 0); + TEST_ASSERT_FALSE(job_should_retain(&j, &cfg)); +} + +static void test_done_job_retained_within_ttl(void) +{ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.finished_retain_s = 3600; + job_t j = make_job(JOB_STAT_DONE, now - 1800, 0); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); +} + +static void test_running_job_always_retained(void) +{ + persist_config_t cfg = persist_config_defaults(); + job_t j = make_job(JOB_STAT_RUN, 0, 0); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); +} + +static void test_pending_job_always_retained(void) +{ + persist_config_t cfg = persist_config_defaults(); + job_t j = make_job(JOB_STAT_PEND, 0, 0); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); +} + +/* ------------------------------------------------------------------ */ +/* worker_should_retain tests */ +/* ------------------------------------------------------------------ */ + +static void test_worker_unavail_retained_within_ttl(void) +{ + /* dead_since = now - 43200s (12 h), retain = 86400s (24 h) → retain */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.worker_retain_s = 86400; + xce_worker_t w = make_worker(XCE_WORKER_UNAVAIL, now - 43200); + TEST_ASSERT_TRUE(worker_should_retain(&w, &cfg)); +} + +static void test_worker_unavail_expired(void) +{ + /* dead_since = now - 100000s, retain = 86400s → do NOT retain */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.worker_retain_s = 86400; + xce_worker_t w = make_worker(XCE_WORKER_UNAVAIL, now - 100000); + TEST_ASSERT_FALSE(worker_should_retain(&w, &cfg)); +} + +static void test_worker_hidden_never_retained(void) +{ + /* XCE_WORKER_HIDDEN → cleanup immediately regardless of dead_since */ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + xce_worker_t w = make_worker(XCE_WORKER_HIDDEN, now - 10); + TEST_ASSERT_FALSE(worker_should_retain(&w, &cfg)); +} + +static void test_worker_healthy_always_retained(void) +{ + persist_config_t cfg = persist_config_defaults(); + xce_worker_t w = make_worker(0, 0); + TEST_ASSERT_TRUE(worker_should_retain(&w, &cfg)); +} + +static void test_worker_unavail_zero_dead_since_always_retained(void) +{ + /* dead_since == 0 → not yet set, do not GC */ + persist_config_t cfg = persist_config_defaults(); + cfg.worker_retain_s = 86400; + xce_worker_t w = make_worker(XCE_WORKER_UNAVAIL, 0); + TEST_ASSERT_TRUE(worker_should_retain(&w, &cfg)); +} + +/* ------------------------------------------------------------------ */ +/* Test suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_gc_retention(void) +{ + RUN_TEST(test_running_job_not_terminal); + RUN_TEST(test_pending_job_not_terminal); + RUN_TEST(test_exit_job_is_terminal); + RUN_TEST(test_done_job_is_terminal); + RUN_TEST(test_perr_job_is_terminal); + RUN_TEST(test_transient_job_not_terminal); + + RUN_TEST(test_transient_job_retained_within_timeout); + RUN_TEST(test_transient_job_expired); + RUN_TEST(test_transient_job_zero_since_always_retained); + + RUN_TEST(test_finished_job_retained_within_ttl); + RUN_TEST(test_finished_job_expired); + RUN_TEST(test_done_job_retained_within_ttl); + RUN_TEST(test_running_job_always_retained); + RUN_TEST(test_pending_job_always_retained); + + RUN_TEST(test_worker_unavail_retained_within_ttl); + RUN_TEST(test_worker_unavail_expired); + RUN_TEST(test_worker_hidden_never_retained); + RUN_TEST(test_worker_healthy_always_retained); + RUN_TEST(test_worker_unavail_zero_dead_since_always_retained); +} diff --git a/src/controller/worker.c b/src/controller/worker.c index bb099f7d..eb90707e 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -1,7 +1,10 @@ #include +#include #include +#include #include "cluster.h" +#include "persist.h" #include "wobj.h" #include "worker.h" #include "xce/cluster_def.h" @@ -519,6 +522,8 @@ int mark_worker_unavail(cluster_t* c, worker_t* w) HOOK_DETAIL_WORKER_DEAD, c, w); w->host.status |= XCE_WORKER_UNAVAIL; + if (w->host.dead_since == 0) + w->host.dead_since = time(NULL); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_DEAD, c, w); @@ -658,3 +663,7 @@ int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) (void)len; return 0; } + +/* worker_should_retain lives in persist.c — + * it only depends on xce_worker_t fields and persist_config_t, + * and must be accessible from the unit test binary which links persist.c. */ diff --git a/src/include/persist.h b/src/include/persist.h index 01afb7c8..2d68f6f0 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -163,3 +163,10 @@ int persist_queue_hook_fun(int type, int situation, int detail, void* cls, va_list ap); int persist_worker_hook_fun(int type, int situation, int detail, void* cls, va_list ap); + +/* GC retention predicates (P0-5a / P0-8 / P0-13) */ +#include +#include "xce/types.h" +bool job_is_terminal(const job_t* j); +bool job_should_retain(const job_t* j, const persist_config_t* cfg); +bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg); diff --git a/src/include/xce/types.h b/src/include/xce/types.h index c62379c5..50d441ba 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -172,6 +172,7 @@ typedef struct { #define XCE_WORKER_LOCKEDW (0x00040000) #define XCE_WORKER_BUSY (0x00080000) #define XCE_WORKER_CLOSED (0x00100000) +#define XCE_WORKER_HIDDEN (0x00200000) /* Admin explicitly removed — invisible but not yet cleaned up */ #define IS_WORKER_UNAVAIL(status) (((status) & XCE_WORKER_UNAVAIL) != 0) #define IS_WORKER_LOCKEDU(status) (((status) & XCE_WORKER_LOCKEDU) != 0) @@ -353,7 +354,8 @@ typedef struct { #define JOB_STAT_PDONE (0x80) /**< Post job process done successfully */ #define JOB_STAT_PERR (0x100) /**< Post job process has error */ -#define JOB_STAT_WAIT (0x200) /**< Chunk job waiting its turn to exec */ +#define JOB_STAT_WAIT (0x200) /**< Chunk job waiting its turn to exec */ +#define JOB_STAT_TRANSIENT (0x400) /**< Job state uncertain after failover — awaiting worker re-join */ #define JOB_STAT_RUNKWN \ 0x8000 /* Flag : Job status is UNKWN caused by losting contact with remote \ cluster */ @@ -381,6 +383,7 @@ typedef struct { #define QUEUE_STAT_DISC 0x10 /**< Remote queue status is disconnected. */ #define QUEUE_STAT_FAIRSHARE_DEFAULT \ 0x20 /**< Fairshare queue with null user share defined*/ +#define QUEUE_STAT_HIDDEN (0x0100) /**< Admin deleted — invisible but not yet cleaned up */ /* FIXME: improve following later */ #define JOB_IS_PEND(state) (((state) & JOB_STAT_PEND) != 0)