feat: JOB_STAT_TRANSIENT, job_should_retain, worker_should_retain, dead_since (P0-5a, P0-8, P0-13)

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)
This commit is contained in:
Rongsong Shen 2026-05-07 12:52:07 +00:00
parent 6f3b98c6b5
commit 3030012f51
9 changed files with 330 additions and 2 deletions

View file

@ -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;
}

View file

@ -1,9 +1,11 @@
#include <errno.h>
#include <signal.h>
#include <stdbool.h>
#include <time.h>
#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. */

View file

@ -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)

View file

@ -3,8 +3,10 @@
#include "config.h"
#include "hook.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include <assert.h>
#include <stdbool.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
@ -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;
}

View file

@ -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();
}

View file

@ -0,0 +1,244 @@
#define _GNU_SOURCE
#include <stdbool.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#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);
}

View file

@ -1,7 +1,10 @@
#include <errno.h>
#include <stdbool.h>
#include <stdlib.h>
#include <time.h>
#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. */

View file

@ -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 <stdbool.h>
#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);

View file

@ -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)
@ -354,6 +355,7 @@ 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_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)