lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

admin-plane breakpoint 3: done-job live GC

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-10-23 15:47:10 +00:00
parent fc33a5da16
commit d028c8c315
10 changed files with 638 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/controller/testsuite/controller_test.c
View file

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

302
src/controller/testsuite/test_job_live_gc.c Normal file
View file

@ -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 <rtti.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#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);
}