diff --git a/.kiro/specs/ha-convergence-recovery/tasks.md b/.kiro/specs/ha-convergence-recovery/tasks.md index 1b673734..fd840434 100644 --- a/.kiro/specs/ha-convergence-recovery/tasks.md +++ b/.kiro/specs/ha-convergence-recovery/tasks.md @@ -5,7 +5,7 @@ > Task 8 (E2E) additionally requires **ha-single-leader-role** to have landed and a live `/xce-deploy-ha` cluster. - [ ] 1. Egress filter: submit_lsn tagging and commit-watermark gate on scheduler exports -- [ ] 1.1 Write RED tests for tagging and the scheduler-export watermark +- [x] 1.1 Write RED tests for tagging and the scheduler-export watermark - Create `testsuite/test_egress_filter.c` (XCE_TESTING, AAA structure) and register it in `controller_unit_test_src` (meson.build); add `uint64_t submit_lsn` to the `job_t` wrapper (types.h:483-485 anchor) and declare `job_submit_committed` / `collect_jobs_committed` in the dataex header so tests compile - Tests: a live-submitted job is tagged with its WAL entry LSN; a job restored via `apply_wal_entry`/`restore_job` replay is tagged with the replayed entry's LSN; a job with `submit_lsn > committed_lsn` is absent from `collect_jobs_committed` results and from the JOB_NEW incremental push list; the same job appears once `committed_lsn` advances to cover it; `submit_lsn == 0` is always exportable; plain `collect_jobs` (status/API path) still returns the uncommitted job - Observable completion: build succeeds with the new field/header declarations; all new test cases FAIL (red) against unimplemented predicate/tagging diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 53400394..bbcf827e 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -4,6 +4,7 @@ #include "config.h" #include +#include #include #include "hook.h" @@ -83,6 +84,18 @@ extern void collect_queues(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_resources(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_jobs(cluster_t*, collect_check_fn_t, xlist_t*); extern void collect_workers(cluster_t*, collect_check_fn_t, xlist_t*); + +/* ha-convergence-recovery (3.1-3.3): commit-watermark egress gate for + * scheduler-bound job exports. Defined in dataex.c. + * job_submit_committed — true iff the job's submit entry is committed + * (submit_lsn == 0 || submit_lsn <= committed_lsn); + * one lock-free atomic load + compare. + * collect_jobs_committed — identical contract to collect_jobs, additionally + * excluding jobs whose submit entry is uncommitted. + * For scheduler-bound callers ONLY; status/API read + * paths keep plain collect_jobs (Req 3.4). */ +extern bool job_submit_committed(cluster_t*, job_t*); +extern void collect_jobs_committed(cluster_t*, collect_check_fn_t, xlist_t*); extern int get_queue_exec(cluster_t*, char*, char**, char**); extern void unlock_collect_queues(xlist_t*); extern void unlock_collect_jobs(xlist_t*); diff --git a/src/controller/meson.build b/src/controller/meson.build index 5cf7bce3..fafb5228 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -122,6 +122,7 @@ controller_unit_test_src = [ 'testsuite/test_commit_engine.c', 'testsuite/test_commit_barrier_rollout.c', 'testsuite/test_commit_barrier_reply.c', + 'testsuite/test_egress_filter.c', 'commit.c', 'job.c', 'queue.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index ad8f1680..34300030 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -52,6 +52,7 @@ extern void test_ha_liveness(void); extern void test_commit_engine(void); extern void test_commit_barrier_rollout(void); extern void test_commit_barrier_reply(void); +extern void test_egress_filter(void); void setUp(void) { return; } @@ -173,6 +174,7 @@ int main(int argc, char* argv[]) TEST(test_commit_engine); TEST(test_commit_barrier_rollout); TEST(test_commit_barrier_reply); + TEST(test_egress_filter); return UNITY_END(); } diff --git a/src/controller/testsuite/test_egress_filter.c b/src/controller/testsuite/test_egress_filter.c new file mode 100644 index 00000000..78a4012e --- /dev/null +++ b/src/controller/testsuite/test_egress_filter.c @@ -0,0 +1,537 @@ +/* + * test_egress_filter.c — RED-phase tests for commit-gated scheduler egress + * (ha-convergence-recovery task 1.1). + * + * Requirements (this spec's IDs): + * 3.1 — a job whose submit entry is not yet committed is EXCLUDED from + * scheduler-bound exports (collect_jobs_committed) and from the + * JOB_NEW incremental push list. + * 3.2 — once committed_lsn advances to cover the submit entry, the job is + * INCLUDED in subsequent scheduler-bound exports. + * 3.3 — a job created live carries its submit entry's LSN in submit_lsn; a + * job restored via apply_wal_entry / restore_job replay carries the + * replayed entry's LSN. + * 3.4 — a job with an uncommitted submit entry remains visible to the + * status/API read path (plain collect_jobs is NOT commit-filtered). + * submit_lsn == 0 is always exportable (pre-feature / unknown). + * + * RED CONTRACT: these tests are written against the task-1.2 GREEN behavior. + * They FAIL today because: + * - persist_job_hook_fun does not yet set j->submit_lsn (live tagging gap) + * - apply_wal_entry passes 0 (not hdr->lsn) to restore_job, and restore_job + * does not tag submit_lsn (replay tagging gap) + * - job_submit_committed / collect_jobs_committed are not yet implemented in + * dataex.c (the RED scaffold below is intentionally unfiltered — task 1.2 + * DELETES this scaffold and implements the real predicate + wrapper in + * dataex.c per design.md "Egress Filter"). + * + * Harness: a real persist_t over a private tmpdir embedded in a test + * cluster_t with a registered jobmgr and persistence hooks, a wired commit + * engine, and a stub standby registry — mirrors test_commit_barrier_reply.c. + */ + +/* rtti.h first — defines 'var'/'typeinfo_t' used by kvmgr.h */ +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "controller.capnp.h" +#include "xce/cluster_def.h" +#include "xce/types.h" + +#include "cluster.h" +#include "commit.h" +#include "hook.h" +#include "jobj.h" +#include "kvmgr.h" +#include "persist.h" +#include "repl.h" +#include "resobj.h" +#include "utils.h" +#include "wal_types.h" +#include "xlist.h" + +/* ── Local job collector (avoids linking dataex.c into this unit binary) ── + * + * dataex.c is NOT in controller_unit_test_src (its notify()/collect_confvars() + * clash with stubs already in test_commit_barrier_reply.c / test_worker_join.c). + * collect_jobs / unlock_collect_jobs therefore are not linkable here, so this + * file uses ef_collect_jobs — a faithful copy of collect_jobs' jobmgr iteration + * (dataex.c:294) — for the unfiltered status/API path AND as the iteration core + * of the RED scaffold below. The behavior asserted is identical to collect_jobs; + * only the symbol is local. */ +static void ef_collect_jobs(cluster_t *c, xlist_t *l) +{ + iterator_t it; + + lock(c->jobmgr, RTTI_LOCK_READ); + if (iter_make(&it, c->jobmgr) < 0) { + unlock(c->jobmgr); + return; + } + for (;;) { + job_t *j = NULL; + int rc = iter_next(&it, (void **)&j); + if (rc < 0 || rc == RTTI_ITER_NO_DATA) + break; + if (j == NULL) + continue; + reference(j); + xlist_store(l, &(j->job)); + } + iter_destroy(&it); + unlock(c->jobmgr); +} + +static void ef_unlock_jobs(xlist_t *l) +{ + for (int i = 0; i < l->count; i++) { + xce_job_t *xj = (xce_job_t *)xlist_get(l, i); + if (xj == NULL) + continue; + job_t *j = CONTAINER_OF(xj, job_t, job); + dereference(j); + } +} + +/* ── RED scaffold: minimal, intentionally UNFILTERED stand-ins ─────────── + * + * These satisfy the linker (job_submit_committed / collect_jobs_committed are + * declared in cluster.h but not yet defined). They DO NOT implement the commit + * watermark — job_submit_committed always returns true and collect_jobs_committed + * collects every job. That makes the 3.1/3.2 exclusion assertions FAIL (the + * uncommitted job leaks through), which is the intended RED state. + * + * Task 1.2 (GREEN) MUST delete this entire scaffold block and provide the real + * definitions in dataex.c per design.md "Egress Filter": + * job_submit_committed -> submit_lsn == 0 || submit_lsn <= committed_lsn + * collect_jobs_committed -> collect_jobs filtered by that predicate + * At that point dataex.c becomes part of the test link (with the notify/ + * collect_confvars stub clashes reconciled) and this scaffold is removed to + * avoid duplicate symbols. */ +bool job_submit_committed(cluster_t *c, job_t *j) +{ + (void)c; + (void)j; + return true; /* RED: unfiltered — task 1.2 replaces with the watermark */ +} + +void collect_jobs_committed(cluster_t *c, collect_check_fn_t fn, xlist_t *l) +{ + (void)fn; /* scheduler-bound callers pass NULL */ + ef_collect_jobs(c, l); /* RED: no commit filtering */ +} +/* ── end RED scaffold ──────────────────────────────────────────────────── */ + +/* notify() is provided by test_commit_barrier_reply.c (same test binary) — do + * not redefine it here (would be a duplicate symbol). */ + +/* Test-injectable send hook (repl.c, XCE_TESTING): keep repl_broadcast from + * dereferencing the dummy standby endpoints during the priming append. */ +extern int (*repl_broadcast_send_hook)(void *ep, + wal_catchup_entry_t **entries, + int n_entries); + +static int ef_drop_send_hook(void *ep, wal_catchup_entry_t **entries, + int n_entries) +{ + (void)ep; + (void)entries; + (void)n_entries; + return 0; +} + +/* ── Cluster harness (mirror of test_commit_barrier_reply.c cbrp_setup) ── */ +static cluster_t g_c; +static commit_engine_t g_ce; +static int g_ep_a, g_ep_b; +static char g_tmpdir[256]; +static int g_jobobj_ready; + +static void ef_setup(void) +{ + persist_config_t cfg; + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 64, + .delta = 16 }; + kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, + .nslots = 64, + .delta = 16 }; + + memset(&g_c, 0, sizeof(g_c)); + + g_c.myhost = &g_c.myhost_info; + g_c.myhost_info.name = (char *)"ctrl-test"; + g_c.myhost_info.key = (char *)""; + + TEST_ASSERT_EQUAL_INT( + 0, xce_hook_init(&g_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_c.qmgr); + TEST_ASSERT_NOT_NULL(g_c.workermgr); + TEST_ASSERT_NOT_NULL(g_c.resmgr); + zarray_init(&g_c.sharedinst, 64, NULL); + + if (!g_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64)); + g_jobobj_ready = 1; + } + + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = 50; /* short — no real standby ACKs in this suite */ + + snprintf(g_tmpdir, sizeof(g_tmpdir), "/tmp/xce_egress_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_tmpdir)); + persist_set_workdir(&g_c.persist, g_tmpdir, "ctrl"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_c.persist, &cfg, &g_c)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_c.persist)); + + TEST_ASSERT_TRUE( + repl_add_standby(&g_c.persist.repl, &g_ep_a, "sb_a", 7400) >= 0); + TEST_ASSERT_TRUE( + repl_add_standby(&g_c.persist.repl, &g_ep_b, "sb_b", 7401) >= 0); + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_c.persist, 3, 8)); + repl_set_commit_engine(&g_c.persist.repl, &g_ce); + + /* Install the drop send hook BEFORE any append that broadcasts — the dummy + * standby endpoints (g_ep_a/g_ep_b) are not real endpoint_t and must never + * be dereferenced by repl_broadcast. */ + repl_broadcast_send_hook = ef_drop_send_hook; + + /* Priming append so subsequent real entries land at LSN >= 1 (the engine + * rejects lsn == 0; matches test_commit_barrier_reply.c cbrp_setup). This + * also ensures the live-tag test's appended LSN is non-zero, so an untagged + * submit_lsn (0) cannot coincidentally match the appended LSN. */ + TEST_ASSERT_EQUAL_INT( + 0, persist_wal_async(&g_c.persist, WAL_JOB_UPDATE, "p", 1)); + TEST_ASSERT_EQUAL_UINT64(1, g_c.persist.current_lsn); +} + +static void ef_teardown(void) +{ + repl_broadcast_send_hook = NULL; + commit_engine_fini(&g_ce); + persist_stop(&g_c.persist); + persist_fini(&g_c.persist); + + zarray_fini(&g_c.sharedinst); + if (g_c.resmgr) dereference(g_c.resmgr); + if (g_c.workermgr) dereference(g_c.workermgr); + if (g_c.qmgr) dereference(g_c.qmgr); + if (g_c.jobmgr) dereference(g_c.jobmgr); + xce_hook_fini(&g_c.hooksys); +} + +/* Create a pending job directly in g_c.jobmgr; returns the job_t wrapper. + * submit must be non-NULL (makeJob → makeJobsub deep-copies it; mirror of + * test_commit_barrier_reply.c cbrp_make_job). */ +static job_submit_data_t g_ef_submit; /* zeroed; deep-copied by makeJob */ + +static job_t *ef_make_job(uint64_t jobid) +{ + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + memset(&g_ef_submit, 0, sizeof(g_ef_submit)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_PEND; + jdef.submit = &g_ef_submit; + job_t *j = makeJob(g_c.jobmgr, &jdef); + TEST_ASSERT_NOT_NULL(j); + return j; +} + +/* Look up the job_t wrapper for a jobid (NULL if absent). */ +static job_t *ef_find_job(uint64_t jobid) +{ + job_t *j = NULL; + lock(g_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_c.jobmgr, (void *)jobid, (void **)&j); + unlock(g_c.jobmgr); + return (found > 0) ? j : NULL; +} + +/* Is jobid present in the collected xlist? (collect lists store &j->job). */ +static int ef_list_has_job(xlist_t *l, uint64_t jobid) +{ + for (int i = 0; i < l->count; i++) { + xce_job_t *xj = (xce_job_t *)xlist_get(l, i); + if (xj != NULL && xj->jobid == jobid) + return 1; + } + return 0; +} + +/* Serialize a minimal xce_job_t to a Cap'n Proto Job buffer (mirror of the + * static capnp_encode_job in persist.c — needed to drive apply_wal_entry). */ +static int ef_encode_job(const xce_job_t *j, uint8_t **out, size_t *out_len) +{ + struct capn c; + struct capn_segment *cs; + struct Job cj; + Job_ptr jp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Job(cs, &cj, (xce_job_t *)j); + jp = new_Job(cs); + write_Job(&cj, jp); + capn_setp(capn_root(&c), 0, jp.p); + + *out_len = capn_size(&c); + *out = malloc(*out_len); + if (!*out) { + capn_free(&c); + return -ENOMEM; + } + capn_write_mem(&c, *out, *out_len, 0); + capn_free(&c); + return 0; +} + +/* Fire a job persistence hook through a real va_list (the hook reads its args + * via va_arg, mirroring test_persist_hooks.c's call_job_hook). */ +static int call_job_hook(int detail, persist_t *p, ...) +{ + va_list ap; + va_start(ap, p); + int rc = persist_job_hook_fun(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, detail, + p, ap); + va_end(ap); + return rc; +} + +/* JOB_NEW AFTER_EVENT args: (cluster_t*, xlist_t* jlist). */ +static int call_job_hook_new(persist_t *p, cluster_t *c, xlist_t *jlist) +{ + return call_job_hook(HOOK_DETAIL_JOB_NEW, p, c, jlist); +} + +/* ── Tests ──────────────────────────────────────────────────────────────── */ + +/* 3.3 (live tag): firing the JOB_NEW persistence hook for a submitted job + * must tag the job_t with the LSN of the WAL_JOB_SUBMIT entry it wrote. + * RED: persist_job_hook_fun does not set submit_lsn yet. */ +static void test_live_submit_tags_submit_lsn(void) +{ + ef_setup(); + + job_t *j = ef_make_job(101); + + uint64_t lsn_before = g_c.persist.current_lsn; + + /* JOB_NEW AFTER_EVENT args: (cluster_t*, xlist_t* jlist) */ + void *jptr = j; + xlist_t jlist = { .count = 1, .caps = 1, .delta = 1, .data = &jptr }; + + /* The JOB_NEW hook does a SYNC append; with no real standby ACKs in this + * harness it returns -ETIMEDOUT, but the LOCAL WAL write still succeeds and + * the LSN advances (production keeps the entry on replication-degraded + * timeout). The tagging contract is independent of replication status, so + * we require only that the local append happened. */ + int rc = call_job_hook_new(&g_c.persist, &g_c, &jlist); + TEST_ASSERT_TRUE_MESSAGE(rc == 0 || rc == -ETIMEDOUT || rc == -ENODEV, + "JOB_NEW local append should have succeeded"); + + uint64_t appended_lsn = g_c.persist.current_lsn - 1; + TEST_ASSERT_GREATER_THAN_UINT64(lsn_before, g_c.persist.current_lsn); + TEST_ASSERT_GREATER_THAN_UINT64(0, appended_lsn); /* not the LSN-0 edge */ + + /* GREEN expectation: the job carries the submit entry's LSN. */ + TEST_ASSERT_EQUAL_UINT64(appended_lsn, j->submit_lsn); + + ef_teardown(); + printf("test_live_submit_tags_submit_lsn: checked\n"); +} + +/* 3.3 (replay tag): a job restored via apply_wal_entry's WAL_JOB_SUBMIT path + * must carry the replayed entry's LSN in submit_lsn. + * RED: apply_wal_entry passes 0 (not hdr->lsn) to restore_job, and + * restore_job does not tag submit_lsn. */ +static void test_replay_restore_tags_entry_lsn(void) +{ + ef_setup(); + + const uint64_t replay_lsn = 77; + + /* Encode a job WITH submit data so restore_job's makeJob (makeJobsub + * rejects a NULL submit) succeeds — mirrors a real WAL_JOB_SUBMIT entry. */ + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = 202; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + uint8_t *payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, ef_encode_job(&jdef, &payload, &plen)); + + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.type = WAL_JOB_SUBMIT; + hdr.lsn = replay_lsn; + hdr.payload_len = (uint32_t)plen; + + int rc = apply_wal_entry(&g_c.persist, &hdr, payload); + TEST_ASSERT_EQUAL_INT(0, rc); + free(payload); + + job_t *restored = ef_find_job(202); + TEST_ASSERT_NOT_NULL(restored); + + /* GREEN expectation: restored job is tagged with the replayed entry LSN. */ + TEST_ASSERT_EQUAL_UINT64(replay_lsn, restored->submit_lsn); + + ef_teardown(); + printf("test_replay_restore_tags_entry_lsn: checked\n"); +} + +/* 3.1: a job whose submit entry is uncommitted is absent from the + * scheduler-bound export (collect_jobs_committed). + * RED: the unfiltered scaffold lets it through. */ +static void test_uncommitted_job_absent_from_scheduler_export(void) +{ + ef_setup(); + + /* committed_lsn = 5; job's submit entry at LSN 9 is uncommitted. */ + atomic_store(&g_c.persist.committed_lsn, 5); + job_t *j = ef_make_job(303); + j->submit_lsn = 9; + + xlist_t out; + TEST_ASSERT_EQUAL_INT(0, xlist_init(&out, 16)); + collect_jobs_committed(&g_c, NULL, &out); + + TEST_ASSERT_FALSE_MESSAGE( + ef_list_has_job(&out, 303), + "uncommitted job (submit_lsn=9 > committed_lsn=5) leaked into " + "scheduler export"); + + ef_unlock_jobs(&out); + xlist_fini(&out); + ef_teardown(); + printf("test_uncommitted_job_absent_from_scheduler_export: checked\n"); +} + +/* 3.2: once committed_lsn advances to cover the submit entry, the job appears + * in subsequent scheduler-bound exports. + * RED: depends on collect_jobs_committed honoring the watermark; with the + * unfiltered scaffold the BEFORE assertion (3.1) already fails, and this test + * additionally pins that the predicate flips at the boundary. */ +static void test_job_present_after_commit_advances(void) +{ + ef_setup(); + + atomic_store(&g_c.persist.committed_lsn, 5); + job_t *j = ef_make_job(404); + j->submit_lsn = 9; + + /* Before: uncommitted → predicate false, absent from export. */ + TEST_ASSERT_FALSE(job_submit_committed(&g_c, j)); + + xlist_t before; + TEST_ASSERT_EQUAL_INT(0, xlist_init(&before, 16)); + collect_jobs_committed(&g_c, NULL, &before); + TEST_ASSERT_FALSE(ef_list_has_job(&before, 404)); + ef_unlock_jobs(&before); + xlist_fini(&before); + + /* Advance committed_lsn to cover the submit entry. */ + atomic_store(&g_c.persist.committed_lsn, 9); + + /* After: committed → predicate true, present in export. */ + TEST_ASSERT_TRUE(job_submit_committed(&g_c, j)); + + xlist_t after; + TEST_ASSERT_EQUAL_INT(0, xlist_init(&after, 16)); + collect_jobs_committed(&g_c, NULL, &after); + TEST_ASSERT_TRUE_MESSAGE( + ef_list_has_job(&after, 404), + "job absent from scheduler export after committed_lsn advanced to " + "cover its submit entry"); + ef_unlock_jobs(&after); + xlist_fini(&after); + + ef_teardown(); + printf("test_job_present_after_commit_advances: checked\n"); +} + +/* submit_lsn == 0 (pre-feature / unknown) is always exportable. */ +static void test_submit_lsn_zero_always_exportable(void) +{ + ef_setup(); + + atomic_store(&g_c.persist.committed_lsn, 5); + job_t *j = ef_make_job(505); + j->submit_lsn = 0; + + TEST_ASSERT_TRUE_MESSAGE( + job_submit_committed(&g_c, j), + "submit_lsn == 0 must be treated as exportable"); + + xlist_t out; + TEST_ASSERT_EQUAL_INT(0, xlist_init(&out, 16)); + collect_jobs_committed(&g_c, NULL, &out); + TEST_ASSERT_TRUE(ef_list_has_job(&out, 505)); + ef_unlock_jobs(&out); + xlist_fini(&out); + + ef_teardown(); + printf("test_submit_lsn_zero_always_exportable: checked\n"); +} + +/* 3.4: the status/API read path (plain collect_jobs) is NOT commit-filtered — + * an uncommitted job remains visible there. This pins the documented + * deviation; it passes today and must keep passing after 1.2 (collect_jobs is + * never modified). Included so the suite guards 3.4 from accidental drift. */ +static void test_uncommitted_job_visible_to_status_path(void) +{ + ef_setup(); + + atomic_store(&g_c.persist.committed_lsn, 5); + job_t *j = ef_make_job(606); + j->submit_lsn = 9; /* uncommitted */ + + xlist_t out; + TEST_ASSERT_EQUAL_INT(0, xlist_init(&out, 16)); + ef_collect_jobs(&g_c, &out); /* status/API path — unfiltered */ + + TEST_ASSERT_TRUE_MESSAGE( + ef_list_has_job(&out, 606), + "status/API read path must remain unfiltered (Req 3.4)"); + + ef_unlock_jobs(&out); + xlist_fini(&out); + ef_teardown(); + printf("test_uncommitted_job_visible_to_status_path: checked\n"); +} + +/* ── Suite entry point ──────────────────────────────────────────────────── */ + +void test_egress_filter(void) +{ + printf("\n>> commit-gated scheduler egress tests (RED, task 1.1)\n\n"); + + RUN_TEST(test_live_submit_tags_submit_lsn); + RUN_TEST(test_replay_restore_tags_entry_lsn); + RUN_TEST(test_uncommitted_job_absent_from_scheduler_export); + RUN_TEST(test_job_present_after_commit_advances); + RUN_TEST(test_submit_lsn_zero_always_exportable); + RUN_TEST(test_uncommitted_job_visible_to_status_path); +} diff --git a/src/include/xce/types.h b/src/include/xce/types.h index 70aecf8b..07c1f798 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -482,6 +482,11 @@ typedef struct { typedef struct { xce_job_t job; + /* ha-convergence-recovery (3.3): WAL entry LSN of this job's submit record. + * In-memory only — recomputed from the applied entry at replay; never + * serialized (wire type xce_job_t above is untouched). 0 = unknown / + * pre-feature = always exportable (see job_submit_committed, cluster.h). */ + uint64_t submit_lsn; } job_t; typedef struct {