From 9e0d87fe5071505321833ed0745c1f9da4c42d61 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 15 Jul 2026 15:56:48 +0000 Subject: [PATCH] feat(job-array-single-wal-record): API-server-side (Go) local expansion of the compact push Adds the Go/CGO receive path for Data.jobArray, mirroring task 5.3's scheduler-side C implementation: rawJobArrayToModels() (pure-Go base_jobid + i expansion), cJobArrayToModels() (CGO adapter sibling to cJobToModel), dispatchJobArray() (dedup via the same store.Jobs.Has() the ordinary Data_data_jobs path already writes through, single confirmation log only when at least one element was newly dispatched), and a new case C.Data_data_jobArray in handleDataset. Elements route through OnJobUpdated, matching the ordinary single-job push's own routing. The CGO boundary needed no changes -- both C.xce_job_submit_t and C.Data_data_jobArray were already exposed from prior tasks, including task 5.3b's extract_data() fix which this receive path depends on identically to the scheduler's. store.ArrayJobs (the array-level rollup aggregate) is deliberately left unpopulated -- confirmed out of this spec's scope, deferred to a future spec per requirements.md's Boundary Context. Task 5.4 of job-array-single-wal-record (Requirements 6.2, 6.4, 6.5, 8.4). --- .../job-array-single-wal-record/tasks.md | 3 +- src/frontend/apiserver/backend/clib.go | 17 +++ src/frontend/apiserver/backend/convert.go | 42 ++++++ .../apiserver/backend/convert_pure.go | 38 ++++- .../apiserver/backend/convert_test.go | 63 ++++++++ src/frontend/apiserver/backend/handler.go | 49 ++++++ .../apiserver/backend/handler_test.go | 142 +++++++++++++++++- 7 files changed, 345 insertions(+), 9 deletions(-) diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index 014ef03..391444c 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -140,7 +140,7 @@ - _Boundary: shared wire transport (`src/lib/request.c`, `src/lib/xcectx.c`) — out of the Compact Scheduler/API-Server Push component's original File Structure Plan; discovered during task 5.3's review (2026-07-15), design.md corrected to record it_ - _Depends: 5.1, 5.2_ -- [ ] 5.4 (P) API-server-side local expansion +- [x] 5.4 (P) API-server-side local expansion - Decode the compact push and expand it locally into N per-element records using the same base-jobid-plus-index arithmetic - Deduplicate by element identity so a redelivered push does not create duplicate records - Log one identifiable confirmation line (base jobid plus element count) once expansion completes @@ -192,3 +192,4 @@ - **Task 5.3**: `sched_on_job_array()` (`handler.c`) decodes `Data_data_jobArray`, derives each element's jobid via `base_jobid + i`, and routes constructed elements through the unmodified per-job pipeline (`sched_on_jobs()`). Each element gets its OWN deep clone (`clone_job_submit_data`/`clone_job_startinfo`) of the shared submit spec — required because `sched_pend_update_killer()`'s per-element teardown (`free_Job_ptr`) frees each element's `submit` independently and a shared pointer across N elements would double-free; independently verified. Dedup reuses `sched_get_job()` against the same `sched->jobs` kvtab `sched_add_job()`'s `key_add` authoritatively guards — sound at the data-integrity level, but review found a narrow, non-corrupting TOCTOU: the dedup pre-check runs synchronously at receive time while the actual insert only happens later when `sched_update()` drains `pend_updates[0]`, so a redelivery landing before that drain can re-construct/re-enqueue a duplicate (harmless) object and double the confirmation log line — the same pre-existing, task-5.3-unrelated pattern as `sched_update_jobs()`'s already-ignored `key_add` `-EEXIST` return on the ordinary single-job path. Not fixed (pre-existing class of gap, narrow window, non-corrupting). **CRITICAL discovery during this task's review (2026-07-15), NOT a defect in task 5.3 itself**: the shared wire dispatcher `extract_data()` (`src/lib/request.c`) has no case for `Data_data_jobArray` and silently drops every real array push before ANY receiver's handler runs (see new task 5.3b, inserted immediately after this task per explicit user decision). Confirmed to affect task 5.4's future API-server receive path identically (both share the same `xcecli_ctx_init()` → `vtask_entry()` → `extract_data()` pipeline). Task 5.3's own tests are unaffected by this gap since they use the same established white-box convention as every sibling `handler.c` test (calling `sched_handle_dataset()` directly, bypassing the transport layer) — this is not a new testing shortcut, it just means task 5.3's passing tests do not, and were never intended to, prove real-wire delivery. - **Task 5.3b**: Added `case Data_data_jobArray:` to `extract_data()` (`request.c`) — sets `dataset->rows`/`nrows=1` directly (no pagination, matching the always-one-shot wire shape) and nulls `getdata->data.job_array` after transferring it. The null is NOT cosmetic — independently verified `free_Request()`'s `Request_body_pushmsg` case calls `free_Data()`, which has a live `Data_data_jobArray` branch that would double-free/UAF without it. Proven end-to-end with a real `xce_encode()`/`xce_decode()` round-trip through `extract_data()` → `sched_handle_dataset()` → `sched_on_job_array()` (task 5.3, unmodified), not just a unit-level check. Also independently confirmed the alternate hypothesis some earlier notes floated — that `scheduler_req_handler()` itself has a `pushmsg`-rejecting branch — is dead code on the live wire path, since `vtask_entry()` always converts `body_which` to `datamsg` before invoking any handler. **New residual gap, self-flagged, NOT fixed (pre-existing since task 5.3, not introduced here, genuinely out of this task's `request.c`-only boundary)**: `dataset->rows` for the jobArray case is never deep-freed downstream — `free_Dataset()` is an empty no-op for all table types, and `sched_handle_dataset()`'s jobArray case (task 5.3) doesn't null it after consuming it. Net effect: one bounded `xce_job_submit_t` + embedded spec leaks per successfully-delivered array push commit (not per element, not unbounded) — real but small; worth a dedicated follow-up before this spec is considered fully closed, alongside the other self-flagged residual gaps (task 3.1's bulk-insert orphan prefix, task 4.2's boot-replay fence gap). +- **Task 5.4**: Go/CGO receive path — `rawJobArrayToModels()` (pure-Go N-way expansion), `cJobArrayToModels()` (CGO adapter sibling to `cJobToModel`), `dispatchJobArray()` (dedup via the SAME `store.Jobs.Has()` the ordinary `Data_data_jobs` path writes through, one confirmation log only on ≥1 newly-dispatched element), new `case C.Data_data_jobArray:` in `handleDataset`. Elements route through `OnJobUpdated` (not `OnJobCreated`), matching the ordinary single-job push's own routing — verified consistent, not a divergence. CGO boundary needed zero changes — `C.xce_job_submit_t` and `C.Data_data_jobArray` were already exposed/generated from prior tasks. A structural TOCTOU exists between the `Has()` dedup check and the eventual store write (two separate lock acquisitions, no compare-and-set primitive) but is narrower than task 5.3's accepted C-side gap (no intervening async hop) and strictly safer than the pre-existing baseline (the ordinary `Data_data_jobs` path has NO dedup check at all today) — not fixed, judged acceptable. `cJobArrayToModels()` has zero direct unit coverage at the CGO-struct level, but this exactly mirrors the pre-existing, repo-wide precedent for `cJobToModel()` itself (also never unit-tested with a real C struct) — reviewer independently reconstructed a full `CGO_ENABLED=1` build+link+test run against the real generated headers to confirm correctness beyond a compile-only check. `store.ArrayJobs` (array-level rollup aggregate) deliberately left unpopulated — confirmed genuinely out of this spec's scope per requirements.md's Boundary Context, which explicitly defers whole-array bulk query/cancel to a future spec; zero code anywhere in the codebase populates it today, including this diff. diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index 86d2fa4..e993c68 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -256,6 +256,23 @@ func handleDataset(d *C.dataset_t) C.int { info := cWorkerToModel(wp) gHandler.OnHostUpdated(info) } + + case C.Data_data_jobArray: + // job-array-single-wal-record task 5.4 (Req 6.2, 6.4, 6.5, 8.4): + // sibling receive path to the Data_data_jobs case above, for the + // compact array descriptor apiserver_push_jobarray() (apiserver.c, + // task 5.2) sends. Unlike the LIST-shaped cases, extract_data() + // (request.c, task 5.3b) always delivers exactly one descriptor + // (dataset->nrows == 1, dataset->rows == the single + // *xce_job_submit_t) -- no per-row slice to walk. Decode it, + // expand locally into N models.JobInfo entries, and hand them to + // dispatchJobArray for dedup + delivery + confirmation logging. + if d.nrows == 0 || d.rows == nil { + return 0 + } + jobArray := (*C.xce_job_submit_t)(d.rows) + elements := cJobArrayToModels(jobArray) + dispatchJobArray(gHandler, elements, uint64(jobArray.jobid), uint64(jobArray.array_size)) } return 0 } diff --git a/src/frontend/apiserver/backend/convert.go b/src/frontend/apiserver/backend/convert.go index 22ddd17..da729be 100644 --- a/src/frontend/apiserver/backend/convert.go +++ b/src/frontend/apiserver/backend/convert.go @@ -28,6 +28,48 @@ func goString(s *C.char) string { return C.GoString(s) } +// cJobArrayToModels decodes a compact Data.jobArray push descriptor +// (*C.xce_job_submit_t: base jobid, array_size, one shared job_submit_data_t +// spec — the same JobSubmit shape apiserver_push_jobarray() sends, +// job-array-single-wal-record task 5.2) and expands it locally into N +// models.JobInfo entries via rawJobArrayToModels, mirroring cJobToModel's +// single-job field extraction below but for the array-shaped descriptor +// (task 5.4, Req 6.2, 6.5). Decodes the shared submit spec exactly once +// (same field-by-field extraction cJobToModel uses for job.submit) since +// every element shares it by construction — never once per element. +func cJobArrayToModels(jobArray *C.xce_job_submit_t) []*models.JobInfo { + if jobArray == nil || jobArray.submit == nil { + return nil + } + + s := jobArray.submit + rd := &rawSubmitData{ + queue: goString(s.queue), + user: goString(s.user), + jobName: goString(s.job_name), + projectName: goString(s.project_name), + userGroup: goString(s.user_group), + jobDesc: goString(s.job_description), + resReq: goString(s.resreq), + minTasks: int(s.mintasks), + maxTasks: int(s.maxtasks), + } + if s.start_info != nil { + si := s.start_info + rd.startInfo = &rawStartInfo{ + cmdline: goString(si.cmdline), + launchType: goString(si.launch_type), + preExec: goString(si.job_pre_exec), + postExec: goString(si.job_post_exec), + execHome: goString(si.exechome), + execCWD: goString(si.execcwd), + execUser: goString(si.execuser), + } + } + + return rawJobArrayToModels(uint64(jobArray.jobid), uint64(jobArray.array_size), rd) +} + // cJobToModel extracts C xce_job_t fields into a rawJobData and maps it to JobInfo. func cJobToModel(job *C.xce_job_t) *models.JobInfo { r := rawJobData{ diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index 48b0048..d4e0f76 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -3,6 +3,7 @@ package backend import ( "apiserver/models" "encoding/json" + "time" ) // JobID bit layout constants — mirrors src/include/config.h. @@ -180,9 +181,9 @@ func rawJobToModel(r rawJobData) *models.JobInfo { } } info.SubmitTime = r.subTime - info.QueueTime = r.queueTime - info.StartTime = r.startTime - info.EndTime = r.finishTime + info.QueueTime = r.queueTime + info.StartTime = r.startTime + info.EndTime = r.finishTime if len(r.execHosts) > 0 { type hostSlot struct { HostName string `json:"host_name"` @@ -202,6 +203,37 @@ func rawJobToModel(r rawJobData) *models.JobInfo { return info } +// rawJobArrayToModels expands one compact (base jobid, array_size, one +// shared spec) array descriptor into N models.JobInfo entries locally, one +// per index in [0, array_size), using base_jobid + i arithmetic — the same +// arithmetic the controller's restore_job_array() (job.c) and the +// scheduler's sched_on_job_array() (handler.c) use to derive each +// element's identity (job-array-single-wal-record task 5.4, Req 6.2, 6.5). +// A freshly-received array element mirrors sched_on_job_array's +// make_array_element: JOB_STAT_PEND, receive-time sub_time — the compact +// wire descriptor carries no per-element timestamps to recover. +// Returns nil for a zero array_size or a nil submit (nothing to expand). +// Pure Go (no CGO) — cJobArrayToModels (convert.go) is the CGO adapter +// that decodes a *C.xce_job_submit_t into the rawSubmitData this consumes. +func rawJobArrayToModels(baseJobID uint64, arraySize uint64, submit *rawSubmitData) []*models.JobInfo { + if arraySize == 0 || submit == nil { + return nil + } + + now := time.Now().Unix() + out := make([]*models.JobInfo, 0, arraySize) + for i := uint64(0); i < arraySize; i++ { + r := rawJobData{ + jobID: int64(baseJobID + i), + state: models.JOBSTAT["PEND"], + subTime: now, + submit: submit, + } + out = append(out, rawJobToModel(r)) + } + return out +} + // rawWorkerToModel maps a rawWorkerData to a models.HostInfo. func rawWorkerToModel(r rawWorkerData) *models.HostInfo { info := &models.HostInfo{ diff --git a/src/frontend/apiserver/backend/convert_test.go b/src/frontend/apiserver/backend/convert_test.go index 6ec9f0c..83d2cdb 100644 --- a/src/frontend/apiserver/backend/convert_test.go +++ b/src/frontend/apiserver/backend/convert_test.go @@ -714,6 +714,69 @@ func TestRawResvalToDistribution_BooleanFalse(t *testing.T) { } } +// --- job-array-single-wal-record task 5.4: rawJobArrayToModels --- +// +// Sibling to cJobToModel's local-expansion contract: given one compact +// (base jobid, array_size, shared spec) descriptor, expand it into N +// models.JobInfo entries using base_jobid + i arithmetic — the same +// arithmetic the controller's restore_job_array() and the scheduler's +// sched_on_job_array() use to derive each element's identity (Req 6.2, +// 6.5). Pure Go (no CGO), mirrors rawJobToModel's existing split. + +func TestRawJobArrayToModels_ExpandsNElementsWithCorrectIdentity(t *testing.T) { + jobNumber := uint64(9) + baseJobID := makeJobID(1, 0, 1700000000, tagJobArray, jobNumber< 0 { + logger.Info("job array push received and expanded", + zap.Uint64("base_jobid", baseJobID), + zap.Uint64("array_size", arraySize), + zap.Int("new_records", nNew)) + } + + return nNew +} + // dispatchConfigPush delivers converted config-push data to the handler (CGO-free, for testability). func dispatchConfigPush( handler NotificationHandler, diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go index dfc609c..3a27740 100644 --- a/src/frontend/apiserver/backend/handler_test.go +++ b/src/frontend/apiserver/backend/handler_test.go @@ -3,7 +3,12 @@ package backend import ( "apiserver/models" "apiserver/store" + "apiserver/utils" "testing" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" ) // resetStores reinitializes all global stores for test isolation. @@ -11,6 +16,17 @@ func resetStores() { store.InitStores() } +// captureLog swaps utils.Log for an observer core so a test can assert on +// exactly which log lines fired, restoring the previous logger on cleanup. +func captureLog(t *testing.T) *observer.ObservedLogs { + t.Helper() + core, logs := observer.New(zapcore.InfoLevel) + prev := utils.Log + utils.Log = zap.New(core) + t.Cleanup(func() { utils.Log = prev }) + return logs +} + func TestOnHostConfigUpdated_UnregisteredHostIgnored(t *testing.T) { // Config-only entries (e.g. "default", hosts whose daemons haven't started) // must not be inserted into the Hosts store. Only OnHostUpdated (worker push) creates entries. @@ -185,10 +201,10 @@ func TestOnHostUpdated_PartialMerge(t *testing.T) { // Simulate a prior config push that set config-sourced fields. store.Hosts.Put("node01", models.HostInfo{ - HostName: "node01", - HostModel: "x86_64", - HostType: "LINUX64", - MaxSlot: 8, + HostName: "node01", + HostModel: "x86_64", + HostType: "LINUX64", + MaxSlot: 8, DisptWindow: "always", }) @@ -347,6 +363,122 @@ func TestDispatchConfigPush_NilClusterIsSkipped(t *testing.T) { } } +// --- job-array-single-wal-record task 5.4: dispatchJobArray --- +// +// dispatchJobArray is the sibling of the single-job Data_data_jobs push +// path (clib.go handleDataset, which calls gHandler.OnJobUpdated per job +// with no separate dedup step because store.Jobs is a map keyed by jobid). +// For the array path, dedup happens BEFORE calling the handler so a +// redelivered push can never clobber a jobid whose state has since moved +// on, mirroring the scheduler's sched_on_job_array() (handler.c) skip-if- +// exists behavior. Requirement 6.4/6.5/8.4: exactly one confirmation log +// line per successful (>=1 new element) expansion, zero on a fully- +// duplicate redelivery. + +func TestDispatchJobArray_NewElementsStoredAndLogged(t *testing.T) { + resetStores() + logs := captureLog(t) + h := &DefaultHandler{} + + elements := []*models.JobInfo{ + {Unique_JobID: 100, JobID: 100, ArrayIdx: 0, QueueName: "normal"}, + {Unique_JobID: 101, JobID: 100, ArrayIdx: 1, QueueName: "normal"}, + {Unique_JobID: 102, JobID: 100, ArrayIdx: 2, QueueName: "normal"}, + } + + nNew := dispatchJobArray(h, elements, 100, 3) + + if nNew != 3 { + t.Errorf("nNew: got %d, want 3", nNew) + } + for _, jobid := range []uint64{100, 101, 102} { + got, ok := store.Jobs.Get(jobid) + if !ok { + t.Errorf("jobid %d should exist in store.Jobs after expansion", jobid) + continue + } + if got.QueueName != "normal" { + t.Errorf("jobid %d: QueueName got %q, want normal", jobid, got.QueueName) + } + } + if got := logs.Len(); got != 1 { + t.Errorf("expected exactly 1 confirmation log line, got %d", got) + } +} + +func TestDispatchJobArray_FullRedeliveryIsNoOpAndDoesNotLog(t *testing.T) { + resetStores() + logs := captureLog(t) + h := &DefaultHandler{} + + elements := []*models.JobInfo{ + {Unique_JobID: 200, JobID: 200, ArrayIdx: 0}, + {Unique_JobID: 201, JobID: 200, ArrayIdx: 1}, + } + + // First delivery: both elements are new. + if nNew := dispatchJobArray(h, elements, 200, 2); nNew != 2 { + t.Fatalf("first delivery nNew: got %d, want 2", nNew) + } + logs.TakeAll() // discard the first delivery's own confirmation line + + // Redelivery of the identical push (e.g. after a reconnect): every + // index already exists, so this must be a pure no-op. + nNew := dispatchJobArray(h, elements, 200, 2) + + if nNew != 0 { + t.Errorf("redelivery nNew: got %d, want 0", nNew) + } + if got := logs.Len(); got != 0 { + t.Errorf("expected zero confirmation log lines on full redelivery, got %d", got) + } + if n := store.Jobs.Count(); n != 2 { + t.Errorf("store.Jobs should still have exactly 2 entries after redelivery, got %d", n) + } +} + +func TestDispatchJobArray_PartialRedeliverySkipsExistingOnly(t *testing.T) { + resetStores() + h := &DefaultHandler{} + + // Pre-seed jobid 300 with state that a stale redelivered array push + // must NOT clobber (dedup-by-identity is a correctness guard, not + // just a log-suppression detail). + store.Jobs.Put(300, models.JobInfo{Unique_JobID: 300, JobStatus: models.JOBSTAT["RUN"]}) + + logs := captureLog(t) + elements := []*models.JobInfo{ + {Unique_JobID: 300, JobID: 300, ArrayIdx: 0, JobStatus: models.JOBSTAT["PEND"]}, + {Unique_JobID: 301, JobID: 300, ArrayIdx: 1, JobStatus: models.JOBSTAT["PEND"]}, + } + + nNew := dispatchJobArray(h, elements, 300, 2) + + if nNew != 1 { + t.Errorf("nNew: got %d, want 1 (only jobid 301 is new)", nNew) + } + got300, _ := store.Jobs.Get(300) + if got300.JobStatus != models.JOBSTAT["RUN"] { + t.Errorf("jobid 300 must not be clobbered by a stale redelivered array push: got status %d, want %d", + got300.JobStatus, models.JOBSTAT["RUN"]) + } + if _, ok := store.Jobs.Get(301); !ok { + t.Error("jobid 301 should have been created") + } + if got := logs.Len(); got != 1 { + t.Errorf("expected exactly 1 confirmation log line for a partial (1 new) expansion, got %d", got) + } +} + +func TestDispatchJobArray_NilHandlerIsNoOp(t *testing.T) { + resetStores() + elements := []*models.JobInfo{{Unique_JobID: 400}} + nNew := dispatchJobArray(nil, elements, 400, 1) + if nNew != 0 { + t.Errorf("nNew: got %d, want 0 for nil handler", nNew) + } +} + func TestDispatchConfigPush_EmptySlicesNoOp(t *testing.T) { resetStores() h := &DefaultHandler{} @@ -367,4 +499,4 @@ func TestDispatchConfigPush_EmptySlicesNoOp(t *testing.T) { if n := store.Clusters.Count(); n != 0 { t.Errorf("Clusters store should be empty, got %d entries", n) } -} \ No newline at end of file +}