diff --git a/docs/left_scheduler_work.md b/docs/left_scheduler_work.md index 14b6e2ec..2c6bda3c 100644 --- a/docs/left_scheduler_work.md +++ b/docs/left_scheduler_work.md @@ -122,7 +122,7 @@ spec 声称 36/36 任务完成,但 re-audit(research.md)发现缺陷。来 | Phase 2:SSE 事件流 | 计划 | apiserver SSE endpoint + EventBroker,schedulerclient `WatchJobs()`,mcp-server 接入。ADR-004 设计完成待实施。估 2-3 天。 | | Phase 3:LSF adapter | 计划 | CLI 解析优先,Python Sidecar 后续。估 +1 月。 | | Phase 4:Slurm + K8s | 计划 | slurmrestd + CLI fallback;client-go。估 +1 月。 | -| modify 工具 | 推迟 | apiserver 当前无 modify 端点(control 仅 kill/stop/resume/checkpoint),需新增后端端点。 | +| modify 工具 | 推迟 | `scheduler_modify(id, changes)` 全链路空白:controller `job_ctrl` 只有 siginfo+checkpoint 无改作业字段 action → 需 controller 新 opcode → apiserver 新端点 → schedulerclient → mcp 工具(见 §八.4)。kill/stop/resume 工具已有。 | | 未决决策 | 未解决 | LLM 模型选型(Claude API/自建/混合);多集群 URM 源识别;air-gapped LLM 访问。 | --- @@ -198,13 +198,15 @@ spec 声称 36/36 任务完成,但 re-audit(research.md)发现缺陷。来 - Phase 4:**Slurm**(slurmrestd)+ **K8s**(client-go)(估 +1 月) ### 4. 推迟 / 未决 -- `scheduler_modify` 工具推迟——apiserver 当前无 modify 端点(control 仅 kill/stop/resume/checkpoint),需新增后端端点 +- `scheduler_modify(id, changes)` 工具推迟(MCP 缺口)——整条链路空白:controller `job_ctrl`(src/controller/job.c:957)仅 siginfo(SIGKILL/SIGSTOP/SIGCONT)+ checkpoint,无改作业字段的 action → 需 **controller 新 job-update opcode**(写回 queue/priority/resreq/user_priority + 触发调度器重新评估,大头)→ **apiserver** `POST /jobs/:jobids/modify`(routers.go 现有 create/get/list/output/control/restart,无 modify)+ 字段/状态校验(哪些字段可改、是否仅 PEND)→ **schedulerclient** `ModifyJob`(write.go 仅 Submit/Cancel/Hold/Release)→ **mcp-server** `scheduler_modify` 工具(~30 行,最小)。注:kill/stop/resume 三件套(`scheduler_cancel`/`hold`/`release`)已有;`scheduler_checkpoint` + `scheduler_restart` 已补(2026-08-30,17 工具,复用控制端 checkpoint action 与 `/jobs/:id/restart` 端点,单测通过)。 - 3 个未决决策:LLM 模型选型(Claude API/自建/混合)、多集群 URM 来源识别、air-gapped LLM 访问 - **PEND reason 分类体系(进行中,2026-08-14 已实施 P0+P1 部分)**:见 `docs/pend-reason-investigation-design.md`。**已完成(commit 4e2ac86)**: - **code 策略已定**:扩展 PREASON_* 枚举到 9 个(BAD_RESREQ/NO_RESOURCE/MATCH_FAIL/LIMIT_GLOBAL/LIMIT_HOST/QUEUE_CLOSED/QUEUE_DRAINING/TIME_WINDOW/HOST_UNAVAIL),capnp UInt32 无限制。 - **P0**:main_reason 修复(首个 detail code 设为主因,集群验证 PEND 作业 main_reason 非 0 + reason_name)+ limit 插件 attach(host 限额→LIMIT_HOST+limit 名,全局限额→LIMIT_GLOBAL+limit 名)。**集群验证**:PEND 作业 API 显示 `main_reason:3 main_reason_name:"match_fail"` + 每条 host/res reason 带 reason_name。 - **P1 部分**:队列状态(sched_queue_is_ready 非 stub,closed/inactive 队列不参选 + fcfs election attach QUEUE_CLOSED/DRAINING);worker 过滤(sched_try_alloc_on_host_ 跳过 UNAVAIL/CLOSED worker + HOST_UNAVAIL)。 - - **剩余**:P1 时间窗(TIME_WINDOW——scheduler 需保留 dispatchwindow 配置 + 自写"now in window"评估,controller 不计算窗口状态);P2 依赖/license/保留/exclusive/fairshare(机制本身未实现);MCP `bjobs -p` 等价工具(未做)。 + - **⚠️ scheduler-framework 合并(2026-08-25)回归掉 P1**:P1 改动点(4e2ac86/ee79179)在旧 `fcfs.c`,框架合并用 `builtin_fcfs` 取代后旧 fcfs.c 成死路径;`resource.c`/`limit.c` 冲突取 theirs 直接丢;merge-fix 只补 dispatchwindow **字段**不补逻辑;验证从未跑。合并后 `sched_queue_is_ready`/`sched_job_is_ready` 是 `return 1` stub(core.c),仅 3 处 preason emit。 + - **✅ P1 已恢复(2026-08-30)**:core.c 恢复 job 状态门 + queue 状态门;scheduler.c allocator 循环注入 `sched_queue_status_reason`(状态→QUEUE_CLOSED/DRAINING,时间窗→TIME_WINDOW,用 `xce_core_now` 虚拟时钟);resource.c 恢复 worker 过滤+HOST_UNAVAIL;limit.c 恢复 LIMIT_GLOBAL/HOST attach。**全部 reason attach 走 allocator 线程(bug-1521 安全)**。单测通过(新增 `test_core_ready_gates.c`,scheduler 套件 52/53,唯一失败是既有的 admin-request 校验顺序问题与 P1 无关)。遗留:持续 closed 队列的作业被 picker 跳过(spec 契约)到不了 allocator → 收不到 reason(合并前 #17 后同样如此,非新回归)。集群验证待做。 + - **剩余(P2,超出 PEND reason 范畴,属 scheduler 特性开发)**:依赖 depend_cond / license / 保留 RsvID / exclusive / fairshare——机制本身在 scheduler 未实现(设计文档:调度器无处理);机制落地后 reason 自动出现,MCP 侧无需改。已收尾:TIME_WINDOW 已实现(ee79179,2026-08-15);`scheduler_pend_reasons` 工具(bjobs -p 等价)已上线(67abbfa,2026-08-15)。 ### 5. 设计层加固(文档已写、代码零实现) - **Auto-pilot 模式**:用户显式授权范围内的自主行为 diff --git a/src/frontend/mcp-server/cmd/main.go b/src/frontend/mcp-server/cmd/main.go index fd34ea94..1637f089 100644 --- a/src/frontend/mcp-server/cmd/main.go +++ b/src/frontend/mcp-server/cmd/main.go @@ -25,7 +25,7 @@ const serverInstructions = "XCE scheduler agent — cluster status, not an incid "root causes, do NOT reference \"known bugs\", and do NOT propose diagnostic or repair actions (checking logs, " + "restarting daemons, opening tickets, ssh'ing to nodes, etc.) unless the user explicitly asks for help or " + "remediation. If you believe action may be warranted, say so in at most one line and let the user decide.\n\n" + - "Write tools (scheduler_submit/hold/release/cancel) are destructive and require the user's explicit intent." + "Write tools (scheduler_submit/cancel/hold/release/checkpoint/restart) are destructive and require the user's explicit intent." func main() { baseURL := os.Getenv("XCE_API_URL") @@ -221,6 +221,22 @@ func main() { mcp.WithDestructiveHintAnnotation(true), ), app.HandleReleaseJob) + // ── scheduler_checkpoint ── + s.AddTool(mcp.NewTool("scheduler_checkpoint", + mcp.WithDescription("Checkpoint a running job by ID (action 'checkpoint', CRIU). Destructive — you will be asked to confirm before execution."), + mcp.WithString("id", mcp.Description("Job ID to checkpoint"), mcp.Required()), + mcp.WithReadOnlyHintAnnotation(false), + mcp.WithDestructiveHintAnnotation(true), + ), app.HandleCheckpointJob) + + // ── scheduler_restart ── + s.AddTool(mcp.NewTool("scheduler_restart", + mcp.WithDescription("Restart (re-queue) a finished job as a new job by ID, resuming from its last checkpoint if one exists. Returns the new job ID. Destructive — you will be asked to confirm before execution."), + mcp.WithString("id", mcp.Description("Job ID to restart"), mcp.Required()), + mcp.WithReadOnlyHintAnnotation(false), + mcp.WithDestructiveHintAnnotation(true), + ), app.HandleRestartJob) + log.Printf("Starting scheduler-agent MCP server (base-url: %s)", baseURL) if err := mcps.ServeStdio(s); err != nil { log.Fatalf("server error: %v", err) diff --git a/src/frontend/mcp-server/internal/app.go b/src/frontend/mcp-server/internal/app.go index c2bc05cf..907bcb88 100644 --- a/src/frontend/mcp-server/internal/app.go +++ b/src/frontend/mcp-server/internal/app.go @@ -288,6 +288,29 @@ func (a *App) HandleReleaseJob(ctx context.Context, req mcp.CallToolRequest) (*m return textResultStr(fmt.Sprintf("job %s released (resume accepted)", id)), nil } +func (a *App) HandleCheckpointJob(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + id := getString(req, "id") + if id == "" { + return mcp.NewToolResultError("job id is required"), nil + } + if err := a.client.CheckpointJob(ctx, id); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("checkpoint job failed: %v", err)), nil + } + return textResultStr(fmt.Sprintf("job %s checkpoint accepted", id)), nil +} + +func (a *App) HandleRestartJob(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + id := getString(req, "id") + if id == "" { + return mcp.NewToolResultError("job id is required"), nil + } + res, err := a.client.RestartJob(ctx, id) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("restart job failed: %v", err)), nil + } + return textResult(res) +} + // ── helpers ───────────────────────────────────────────────────────────────── // parseStatuses parses a comma-separated list of URM status strings. diff --git a/src/frontend/mcp-server/internal/app_test.go b/src/frontend/mcp-server/internal/app_test.go index 26effd7b..839bd28d 100644 --- a/src/frontend/mcp-server/internal/app_test.go +++ b/src/frontend/mcp-server/internal/app_test.go @@ -18,12 +18,16 @@ type fakeClient struct { submitSpec *sclient.JobSpec submitErr error - cancelID string - cancelErr error - holdID string - holdErr error - releaseID string - releaseErr error + cancelID string + cancelErr error + holdID string + holdErr error + releaseID string + releaseErr error + checkpointID string + checkpointErr error + restartID string + restartErr error } func (f *fakeClient) SubmitJob(ctx context.Context, spec sclient.JobSpec) (*sclient.SubmitJobResult, error) { @@ -46,6 +50,19 @@ func (f *fakeClient) ReleaseJob(ctx context.Context, id string) error { return f.releaseErr } +func (f *fakeClient) CheckpointJob(ctx context.Context, id string) error { + f.checkpointID = id + return f.checkpointErr +} + +func (f *fakeClient) RestartJob(ctx context.Context, id string) (*sclient.RestartJobResult, error) { + f.restartID = id + if f.restartErr != nil { + return nil, f.restartErr + } + return &sclient.RestartJobResult{JobID: 999, QueueName: "normal", Message: "ok"}, nil +} + func newTestApp(client sclient.Client) *App { a := &App{client: client, templates: NewTemplateRegistry()} return a @@ -288,6 +305,92 @@ func TestHandleReleaseJob(t *testing.T) { } } +func TestHandleCheckpointJob(t *testing.T) { + fake := &fakeClient{} + a := newTestApp(fake) + + res, err := a.HandleCheckpointJob(context.Background(), idRequest("scheduler_checkpoint", "101")) + if err != nil { + t.Fatalf("HandleCheckpointJob error: %v", err) + } + if res.IsError { + t.Fatalf("expected success, got error: %s", resultText(t, res)) + } + if fake.checkpointID != "101" { + t.Errorf("checkpointID = %q, want 101", fake.checkpointID) + } + if !strings.Contains(resultText(t, res), "checkpoint accepted") { + t.Errorf("result should mention checkpoint accepted, got: %s", resultText(t, res)) + } +} + +func TestHandleCheckpointJob_MissingID(t *testing.T) { + fake := &fakeClient{} + a := newTestApp(fake) + + res, err := a.HandleCheckpointJob(context.Background(), idRequest("scheduler_checkpoint", "")) + if err != nil { + t.Fatalf("HandleCheckpointJob error: %v", err) + } + if !res.IsError { + t.Fatal("expected error for missing id") + } + if fake.checkpointID != "" { + t.Errorf("client should not be called, got checkpointID = %q", fake.checkpointID) + } +} + +func TestHandleRestartJob(t *testing.T) { + fake := &fakeClient{} + a := newTestApp(fake) + + res, err := a.HandleRestartJob(context.Background(), idRequest("scheduler_restart", "123")) + if err != nil { + t.Fatalf("HandleRestartJob error: %v", err) + } + if res.IsError { + t.Fatalf("expected success, got error: %s", resultText(t, res)) + } + if fake.restartID != "123" { + t.Errorf("restartID = %q, want 123", fake.restartID) + } + if !strings.Contains(resultText(t, res), `"jobid": 999`) { + t.Errorf("result should include the new job id 999, got: %s", resultText(t, res)) + } +} + +func TestHandleRestartJob_ClientError(t *testing.T) { + fake := &fakeClient{restartErr: &fakeError{msg: "restart failed"}} + a := newTestApp(fake) + + res, err := a.HandleRestartJob(context.Background(), idRequest("scheduler_restart", "123")) + if err != nil { + t.Fatalf("HandleRestartJob error: %v", err) + } + if !res.IsError { + t.Fatal("expected error result when client fails") + } + if !strings.Contains(resultText(t, res), "restart failed") { + t.Errorf("result should mention the client error, got: %s", resultText(t, res)) + } +} + +func TestHandleRestartJob_MissingID(t *testing.T) { + fake := &fakeClient{} + a := newTestApp(fake) + + res, err := a.HandleRestartJob(context.Background(), idRequest("scheduler_restart", "")) + if err != nil { + t.Fatalf("HandleRestartJob error: %v", err) + } + if !res.IsError { + t.Fatal("expected error for missing id") + } + if fake.restartID != "" { + t.Errorf("client should not be called, got restartID = %q", fake.restartID) + } +} + // fakeError is a minimal error for tests. type fakeError struct{ msg string } diff --git a/src/frontend/schedulerclient/client.go b/src/frontend/schedulerclient/client.go index 33a66e42..3ab7d68b 100644 --- a/src/frontend/schedulerclient/client.go +++ b/src/frontend/schedulerclient/client.go @@ -51,6 +51,13 @@ type Client interface { // ReleaseJob resumes a suspended job (action "resume", default SIGCONT). ReleaseJob(ctx context.Context, id string) error + // CheckpointJob checkpoints a running job (action "checkpoint", CRIU). + CheckpointJob(ctx context.Context, id string) error + + // RestartJob re-queues a finished job as a new job (resuming from its last + // checkpoint when one exists) and returns the new job's identity. + RestartJob(ctx context.Context, id string) (*RestartJobResult, error) + // ── Event stream (Phase 2) ── // WatchJobs subscribes to the live job-event SSE stream and returns a diff --git a/src/frontend/schedulerclient/write.go b/src/frontend/schedulerclient/write.go index 2f319d90..f8fa2bb2 100644 --- a/src/frontend/schedulerclient/write.go +++ b/src/frontend/schedulerclient/write.go @@ -51,6 +51,12 @@ type JobControlRequest struct { Method string `json:"method,omitempty"` } +// JobRestartRequest is the body sent to POST /xce/v1/jobs/:jobids/restart. +// All fields are optional; an empty request restarts to any eligible host. +type JobRestartRequest struct { + AskedHosts string `json:"asked_hosts,omitempty"` +} + // ControlResult is the apiserver's per-item control response. type ControlResult struct { Status string `json:"status"` // accepted | failed | partially_accepted | no_items_processed @@ -58,6 +64,15 @@ type ControlResult struct { Failed []ControlItem `json:"failed"` } +// RestartJobResult is the apiserver's response to a job restart — a new job +// is created (optionally resumed from the old job's last checkpoint). +type RestartJobResult struct { + JobID int64 `json:"jobid"` + JobIDHuman string `json:"job_id_human"` + QueueName string `json:"queue_name"` + Message string `json:"message"` +} + // ControlItem reports the outcome for one job (or array element). type ControlItem struct { UniqueJobID int64 `json:"unique_job_id"` @@ -223,3 +238,29 @@ func (hc *HTTPClient) HoldJob(ctx context.Context, id string) error { func (hc *HTTPClient) ReleaseJob(ctx context.Context, id string) error { return hc.controlJob(ctx, id, "resume") } + +// CheckpointJob checkpoints a running job (action "checkpoint", CRIU). +func (hc *HTTPClient) CheckpointJob(ctx context.Context, id string) error { + return hc.controlJob(ctx, id, "checkpoint") +} + +// RestartJob re-queues a finished job as a new job, resuming from its last +// checkpoint when one exists. Returns the new job's identity. The apiserver +// backs this with the same submit path, so a restart is accepted with 201. +func (hc *HTTPClient) RestartJob(ctx context.Context, id string) (*RestartJobResult, error) { + if strings.TrimSpace(id) == "" { + return nil, fmt.Errorf("restart job: job id is required") + } + + url := hc.baseURL + "/xce/v1/jobs/" + id + "/restart" + env, err := hc.doPost(ctx, url, JobRestartRequest{}, http.StatusCreated) + if err != nil { + return nil, fmt.Errorf("restart job %s: %w", id, err) + } + + var res RestartJobResult + if err := json.Unmarshal(env.Data, &res); err != nil { + return nil, fmt.Errorf("restart job %s: parse result: %w", id, err) + } + return &res, nil +} diff --git a/src/frontend/schedulerclient/write_test.go b/src/frontend/schedulerclient/write_test.go index 5a8f3897..8a3f5db3 100644 --- a/src/frontend/schedulerclient/write_test.go +++ b/src/frontend/schedulerclient/write_test.go @@ -282,6 +282,85 @@ func TestReleaseJob_SendsResume(t *testing.T) { } } +// ── Checkpoint / Restart ────────────────────────────────────────────────────── + +func TestCheckpointJob_SendsCheckpoint(t *testing.T) { + hc, capture := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{"code":202,"msg":"accepted","data":{"status":"accepted","accepted":[],"failed":[]}}`) + }) + + if err := hc.CheckpointJob(context.Background(), "101"); err != nil { + t.Fatalf("CheckpointJob returned error: %v", err) + } + + _, path, _, body := capture() + if path != "/xce/v1/jobs/101/control" { + t.Errorf("path = %q, want /xce/v1/jobs/101/control", path) + } + if strings.TrimSpace(body) != `{"action":"checkpoint"}` { + t.Errorf("body = %s, want {\"action\":\"checkpoint\"}", body) + } +} + +func TestCheckpointJob_StatusFailed(t *testing.T) { + hc, _ := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{"code":202,"msg":"failed","data":{"status":"failed","accepted":[],"failed":[{"unique_job_id":101,"job_id":101,"array_idx":0,"msg":"job 101 is in state DONE, cannot be checkpointed"}]}}`) + }) + + err := hc.CheckpointJob(context.Background(), "101") + if err == nil { + t.Fatal("expected error when control status is failed") + } + if !strings.Contains(err.Error(), "101") { + t.Errorf("error = %q, want to mention job id 101", err) + } +} + +func TestRestartJob_Success(t *testing.T) { + hc, capture := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{"code":201,"msg":"success","data":{"jobid":999,"queue_name":"normal","message":"Job <123> will be restarted with new job <999> to default queue ."}}`) + }) + + res, err := hc.RestartJob(context.Background(), "123") + if err != nil { + t.Fatalf("RestartJob returned error: %v", err) + } + + method, path, _, body := capture() + if method != http.MethodPost { + t.Errorf("method = %q, want POST", method) + } + if path != "/xce/v1/jobs/123/restart" { + t.Errorf("path = %q, want /xce/v1/jobs/123/restart", path) + } + if strings.TrimSpace(body) != "{}" { + t.Errorf("body = %s, want {}", body) + } + if res.JobID != 999 { + t.Errorf("JobID = %d, want 999", res.JobID) + } + if res.QueueName != "normal" { + t.Errorf("QueueName = %q, want normal", res.QueueName) + } +} + +func TestRestartJob_EmptyID(t *testing.T) { + hc, _ := newTestServer(t, func(w http.ResponseWriter, r *http.Request) { + t.Error("server should not be called when id is empty") + }) + + _, err := hc.RestartJob(context.Background(), " ") + if err == nil { + t.Fatal("expected error for empty id") + } + if !strings.Contains(err.Error(), "job id is required") { + t.Errorf("error = %q, want 'job id is required'", err) + } +} + // ── conversion helpers ─────────────────────────────────────────────────────── func TestFormatWalltimeSeconds(t *testing.T) { diff --git a/src/scheduler/src/core.c b/src/scheduler/src/core.c index 82d3596f..4bd747e0 100644 --- a/src/scheduler/src/core.c +++ b/src/scheduler/src/core.c @@ -432,8 +432,41 @@ xce_sched_project_t* sched_get_project(xce_sched_core_t* core, char* name) return make_sched_project(core, name); } -/* FIXME this is just stub, we need to add more */ -int sched_job_is_ready(xce_sched_job_t* j) { return 1; } +/* A job is electable only while it is PENDING and not suspended. Once + * dispatched (RUN), finished (EXIT/DONE), or suspended (PSUSP/USUSP/SSUSP) + * it must never be re-elected — bug-1498 (restored from the pre-framework + * scheduler.c after the scheduler-framework merge dropped it). */ +int sched_job_is_ready(xce_sched_job_t* j) +{ + if (j == NULL) { + return 0; + } + if ((j->state & JOB_STAT_PEND) == 0) { + return 0; /* already dispatched/running or terminal */ + } + if (j->state & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) { + return 0; /* suspended — not electable */ + } + return 1; +} -/* FIXME this is just stub, we need to add more */ -int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; } +/* A queue is electable only while it accepts new jobs. The controller + * maintains q->status (closed / inactivated-by-admin / inactive-by-window). + * The dispatch window (q->dispatchwindow, from queues.yaml) is NOT evaluated + * here: core.c is isolated from wall-clock time (test_core_isolation.c) and + * the window can only be judged against the core's virtual clock + * (xce_core_now), which this signature has no access to. Windowed-out queues + * therefore pass this gate and are re-pended with PREASON_TIME_WINDOW by the + * allocator (scheduler.c, where xce_core_now() is reachable) — same net + * effect, with a safe (allocator-thread) reason attach. PEND-reason P1. */ +int sched_queue_is_ready(xce_sched_queue_t* q) +{ + if (q == NULL) { + return 0; + } + if (q->status + & (QUEUE_STAT_CLOSED | QUEUE_STAT_INACT_ADMIN | QUEUE_STAT_INACT_WINDOW)) { + return 0; + } + return 1; +} diff --git a/src/scheduler/src/core_events.c b/src/scheduler/src/core_events.c index 78f1d1e2..0de39f9a 100644 --- a/src/scheduler/src/core_events.c +++ b/src/scheduler/src/core_events.c @@ -284,6 +284,15 @@ static xce_sched_queue_t* core_build_queue(xce_queue_t* q) queue->priority = q->priority; queue->accept_interval = q->accept_interval; queue->sched_delay = q->sched_delay; + /* PEND-reason P1 fix (2026-08-30): the scheduler-framework merge + * dropped the dispatch-window config here — make_sched_queue (schedobj.c) + * duplicates q->dispatchwindow_config, but this framework core path did + * not, so the TIME_WINDOW PEND reason could never fire on the live + * default path. Mirror make_sched_queue: dup the window (freed by + * kill_sched_queue via core->queues' datakiller). */ + if (q->dispatchwindow_config != NULL) { + queue->dispatchwindow = dup_timewindow_config(q->dispatchwindow_config); + } /* Same scope note as make_sched_queue's own (schedobj.c): project_name/ * job_pick_plugin have no wire-config source yet, and (matching * existing behavior) resreq is not populated here either -- wiring @@ -336,6 +345,15 @@ static void core_upsert_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker FREEUP(q->resreq); q->resreq = makeString(updated->resreq); } + + /* PEND-reason P1 fix (2026-08-30): same merge regression as + * core_build_queue above — the update branch never carried the + * dispatch window. Refresh it so a QUEUE_UPDATE that changes the + * window (or a re-push after the initial build) takes effect. */ + if (updated->dispatchwindow_config != NULL) { + free_timewindow_config(q->dispatchwindow); + q->dispatchwindow = dup_timewindow_config(updated->dispatchwindow_config); + } } if (picker != NULL) { diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 0067f669..43846058 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -6,6 +6,7 @@ #include "kvmgr.h" #include "plugin.h" +#include "preason.h" #include "sched/types.h" #include "scheduler.h" #include "utils.h" @@ -208,6 +209,17 @@ static int match_job_global_limit_(xce_sched_plugin_t* plugin, TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__, limit->name, obj->user, obj->queue, obj->project, rc); if (rc <= 0) { + /* global quota hit — report the specific limit (PEND-reason P1, + * restored from 4e2ac86). The election return value is ignored by + * the main loop, so the reason is carried on ctx->reason and moved + * to obj->reason by limit_job_election. */ + if (ctx->reason == NULL) { + ctx->reason = make_pendreason(0); + } + if (ctx->reason != NULL) { + add_job_preason(ctx->reason->reasons, PREASON_LIMIT_GLOBAL, NULL, + limit->name); + } return rc; } @@ -299,6 +311,14 @@ static int limit_job_election(void* arg, xce_sched_job_t* job) iter_destroy(&it); + /* Carry any global-quota reason (PREASON_LIMIT_GLOBAL) attached by + * match_job_global_limit_ onto the job so it survives to the PEND + * decision. The election's rc itself is ignored by the main loop. + * (PEND-reason P1, restored from 4e2ac86.) */ + if (ctx.reason != NULL) { + MOVE_PTR(obj->reason, ctx.reason); + } + obj->configver = sched->core.configver; return rc; @@ -358,6 +378,16 @@ static int match_job_host_limit_(xce_sched_plugin_t* plugin, TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__, limit->name, obj->user, obj->queue, obj->project, rc); if (rc <= 0) { + /* host-based quota hit — report the specific limit (PEND-reason P1, + * restored from 4e2ac86; previously the block was silent and the job + * just stayed PEND). */ + if (ctx->reason == NULL) { + ctx->reason = make_pendreason(0); + } + if (ctx->reason != NULL) { + add_job_preason(ctx->reason->reasons, PREASON_LIMIT_HOST, + worker->name, limit->name); + } return rc; } diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index 72b55551..5b5b3454 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -484,6 +484,23 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) int rc; xvec_t* uncomit = NULL; + /* Worker down/closed — never eligible. Attach HOST_UNAVAIL instead of + * trying to allocate on it (PEND-reason P1, E17 — restored from + * 4e2ac86 after the scheduler-framework merge dropped it). Runs in the + * allocator thread (sched_try_alloc_on_hosts), so ctx->reason is safe + * to touch here. */ + if (w->status + & (XCE_WORKER_UNAVAIL | XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN)) { + if (ctx->reason == NULL) { + ctx->reason = make_pendreason(0); + } + if (ctx->reason != NULL) { + add_job_preason(ctx->reason->reasons, PREASON_HOST_UNAVAIL, + w->name, NULL); + } + return 0; + } + rc = sched_meet_host_selection(sched, w, primary->select); if (rc == 0) { if (ctx->reason == NULL) { diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 9a7b92ff..785baa42 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -5,6 +5,7 @@ #include #include #include +#include #include "client.h" #include "controller.capnp.h" @@ -266,6 +267,39 @@ static int sched_make_alter_allocation(xce_scheduler_t* sched, return 0; } +/* Map a queue to the PEND reason it should report (PEND-reason P1 restore, + * resurrected from the pre-framework scheduler.c). Status flags take + * precedence, then the dispatch window. The window is evaluated against the + * core's virtual clock (xce_core_now, /1e9 → seconds) so live and sim modes + * agree; a NULL/empty window is always open. Runs only from the allocator + * thread (see the gate at the top of the allocator loop), so its reason + * attachment cannot race MOVE_PTR(obj->reason, ...) (bug-1521). */ +static int sched_queue_status_reason(xce_scheduler_t* sched, + xce_sched_queue_t* q) +{ + if (q == NULL) { + return 0; + } + if (q->status & QUEUE_STAT_CLOSED) { + return PREASON_QUEUE_CLOSED; + } + if (q->status & QUEUE_STAT_INACT_ADMIN) { + return PREASON_QUEUE_DRAINING; + } + if (q->status & QUEUE_STAT_INACT_WINDOW) { + return PREASON_TIME_WINDOW; + } + if (q->dispatchwindow != NULL) { + uint64_t now_ns = xce_core_now(&(sched->core)); + + if (!xce_timewindow_is_open(q->dispatchwindow, + (time_t)(now_ns / 1000000000ULL))) { + return PREASON_TIME_WINDOW; + } + } + return 0; +} + static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_obj_t* obj) { int rc = 0; @@ -326,6 +360,7 @@ static void* sched_allocator(void* arg) xce_decision_t* decision = NULL; xce_sched_job_t* job = NULL; int alloc_rc = 0; + int qr = 0; pthread_testcancel(); @@ -346,6 +381,35 @@ static void* sched_allocator(void* arg) plugin_job_election(sched, job); + /* PEND-reason P1 restore (bug-1521-safe): a job whose queue is not + * ready (closed / drained / outside its dispatch window) must not be + * dispatched. The gate and reason attach live HERE (allocator thread) + * rather than in the picker: the picker thread writing + * job->schedobj->reason races this thread's MOVE_PTR(obj->reason, ...) + * (bug-1521 heap corruption). The picker still skips persistently- + * closed queues per its spec contract, so this path fires for queues + * that close or window-out between pick and allocation, and for + * windowed-out queues that pass the picker's status-only gate. */ + qr = sched_queue_status_reason(sched, job->queue); + if (qr != 0) { + xce_pendreason_t* qreason = make_pendreason(0); + + job_picker->picker_iface->on_pend(job_picker, jobid); + + if (qreason != NULL) { + qreason->jobid = job->jobid; + add_job_preason(qreason->reasons, qr, + job->queue != NULL ? job->queue->name : NULL, + NULL); + decision = make_pend_decision(qreason); + } + if (decision != NULL) { + commit_decision(sched, decision); + cbuf_put(&(sched->core.decisions), &decision); + } + continue; + } + /* calculate decision for jobid */ alloc_rc = sched_make_allocation(sched, job->schedobj); if (alloc_rc < 0) { diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index e8d72cc9..2c46866c 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -264,6 +264,19 @@ sched_core_jbuf_capacity_pin_test = executable( + '"'], install: false) +# Regression guard for the PEND-reason P1 readiness gates restored after +# the scheduler-framework merge dropped them (sched_job_is_ready / +# sched_queue_is_ready were `return 1` stubs in core.c). Same white-box +# #include pattern as test_core_jbuf_capacity_pin.c (pulls the real +# ../src/core.c into this TU); exercises the REAL gate functions, not a +# re-implementation. +sched_core_ready_gates_test = executable( + 'sched_core_ready_gates_test', + ['test_core_ready_gates.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + # Contract test for the core event interface and batch event push # (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8): xce_core_push_events() # fires the resolved picker's corresponding state-change hook exactly once @@ -1235,6 +1248,9 @@ test('scheduler no bare printf test', test('scheduler core jbuf capacity pin test', sched_core_jbuf_capacity_pin_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler core ready gates test', + sched_core_ready_gates_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) test('scheduler core events test', sched_core_events_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_core_ready_gates.c b/src/scheduler/testsuite/test_core_ready_gates.c new file mode 100644 index 00000000..ef3aecf0 --- /dev/null +++ b/src/scheduler/testsuite/test_core_ready_gates.c @@ -0,0 +1,121 @@ +/* + * test_core_ready_gates.c - contract test for the PEND-reason P1 readiness + * gates restored after the scheduler-framework merge dropped them + * (sched_job_is_ready / sched_queue_is_ready were `return 1` stubs in + * core.c; the pre-framework scheduler.c implementations were lost). + * + * Observable under test: + * - sched_job_is_ready(): a job is electable only while PENDING and not + * suspended. NULL / RUN / EXIT / DONE / PSUSP / USUSP / SSUSP -> 0. + * PEND -> 1. (bug-1498: a running job lingering in the pendJobs queue + * must never be re-elected and re-dispatched.) + * - sched_queue_is_ready(): a queue is electable only while it accepts + * new jobs. NULL / QUEUE_STAT_CLOSED / QUEUE_STAT_INACT_ADMIN / + * QUEUE_STAT_INACT_WINDOW -> 0. Status 0 -> 1. The dispatch-window + * half of the gate intentionally lives in the allocator (scheduler.c, + * xce_core_now) not here: core.c is isolated from wall-clock time, and + * the queue has no back-pointer to the core (see the function's own + * comment in core.c). + * + * White-box #include of ../src/core.c (same convention as + * test_core_jbuf_capacity_pin.c): these tests exercise the REAL core.c + * functions, not a re-implementation. Stub surface is exactly the seven + * kvmgr datakillers/plugin_unregister core.c references as addresses in + * xce_sched_core_init() (never actually invoked -- fixtures aren't torn + * down, same convention as the sibling core tests). + */ +#include + +#include "sched/types.h" +#include "xce/types.h" + +#include "../src/core.c" + +/* Linker stubs for core.c's extern datakillers (never invoked). */ +void kill_sched_job(void* arg) { (void)arg; } +void kill_sched_queue(void* arg) { (void)arg; } +void kill_sched_worker(void* arg) { (void)arg; } +void kill_sched_user(void* arg) { (void)arg; } +void kill_sched_project(void* arg) { (void)arg; } +void kill_sched_resval(xce_val_t* val) { (void)val; } +void plugin_unregister(void* arg) { (void)arg; } + +void setUp(void) { return; } +void tearDown(void) { return; } + +static void test_job_is_ready_rejects_null(void) +{ + TEST_ASSERT_EQUAL_INT(0, sched_job_is_ready(NULL)); +} + +static void test_job_is_ready_accepts_pend(void) +{ + xce_sched_job_t j = { .state = JOB_STAT_PEND }; + + TEST_ASSERT_EQUAL_INT(1, sched_job_is_ready(&j)); +} + +static void test_job_is_ready_rejects_non_pend(void) +{ + const uint32_t states[] = { 0, JOB_STAT_RUN, JOB_STAT_DONE, JOB_STAT_EXIT }; + size_t i; + + for (i = 0; i < sizeof(states) / sizeof(states[0]); i++) { + xce_sched_job_t j = { .state = states[i] }; + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, sched_job_is_ready(&j), + "non-PEND job must not be electable"); + } +} + +static void test_job_is_ready_rejects_suspended(void) +{ + const uint32_t susp[] = { JOB_STAT_PSUSP, JOB_STAT_USUSP, JOB_STAT_SSUSP }; + size_t i; + + for (i = 0; i < sizeof(susp) / sizeof(susp[0]); i++) { + xce_sched_job_t j = { .state = JOB_STAT_PEND | susp[i] }; + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, sched_job_is_ready(&j), + "suspended job must not be electable"); + } +} + +static void test_queue_is_ready_rejects_null(void) +{ + TEST_ASSERT_EQUAL_INT(0, sched_queue_is_ready(NULL)); +} + +static void test_queue_is_ready_accepts_open(void) +{ + xce_sched_queue_t q = { .status = 0 }; + + TEST_ASSERT_EQUAL_INT(1, sched_queue_is_ready(&q)); +} + +static void test_queue_is_ready_rejects_closed_or_inactive(void) +{ + const uint32_t statuses[] = { QUEUE_STAT_CLOSED, QUEUE_STAT_INACT_ADMIN, + QUEUE_STAT_INACT_WINDOW }; + size_t i; + + for (i = 0; i < sizeof(statuses) / sizeof(statuses[0]); i++) { + xce_sched_queue_t q = { .status = statuses[i] }; + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, sched_queue_is_ready(&q), + "closed/inactive queue must not be electable"); + } +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_job_is_ready_rejects_null); + RUN_TEST(test_job_is_ready_accepts_pend); + RUN_TEST(test_job_is_ready_rejects_non_pend); + RUN_TEST(test_job_is_ready_rejects_suspended); + RUN_TEST(test_queue_is_ready_rejects_null); + RUN_TEST(test_queue_is_ready_accepts_open); + RUN_TEST(test_queue_is_ready_rejects_closed_or_inactive); + return UNITY_END(); +}