diff --git a/docs/Job-Finish-Robustness-设计文档.md b/docs/Job-Finish-Robustness-设计文档.md new file mode 100644 index 00000000..d3189d86 --- /dev/null +++ b/docs/Job-Finish-Robustness-设计文档.md @@ -0,0 +1,255 @@ +# 作业终态可靠性设计(Job Finish Robustness) + +- 状态:**v2(已含 agent review 结论,2026-08-06)** +- 日期:2026-08-06 +- 关联问题:作业在 controller 端永久停留 RUN(如 `noexec-test`/`newcfg-test`/`verify-fix`) + +## 0. Review 结论(agent review 2026-08-06,结论:approve with changes) + +10 条代码断言全部核实为真。发现的必改项、强建议项、边界情况见本节;对应修订已回写进正文各节(§3/§4/§7/§9/§10 用 🔧 标注)。 + +### 0.1 必改项 + +| ID | 严重度 | 问题 | 修订 | +|----|--------|------|------| +| **A1** | HIGH | `record_job_finish_fact` 文档要求调用方持 `job` 的 `RTTI_LOCK_WRITE`(job.c:1737-1743),但 Part A 在 `apply_committed_entry` 取到 `j` 后未加锁(persist.c:3517-3521)。job_ctrl/scheduler 的直接 mutator 不受 apply_pump_lock 串行化 → 数据竞争。 | 🔧 §3.4:take + record 包进 `lock(j, RTTI_LOCK_WRITE)` | +| **B2** | HIGH | `transient_since` 未持久化到 WAL(WAL_JOB_UPDATE 只带 `wal_jobstate_t{jobid,state,_pad}`)。重启后 B1 标记的作业恢复为 `RUN\|UNKWN\|TRANSIENT` 但 `transient_since=0` → `job_should_retain` 永不 GC。 | 🔧 §4.1/§4.3:B3 重标为显式持久化机制(每次重启重设 transient_since),并明确该机制为兜底;Phase 2 起在 WAL 携带 transient_since(新字段/条目) | +| **B3** | MED/HIGH | `resend_job_status` 只在 worker 启动时调用(xcecd.c:522),join/rejoin 路径不重触发。分区后重连的 worker 不再上报运行清单 → UNKWN 作业被 grace 误杀。 | 🔧 §4.2/§4.4:每次 join/rejoin 都调用(修复后的)`resend_job_status` | + +### 0.2 强建议项 + +| ID | 严重度 | 问题 | 修订 | +|----|--------|------|------| +| **B1** | HIGH | grace 窗口把"仅分区"(TCP half-open,作业仍在跑)的 worker 的作业 EXIT;终态后 worker 的 RUNNING 认领被 `job_state_is_backward` 拒绝 → 孤儿进程 + 永久错误终态。 | 🔧 §4.4:grace 放宽/可配;区分"确认死亡"vs"未知";加 EXIT 后核对(worker 认领则复活或杀孤儿) | +| **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 作提示 | + +### 0.3 其他 / 边界 + +- **A4**:FINISH-only 注册是范围裁剪——START 对 fast-job 同样会被丢弃(作业永不显示 RUN、start_time=0、无 WAL_JOB_START)。本次接受,后果记录在 §3.5。 +- **A5**:TTL 必须 > 最大 commit→apply 延迟,防合法 deferred finish 被清扫;注册表需设容量上限防 bogus-finish 洪泛。 +- **B5**:B1 在 `handle_worker_leave` 持 worker 锁时扫 jobmgr 有锁序反转风险 → 先放 worker 锁再扫 jobmgr。 +- **B6**:B3 重启全标 UNKWN 后,apiserver 渲染为 "TRANSIENT"(可接受);但 `xcecli_jobcontrol` 对 `allocs==NULL` 的 UNKWN 作业的行为需验证(Phase 2)。 +- **M1**:已有残留 RUN 作业的迁移路径 = B3 重启重标 + B4/B2 重报自愈,需显式测试。 +- **M2**:standby catch-up 应用 Part A 的 WAL_JOB_SUBMIT 时注册表为空 → 无操作(安全);但须只在权威 primary 上追加/消费(failover 窗口防护)。 +- **M3**:作业完成后 worker 才重启:DB 行已是 WORKER_DONE → `get_running_workers`(筛 RUNNING)不包含它 → 丢失的 finish 不被重发。这是 §7.6 已承认的局限。 +- **M4**:重复 finish:作业实体化后重复 finish 走终态→终态 no-op(`record_job_finish_fact` peek,job.c:1744-1747)——安全。 +- **M5**:grace EXIT 的作业 exit code 未设置,与真实失败不可区分(`apply_job_fail_transition` 非 post 阶段设裸 EXIT)——可接受。 + +## 1. 问题与目标 + +### 1.1 症状 + +若干作业已提交、运行、并在 worker 端确实执行完毕,但 controller 端的作业状态永久停留在 `RUN`。即使多次重启 controller/worker 也依旧出现(因为状态已持久化在 WAL,重启后回放恢复成 RUN)。 + +### 1.2 根因 + +worker 上报作业完成(finish 通知)后,controller 未能把作业收敛到终态(`DONE`/`EXIT`)。两个子场景: + +- **场景 A(controller 收到但丢弃)**:finish 到达 controller 时,作业尚未实体化(submit 已 commit、但 `apply_committed_entry` 尚未 `restore_job` 进 jobmgr)。`on_job_finish`(`src/controller/job.c:1777`)在 `key_find` 找不到作业时仅 `ERROR("can not find job %ld")` 直接返回——**finish 被丢弃,无 WAL 追加、无重试、无补救**。 + - 触发窗口:作业构造被延迟到 apply 时(`apply_committed_entry` 的 WAL_JOB_SUBMIT case,commit 之后才 `restore_job`)。短作业(如 `sleep 10`)极易在"commit → apply"窗口内跑完并上报 finish。 +- **场景 B(传输中丢失)**:finish 在网络上丢失,或 worker 崩溃/失联,controller 从未收到。worker 不会重发(worker 认为作业已完成并已从自身跟踪中移除)。 + +### 1.3 目标 + +任何情况下,作业最终收敛到终态(`DONE`/`EXIT`),不再出现"永远 RUN"。方案分 A、B 两部分,分别覆盖两个子场景。 + +--- + +## 2. 现状分析(已核实) + +### 2.1 状态机已具备所需状态(但未接线) + +| 状态 | 值 | 语义 | 当前使用 | +|------|-----|------|---------| +| `JOB_STAT_UNKWN` | 0x10000 | worker 失联,作业状态未知 | **C 代码零使用** | +| `JOB_STAT_RUNKWN` | 0x8000 | 作业状态未知(与远程集群失去联系) | **零使用** | +| `JOB_STAT_TRANSIENT` | 0x400 | failover 后状态不确定,等待 worker 重连 | persist.c 中仅有读取(GC 判断) | + +- apiserver **已支持显示** `UNKWN`/`RUNKWN`/`TRANSIENT`(`src/frontend/apiserver/models/jobs_test.go` 有映射断言)。 +- `job_should_retain`(`src/controller/persist.c:6730`)对 `TRANSIENT` 作业:`transient_since` 超时后不保留(自动 GC)。但 `TRANSIENT` 从未被设置,故该机制永不触发。 + +### 2.2 worker 失联处理缺作业标记 + +- `handle_worker_leave`(`src/controller/worker.c:693`)→ `mark_worker_unavail`(`worker.c:661`):只清 `ep`、置 `XCE_WORKER_UNAVAIL`、重置 host attrs/resvals/slots。**不碰该 worker 的 RUN 作业**。 +- `apiserver_worker_dead` 与 `sched_worker_dead` 均有 `/* FIXME: need to handle job status in this worker */`(`apiserver.c`、`sched.c`)。 +- **无 worker 存活监控**:worker 失联仅靠 TCP RDHUP 触发 `handle_worker_leave`。静默失联(网络分区、TCP 不断)检测不到。 + +### 2.3 worker 侧重连上报(已存在,但有缺陷) + +- worker 运行作业时把 `(pid, status, job_id, job_status)` 写入 sqlite(`/run/xcecd.db`,`src/worker/src/xcecd_db.c`,`DBUS_MANAGER` 构建启用)。 +- worker 崩溃后重启:`xcecd.c:502` `xcecd_db_get_running_workers` 恢复 worker_list → `xcecd.c:522` `resend_job_status`(`xcecd_worker_utils.c:447`)逐一对 `WORKER_RUNNING` 的 worker 调 `xcecd_reply_startjob`(即 `xcecli_notify_jobstart`)重报 RUNNING。 +- **缺陷**:`resend_job_status` 盲目上报 `JOB_RUNNING`(不查进程死活、`job_pid=0`),导致已死的作业也被重报为 RUNNING。 + +### 2.4 finish 处理链路(已核实正确,缺口在丢弃) + +``` +worker → OPCode_notify(jobFinishInfo) → do_notify(cluster.c:130) → on_job_finish + → key_find(jobmgr) + → 找到: record_job_finish_fact → 追加 WAL_JOB_FINISH → commit → apply → apply_job_update_transition → DONE/EXIT ✓ + → 找不到: ERROR "can not find job" → 丢弃 ✗(无重试) +``` + +- `apply_job_update_transition`(`job.c:1895`)对 RUN/PEND → DONE/EXIT 判定为非 backward(`job_state_is_backward` 在 `job.c:43`,仅当 cur 是终态且 next 非终态才 backward)——转换守卫无误。 +- 分发路由(`do_notify`)、接收(`on_job_finish`)、WAL 追加(`record_job_finish_fact`)均正确;缺口仅在"作业未实体化时丢弃"。 + +### 2.5 重启后 exec host(allocs)丢失 + +- `xce_job_t.allocs`(含 exec host)**不持久化**到 WAL;controller 重启后 `j->job.allocs == NULL`。因此重启后 controller 无法将 RUN 作业映射到具体 worker。 +- 设计含义:Part B 采用 **worker-push 模型**(worker 上报自己的运行作业),不依赖 exec-host 映射。 + +--- + +## 3. Part A:Unknown Finish 注册表(controller 端,修复"收到但丢弃") + +### 3.1 目标 + +场景 A 下,worker 的 finish 不再被丢弃:controller 暂存它,等作业实体化后补应用。 + +### 3.2 数据结构(`src/controller/job.c`,仿照 `fetchfile_pending` 模式) + +```c +typedef struct unknown_finish { + uint64_t jobid; + xce_jobfinish_t finish; /* 深拷贝,本表持有 */ + 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; +``` + +配套辅助函数:`unknown_finish_register()`、`unknown_finish_take(jobid)`(取出并摘除)、`unknown_finish_sweep_ttl()`(清理超时未实体化的条目,TTL 建议 60s,防 submit 永不被 apply 时泄漏)。 + +### 3.3 `on_job_finish` 改动(`job.c:1762`) + +- 作业找到 → 现路径不变(`record_job_finish_fact`)。 +- 作业未找到 → 深拷贝 `xce_jobfinish_t` 登记到 `g_unknown_finish_list`,`INFO` 日志("finish for un-materialized job %ld queued in unknown-finish registry"),`return 0`(controller 已接受责任,不视为错误)。**不追加 WAL**。 + +### 3.4 submit apply 改动(`apply_committed_entry` WAL_JOB_SUBMIT case,`persist.c:3488`,`restore_job` 之后) + +1. `restore_job` 已把作业实体化进 jobmgr。 +2. `unknown_finish_take(jobid)`:若有该 jobid 的 finish → 取出。 +3. 🔧 **持作业锁**:`lock(j, RTTI_LOCK_WRITE)` 包住 take + `record_job_finish_fact`(A1:该函数要求调用方持 job 锁,job.c:1737-1743;apply pump 锁不串行化 job_ctrl/scheduler 的直接 mutator)。 +4. 用取出的 finish 调 `record_job_finish_fact(c, j, &finish)`:追加 WAL_JOB_FINISH(持久化),之后该条目经 commit + apply 收敛作业到 DONE/EXIT。 +5. `unlock(j)`,释放 finish 拷贝。 + +**顺序正确性**:finish 的 WAL 条目在 submit 的 WAL 之后追加,apply 顺序保证 job 先实体化(PEND/RUN)再收敛终态。`apply_job_update_transition` 对 PEND → DONE 亦非 backward,安全。 + +**并发/重入(🔧 A2 修订)**:submit apply 在 apply pump 的 walk 中(持 `apply_pump_lock`);`record_job_finish_fact` 内部 `persist_submit_job_fact` 取 `p->lock`(不同锁,无死锁)。但注意:RETAIN 模式的 `commit_submit` 立即返回、不推进 `committed_lsn`(commit.c:497-506),追加的 WAL_JOB_FINISH 需等 majority ack 后由后续 pump kick 应用——**不在同一 pass 内应用**。pump kick 是异步池 vtask(commit.c:834, persist.c:4415-4433),无自死锁。实现需补:从 apply 内追加的锁序/重入分析 + 回归测试(A2)。 + +**A3 残留风险(🔧 文档化)**:unknown 注册表是进程内的,不跨 failover。primary 登记 finish 后 failover,新 primary 重放 submit(作业 PEND)但 finish 丢失 → 作业 PEND 卡住(B3 只覆盖 RUN,不救 PEND)。commit→apply 窗口很小,概率低;Phase 1 接受为残留风险,后续实现持久化 deferred finish 或把 B3/B4 扩展到 PEND。 + +### 3.5 通用化(可选扩展) + +START/FAIL/STOP/RESUME 等 worker 事实通知存在同样的"作业未实体化"竞态。可将注册表泛化为 `jobid → 事件队列(event_type + payload)`。本次实现仅覆盖 FINISH(对应本 bug),其余留作后续。 + +🔧 **A4 范围裁剪后果(已接受)**:FINISH-only 注册下,fast-job 的 START 仍会被丢弃(同 "can not find job" 路径,job.c:1671-1674)→ 作业从不显示 RUN、`start_time=0`、回放时 PEND→DONE。对本次 bug 无影响(终态正确),但会计/展示不完整。后续扩展注册 START 等。 + +--- + +## 4. Part B:Worker 存活监控 + 重连核对(修复"传输丢失 / worker 失联") + +### 4.1 B1:Worker 存活监控 + +- 每 worker 增加 `ts_last_recv` 字段(`worker_t`),在任何入站 worker 消息上更新(`update_job_rusages`、`do_notify`、rusage 等;worker 每 ~10s 发一次 rusage)。 +- controller 增加周期清扫(delaymgr/vtaskmgr 定时任务):`now - ts_last_recv > WORKER_LIVENESS_TIMEOUT`(建议 30~60s,约 3 个更新周期)的 worker 标记 dead,走 `handle_worker_leave` 路径。 +- **关键改动**:`handle_worker_leave`/`mark_worker_unavail` 中,遍历该 worker 的 RUN 作业(运行时 `job.allocs` 可用,能定位),逐一对 `mark_job_update(c, j, state|JOB_STAT_UNKWN|JOB_STAT_TRANSIENT)` 并设 `transient_since = now`。填补两处 `FIXME: need to handle job status in this worker`。 +- 🔧 **B2(transient_since 持久化)**:WAL_JOB_UPDATE 只带 `wal_jobstate_t{jobid,state,_pad}`,**不持久化 transient_since**。因此 B1 的 `transient_since = now` 在 controller 重启后会丢(恢复成 RUN|UNKWN|TRANSIENT 但 transient_since=0 → `job_should_retain` 永不 GC,persist.c:6736-6741)。Phase 1 的兜底是 **B3 每次重启重标(重设 transient_since)**;Phase 2 起在 WAL 携带 transient_since(新字段/条目)真正持久化。 +- 🔧 **B5(锁序)**:`handle_worker_leave` 持 `lock(w, RTTI_LOCK_WRITE)`(worker.c:721-727)时扫 jobmgr 有锁序反转风险 → 先放 worker 锁再扫 jobmgr。 + +### 4.2 B2:修复 worker 侧重连上报(`resend_job_status`) + +- 现状:`xcecd_worker_utils.c:447` `resend_job_status` 对恢复的 worker 盲目报 `JOB_RUNNING`(`job_pid=0`,不查进程)。 +- 改动:上报前用 process_manager 查每个 job 的进程实际状态: + - 进程活着 → `xcecd_reply_startjob`(报 RUN) + - 进程已死 / systemd unit 不存在 → 直接调 `xcecli_notify_jobfinish`(报 DONE/EXIT),不再依赖后续轮询的二次纠正。 +- 🔧 **B3(每次 join 都上报)**:`resend_job_status` 不能只在启动调用(xcecd.c:522),必须在**每次 join/rejoin**(`xcecd_join_cluster`)成功后调用——否则分区后重连的 worker 不再上报,UNKWN 作业被 grace 误杀。 +- 🔧 **B4(exit code 精度)**:sqlite 不存 exit code(resend 硬编码 exit_code=0),且存的 pid 是 xcecd-worker 子进程而非作业命令树。上报前从 systemd unit 状态 / reaped wait status 推导真实 exit code,并用存储的 `job_status` 作提示;MPI/多进程作业需 pgid 级存活检查。 + +### 4.3 B3:Controller 重启处理 + +- `persist_recover` 回放完成后,把所有 **RUN 状态**的作业置 `JOB_STAT_UNKWN` + `JOB_STAT_TRANSIENT` + `transient_since = now`(重启后与 worker 连通性未知)。 +- 解决"重启后 RUN 作业(无 allocs)永远卡住"的问题;后续由 B4 重连核对收敛。 + +### 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 后来认领则复活作业或杀孤儿。 + +--- + +## 5. 状态模型与转换 + +| 状态 | 设置时机 | 清除时机 | +|------|---------|---------| +| `RUN` | submit apply + START / B4 确认存活 | finish 应用 | +| `UNKWN`(+`TRANSIENT`) | B1 worker dead、B3 controller 重启 | B4 确认存活 → RUN;finish → 终态;TTL → GC | +| `DONE`/`EXIT` | finish 应用(A 或 B) | 保留期后 GC | + +转换表: +- `RUN/UNKWN → DONE/EXIT`:finish 应用(Part A 注册表补应用 / 正常路径 / B2 worker 上报)。 +- `UNKWN → RUN`:B4 重连确认存活。 +- `UNKWN → EXIT`:B4 grace 窗口无人认领。 +- `UNKWN → GC`:`job_should_retain` 的 `transient_job_timeout_s` 到期(兜底)。 + +## 6. 协议改动 + +- **无新 opcode**。Part A 复用现有 WAL_JOB_FINISH;Part B 复用现有 `notify_jobstart`/`notify_jobfinish` 协议与 join 流程。本设计全部复用现有机制,只是正确接线 + 补缺失路径。 + +## 7. 边界情况 + +1. **作业完成与 worker 死亡竞态**:UNKWN 标记基于陈旧 RUN 状态是安全的,重连核对纠正。 +2. **worker 也重启(进程丢失)**:B2 上报查进程 → 报 finish/EXIT。 +3. **短暂网络抖动**:worker 在 grace 窗口内回来 → 核对恢复 RUN;TTL 足够长不误 GC。 +4. **多 worker 重连风暴**:核对上报批量处理/限流。 +5. **submit 永不被 apply**:unknown 注册表条目 TTL 清扫。 +6. **已知局限(Phase 2 候选)**:worker 存活且 finish 在传输中丢失、无重连无失联的场景——A 覆盖"controller 见过",B 覆盖"worker 失联";纯"传输丢失且 worker 一直健康"需 worker 周期上报运行清单或 rusage 携带作业状态。 + +## 8. 测试计划 + +### 8.1 Part A 单测(`testsuite/test_*`) + +- `on_job_finish` 作业未实体化 → 注册到 unknown-finish 表 → 断言条目存在、`return 0`。 +- `apply_committed_entry` WAL_JOB_SUBMIT 应用后 → 查表取 finish → 断言作业收敛 DONE/EXIT、WAL_JOB_FINISH 追加。 +- 模拟 fast-job:提交 WAL 已 commit 但作业未 apply 时收到 finish → 断言不丢。 + +### 8.2 Part B 单测 + +- `mark_worker_unavail` → 该 worker 的 RUN 作业置 UNKWN + TRANSIENT + transient_since。 +- `persist_recover` 回放后 RUN 作业 → UNKWN。 +- worker 上报运行清单 → controller 清 UNKWN;无人认领 → grace 后 EXIT。 +- worker 端 `resend_job_status` 对进程已死的作业报 finish 而非 RUN。 + +### 8.3 集成测试(集群验证) + +- kill worker(作业运行中)→ controller 标记作业 UNKWN → 重启 worker → 核对:作业恢复 RUN(进程存活)或 EXIT(进程已丢)。 +- controller 重启(作业 RUN 中)→ 全部 RUN 置 UNKWN → worker 重连 → 核对收敛。 +- 完整作业验证:submit → RUN → DONE,peek 正常,所有 daemon 健康。 + +## 9. 实施顺序 + +| Phase | 内容 | 风险 | 验证 | +|-------|------|------|------| +| **Phase 1** | Part A:unknown finish 注册表 + 单测 | 低(controller 端独立改动) | 单测 + 完整作业验证 | +| **Phase 2** | B1+B3:worker 存活监控 + handle_worker_leave/recover 置 UNKWN + 单测 | 中(新增定时任务) | 单测 + kill worker 集成 | +| **Phase 3** | B2+B4:resend_job_status 修复 + 重连核对 + 集成测试 | 中(worker 端改动) | 集成测试 + 完整验证 | + +## 10. 改动点清单 + +| 文件 | 函数 | 改动 | +|------|------|------| +| `src/controller/job.c` | `on_job_finish` | 作业未找到 → 登记 unknown-finish(替代丢弃) | +| `src/controller/job.c` | 新增 `unknown_finish_*` | 注册/取出/TTL 清扫 | +| `src/controller/persist.c` | `apply_committed_entry` WAL_JOB_SUBMIT | restore_job 后查 unknown-finish 补应用(🔧 A1:持 job RTTI_LOCK_WRITE) | +| `src/controller/worker.c` | `mark_worker_unavail`/`handle_worker_leave` | 遍历该 worker RUN 作业置 UNKWN+TRANSIENT | +| `src/controller/worker.c` | `worker_t` | 新增 `ts_last_recv` | +| `src/controller/*` | 周期清扫任务 | worker 存活监控 | +| `src/controller/persist.c` | `persist_recover` | 回放后 RUN 作业置 UNKWN+TRANSIENT | +| `src/worker/src/xcecd_worker_utils.c` | `resend_job_status` | 查进程实际状态后上报 | +| `src/worker/src/*` | process_manager 查询接口 | 供 B2 查进程死活 | +| `src/controller/job.c` | 重连核对处理 | 清 UNKWN / grace 后 EXIT | diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 98955f14..3c531378 100644 --- a/src/controller/include/cluster.h +++ b/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 diff --git a/src/controller/job.c b/src/controller/job.c index 8c31fa2a..56bbdb73 100644 --- a/src/controller/job.c +++ b/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; } diff --git a/src/controller/persist.c b/src/controller/persist.c index d3e85deb..aa7efdd1 100644 --- a/src/controller/persist.c +++ b/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; } diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c index db910b67..7cacd02f 100644 --- a/src/controller/testsuite/test_apply_pump.c +++ b/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);