fix(controller): mark RUN jobs unknown on worker loss and restart (bug-1509 P2)

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

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

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

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

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

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

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

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