fix(controller,worker): dual-state worker loss + resend actual job state (bug-1509 P3)

Phase 3 (B1 dual-state + B2): a job never lingers as RUN when its worker
is lost.

Dual-state B1 (user-approved): RDHUP (worker TCP dropped = confirmed dead)
marks its RUN jobs EXIT immediately (mark_worker_jobs_exit). Heartbeat
timeout (unknown) keeps them UNKWN+TRANSIENT awaiting reconnect
reconciliation; the TRANSIENT TTL GC is the backstop. handle_worker_leave
now calls mark_worker_jobs_exit (was mark_worker_jobs_unknown in Phase 2).

B2 (worker side): resend_job_status now checks the recovered process is
alive (kill(pid,0)) before re-reporting RUN — a dead process is reported
finished so the controller converges to EXIT instead of a stale RUN. And
resend_job_status runs on EVERY rejoin (checking_connection), not just
startup, so a reconnecting worker's jobs are reconciled (clears UNKWN via
the existing notify_jobstart claim path).

Design: docs/Job-Finish-Robustness-设计文档.md §4.4 (dual-state decision).

Unit test: worker RDHUP marks jobs EXIT (not UNKWN).

src/controller/include/cluster.h

@@ -57,6 +57,10 @@ 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*);
+/* bug-1509 Phase 3 (B1 dual-state): worker TCP dropped (RDHUP, confirmed dead)
+ * → mark its RUN jobs EXIT immediately. Caller must not hold any
+ * worker/jobmgr lock (B5). */
+extern void mark_worker_jobs_exit(cluster_t*, const char* hostname);
 
 #ifdef XCE_TESTING
 /* Direct probes of the bug-1509 unknown-finish registry (test binary only). */

src/controller/job.c

@@ -2183,6 +2183,69 @@ void mark_all_run_jobs_unknown_in_memory(cluster_t* c)
     mark_jobs_unknown_core(c, NULL, 0);
 }
 
+/* bug-1509 Phase 3 (B1 dual-state): a worker whose TCP connection dropped
+ * (RDHUP) is CONFIRMED dead from the controller's perspective — mark its RUN
+ * jobs EXIT immediately (terminal) instead of leaving them unknown. This is
+ * distinct from the heartbeat-timeout path (mark_worker_jobs_unknown), which
+ * keeps the job UNKWN awaiting reconnect reconciliation. Caller must not hold
+ * any worker/jobmgr lock (B5). */
+void mark_worker_jobs_exit(cluster_t* c, const char* hostname)
+{
+    iterator_t it;
+    job_t*     j;
+
+    if (c == NULL || hostname == 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;
+        }
+
+        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)) {
+            if (j->job.exit_info == NULL) {
+                j->job.exit_info =
+                    (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
+            }
+            if (j->job.exit_info != NULL) {
+                j->job.exit_info->exitcode = -1; /* worker lost */
+            }
+            j->job.finish_time = time(NULL);
+            mark_job_update(c, j, JOB_STAT_EXIT);
+        }
+        unlock(j);
+    }
+    iter_destroy(&it);
+    unlock(c->jobmgr);
+}
+
 /* 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/testsuite/test_apply_pump.c

@@ -1832,11 +1832,37 @@ static void test_restart_marks_all_run_jobs_unknown(void)
     printf("test_restart_marks_all_run_jobs_unknown: PASS\n");
 }
 
+/* Phase 3 dual-state (B1): RDHUP (confirmed worker death) marks RUN jobs
+ * EXIT immediately — distinct from the heartbeat-timeout path (UNKWN). */
+static void test_worker_rdhup_marks_jobs_exit(void)
+{
+    tap_setup();
+
+    job_t* j = tap_seed_job(1007, JOB_STAT_RUN);
+    TEST_ASSERT_NOT_NULL(j);
+    lock(j, RTTI_LOCK_WRITE);
+    tap_set_job_allocs(j, "t1");
+    unlock(j);
+
+    mark_worker_jobs_exit(&g_tap_c, "t1");
+
+    TEST_ASSERT_TRUE_MESSAGE(j->job.state & JOB_STAT_EXIT,
+                             "job must be EXIT after confirmed worker death");
+    TEST_ASSERT_FALSE_MESSAGE(j->job.state & JOB_STAT_UNKWN,
+                              "confirmed-dead path is EXIT, not UNKWN");
+    TEST_ASSERT_TRUE_MESSAGE(j->job.finish_time > 0,
+                             "finish_time must be set");
+
+    tap_teardown();
+    printf("test_worker_rdhup_marks_jobs_exit: PASS\n");
+}
+
 /* ── Test group entry point ───────────────────────────────────────────── */
 
 void test_apply_pump(void)
 {
     UnitySetTestFile(__FILE__);
+    RUN_TEST(test_worker_rdhup_marks_jobs_exit);
     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);

