From fc091348e81f42c3cf2f24a6d48b6b4cc4f9816a Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Thu, 6 Aug 2026 05:57:59 +0000 Subject: [PATCH] fix(controller): mark RUN jobs unknown on worker loss and restart (bug-1509 P2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of the job-finish robustness design (Part B, B1+B3). B1 (worker liveness): a worker that stops reporting (heartbeat timeout, already detected by check_worker_timeout_) or disconnects now has its RUN jobs marked JOB_STAT_UNKWN|JOB_STAT_TRANSIENT (transient_since set, durable WAL_JOB_UPDATE) instead of lingering as RUN forever. Marking happens AFTER the worker lock is released (B5: avoids worker->jobmgr inversion with the scheduler's jobmgr->worker order). B3 (restart): persist_recover marks all recovered RUN jobs UNKWN+TRANSIENT in-memory after replay — a controller restart leaves worker connectivity unknown. In-memory only (no WAL append) because persist_recover may run on a future standby that must never append. Unit tests (red->green): worker-loss marks its RUN jobs unknown, other-host jobs untouched, restart marks all RUN jobs. test_replay_apply_seam updated: replay still drives facts to RUN, then B3 adds UNKWN+TRANSIENT. --- src/controller/cluster.c | 12 ++- src/controller/include/cluster.h | 8 ++ src/controller/job.c | 85 ++++++++++++++++ src/controller/persist.c | 9 ++ src/controller/testsuite/test_apply_pump.c | 99 +++++++++++++++++++ .../testsuite/test_replay_apply_seam.c | 10 +- src/controller/worker.c | 5 + 7 files changed, 224 insertions(+), 4 deletions(-) diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 1146620..79b1e82 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -1731,7 +1731,9 @@ static int check_worker_timeout_(cluster_t* c, worker_t* w) mark_worker_unavail(c, w); - return 0; + /* bug-1509 B1: return 1 so the caller (outside the worker lock) can + * mark this worker's RUN jobs as UNKWN+TRANSIENT (B5 lock order). */ + return 1; } else { if (IS_WORKER_UNAVAIL(w->host.status)) { mark_worker_back(c, w); @@ -1811,9 +1813,15 @@ static int check_workers_timeout(void* args) } lock(w, RTTI_LOCK_WRITE); - check_worker_timeout_(c, w); + int became_unavail = check_worker_timeout_(c, w); unlock(w); + /* bug-1509 B1: worker timed out — mark its RUN jobs unknown AFTER + * releasing the worker lock (B5: avoids worker→jobmgr inversion). */ + if (became_unavail) { + mark_worker_jobs_unknown(c, w->host.name); + } + dereference(w); } diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 3c53137..7bb0be5 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -50,6 +50,14 @@ extern int restore_job(cluster_t*, uint64_t, const void*, size_t, uint64_t); * from apply_committed_entry's WAL_JOB_SUBMIT case after restore_job. */ extern int apply_registered_finish_if_any(cluster_t*, job_t*); +/* bug-1509 Phase 2 (B1): mark RUN jobs as UNKWN+TRANSIENT when their worker + * is lost (hostname set, durable WAL). Caller must not hold any + * worker/jobmgr lock (B5). */ +extern void mark_worker_jobs_unknown(cluster_t*, const char* hostname); +/* bug-1509 Phase 2 (B3): after restart, mark ALL RUN jobs UNKWN+TRANSIENT + * in-memory only (no WAL — persist_recover may run on a future standby). */ +extern void mark_all_run_jobs_unknown_in_memory(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 56bbdb7..99bf839 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -2098,6 +2098,91 @@ static int mark_job_update(cluster_t* c, job_t* job, int state) return 0; } +/* bug-1509 Phase 2 (B1): a worker that stops reporting (liveness timeout) or + * disconnects is marked UNAVAIL, but its RUN jobs must not linger as RUN + * forever — mark them UNKWN+TRANSIENT (transient_since = now) so (a) the + * state is visibly "unknown" and (b) job_should_retain's TRANSIENT TTL can GC + * them if the worker never comes back. hostname == NULL marks ALL RUN jobs + * (used by persist_recover's restart marking, B3). + * + * Lock discipline (B5): the caller must NOT hold any worker or jobmgr lock. + * This function takes the jobmgr read lock and per-job write locks, which + * would invert the scheduler's jobmgr→worker order if a worker lock were held + * here. */ +/* Shared core of the bug-1509 unknown-marking. persist=1 writes a durable + * WAL_JOB_UPDATE (worker-death path, B1); persist=0 mutates in-memory only + * (restart path, B3 — persist_recover runs before the controller knows it + * will be primary, and a standby must never append). */ +static void mark_jobs_unknown_core(cluster_t* c, const char* hostname, + int persist) +{ + iterator_t it; + job_t* j; + + if (c == NULL) { + return; + } + + lock(c->jobmgr, RTTI_LOCK_READ); + if (iter_make(&it, c->jobmgr) < 0) { + unlock(c->jobmgr); + return; + } + for (;;) { + int rc = iter_next(&it, (void**)&j); + if (rc < 0 || rc == RTTI_ITER_NO_DATA) { + break; + } + if (j == NULL) { + continue; + } + + if (hostname != NULL) { + int on_host = 0; + if (j->job.allocs != NULL && j->job.allocs->allocs != NULL) { + for (int i = 0; i < j->job.allocs->nhosts; i++) { + xce_host_alloc_t* ha = j->job.allocs->allocs[i]; + if (ha != NULL && ha->host != NULL + && strcmp(ha->host, hostname) == 0) { + on_host = 1; + break; + } + } + } + if (!on_host) { + continue; + } + } + + lock(j, RTTI_LOCK_WRITE); + if (JOB_IS_RUN(j->job.state) + && !job_state_is_terminal(j->job.state)) { + j->job.transient_since = time(NULL); + if (persist) { + mark_job_update(c, j, j->job.state | JOB_STAT_UNKWN + | JOB_STAT_TRANSIENT); + } else { + j->job.state |= JOB_STAT_UNKWN | JOB_STAT_TRANSIENT; + } + } + unlock(j); + } + iter_destroy(&it); + unlock(c->jobmgr); +} + +void mark_worker_jobs_unknown(cluster_t* c, const char* hostname) +{ + mark_jobs_unknown_core(c, hostname, 1); +} + +/* bug-1509 B3: restart path — after replay, RUN jobs' state is uncertain + * (worker connectivity unknown). In-memory only; idempotent. */ +void mark_all_run_jobs_unknown_in_memory(cluster_t* c) +{ + mark_jobs_unknown_core(c, NULL, 0); +} + /* Guarded state-transition core for a postdone fact, extracted from * on_job_postdone's old inline mutation (raft-apply-decoupling task 6.1 * round 2, Req 3.3/7.1-7.3). Unlike a generic UPDATE, postdone's formula diff --git a/src/controller/persist.c b/src/controller/persist.c index aa7efdd..6a5cd5b 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -5381,6 +5381,15 @@ int persist_recover(persist_t* p) * own walk drains fully from last_applied_lsn+1 through the current * committed_lsn, so this is a complete, ordered catch-up, not a partial * one; a harmless no-op if nothing queued. */ + + /* bug-1509 B3: after restart, RUN jobs' state is uncertain (worker + * connectivity unknown) — mark them UNKWN+TRANSIENT in-memory only. + * No WAL append: persist_recover runs before the controller knows it + * will be primary, and a standby must never append. Idempotent. */ + if (p->cluster) { + mark_all_run_jobs_unknown_in_memory(p->cluster); + } + atomic_store_explicit(&p->recovery_complete, true, memory_order_release); /* n_vcpus==0 is this codebase's own established "not yet initialized" * sentinel for vtaskmgr_t (mirrors cluster_init_phase2's own diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c index 7cacd02..006f0d0 100644 --- a/src/controller/testsuite/test_apply_pump.c +++ b/src/controller/testsuite/test_apply_pump.c @@ -1736,11 +1736,110 @@ static void test_unknown_finish_drain_noop_without_registration(void) printf("test_unknown_finish_drain_noop_without_registration: PASS\n"); } +/* ── 6. bug-1509 Phase 2: B1 (worker loss marks its RUN jobs UNKWN+TRANSIENT) + * ── and B3 (restart marks ALL RUN jobs UNKWN+TRANSIENT). */ + +static void tap_set_job_allocs(job_t* j, const char* host) +{ + xce_host_alloc_t* ha = calloc(1, sizeof(*ha)); + TEST_ASSERT_NOT_NULL(ha); + ha->host = strdup(host); + + xce_job_alloc_t* ja = calloc(1, sizeof(*ja)); + TEST_ASSERT_NOT_NULL(ja); + ja->jobid = j->job.jobid; + ja->nhosts = 1; + ja->allocs = calloc(1, sizeof(xce_host_alloc_t*)); + TEST_ASSERT_NOT_NULL(ja->allocs); + ja->allocs[0] = ha; + j->job.allocs = ja; +} + +static void test_worker_loss_marks_its_run_jobs_unknown(void) +{ + tap_setup(); + + job_t* j = tap_seed_job(1001, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + lock(j, RTTI_LOCK_WRITE); + tap_set_job_allocs(j, "t1"); + unlock(j); + + mark_worker_jobs_unknown(&g_tap_c, "t1"); + + TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_UNKWN, + "job must be UNKWN after worker loss"); + TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_TRANSIENT, + "job must be TRANSIENT after worker loss"); + TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_RUN, + "RUN bit must be preserved"); + TEST_ASSERT_TRUE_MESSAGE(j->job.transient_since > 0, + "transient_since must be set"); + + tap_teardown(); + printf("test_worker_loss_marks_its_run_jobs_unknown: PASS\n"); +} + +static void test_worker_loss_does_not_touch_other_host_jobs(void) +{ + tap_setup(); + + job_t* j_t1 = tap_seed_job(1002, JOB_STAT_RUN); + job_t* j_t2 = tap_seed_job(1003, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j_t1); + TEST_ASSERT_NOT_NULL(j_t2); + lock(j_t1, RTTI_LOCK_WRITE); + tap_set_job_allocs(j_t1, "t1"); + unlock(j_t1); + lock(j_t2, RTTI_LOCK_WRITE); + tap_set_job_allocs(j_t2, "t2"); + unlock(j_t2); + + mark_worker_jobs_unknown(&g_tap_c, "t1"); + + TEST_ASSERT_TRUE_MESSAGE(j_t1->job.state & JOB_STAT_UNKWN, + "t1 job must be marked"); + TEST_ASSERT_FALSE_MESSAGE(j_t2->job.state & JOB_STAT_UNKWN, + "t2 job must be untouched"); + + tap_teardown(); + printf("test_worker_loss_does_not_touch_other_host_jobs: PASS\n"); +} + +static void test_restart_marks_all_run_jobs_unknown(void) +{ + tap_setup(); + + job_t* j_run1 = tap_seed_job(1004, JOB_STAT_RUN); + job_t* j_run2 = tap_seed_job(1005, JOB_STAT_RUN); + job_t* j_done = tap_seed_job(1006, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j_run1); + TEST_ASSERT_NOT_NULL(j_run2); + TEST_ASSERT_NOT_NULL(j_done); + + mark_all_run_jobs_unknown_in_memory(&g_tap_c); + + TEST_ASSERT_TRUE_MESSAGE(j_run1->job.state & JOB_STAT_UNKWN, + "run job 1 must be marked"); + TEST_ASSERT_TRUE_MESSAGE(j_run2->job.state & JOB_STAT_UNKWN, + "run job 2 must be marked"); + TEST_ASSERT_FALSE_MESSAGE(j_done->job.state & JOB_STAT_UNKWN, + "done job must be untouched"); + TEST_ASSERT_TRUE_MESSAGE(j_run1->job.transient_since > 0, + "transient_since must be set on run job"); + + tap_teardown(); + printf("test_restart_marks_all_run_jobs_unknown: PASS\n"); +} + /* ── Test group entry point ───────────────────────────────────────────── */ void test_apply_pump(void) { UnitySetTestFile(__FILE__); + RUN_TEST(test_worker_loss_marks_its_run_jobs_unknown); + RUN_TEST(test_worker_loss_does_not_touch_other_host_jobs); + RUN_TEST(test_restart_marks_all_run_jobs_unknown); RUN_TEST(test_unknown_finish_registry_register_and_take); RUN_TEST(test_unknown_finish_applied_once_job_materializes); RUN_TEST(test_unknown_finish_exit_code_maps_to_exit); diff --git a/src/controller/testsuite/test_replay_apply_seam.c b/src/controller/testsuite/test_replay_apply_seam.c index 130eb10..6cbaae8 100644 --- a/src/controller/testsuite/test_replay_apply_seam.c +++ b/src/controller/testsuite/test_replay_apply_seam.c @@ -361,9 +361,15 @@ static void test_replay_applies_every_classified_type_via_apply_seam(void) TEST_ASSERT_NOT_NULL_MESSAGE(j, "replay must construct the job via the apply seam's " "WAL_JOB_SUBMIT route"); - TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, + /* bug-1509 B3: after recovery, RUN jobs' state is uncertain (worker + * connectivity unknown), so persist_recover marks them UNKWN+TRANSIENT + * in-memory. The replay itself must still drive the fact to RUN; the + * UNKWN/TRANSIENT bits come from B3's post-replay marking. */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN | JOB_STAT_UNKWN + | JOB_STAT_TRANSIENT, + j->job.state, "replay must apply the job-state fact via the apply seam's " - "guarded transition core, reaching the same end state as before"); + "guarded transition core, then B3 marks the RUN job UNKWN"); queue_t* q = trr_find_queue("rq1"); TEST_ASSERT_NOT_NULL(q); diff --git a/src/controller/worker.c b/src/controller/worker.c index e0c4e3d..cf858fb 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -726,6 +726,11 @@ int handle_worker_leave(cluster_t* c, int idx) unlock(w); + /* bug-1509 B1: worker lost — mark its RUN jobs UNKWN+TRANSIENT AFTER + * releasing the worker lock (B5: avoids worker→jobmgr inversion with the + * scheduler's jobmgr→worker order). */ + mark_worker_jobs_unknown(c, w->host.name); + return 0; }