From d3a92f103315b4508ed8219be94cb156603d2db8 Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Thu, 6 Aug 2026 06:49:24 +0000 Subject: [PATCH] fix(controller,worker): dual-state worker loss + resend actual job state (bug-1509 P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- docs/Job-Finish-Robustness-设计文档.md | 19 ++++--- src/controller/include/cluster.h | 4 ++ src/controller/job.c | 63 ++++++++++++++++++++++ src/controller/testsuite/test_apply_pump.c | 26 +++++++++ src/controller/worker.c | 8 +-- src/worker/src/controller_connector.c | 6 +++ src/worker/src/xcecd_worker_utils.c | 27 +++++++--- 7 files changed, 135 insertions(+), 18 deletions(-) diff --git a/docs/Job-Finish-Robustness-设计文档.md b/docs/Job-Finish-Robustness-设计文档.md index d3189d86..ebbc0ca4 100644 --- a/docs/Job-Finish-Robustness-设计文档.md +++ b/docs/Job-Finish-Robustness-设计文档.md @@ -20,7 +20,7 @@ | ID | 严重度 | 问题 | 修订 | |----|--------|------|------| -| **B1** | HIGH | grace 窗口把"仅分区"（TCP half-open，作业仍在跑）的 worker 的作业 EXIT；终态后 worker 的 RUNNING 认领被 `job_state_is_backward` 拒绝 → 孤儿进程 + 永久错误终态。 | 🔧 §4.4：grace 放宽/可配；区分"确认死亡"vs"未知"；加 EXIT 后核对（worker 认领则复活或杀孤儿） | +| **B1** | HIGH | grace 窗口把"仅分区"（TCP half-open，作业仍在跑）的 worker 的作业 EXIT；终态后 worker 的 RUNNING 认领被 `job_state_is_backward` 拒绝 → 孤儿进程 + 永久错误终态。 | ✅ 已定双态（§4.4）：RDHUP（确认死亡）→ 立即 EXIT；心跳超时（未知）→ UNKWN 等重连核对，不 EXIT（TRANSIENT TTL 兜底）。孤儿风险已接受 | | **A3** | MED/HIGH | unknown 注册表进程内、不跨 failover：primary 登记 finish 后 failover，新 primary 重放 submit（PEND）但 finish 丢失 → PEND 永久卡住（B3 只覆盖 RUN）。 | 🔧 §3.6：Phase 1 文档化为残留风险；后续持久化 deferred finish 或把 B3/B4 扩展到 PEND | | **B4** | MED | sqlite 不存 exit code，resend 硬编码 exit_code=0；存的 pid 是 xcecd-worker 子进程（非作业命令树）；MPI/多进程作业需 pgid 级存活。 | 🔧 §4.2：上报前从 systemd unit 状态 / reaped wait status 推导 exit code；用存储的 job_status 作提示 | @@ -172,14 +172,17 @@ START/FAIL/STOP/RESUME 等 worker 事实通知存在同样的"作业未实体化 - `persist_recover` 回放完成后，把所有 **RUN 状态**的作业置 `JOB_STAT_UNKWN` + `JOB_STAT_TRANSIENT` + `transient_since = now`（重启后与 worker 连通性未知）。 - 解决"重启后 RUN 作业（无 allocs）永远卡住"的问题；后续由 B4 重连核对收敛。 -### 4.4 B4：重连核对（worker 上报 → controller 交叉核对） +### 4.4 B4：重连核对（worker 上报 → controller 交叉核对）【已定双态】 -- worker（重）join 后，修复后的 `resend_job_status` 上报其运行 jobid 清单。 -- controller 收到上报后： - - 被上报的作业（UNKWN 态）→ 清 UNKWN/TRANSIENT → 保持 RUN（确认存活）。 - - 过了 **grace 窗口**（建议 60s，等所有 worker 重连）仍未被任何 worker 认领的 UNKWN 作业 → `mark_job_fail`（EXIT）。 -- worker-push 模型无需 exec-host 映射：worker 自己报运行作业，controller 交叉核对。 -- 🔧 **B1（grace 误杀防护）**：grace 窗口内把"仅分区"（TCP half-open、作业仍在跑）的 worker 的作业 EXIT 会**留下孤儿进程**——终态后 worker 的 RUNNING 认领被 `job_state_is_backward` 拒绝（job.c:43-46，record_job_start_fact job.c:1590）。措施：grace 放宽/可配（建议 ≥ 数个存活超时周期）；区分"worker 确认死亡"vs"状态未知"；加 EXIT 后核对——worker 后来认领则复活作业或杀孤儿。 +**双态决策（2026-08-06，用户确认方案 3）**： +- **RDHUP（TCP 断开，确认死亡）** → `mark_worker_jobs_exit`：该 worker 的 RUN 作业**立即置 EXIT**（终态，finish_time/exit_info 设置）。不等待核对。 +- **心跳超时（未知）** → `mark_worker_jobs_unknown`：RUN 作业置 `UNKWN+TRANSIENT`，**等重连核对，不 EXIT**；worker 永不回来则由 TRANSIENT TTL（`transient_job_timeout_s` 默认 300s）GC 兜底。 +- **认领路径（已天然工作）**：worker（重）连上报运行作业（`notify_jobstart`）→ WAL_JOB_START apply → `apply_job_update_transition` 设 state=RUN，**自动清除 UNKWN/TRANSIENT**。 + +worker 侧重连上报（B2）： +- `resend_job_status` **每次 join/rejoin 都调用**（不只启动），上报运行作业清单。 +- 上报前用 `kill(pid,0)` 查进程实际状态：活着 → 报 RUN；已死 → 报 finish（EXIT），不再盲目报 RUN。 +- 🔧 B1 已按上述双态解决（不再有"grace 窗口误杀"——未知态不 EXIT）。遗留孤儿风险（RDHUP→EXIT 但 worker 侧 systemd unit 可能仍在跑）：终态后 worker 认领被 `job_state_is_backward` 拒绝（job.c:43-46），孤儿进程最终自行结束或由后续清理。已接受（用户决策）。 --- diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 7bb0be51..d2d1dea6 100644 --- a/src/controller/include/cluster.h +++ b/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). */ diff --git a/src/controller/job.c b/src/controller/job.c index 99bf839f..17b093d6 100644 --- a/src/controller/job.c +++ b/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 diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c index 006f0d0a..0c259841 100644 --- a/src/controller/testsuite/test_apply_pump.c +++ b/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); diff --git a/src/controller/worker.c b/src/controller/worker.c index cf858fbc..616a43cc 100644 --- a/src/controller/worker.c +++ b/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; } diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 028751c0..6a466dbc 100755 --- a/src/worker/src/controller_connector.c +++ b/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; diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index 2d5dc785..f09cdf90 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -4,6 +4,7 @@ #include "xcecd_utils.h" #include "xcelog.h" #include +#include #include #include #include @@ -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; }