fix(controller): apply worker finish arriving before job materializes (bug-1509)

A worker-reported finish may arrive before the job exists: job
construction is deferred to apply_committed_entry (the submit's WAL
entry), leaving a commit->apply window. on_job_finish dropped such
finishes ('can not find job'), so a fast job that finished in that
window stayed RUN forever (e.g. noexec-test/newcfg-test/verify-fix).

Add an in-memory unknown-finish registry: on_job_finish registers the
finish when the job isn't materialized, and apply_committed_entry's
WAL_JOB_SUBMIT case drains it right after restore_job — appending the
durable WAL_JOB_FINISH and mutating the job to DONE/EXIT immediately so
the scheduler never dispatches a job that already finished.

Design: docs/Job-Finish-Robustness-设计文档.md (Part A of A+B; Part B
worker-liveness + reconnect reconciliation deferred to later phases).

Unit tests (red->green): registry register/take, apply-once-job-
materializes (DONE), exit-code maps to EXIT, drain no-op without
registration.

src/controller/include/cluster.h

@@ -45,6 +45,16 @@ extern int update_resvals(cluster_t*, worker_t*, int, xce_resval_t*);
 
 extern int load_jobs(void*);
 extern int restore_job(cluster_t*, uint64_t, const void*, size_t, uint64_t);
