From d028c8c3152edb3f00b35ab5624c5e6dd935ee5b Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Fri, 23 Oct 2026 15:47:10 +0000 Subject: [PATCH] admin-plane breakpoint 3: done-job live GC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Terminal jobs (DONE/EXIT/PERR) were only filtered from checkpoints and WAL replay (job_should_retain, finished_retain_s=3600) — the live jobmgr never dropped them, so joblist and the scheduler's core->jobs view grew without bound across a controller's uptime (N4). Three pieces: 1. kvmgr delete path (src/lib/kvmgr.c): xkvmgr_type had .del_key=NULL and .vclear=NULL, so key_del()/vclear() on a kvmgr only returned -EINVAL. New xkvmgr_del_key (key_find to locate the slot idx -> key_del — kvtab's val_killer is NULL so the idx scalar is untouched -> vclear(data, idx)) and xkvmgr_vclear forwarder, wired into the typeinfo. A missing key propagates -ENOENT (same contract as xkvmgr_find). 2. Controller periodic sweep (job.c): job_live_gc_sweep — primary-only (standbys rebuild their live table from WAL replay; deleting there diverges), collects !job_should_retain jobs under the jobmgr READ lock with reference() (check_workers_timeout collect pattern), then per-job WRITE-locked key_del. TWO dereferences per purge: the collection reference and the table's own makeJob reference — the jobmgr's datakiller (job_obj_killer) is a deliberate no-op, so key_del transfers the table's ownership duty to the caller; one dereference would leak at refcount 1 forever (the very accumulation this sweep drains). Registered by cluster_start_job_live_gc with ha_start_primary_heartbeat's idempotent static-arm pattern + delaymgr.running guard (60s, runcount=-1). 3. Registration on both promotion tails (ha.c): raft road after ha_start_primary_heartbeat and the non-raft ha_promote_to_primary. NOT cluster_start — it is orphaned on the HA myapp_run path. Safety: apply_committed_entry's jobstate group and WAL_JOB_UPDATE already treat a key_find miss as a benign skip (bug-1970, which explicitly anticipates post-purge facts); persist_verify_apply_correctness counts only within the boot replay window. The TRANSIENT TTL now takes live effect too (previously checkpoint-exclusion only). UTs: kvmgr del INT+STRING (miss propagation, sibling intact, freed-slot reuse, killer invocation), controller test_job_live_gc.c (expired DONE purged + refcount 0, fresh/RUN retained, idempotent re-run, EXIT/PERR, TRANSIENT TTL + transient_since==0 guard, standby no-op, NULL-safe, arming guard). xcelib 761/761, controller 1354/1354. E2E (t1): promote arms the sweep; the pre-restart DONE array jobs were purged across five sweeps (33 jobs) as their 3600s windows elapsed; submit→RUN→END flows and all daemons healthy throughout. Co-Authored-By: Claude Code --- admin-plane-gaps.md | 26 +- src/controller/ha.c | 9 + src/controller/ha.h | 9 + src/controller/include/cluster.h | 8 + src/controller/job.c | 141 +++++++++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_job_live_gc.c | 302 ++++++++++++++++++++ src/lib/kvmgr.c | 39 ++- src/lib/testsuite/test_kvmgr.c | 104 +++++++ 10 files changed, 638 insertions(+), 3 deletions(-) create mode 100644 src/controller/testsuite/test_job_live_gc.c diff --git a/admin-plane-gaps.md b/admin-plane-gaps.md index bebd110d..9d61683b 100644 --- a/admin-plane-gaps.md +++ b/admin-plane-gaps.md @@ -222,4 +222,28 @@ **运维观察（未修，记录）**：CLI `config push` 单节点 ackedCount==0 误报拒绝（push_config_to_standbys 只数 standby 传输层 ACK，controller 端实际成功——以 controller 日志为准）；scheduler 对已注册资源定义变更报 WARN 不 apply（重启拾取——本次 scheduler 重启后 staticres4/res1 正常注册，作业可调度）；饱和时 PEND reason 显示 `-`（PEND-reason 发布链路对 NO_RESOURCE 场景仍未出端到端值，P1 遗留）；t1 controller 无 core dump（ulimit 0）——12:41 segfault 现场只能靠 objdump 模式匹配定位。 -### 断点 ③：done 作业 live GC（待做） +### 断点 ③：done 作业 live GC（已完成 2026-10-23） + +**问题回顾**：N4 —— 终态作业（DONE/EXIT/PERR）只有 checkpoint/WAL 回放侧的 `job_should_retain` 过滤（`finished_retain_s`，默认 3600s），live jobmgr 从不删除：controller 不重启，joblist 和 scheduler 的 `core->jobs` 表无界增长。 + +**实现**（三处）： + +1. **kvmgr 补齐删除接口**（src/lib/kvmgr.c）：`xkvmgr_type` 原本 `.del_key = NULL`、`.vclear = NULL`，`key_del()`/`vclear()` 落到 kvmgr 对象只会 -EINVAL。新增 `xkvmgr_del_key`（先 `key_find` 定位 idx → `key_del`（kvtab val_killer 为 NULL，只 unlink 标量）→ `vclear(data, idx)` 清槽）与 `xkvmgr_vclear`（转发内部 varray），接入 typeinfo。缺失 key 传播 -ENOENT（与 `xkvmgr_find` 同契约）。 + +2. **controller 侧周期清理**（src/controller/job.c 尾部）： + - `job_live_gc_sweep(c)`：主角色 gate（`ha_is_primary`，standby 靠 WAL 回放重建 live 表，删除会分叉）→ jobmgr READ 锁下迭代收集 `!job_should_retain(j, &c->persist.config)` 的作业并 `reference()`（check_workers_timeout 同款 collect 模式）→ 逐个 jobmgr WRITE 锁 + `key_del` 解链。 + - **所有权要点**：jobmgr 的 datakiller（`job_obj_killer`）是刻意空操作，`vclear` 不会释放 job —— 每次清理需 **两次 dereference**：①收集时的引用 ②表本身的 makeJob 引用（key_del 把表的持有责任转移给调用方；少一次就是 refcount 1 永久泄漏——恰是本机制要治的积累）。job_destroy 在 refcount 0 触发完整 teardown。 + - 锁纪律（B5）：调用方不得持有任何 jobmgr 锁；key_del 需要的 WRITE 锁不能嵌套在收集期的 READ 锁内（check_workers_timeout 模式的直接原因）。 + - `cluster_start_job_live_gc(c)`：幂等注册（`ha_start_primary_heartbeat` 同款 `static _Atomic` CAS + `delaymgr.running` 守卫——UT 零值 cluster_t 安全），`delay_task` 60s 周期、runcount=-1 无限重复。 + +3. **注册点 = 两条 promotion 尾**（ha.c）：raft road `ha_do_promote` 尾（`ha_start_primary_heartbeat` 之后）与 non-raft `ha_promote_to_primary` 尾各调一次 `cluster_start_job_live_gc`。**不能挂在 cluster_start**——HA 路径下它是孤儿函数（myapp_run 不调）。幂等约定与 heartbeat 相同：重复调用不叠任务。 + +**安全性论证**：`apply_committed_entry` jobstate 组与 `WAL_JOB_UPDATE` case 对 key_find miss 已返回 0（bug-1970 benign skip，注释明确假设"finished_retain_s 过期后 purge 已发生"）；`persist_verify_apply_correctness` 只在 boot 回放窗口内比较计数（recovery_complete gated）——post-boot GC 与两者正交。TRANSIENT TTL（`transient_since` + `transient_job_timeout_s`）此前只在 checkpoint 排除侧生效，live GC 使其真正生效。 + +**单测**（全绿）： +- lib：`test_kvmgr.c` 新增 `test_kvmgr_del_key`（STRING 键：删后 find -ENOENT、sibling 完好、双删 -ENOENT、datakiller 清槽触发释放——栈上 key 缓冲避免 UAF）+ `test_kvmgr_del_key_int`（INT 键：idx 标量清槽、释放槽可复用再 add）；xcelib 761/761。 +- controller：新增 `testsuite/test_job_live_gc.c`（test_job_array_apply 的 bare-cluster fixture 模式）：过期 DONE 清除+refcount 归零断言（`job_get_refcount`）、fresh DONE/RUN 存活、幂等重跑、EXIT/PERR、TRANSIENT（TTL 过期清除 / transient_since==0 保留）、standby 不动表、NULL/mgr 缺失安全、arming 无 delaymgr 安全；controller 1354/1354。 + +**E2E 结果（2026-10-23 t1）**：新 controller 部署重启 → 15:13:41 promote → `job live GC armed (interval=60s, rc=0)`；重启前 14:2x 的 DONE 数组作业跨过 3600s 保留窗后被逐波清掉（15:24–15:27 五波 1+1+1+10+20 = 33 个，与保留时间线吻合）；期间新作业 submit→RUN→DONE 全流程正常，apiserver/scheduler/controller 全健康。近期作业在保留窗内时 sweep 静默（只在 purged>0 打 INFO）——正确。 + +**下游可见性（记录，未改）**：GC 只清 controller live 表。scheduler 的 `core->jobs` 与 apiserver 的 `store.Jobs` 仍各自持有 DONE 作业副本直到自身重启（scheduler 全量 JOBS_PUSH 也不会撤销已删项；Go 侧 `OnJobDeleted` 已实现但 capnp NotifyData 无 jobDelete 字段、C 侧无发送方）。EDA 习惯观察后再决定是否补 delete 通知链路。 diff --git a/src/controller/ha.c b/src/controller/ha.c index 85902351..e71a1a53 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2337,6 +2337,11 @@ static void raft_cb_become_leader(void *ctx, uint64_t term) * at fn top) prevents double-arming. */ ha_start_primary_heartbeat(c); + /* admin-plane-gaps 断点③: arm the periodic done-job live-GC sweep + * on the primary. Same once-per-process arming contract as the + * heartbeat above, so both promotion roads can call it freely. */ + cluster_start_job_live_gc(c); + /* ha-single-leader-role task 2.2 (Req 1.4, 2.1, 2.2): append the * own-term WAL_NOOP and arm the commit engine's writable_lsn gate. * This is the shared promotion tail both roads to PRIMARY run. @@ -5438,6 +5443,10 @@ int ha_promote_to_primary(cluster_t *c) * the cluster is idle (no WAL entries or config pushes). */ ha_start_primary_heartbeat(c); + /* admin-plane-gaps 断点③: arm the periodic done-job live-GC sweep + * (same shared promotion tail as the raft road; idempotent). */ + cluster_start_job_live_gc(c); + /* Own-term no-op + writable_lsn gate (Req 1.4, 2.1, 2.2). */ ha_arm_own_term_gate(c); diff --git a/src/controller/ha.h b/src/controller/ha.h index be760155..933a3581 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -1800,6 +1800,15 @@ void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha); */ void ha_start_primary_heartbeat(cluster_t *c); +/* + * cluster_start_job_live_gc: arm a repeating delaymgr task (admin-plane-gaps + * 断点③) that unlinks terminal jobs whose finished_retain_s has elapsed from + * the live jobmgr. Same once-per-process idempotency contract as + * ha_start_primary_heartbeat; both promotion roads call it. No-op when + * c == NULL or delaymgr is not running (unit tests). + */ +void cluster_start_job_live_gc(cluster_t *c); + /* * ha_set_reconnect_offload_for_test / ha_reconnect_offload_enabled: * Feature-flag accessors (raft-drive-async Task 2.1). The offload flag gates diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 9eeb6641..e2d22724 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -68,6 +68,14 @@ extern void mark_all_run_jobs_unknown_in_memory(cluster_t*); * worker/jobmgr lock (B5). */ extern void mark_worker_jobs_exit(cluster_t*, const char* hostname); +/* admin-plane-gaps 断点③: live-table sweep that unlinks terminal jobs once + * job_should_retain goes false (finished_retain_s elapsed past finish_time). + * Primary-only; caller must not hold any jobmgr lock (B5). Non-static also + * for the unit test binary; the periodic driver is + * cluster_start_job_live_gc below. */ +extern void job_live_gc_sweep(cluster_t*); +extern void cluster_start_job_live_gc(cluster_t*); + #ifdef XCE_TESTING /* Direct probes of the bug-1509 unknown-finish registry (test binary only). */ extern int test_unknown_finish_register(const xce_jobfinish_t*); diff --git a/src/controller/job.c b/src/controller/job.c index 35ae9c52..294ea735 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -4676,3 +4676,144 @@ int on_fetchfile(cluster_t* c, endpoint_t* caller_ep, xce_request_t* request) /* 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. */ + +/* ============================================================ + * admin-plane-gaps 断点③: done-job live GC + * + * Terminal jobs (DONE/EXIT/PERR) were purged from checkpoints/WAL replay + * by job_should_retain, but never from the live jobmgr — every finished + * job stayed resident in c->jobmgr until restart, so joblist and the + * scheduler's core->jobs table grew without bound across a controller's + * uptime. This sweep applies the same retention predicate to the live + * table: once finished_retain_s has elapsed past finish_time, the job is + * unlinked (key_del) and its reference dropped. + * + * Ownership: the jobmgr's datakiller (job_obj_killer, cluster.c) is a + * deliberate no-op — job_t is refcounted RTTI and its teardown runs via + * dereference() only. So each purge is: key_del (unlink + clear the + * varray slot, which does NOT free the value) followed by dereference. + * + * Lock discipline (B5, mirrors mark_jobs_unknown_core's comment): the + * caller must not hold any jobmgr lock. Collect expired jobs under the + * jobmgr READ lock (referencing each), then re-lock per job in WRITE + * mode to unlink — a jobmgr WRITE lock cannot nest inside the READ hold + * we need for iteration. A job finishing between the collect and the + * unlink is harmless: it was already terminal when collected. + * ============================================================ */ + +/* delaymgr registration interval for the live-GC sweep. The sweep itself + * is idempotent (expired jobs are simply absent), so a fixed 60s cadence + * adds at most a minute of retention slack on top of finished_retain_s. */ +#define JOB_LIVE_GC_INTERVAL_S 60 + +void job_live_gc_sweep(cluster_t* c) +{ + iterator_t it; + xlist_t doomed; + job_t* j; + int purged = 0; + + if (c == NULL || c->jobmgr == NULL) { + return; + } + + /* Standbys rebuild their live table from WAL replay; deleting there + * would diverge from the replay stream (apply_committed_entry's + * jobstate group re-materializes on demand). Primary-only, matching + * the retention predicate's checkpoint-side role. */ + if (!ha_is_primary(&c->ha)) { + return; + } + + xlist_init(&doomed, 64); + + /* Phase 1 — collect under READ, referencing each match so the + * pointer stays valid after the lock drops. Mirrors + * check_workers_timeout's collect pattern. */ + lock(c->jobmgr, RTTI_LOCK_READ); + if (iter_make(&it, c->jobmgr) < 0) { + unlock(c->jobmgr); + xlist_fini(&doomed); + return; + } + for (;;) { + int rc = iter_next(&it, (void**)&j); + if (rc < 0 || rc == RTTI_ITER_NO_DATA) { + break; + } + if (j == NULL) { + continue; + } + if (!job_should_retain(j, &c->persist.config)) { + reference(j); + xlist_store(&doomed, j); + } + } + iter_destroy(&it); + unlock(c->jobmgr); + + /* Phase 2 — per-job WRITE-locked unlink. key_del needs the jobmgr + * WRITE lock, which cannot nest inside the phase-1 READ hold. + * + * Two dereferences per purge: (1) the collection reference taken in + * phase 1, and (2) the TABLE's own reference. makeJob gave the table + * one reference (refcount 1) and the jobmgr's datakiller + * (job_obj_killer, cluster.c) is a deliberate no-op, so unlinking via + * key_del transfers the table's ownership duty to this caller — + * without the second dereference the job_t would leak at refcount 1 + * forever (the very accumulation this sweep exists to drain). */ + for (int i = 0; i < doomed.count; i++) { + j = xlist_get(&doomed, i); + if (j == NULL) { + continue; + } + + lock(c->jobmgr, RTTI_LOCK_WRITE); + key_del(c->jobmgr, (void*)j->job.jobid); + unlock(c->jobmgr); + + purged++; + dereference(j); /* phase-1 collection reference */ + dereference(j); /* the table's makeJob reference */ + } + + xlist_fini(&doomed); + + if (purged > 0) { + INFO("job live GC: purged %d finished job(s) from the live table", + purged); + } +} + +/* Idempotent periodic arming for job_live_gc_sweep — same + * once-per-process pattern as ha_start_primary_heartbeat (with + * runcount=-1 the delaymgr re-enqueues the task itself, so duplicate + * calls on re-election must not stack tasks). Safe on a zeroed + * cluster_t (unit tests): the delaymgr.running guard skips arming. */ +void cluster_start_job_live_gc(cluster_t* c) +{ + static _Atomic int s_armed = 0; + int expected = 0; + + if (c == NULL) { + return; + } + + if (!atomic_compare_exchange_strong(&s_armed, &expected, 1)) { + return; /* already armed — task is still running */ + } + + if (!c->delaymgr.running) { + INFO("job live GC skipped (delaymgr not running)"); + return; + } + + int rc = delay_task(&c->delaymgr, JOB_LIVE_GC_INTERVAL_S, -1, + (delay_task_entry_fn_t)job_live_gc_sweep, c); + if (rc < 0) { + WARN("cluster_start_job_live_gc: delay_task failed: %d", rc); + } else { + INFO("job live GC armed (interval=%ds, rc=%d)", + JOB_LIVE_GC_INTERVAL_S, rc); + } +} diff --git a/src/controller/meson.build b/src/controller/meson.build index de90a38a..781b9ea0 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -72,6 +72,7 @@ controller_unit_test_src = [ 'testsuite/test_e2e_checkpoint.c', 'testsuite/test_e2e_hooks.c', 'testsuite/test_e2e_gc.c', + 'testsuite/test_job_live_gc.c', 'testsuite/test_e2e_capnp.c', 'testsuite/test_ha_role.c', 'testsuite/test_ha_role_lifecycle.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 1072edd9..d3cf9140 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -101,6 +101,7 @@ extern void test_apply_pump(void); extern void test_apply_pump_integration(void); extern void test_persist_checkpoint_offload(void); extern void test_job_array_apply(void); +extern void test_job_live_gc(void); extern void test_job_array_fence(void); extern void test_job_array_replay_fence(void); extern void test_upgrade_snapshot_restore(void); @@ -316,6 +317,7 @@ int main(int argc, char* argv[]) TEST(test_apply_pump_integration); TEST(test_persist_checkpoint_offload); TEST(test_job_array_apply); + TEST(test_job_live_gc); TEST(test_job_array_fence); TEST(test_job_array_replay_fence); TEST(test_upgrade_snapshot_restore); diff --git a/src/controller/testsuite/test_job_live_gc.c b/src/controller/testsuite/test_job_live_gc.c new file mode 100644 index 00000000..2a820774 --- /dev/null +++ b/src/controller/testsuite/test_job_live_gc.c @@ -0,0 +1,302 @@ +/* test_job_live_gc.c — admin-plane-gaps 断点③ + * + * Unit tests for the done-job live GC: + * - job_live_gc_sweep() (job.c): unlinks terminal jobs whose + * finished_retain_s has elapsed past finish_time from the live jobmgr, + * while non-terminal jobs and still-within-retention terminal jobs + * survive. Primary-only (a standby-role fixture is skipped). + * - cluster_start_job_live_gc() (job.c): idempotent arming contract — + * safe on a zeroed cluster_t (delaymgr not running → skip, no crash). + * + * Fixture: a minimal cluster_t (jobmgr + ha + persist config) mirroring + * test_job_array_apply.c's "bare fixture, no more than the function under + * test needs" philosophy. The jobmgr's datakiller is left unset exactly + * like that sibling fixture — job_t values are not individually released + * (accepted under ASAN_OPTIONS=detect_leaks=0); the sweep itself never + * frees the job, it only unlinks and dereferences. + * + * The sweep's primary gate uses ha_is_primary(), which reads + * c->ha.role (atomic, CTRL_PRIMARY == 0 — a zeroed cluster_t IS primary, + * so the positive-path tests need no explicit role setup; the negative + * test sets CTRL_STANDBY explicitly). + */ + +#include + +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "ha.h" +#include "jobj.h" +#include "persist.h" +#include "xce/cluster_def.h" +#include "xce/types.h" + +static int g_jlg_jobobj_ready; +static int g_jlg_xstring_ready; +static cluster_t g_jlg_c; +static persist_t g_jlg_persist; +static unsigned int g_jlg_retain_s; + +static void jlg_setup(void) +{ + kvmgr_params_t int_params + = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; + + if (!g_jlg_xstring_ready) { + TEST_ASSERT_TRUE(xstring_table_init(64) >= 0); + g_jlg_xstring_ready = 1; + } + if (!g_jlg_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_jlg_jobobj_ready = 1; + } + + memset(&g_jlg_c, 0, sizeof(g_jlg_c)); + g_jlg_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_jlg_c.jobmgr); + + /* persist.config defaults (finished_retain_s = 3600), attached where + * the sweep reads it. Zeroed persist_t fields the sweep touches are + * fine — the sweep only dereferences c->persist.config. */ + memset(&g_jlg_persist, 0, sizeof(g_jlg_persist)); + g_jlg_persist.config = persist_config_defaults(); + g_jlg_c.persist = g_jlg_persist; + g_jlg_retain_s = g_jlg_persist.config.finished_retain_s; + + /* Zeroed cluster_t has ha.role == CTRL_PRIMARY already. */ + TEST_ASSERT_EQUAL_INT(CTRL_PRIMARY, atomic_load(&g_jlg_c.ha.role)); +} + +static void jlg_teardown(void) +{ + /* Same fixture convention as test_job_array_apply.c's taj_teardown: + * job_t values still held by the jobmgr are not individually released + * (controller suite runs with ASAN_OPTIONS=detect_leaks=0). */ + if (g_jlg_c.jobmgr) + dereference(g_jlg_c.jobmgr); +} + +/* Seeds a job directly into the fixture's jobmgr with explicit state and + * finish_time. Returns the new job_t* (still table-owned; refcount 1). */ +static job_submit_data_t g_jlg_seed_submit; + +static job_t* jlg_seed_job(uint64_t jobid, unsigned int state, + time_t finish_time) +{ + xce_job_t jdef; + + memset(&g_jlg_seed_submit, 0, sizeof(g_jlg_seed_submit)); + g_jlg_seed_submit.queue = (char*)"gc-q"; + g_jlg_seed_submit.user = (char*)"gc-user"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.finish_time = finish_time; + jdef.submit = &g_jlg_seed_submit; + + return makeJob(g_jlg_c.jobmgr, &jdef); +} + +static job_t* jlg_find_job(uint64_t jobid) +{ + job_t* j = NULL; + lock(g_jlg_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_jlg_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_jlg_c.jobmgr); + return (found > 0) ? j : NULL; +} + +static int jlg_job_count(void) +{ + int n = 0; + lock(g_jlg_c.jobmgr, RTTI_LOCK_READ); + iterator_t it; + if (iter_make(&it, g_jlg_c.jobmgr) >= 0) { + for (;;) { + void* e = NULL; + int rc = iter_next(&it, &e); + if (rc < 0 || rc == RTTI_ITER_NO_DATA) + break; + if (e != NULL) + n++; + } + iter_destroy(&it); + } + unlock(g_jlg_c.jobmgr); + return n; +} + +extern void cluster_start_job_live_gc(cluster_t*); +extern void job_live_gc_sweep(cluster_t*); + +/* ── expired DONE job is purged; fresh DONE and RUN jobs survive ── */ + +static void test_sweep_purges_expired_done_job(void) +{ + jlg_setup(); + + time_t now = time(NULL); + + /* DONE, finished well past retention. */ + job_t* expired = jlg_seed_job(81001, JOB_STAT_DONE, + now - (time_t)g_jlg_retain_s - 120); + TEST_ASSERT_NOT_NULL(expired); + + /* DONE, finished within retention. */ + job_t* fresh = jlg_seed_job(81002, JOB_STAT_DONE, now - 10); + TEST_ASSERT_NOT_NULL(fresh); + + /* RUN (non-terminal) — always retained regardless of finish_time. */ + job_t* running = jlg_seed_job(81003, JOB_STAT_RUN, 0); + TEST_ASSERT_NOT_NULL(running); + + TEST_ASSERT_EQUAL_INT(3, jlg_job_count()); + + job_live_gc_sweep(&g_jlg_c); + + /* The expired job is gone from the table... */ + TEST_ASSERT_NULL(jlg_find_job(81001)); + /* ...and its refcount ran down to destruction (seed refcount 1, sweep + * added 1, dereference after unlink leaves 0 → job_destroy ran). */ + TEST_ASSERT_EQUAL_UINT32(0, job_get_refcount(expired)); + + /* The others are untouched. */ + TEST_ASSERT_NOT_NULL(jlg_find_job(81002)); + TEST_ASSERT_NOT_NULL(jlg_find_job(81003)); + TEST_ASSERT_EQUAL_INT(2, jlg_job_count()); + + /* A re-run is idempotent: nothing left to purge. */ + job_live_gc_sweep(&g_jlg_c); + TEST_ASSERT_EQUAL_INT(2, jlg_job_count()); + + jlg_teardown(); +} + +/* ── EXIT and PERR terminal states are also purged once expired ── */ + +static void test_sweep_purges_expired_exit_and_perr(void) +{ + jlg_setup(); + + time_t now = time(NULL); + time_t stale = now - (time_t)g_jlg_retain_s - 60; + + job_t* exited = jlg_seed_job(81011, JOB_STAT_EXIT, stale); + job_t* perr = jlg_seed_job(81012, JOB_STAT_PERR, stale); + TEST_ASSERT_NOT_NULL(exited); + TEST_ASSERT_NOT_NULL(perr); + + job_live_gc_sweep(&g_jlg_c); + + TEST_ASSERT_NULL(jlg_find_job(81011)); + TEST_ASSERT_NULL(jlg_find_job(81012)); + + jlg_teardown(); +} + +/* ── a TRANSIENT job whose TTL expired is purged (predicate parity) ── */ + +static void test_sweep_purges_expired_transient(void) +{ + jlg_setup(); + + job_t* transient = jlg_seed_job(81021, JOB_STAT_TRANSIENT, 0); + TEST_ASSERT_NOT_NULL(transient); + /* seed_job cannot set transient_since via finish_time — set it under + * the job's own lock, mirroring mark_jobs_unknown_core. */ + lock(transient, RTTI_LOCK_WRITE); + transient->job.transient_since + = (uint64_t)(time(NULL) + - (time_t)g_jlg_persist.config.transient_job_timeout_s + - 30); + unlock(transient); + + /* A TRANSIENT job with the timestamp NOT set must NOT be purged + * (predicate: transient_since == 0 → retain). */ + job_t* unmarked = jlg_seed_job(81022, JOB_STAT_TRANSIENT, 0); + TEST_ASSERT_NOT_NULL(unmarked); + + job_live_gc_sweep(&g_jlg_c); + + TEST_ASSERT_NULL(jlg_find_job(81021)); + TEST_ASSERT_NOT_NULL(jlg_find_job(81022)); + + jlg_teardown(); +} + +/* ── standby role: sweep must not touch the table at all ── */ + +static void test_sweep_skipped_on_standby(void) +{ + jlg_setup(); + + atomic_store(&g_jlg_c.ha.role, CTRL_STANDBY); + + time_t now = time(NULL); + TEST_ASSERT_NOT_NULL(jlg_seed_job( + 81031, JOB_STAT_DONE, now - (time_t)g_jlg_retain_s - 120)); + TEST_ASSERT_EQUAL_INT(1, jlg_job_count()); + + job_live_gc_sweep(&g_jlg_c); + + /* Standby builds its live table from WAL replay — nothing purged. */ + TEST_ASSERT_NOT_NULL(jlg_find_job(81031)); + TEST_ASSERT_EQUAL_INT(1, jlg_job_count()); + + jlg_teardown(); +} + +/* ── missing jobmgr / NULL cluster are harmless ── */ + +static void test_sweep_null_safe(void) +{ + jlg_setup(); + + time_t now = time(NULL); + TEST_ASSERT_NOT_NULL(jlg_seed_job( + 81041, JOB_STAT_DONE, now - (time_t)g_jlg_retain_s - 120)); + + job_live_gc_sweep(NULL); + + var saved = g_jlg_c.jobmgr; + g_jlg_c.jobmgr = NULL; + job_live_gc_sweep(&g_jlg_c); + g_jlg_c.jobmgr = saved; + + TEST_ASSERT_EQUAL_INT(1, jlg_job_count()); + + jlg_teardown(); +} + +/* ── arming: idempotent + safe when delaymgr is not running ── */ + +static void test_arm_skips_without_delaymgr(void) +{ + cluster_t c; + + memset(&c, 0, sizeof(c)); + TEST_ASSERT_FALSE(c.delaymgr.running); + + /* Must not crash and must not arm anything. */ + cluster_start_job_live_gc(&c); + cluster_start_job_live_gc(NULL); +} + +void test_job_live_gc(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_sweep_purges_expired_done_job); + RUN_TEST(test_sweep_purges_expired_exit_and_perr); + RUN_TEST(test_sweep_purges_expired_transient); + RUN_TEST(test_sweep_skipped_on_standby); + RUN_TEST(test_sweep_null_safe); + RUN_TEST(test_arm_skips_without_delaymgr); +} diff --git a/src/lib/kvmgr.c b/src/lib/kvmgr.c index 72c5dffc..f938a2de 100644 --- a/src/lib/kvmgr.c +++ b/src/lib/kvmgr.c @@ -149,6 +149,41 @@ static int xkvmgr_find(meta_obj_t* obj, void* key, void** value) return rc; } +static int xkvmgr_del_key(meta_obj_t* obj, void* key) +{ + xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); + int rc; + void* data; + + /* Locate the slot first so the value array entry can be cleared once + * the kvtab unlink succeeds. kvtab's val_killer is NULL in this manager + * (xkvmgr_init), so key_del only unlinks the idx scalar — any extra + * teardown of the value itself belongs to the caller. Callers whose + * values are refcounted RTTI objects (e.g. the controller's jobmgr, + * whose datakiller is a no-op) MUST dereference the value after a + * successful delete. */ + rc = key_find(kvm->kvtab, key, &data); + if (rc < 0) { + return rc; + } + + rc = key_del(kvm->kvtab, key); + if (rc < 0) { + return rc; + } + + vclear(kvm->data, (int)(long)data); + + return 1; +} + +static int xkvmgr_vclear(meta_obj_t* obj, int idx) +{ + xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); + + return vclear(kvm->data, idx); +} + static int xkvmgr_make_iter(iterator_t* iter, va_list ap) { meta_obj_t* obj = iter->obj; @@ -276,14 +311,14 @@ typeinfo_t xkvmgr_type = { .destroy = (obj_destroy_fn_t)xkvmgr_destroy, .interfaces = { .kv = &(intf_kv_t) { .add_key = (key_add_fn_t)xkvmgr_add, - .del_key = (key_del_fn_t)NULL, + .del_key = (key_del_fn_t)xkvmgr_del_key, .find_key = (key_find_fn_t)xkvmgr_find }, .varray = &(intf_varray_t) { .vstore = (vstore_fn_t)NULL, .vget = (vget_fn_t)xkvmgr_get, .vget_vec = (vget_vec_fn_t)xkvmgr_get_vec, .vput = (vput_fn_t)xkvmgr_put, .vput_vec = (vput_vec_fn_t)xkvmgr_put_vec, - .vclear = (vclear_fn_t)NULL, + .vclear = (vclear_fn_t)xkvmgr_vclear, .vfreeze = NULL, /* wired in P0-2 */ .vthaw = NULL /* wired in P0-2 */ }, .iter diff --git a/src/lib/testsuite/test_kvmgr.c b/src/lib/testsuite/test_kvmgr.c index d663a775..455d5c2b 100644 --- a/src/lib/testsuite/test_kvmgr.c +++ b/src/lib/testsuite/test_kvmgr.c @@ -117,10 +117,114 @@ static void test_kvmgr_iter(void) destroy(kvmgr); } +/* admin-plane-gaps 断点③: xkvmgr del_key / vclear wiring. key_del unlinks + * the kvtab entry AND clears the backing varray slot; a subsequent find + * misses, sibling entries stay intact, and a double delete propagates the + * miss. With a datakiller wired, clearing the slot frees the value (the + * same path the controller's jobmgr purge relies on, except there the + * killer is a no-op and the caller dereferences the job itself). */ +static void test_kvmgr_del_key(void) +{ + kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING, + .nslots = 4, + .delta = 4, + .datakiller = my_user_killer }; + var kvmgr = makeInstance(&xkvmgr_type, ¶ms); + char key1[] = "user1"; + char key2[] = "user2"; + user_t* u1 = NULL; + user_t* u2 = NULL; + int rc = 0; + void* data = NULL; + + TEST_ASSERT_TRUE(kvmgr != NULL); + + u1 = calloc(1, sizeof(*u1)); + u2 = calloc(1, sizeof(*u2)); + TEST_ASSERT_TRUE(u1 != NULL && u2 != NULL); + u1->name = strdup(key1); + u2->name = strdup(key2); + TEST_ASSERT_TRUE(u1->name != NULL && u2->name != NULL); + u1->age = 30; + u2->age = 40; + + rc = key_add(kvmgr, (void*)key1, (void*)u1); + TEST_ASSERT_TRUE(rc >= 0); + rc = key_add(kvmgr, (void*)key2, (void*)u2); + TEST_ASSERT_TRUE(rc >= 0); + + /* Delete the first key. With a datakiller wired, the cleared slot + * frees u1 — so only the stack key buffers are used from here on. */ + rc = key_del(kvmgr, (void*)key1); + TEST_ASSERT_EQUAL_INT(1, rc); + + /* The deleted key no longer resolves. */ + rc = key_find(kvmgr, (void*)key1, &data); + TEST_ASSERT_TRUE(rc < 0); + + /* The sibling entry is untouched. */ + rc = key_find(kvmgr, (void*)key2, &data); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_TRUE(data == u2); + + /* Double delete misses — xkvmgr_del_key propagates key_find's + * -ENOENT for an absent key (same contract as xkvmgr_find). */ + rc = key_del(kvmgr, (void*)key1); + TEST_ASSERT_EQUAL_INT(-ENOENT, rc); + + destroy(kvmgr); +} + +static void test_kvmgr_del_key_int(void) +{ + kvmgr_params_t params = { .key_type = KVMGR_KEY_INT, + .nslots = 4, + .delta = 4, + .datakiller = NULL }; + var kvmgr = makeInstance(&xkvmgr_type, ¶ms); + int rc = 0; + void* data = NULL; + + TEST_ASSERT_TRUE(kvmgr != NULL); + + rc = key_add(kvmgr, (void*)(long)101, (void*)(long)10100); + TEST_ASSERT_TRUE(rc >= 0); + rc = key_add(kvmgr, (void*)(long)102, (void*)(long)10200); + TEST_ASSERT_TRUE(rc >= 0); + + rc = key_del(kvmgr, (void*)(long)101); + TEST_ASSERT_EQUAL_INT(1, rc); + + rc = key_find(kvmgr, (void*)(long)101, &data); + TEST_ASSERT_TRUE(rc < 0); + + /* The int-keyed table's idx scalar was cleared with the entry — the + * slot the deleted key occupied now stores nothing. */ + rc = key_find(kvmgr, (void*)(long)102, &data); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_EQUAL_INT(10200, (int)(long)data); + + /* Delete the sibling too, then re-add: the freed slot must be + * reusable (vclear reset it in place). */ + rc = key_del(kvmgr, (void*)(long)102); + TEST_ASSERT_EQUAL_INT(1, rc); + + rc = key_add(kvmgr, (void*)(long)103, (void*)(long)10300); + TEST_ASSERT_TRUE(rc >= 0); + + rc = key_find(kvmgr, (void*)(long)103, &data); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_EQUAL_INT(10300, (int)(long)data); + + destroy(kvmgr); +} + void test_kvmgr(void) { printf("\n>> kvmgr test\n"); RUN_TEST(test_kvmgr_create); RUN_TEST(test_kvmgr_add); RUN_TEST(test_kvmgr_iter); + RUN_TEST(test_kvmgr_del_key); + RUN_TEST(test_kvmgr_del_key_int); }