src/controller/worker.c

@@ -726,10 +726,12 @@ 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
+    /* bug-1509 Phase 3 (B1 dual-state): RDHUP means the worker is CONFIRMED
+     * dead — mark its RUN jobs EXIT immediately (not UNKWN; the
+     * heartbeat-timeout path uses mark_worker_jobs_unknown). 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);
+    mark_worker_jobs_exit(c, w->host.name);
 
     return 0;
 }

src/worker/src/controller_connector.c

@@ -205,6 +205,12 @@ int checking_connection()
                       "Rejoined cluster successfully after retry %d",
                       retry_count);
 
+            /* bug-1509 Phase 3 (B2/B3): re-report running jobs on every
+             * rejoin, not just at startup, so the controller reconciles
+             * their state (clears UNKWN on a live claim; converges to a
+             * terminal state for processes that are gone). */
+            (void)resend_job_status(*cc_ctx->worker_list);
+
             // api_ctx_t* api_ctx = (api_ctx_t*)xcecli_ctx;
             // cc_ctx->xce_ctx  = &(api_ctx->ctx);
             return 0;

src/worker/src/xcecd_worker_utils.c

@@ -4,6 +4,7 @@
 #include "xcecd_utils.h"
 #include "xcelog.h"
 #include <errno.h>
+#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -451,19 +452,31 @@ bool resend_job_status(xcecd_worker_t* worker_list)
         return true;
     }
 
-    xcecd_job_t job = { .job_id     = 0,
-                        .job_pid    = 0,
-                        .job_status = JOB_RUNNING,
-                        .exit_code  = 0 };
-
     xcecd_worker_t* worker = worker_list;
     while (worker) {
         if (worker->worker_status == WORKER_RUNNING) {
             XCECD_LOG(XCE_LOG_DEBUG, "Resending job %lu status to controller.",
                       worker->job_info.jobid);
 
-            job.job_id = worker->job_info.jobid;
-            xcecd_reply_startjob(&job);
+            xcecd_job_t job;
+            memset(&job, 0, sizeof(job));
+            job.job_id  = worker->job_info.jobid;
+            job.job_pid = worker->worker_pid;
+
+            /* bug-1509 B2: report the ACTUAL job state, not a blind RUN. The
+             * recovered worker's process may be gone (the worker crashed while
+             * the job was running) — in that case report it finished so the
+             * controller converges to a terminal state instead of a stale RUN. */
+            if (worker->worker_pid > 0 && kill(worker->worker_pid, 0) == 0) {
+                job.job_status = JOB_RUNNING;
+                job.stage      = JOB_EXEC_STAGE_MAIN;
+                xcecd_reply_startjob(&job);
+            } else {
+                job.job_status = JOB_EXITED;
+                job.stage      = JOB_EXEC_STAGE_MAIN;
+                job.exit_code  = 1; /* process lost — terminal EXIT */
+                xcecd_job_finished(&job);
+            }
         }
         worker = worker->next;
     }