+/* bug-1509: drain a worker finish that arrived before the job was
+ * materialized (registered by on_job_finish's unknown-finish path). Called
+ * from apply_committed_entry's WAL_JOB_SUBMIT case after restore_job. */
+extern int apply_registered_finish_if_any(cluster_t*, job_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*);
+extern int test_unknown_finish_take(uint64_t, xce_jobfinish_t*);
+#endif
 /* job-array-single-wal-record task 3.1 (design.md Array Apply Expansion):
  * shared N-way expansion primitive for a committed WAL_JOB_ARRAY_SUBMIT
  * payload -- see the doc comment on the definition (job.c) for the full

src/controller/job.c

@@ -1759,6 +1759,140 @@ static int record_job_finish_fact(cluster_t* c, job_t* job,
     return 0;
 }
 
+/* ── Unknown-finish registry (bug-1509) ────────────────────────────────────
+ *
+ * A worker-reported finish may arrive at the controller BEFORE the job is
+ * materialized: job construction is deferred to apply_committed_entry (the
+ * submit's WAL entry), so there is a commit→apply window during which
+ * key_find(jobmgr) misses the job. on_job_finish used to drop such finishes
+ * ("can not find job"), leaving the job RUN forever. This registry holds the
+ * finish until the job materializes; apply_committed_entry's WAL_JOB_SUBMIT
+ * case then drains it via apply_registered_finish_if_any().
+ *
+ * The registry is process-local and NOT durable across failover (a finish
+ * registered but unconsumed when the primary fails over is lost; the new
+ * primary replays the submit with the job stuck PEND). Documented residual
+ * risk (design §3.6/A3) — the commit→apply window is small. */
+typedef struct unknown_finish {
+    uint64_t               jobid;
+    xce_jobfinish_t        finish;    /* owned copy (2 scalar fields) */
+    time_t                 queued_at;
+    struct unknown_finish* next;
+} unknown_finish_t;
+
+static pthread_mutex_t   g_unknown_finish_mu = PTHREAD_MUTEX_INITIALIZER;
+static unknown_finish_t* g_unknown_finish_list = NULL;
+
+#define UNKNOWN_FINISH_TTL_S (60) /* must exceed max commit→apply lag */
+
+static void unknown_finish_sweep_locked(void)
+{
+    unknown_finish_t** pp = &g_unknown_finish_list;
+    time_t            now = time(NULL);
+    while (*pp != NULL) {
+        if (now - (*pp)->queued_at > UNKNOWN_FINISH_TTL_S) {
+            unknown_finish_t* dead = *pp;
+            *pp = dead->next;
+            free(dead);
+        } else {
+            pp = &(*pp)->next;
+        }
+    }
+}
+
+static int unknown_finish_register(const xce_jobfinish_t* finish)
+{
+    unknown_finish_t* uf;
+
+    if (finish == NULL) {
+        return -EINVAL;
+    }
+    uf = (unknown_finish_t*)calloc(1, sizeof(*uf));
+    if (uf == NULL) {
+        return -ENOMEM;
+    }
+    uf->jobid     = finish->jobid;
+    uf->finish    = *finish;
+    uf->queued_at = time(NULL);
+
+    pthread_mutex_lock(&g_unknown_finish_mu);
+    unknown_finish_sweep_locked(); /* opportunistic TTL bound */
+    uf->next = g_unknown_finish_list;
+    g_unknown_finish_list = uf;
+    pthread_mutex_unlock(&g_unknown_finish_mu);
+    return 0;
+}
+
+/* Take (and remove) the registered finish for jobid. Returns 1 if found, 0
+ * if not. `out` (optional) receives a copy. */
+static int unknown_finish_take(uint64_t jobid, xce_jobfinish_t* out)
+{
+    unknown_finish_t** pp = &g_unknown_finish_list;
+    int               found = 0;
+
+    pthread_mutex_lock(&g_unknown_finish_mu);
+    while (*pp != NULL) {
+        if ((*pp)->jobid == jobid) {
+            unknown_finish_t* uf = *pp;
+            *pp = uf->next;
+            if (out != NULL) {
+                *out = uf->finish;
+            }
+            free(uf);
+            found = 1;
+            break;
+        }
+        pp = &(*pp)->next;
+    }
+    pthread_mutex_unlock(&g_unknown_finish_mu);
+    return found;
+}
+
+/* Called from apply_committed_entry's WAL_JOB_SUBMIT case right after the job
+ * is materialized: if a worker finish was registered for this jobid while the
+ * job did not yet exist, append the durable WAL_JOB_FINISH and mutate the job
+ * to its terminal state immediately — so the scheduler never sees a
+ * PEND/RUN job that already finished. The later WAL apply of the appended
+ * entry is a terminal→terminal no-op. Returns 1 if a deferred finish was
+ * applied, 0 otherwise. */
+int apply_registered_finish_if_any(cluster_t* c, job_t* j)
+{
+    xce_jobfinish_t finish;
+
+    if (c == NULL || j == NULL) {
+        return 0;
+    }
+    if (!unknown_finish_take(j->job.jobid, &finish)) {
+        return 0;
+    }
+
+    lock(j, RTTI_LOCK_WRITE);
+    /* Job is PEND here (not terminal), so record_job_finish_fact appends the
+     * durable WAL_JOB_FINISH rather than skipping it as a duplicate. */
+    record_job_finish_fact(c, j, &finish);
+    /* Mutate to terminal immediately; later WAL apply is a no-op. */
+    apply_job_finish_transition(j, &finish);
+    unlock(j);
+
+    INFO("job %ld: applied deferred worker finish (exit_code=%d) at submit "
+         "apply",
+         j->job.jobid, finish.exit_code);
+    return 1;
+}
+
+#ifdef XCE_TESTING
+/* Exposed for the controller unit test binary (compiled with -DXCE_TESTING):
+ * direct probes of the bug-1509 unknown-finish registry. */
+int test_unknown_finish_register(const xce_jobfinish_t* f)
+{
+    return unknown_finish_register(f);
+}
+int test_unknown_finish_take(uint64_t jobid, xce_jobfinish_t* out)
+{
+    return unknown_finish_take(jobid, out);
+}
+#endif
+
 int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
 {
     job_t* job = NULL;
@@ -1774,6 +1908,17 @@ int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
     unlock(c->jobmgr);
 
     if (job == NULL) {
+        /* bug-1509: finish arrived before the job was materialized (submit
+         * committed but not yet applied — job construction is deferred to
+         * apply_committed_entry). Register it in the unknown-finish registry;
+         * the submit's apply drains it and converges the job to DONE/EXIT.
+         * Previously this finish was dropped, leaving the job RUN forever. */
+        if (unknown_finish_register(finish) == 0) {
+            INFO("job %ld: finish for un-materialized job queued in "
+                 "unknown-finish registry",
+                 finish->jobid);
+            return 0;
+        }
         ERROR("can not find job %ld", finish->jobid);
         return -errno;
     }

src/controller/persist.c

@@ -3528,6 +3528,11 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
                       HOOK_DETAIL_JOB_NEW, c, &jlist);
         t_apply_seam_notifying = false;
 
+        /* bug-1508: a worker finish may have arrived before this job was
+         * materialized (the commit→apply window). Drain it now so the job
+         * converges to DONE/EXIT instead of lingering as RUN/PEND. */
+        (void)apply_registered_finish_if_any(c, j);
+
         return 1;
     }
 

src/controller/testsuite/test_apply_pump.c

@@ -1635,11 +1635,116 @@ static void test_standby_truncate_before_commit_leaves_no_phantom_job(void)
           "PASS\n");
 }
 
+/* ── 5. bug-1509: unknown-finish registry (finish arriving before the job is
+ * ──    materialized must not be dropped — it must be applied once the job
+ * ──    materializes, converging to DONE/EXIT instead of lingering as RUN). */
+
+static void test_unknown_finish_registry_register_and_take(void)
+{
+    tap_setup();
+
+    xce_jobfinish_t out;
+
+    /* Register a finish for a not-yet-materialized job. */
+    xce_jobfinish_t f = { .jobid = 903, .exit_code = 0 };
+    TEST_ASSERT_EQUAL_INT(0, test_unknown_finish_register(&f));
+
+    /* Take it back — must be found with the exact payload. */
+    TEST_ASSERT_EQUAL_INT(1, test_unknown_finish_take(903, &out));
+    TEST_ASSERT_EQUAL_UINT64(903, out.jobid);
+    TEST_ASSERT_EQUAL_INT(0, out.exit_code);
+
+    /* Take again — nothing left. */
+    TEST_ASSERT_EQUAL_INT(0, test_unknown_finish_take(903, &out));
+
+    /* A second jobid coexists independently. */
+    xce_jobfinish_t g = { .jobid = 904, .exit_code = 7 };
+    TEST_ASSERT_EQUAL_INT(0, test_unknown_finish_register(&g));
+    TEST_ASSERT_EQUAL_INT(1, test_unknown_finish_take(904, &out));
+    TEST_ASSERT_EQUAL_INT(7, out.exit_code);
+    TEST_ASSERT_EQUAL_INT(0, test_unknown_finish_take(903, &out));
+
+    tap_teardown();
+    printf("test_unknown_finish_registry_register_and_take: PASS\n");
+}
+
+/* The real bug: a worker finish for a job the controller has not yet
+ * materialized (submit committed but not applied). on_job_finish must
+ * register it (return 0), and once the job materializes,
+ * apply_registered_finish_if_any must converge it to DONE. */
+static void test_unknown_finish_applied_once_job_materializes(void)
+{
+    tap_setup();
+
+    /* No job 905 exists yet. */
+    TEST_ASSERT_NULL(tap_find_job(905));
+
+    xce_jobfinish_t finish = { .jobid = 905, .exit_code = 0 };
+    TEST_ASSERT_EQUAL_INT(0, on_job_finish(&g_tap_c, &finish));
+
+    /* Still no job (on_job_finish must not fabricate one). */
+    TEST_ASSERT_NULL(tap_find_job(905));
+
+    /* The submit's apply materializes the job (simulated restore_job). */
+    job_t* j = tap_seed_job(905, JOB_STAT_PEND);
+    TEST_ASSERT_NOT_NULL(j);
+
+    /* Draining applies the deferred finish: exit_code 0 -> DONE. */
+    TEST_ASSERT_EQUAL_INT(1, apply_registered_finish_if_any(&g_tap_c, j));
+    TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_DONE,
+                             "job must converge to DONE after deferred finish");
+
+    /* Draining again is a no-op. */
+    TEST_ASSERT_EQUAL_INT(0, apply_registered_finish_if_any(&g_tap_c, j));
+
+    tap_teardown();
+    printf("test_unknown_finish_applied_once_job_materializes: PASS\n");
+}
+
+/* Non-zero exit_code converges to EXIT, not DONE. */
+static void test_unknown_finish_exit_code_maps_to_exit(void)
+{
+    tap_setup();
+
+    xce_jobfinish_t finish = { .jobid = 906, .exit_code = 3 };
+    TEST_ASSERT_EQUAL_INT(0, on_job_finish(&g_tap_c, &finish));
+
+    job_t* j = tap_seed_job(906, JOB_STAT_PEND);
+    TEST_ASSERT_NOT_NULL(j);
+    TEST_ASSERT_EQUAL_INT(1, apply_registered_finish_if_any(&g_tap_c, j));
+    TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_EXIT,
+                             "exit_code!=0 must converge to EXIT");
+    TEST_ASSERT_FALSE_MESSAGE(j->job.state & JOB_STAT_DONE,
+                              "EXIT and DONE must not both be set");
+
+    tap_teardown();
+    printf("test_unknown_finish_exit_code_maps_to_exit: PASS\n");
+}
+
+/* The drain must not touch a job with no registered finish. */
+static void test_unknown_finish_drain_noop_without_registration(void)
+{
+    tap_setup();
+
+    job_t* j = tap_seed_job(907, JOB_STAT_RUN);
+    TEST_ASSERT_NOT_NULL(j);
+    TEST_ASSERT_EQUAL_INT(0, apply_registered_finish_if_any(&g_tap_c, j));
+    TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_RUN,
+                             "job without a deferred finish must be untouched");
+
+    tap_teardown();
+    printf("test_unknown_finish_drain_noop_without_registration: PASS\n");
+}
+
 /* ── Test group entry point ───────────────────────────────────────────── */
 
 void test_apply_pump(void)
 {
     UnitySetTestFile(__FILE__);
+    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);
+    RUN_TEST(test_unknown_finish_drain_noop_without_registration);
     RUN_TEST(test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_order);
     RUN_TEST(test_apply_pump_run_applies_mixed_submit_and_fact_entries);
     RUN_TEST(test_apply_pump_run_halts_at_failed_lsn_without_advancing);