From b7b90e9d214e46540bea3e0d14cbd8898fc74d64 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 15 Jul 2026 17:27:34 +0000 Subject: [PATCH] test(job-array-single-wal-record): end-to-end integration test for the compact push Cross-daemon integration test, test-only, spanning the scheduler (C) and API-server (Go/CGO) receive paths built in tasks 5.1-5.4/5.3b. Drives real xce_encode()/xce_decode() wire bytes through the actual production dispatch chain on both daemons -- extract_data() -> scheduler_req_handler() -> sched_handle_dataset() -> sched_on_job_array() and go_push_handler() -> handleDataset() -> cJobArrayToModels()/dispatchJobArray() -- skipping only vtask_entry()'s ~10 lines of static glue, mirrored verbatim in test code (the if (callnext) treats-any-nonzero-as-proceed behavior that caused 5.3b's bug is reproduced exactly). Redelivery is tested via genuinely independent fresh decodes of separately-encoded bytes, not a decode-once-call-twice shortcut. Bug-catching capability proven by temporarily reverting task 5.3b's fix and confirming both the C and Go tests fail meaningfully, then restoring to green -- confirmed independently by both the implementer and reviewer. Task 5.5 of job-array-single-wal-record (Requirements 6.1-6.5, 8.4). This completes Phase 5 (Compact Scheduler/API-Server Push). --- .../job-array-single-wal-record/tasks.md | 3 +- .../backend/wire_dispatch_support.go | 193 ++++++++ .../apiserver/backend/wire_dispatch_test.go | 138 ++++++ src/scheduler/testsuite/meson.build | 16 + .../testsuite/test_job_array_wire_e2e.c | 420 ++++++++++++++++++ 5 files changed, 769 insertions(+), 1 deletion(-) create mode 100644 src/frontend/apiserver/backend/wire_dispatch_support.go create mode 100644 src/frontend/apiserver/backend/wire_dispatch_test.go create mode 100644 src/scheduler/testsuite/test_job_array_wire_e2e.c diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index 391444c..4b0204c 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -149,7 +149,7 @@ - _Boundary: Compact Scheduler/API-Server Push (apiserver receive path)_ - _Depends: 5.2_ -- [ ] 5.5 Integration test: end-to-end compact push +- [x] 5.5 Integration test: end-to-end compact push - Commit one array record end-to-end and confirm the scheduler and the API server each receive exactly one push and each expand it into N correctly-addressable entries - Confirm a simulated redelivery (reconnect) to either receiver does not double-expand - Observable: a passing integration test covering both receivers' one-push/N-entries/no-double-expand behavior, including both receivers' confirmation log lines @@ -193,3 +193,4 @@ **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. +- **Task 5.5**: Cross-daemon integration test, test-only (zero net production diff, confirmed on `request.c` and every task 5.1-5.4 file). Drives real `xce_encode()`/`xce_decode()` wire bytes through the REAL production dispatch chain on both daemons — `extract_data()` → `scheduler_req_handler()` → `sched_handle_dataset()` → `sched_on_job_array()` (C, via `#include "../src/handler.c"` in the test) and `go_push_handler()` → `handleDataset()` → `cJobArrayToModels()`/`dispatchJobArray()` (Go/CGO) — skipping only `vtask_entry()`'s ~10 lines of `static` glue, which the test mirrors verbatim (independently verified line-by-line accurate on the one behavior that mattered: `if (callnext)` treating any nonzero `extract_data()` return, including a negative errno, as "proceed"). Redelivery is tested via genuinely independent fresh decodes of separately-encoded bytes, not a cheaper decode-once-call-twice shortcut. Bug-catching capability independently proven twice (implementer, then reviewer): temporarily reverting task 5.3b's fix makes both the C and Go tests fail meaningfully; restoring returns both to green. Go/cgo toolchain constraint discovered: `import "C"` is disallowed in any `_test.go` file, forcing a two-file split (`wire_dispatch_support.go` for cgo-preamble helpers, `wire_dispatch_test.go` for the actual tests) — logged as bug-1440/1441/1442. One low-severity, non-blocking nit (not fixed): the C test file's own header comment misdescribes which assertion fails first under the RED-phase revert (claims the `callnext != 0` check; actually a downstream `>= 0` check) — cosmetic only, the test's real bug-catching property is independently confirmed accurate. diff --git a/src/frontend/apiserver/backend/wire_dispatch_support.go b/src/frontend/apiserver/backend/wire_dispatch_support.go new file mode 100644 index 0000000..3aaf183 --- /dev/null +++ b/src/frontend/apiserver/backend/wire_dispatch_support.go @@ -0,0 +1,193 @@ +//go:build cgo + +package backend + +/* +#include +#include +#include "client.h" +#include "netio.h" +#include "request.h" +#include "controller.capnp.h" + +// job-array-single-wal-record task 5.5: xce_data_t.data and xce_request_t.body +// are C unions -- cgo represents these as opaque byte arrays, not named +// fields, so Go code cannot write/read them directly. These test-support +// helpers mirror the established pattern this package already uses for the +// SAME problem (clib.go's jobctrl_set_siginfo/jobctrl_set_ckptoption, +// val_helpers.h's xce_val_t accessors) -- new test-support C code, no +// production file touched. +static void wire_test_data_set_job_array(xce_data_t *d, xce_job_submit_t *ja) { + d->data.job_array = ja; +} +static void wire_test_request_set_pushmsg(xce_request_t *r, xce_data_t *d) { + r->body_which = Request_body_pushmsg; + r->body.pushmsg = d; +} +static xce_data_t* wire_test_request_get_pushmsg(xce_request_t *r) { + return r->body.pushmsg; +} +static void wire_test_request_set_datamsg(xce_request_t *r, dataset_t *ds) { + r->body_which = Request_body_datamsg; + r->body.datamsg = ds; +} +*/ +import "C" + +import ( + "fmt" + "unsafe" +) + +// wire_dispatch_support.go backs wire_dispatch_test.go (job-array-single- +// wal-record task 5.5). It is NOT named *_test.go despite being test-only in +// purpose: the Go toolchain hard-rejects `import "C"` in any file whose name +// ends in _test.go ("use of cgo in test ... not supported", golang/go#13938 +// -- confirmed against this repo's actual go toolchain source, +// cmd/go/internal/modindex/read.go). Every function here is unexported and +// never called by production code (RegisterToBackend, SubmitJob, etc.) -- +// it compiles into the production apiserver binary whenever CGO is enabled +// (same as any other file in this package) but is inert dead code there, +// the standard accepted trade-off for driving a real CGO struct through a +// Go test. wire_dispatch_test.go itself imports nothing from "C" -- it only +// calls the plain-Go-signature wrappers below. + +// wireDispatchBuildPush constructs the exact wire message +// apiserver_push_jobarray() (apiserver.c, task 5.2) sends -- OPCode_push / +// Request_body_pushmsg carrying one Data.jobArray descriptor -- and encodes +// it via the REAL C.xce_encode(), mirroring the C-side test's +// build_encoded_push() (test_job_array_wire_e2e.c, task 5.5) and +// test_request.c's test_extract_data_job_array_passes_through_intact (task +// 5.3b). Returns an opaque handle to the encoded io_packet_t (owned by the +// caller until wireDispatchDeliver consumes it) usable only with +// wireDispatchDeliver. +func wireDispatchBuildPush(baseJobID, arraySize uint64) (unsafe.Pointer, error) { + submit := (*C.job_submit_data_t)(C.calloc(1, C.size_t(unsafe.Sizeof(C.job_submit_data_t{})))) + submit.queue = C.CString("normal") + submit.user = C.CString("tester") + submit.resreq = C.CString("type == local") + submit.mintasks = 1 + submit.maxtasks = 1 + + jobArray := (*C.xce_job_submit_t)(C.calloc(1, C.size_t(unsafe.Sizeof(C.xce_job_submit_t{})))) + jobArray.jobid = C.uint64_t(baseJobID) + jobArray.array_size = C.uint64_t(arraySize) + jobArray.submit = submit + + data := (*C.xce_data_t)(C.calloc(1, C.size_t(unsafe.Sizeof(C.xce_data_t{})))) + data.table = C.Data_data_jobArray + C.wire_test_data_set_job_array(data, jobArray) + + var request C.xce_request_t + request.opcode = C.OPCode_push + C.wire_test_request_set_pushmsg(&request, data) + + pkt := (*C.io_packet_t)(C.calloc(1, C.size_t(unsafe.Sizeof(C.io_packet_t{})))) + if rc := C.xce_encode(&request, pkt); rc != 1 { + C.free(unsafe.Pointer(pkt)) + C.free(unsafe.Pointer(submit.queue)) + C.free(unsafe.Pointer(submit.user)) + C.free(unsafe.Pointer(submit.resreq)) + C.free(unsafe.Pointer(submit)) + C.free(unsafe.Pointer(jobArray)) + C.free(unsafe.Pointer(data)) + return nil, fmt.Errorf("xce_encode failed: rc=%d", int(rc)) + } + + // The hand-built C structs are no longer needed once encoded into pkt's + // wire bytes -- xce_decode() will build entirely fresh copies from that + // buffer. Free them now so this test doesn't leak. + C.free(unsafe.Pointer(submit.queue)) + C.free(unsafe.Pointer(submit.user)) + C.free(unsafe.Pointer(submit.resreq)) + C.free(unsafe.Pointer(submit)) + C.free(unsafe.Pointer(jobArray)) + C.free(unsafe.Pointer(data)) + + return unsafe.Pointer(pkt), nil +} + +// wireDispatchDeliver mirrors vtask_entry()'s (xcectx.c, static, not +// exported) push-handling glue: decode -> extract_data() (task 5.3b's real +// fix) -> reassemble into Request_body_datamsg exactly as vtask_entry()'s +// MOVE_PTR does -> call the REAL go_push_handler() (clib.go), the top-level +// CGO entry point xce_backend_handler calls from C, never exercised by any +// prior apiserver test. +// +// ds is allocated via C.calloc (C heap), never a Go stack/heap value, +// because it is embedded into decoded.body.datamsg and decoded is then +// handed to C -- storing a Go pointer inside Go memory passed to C violates +// cgo's pointer-passing rules and panics under the (default-enabled) cgo +// pointer checker. +func wireDispatchDeliver(pktHandle unsafe.Pointer) (int, error) { + pkt := (*C.io_packet_t)(pktHandle) + + var decoded C.xce_request_t + if rc := C.xce_decode(pkt, &decoded); rc != 0 { + return 0, fmt.Errorf("xce_decode failed: rc=%d", int(rc)) + } + if decoded.body_which != C.Request_body_pushmsg { + return 0, fmt.Errorf("decoded body_which = %d, want Request_body_pushmsg", decoded.body_which) + } + + pushmsg := C.wire_test_request_get_pushmsg(&decoded) + if pushmsg == nil { + return 0, fmt.Errorf("decoded pushmsg is nil") + } + + ds := (*C.dataset_t)(C.calloc(1, C.size_t(unsafe.Sizeof(C.dataset_t{})))) + ds.table = pushmsg.table + + callnext := C.extract_data(ds, pushmsg) + if callnext == 0 { + C.free(unsafe.Pointer(ds)) + return 0, fmt.Errorf("extract_data() returned 0 (reassembly incomplete) -- test harness broken, not exercising the bug class") + } + + // FREEUP(decoded.body.pushmsg) equivalent: extract_data() already + // transferred/nulled the payload pointer inside pushmsg, so this is a + // safe shallow free of the now-hollow xce_data_t container -- mirrors + // vtask_entry() (xcectx.c) exactly. + C.free(unsafe.Pointer(pushmsg)) + C.wire_test_request_set_datamsg(&decoded, ds) + + // go_push_handler is defined as a plain (albeit //export-annotated) Go + // function in clib.go, in the SAME package -- called directly as a + // normal same-package Go function, not via C.go_push_handler (it has no + // extern declaration in this file's own cgo preamble; cgo symbol + // resolution is per-file, so C.go_push_handler here would fail to + // resolve even though both files are in the same package and both carry + // //go:build cgo). *C.xce_request_t from this file's preamble is the + // same underlying C type as clib.go's parameter (both derive from the + // same request.h/controller.capnp.h), so passing &decoded across the + // file boundary is type-correct. + // Capture ds.rows BEFORE calling go_push_handler: its internal + // `defer C.free_Request(req)` runs free_Request(&decoded), whose + // Request_body_datamsg case calls free_Dataset_ptr(&decoded.body.datamsg) + // -- free_Dataset_ptr() unconditionally free()s the dataset_t pointer it + // is given (confirmed by reading controller.capnp.c's generated bodies), + // even though free_Dataset() ITSELF is an empty no-op (never touches + // ->rows). So `ds` (the container) is freed automatically the instant + // go_push_handler returns -- ds is a dangling pointer after that call. + // Reading ds.rows afterward would be a use-after-free; separately + // free()-ing ds itself would be a double-free (this exact double-free, + // on the 16-byte dataset_t region, was caught by ASan -- bug-1442). + rows := ds.rows + + rc := go_push_handler(&decoded, nil) + + // go_push_handler already freed *ds (see above) -- only the payload + // xce_job_submit_t it pointed to still needs freeing: free_Dataset() + // never touches ->rows, and handleDataset()'s Data_data_jobArray case + // (clib.go) never nulls d.rows after use (every element is its own + // independent Go-native copy via cJobArrayToModels, mirroring the C + // scheduler side's sched_handle_dataset() jobArray case). Freeing it now + // is safe since cJobArrayToModels already fully copied every field into + // Go-native models.JobInfo before go_push_handler returned. + if rows != nil { + jobArrayPtr := (*C.xce_job_submit_t)(rows) + C.free_JobSubmit_ptr(&jobArrayPtr) + } + + return int(rc), nil +} diff --git a/src/frontend/apiserver/backend/wire_dispatch_test.go b/src/frontend/apiserver/backend/wire_dispatch_test.go new file mode 100644 index 0000000..838bef5 --- /dev/null +++ b/src/frontend/apiserver/backend/wire_dispatch_test.go @@ -0,0 +1,138 @@ +//go:build cgo + +package backend + +import ( + "testing" + + "apiserver/store" +) + +// wire-dispatch integration test (job-array-single-wal-record task 5.5): +// proves the chain +// +// xce_encode() (real wire bytes, same as apiserver_push_jobarray() sends) +// -> xce_decode() +// -> extract_data() (task 5.3b's fix) +// -> go_push_handler() (the REAL top-level CGO entry point xce_backend_ +// handler calls from C -- clib.go -- never +// exercised by any prior apiserver test, which +// all call dispatchJobArray()/handleDataset() +// directly with hand-built Go/C structs) +// -> handlePush() -> handleDataset() -> cJobArrayToModels() (first-ever +// coverage at the CGO-struct level -- see task 5.4's Implementation +// Notes: "cJobArrayToModels() has zero direct unit coverage at the +// CGO-struct level") +// -> dispatchJobArray() (already well covered by TestDispatchJobArray_* +// in handler_test.go -- exercised here for the first time via the +// REAL CGO/wire path rather than a hand-built []*models.JobInfo) +// +// delivers exactly N correctly-addressed records and fires task 5.4's +// confirmation log line exactly once; and that an identical byte-for-byte +// SECOND delivery (independently re-encoded/re-decoded, the wire-level +// equivalent of "the same push arrives twice", e.g. after an apiserver +// reconnect) creates zero new records and logs zero additional lines. +// +// The actual cgo plumbing (encode/decode/extract_data/go_push_handler +// calls) lives in wire_dispatch_support.go, NOT in this file -- the Go +// toolchain hard-rejects `import "C"` in any *_test.go file (golang/go#13938, +// confirmed against this repo's go toolchain: cmd/go/internal/modindex/ +// read.go's `if isTest { badGoFile(...) }` check), so this file only calls +// the plain-Go-signature wrappers wireDispatchBuildPush/wireDispatchDeliver. +// +// vtask_entry() (xcectx.c, C side) is `static` and not exported for +// testing; wireDispatchDeliver mirrors its few lines of push-handling glue +// verbatim -- every OTHER step (encode, decode, extract_data, +// go_push_handler and everything it calls) is the real, unmodified +// production function. + +func TestWireDispatch_OnePushProducesNRecordsAndLogsOnce(t *testing.T) { + resetStores() + prevHandler := gHandler + gHandler = &DefaultHandler{} + defer func() { gHandler = prevHandler }() + logs := captureLog(t) + + pkt, err := wireDispatchBuildPush(52000, 5) + if err != nil { + t.Fatalf("wireDispatchBuildPush failed: %v", err) + } + rc, err := wireDispatchDeliver(pkt) + if err != nil { + t.Fatalf("wireDispatchDeliver failed: %v", err) + } + if rc != 0 { + t.Fatalf("go_push_handler rc = %d, want 0", rc) + } + + for i := uint64(0); i < 5; i++ { + jobid := uint64(52000) + i + got, ok := store.Jobs.Get(jobid) + if !ok { + t.Errorf("jobid %d should exist in store.Jobs after one real-wire array push", jobid) + continue + } + if got.QueueName != "normal" { + t.Errorf("jobid %d: QueueName = %q, want normal", jobid, got.QueueName) + } + } + + if got := logs.Len(); got != 1 { + t.Errorf("expected exactly 1 confirmation log line, got %d", got) + } +} + +func TestWireDispatch_RedeliveryCreatesNoNewRecordsAndLogsNothing(t *testing.T) { + resetStores() + prevHandler := gHandler + gHandler = &DefaultHandler{} + defer func() { gHandler = prevHandler }() + logs := captureLog(t) + + // Two independent build+encode calls with identical inputs, so each + // delivery genuinely goes through its own encode and decode -- not one + // decode fed to the handler twice. + pkt1, err := wireDispatchBuildPush(53000, 3) + if err != nil { + t.Fatalf("wireDispatchBuildPush (1st) failed: %v", err) + } + pkt2, err := wireDispatchBuildPush(53000, 3) + if err != nil { + t.Fatalf("wireDispatchBuildPush (2nd) failed: %v", err) + } + + rc, err := wireDispatchDeliver(pkt1) + if err != nil { + t.Fatalf("first delivery failed: %v", err) + } + if rc != 0 { + t.Fatalf("first delivery rc = %d, want 0", rc) + } + for i := uint64(0); i < 3; i++ { + if _, ok := store.Jobs.Get(53000 + i); !ok { + t.Fatalf("jobid %d should exist after first delivery", 53000+i) + } + } + if got := logs.Len(); got != 1 { + t.Fatalf("expected exactly 1 confirmation log line after first delivery, got %d", got) + } + logs.TakeAll() // discard the first delivery's own confirmation line + + // Second, byte-identical delivery (simulated redelivery/reconnect): + // every index (53000..53002) already exists in store.Jobs -- must be a + // pure no-op, and must NOT log a second confirmation line. + rc, err = wireDispatchDeliver(pkt2) + if err != nil { + t.Fatalf("second (redelivered) delivery failed: %v", err) + } + if rc != 0 { + t.Fatalf("second (redelivered) delivery rc = %d, want 0", rc) + } + + if n := store.Jobs.Count(); n != 3 { + t.Errorf("store.Jobs count after redelivery = %d, want 3 (no new entries)", n) + } + if got := logs.Len(); got != 0 { + t.Errorf("expected zero confirmation log lines on redelivery, got %d", got) + } +} diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index 25ccb19..b9b94b5 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -42,6 +42,20 @@ sched_job_array_test = executable('sched_job_array_test', dependencies: [scheduler_test_deps], install: false) +# Cross-daemon integration test (job-array-single-wal-record task 5.5): the +# REAL xce_encode()/xce_decode()/extract_data() -> scheduler_req_handler() +# chain (the top-level function vtask_entry() registers as its handler), +# proving one-push/N-tasks/log-once and redelivery-safety under genuinely +# re-decoded wire bytes -- one layer more real than sched_job_array_test's +# own wire-dispatch case, which stops at sched_handle_dataset(). Kept as its +# own binary for the same reason sched_job_array_test is separate from +# sched_handlers_test: it needs a REAL sched->jobs kvmgr for the dedup check. +sched_job_array_wire_e2e_test = executable('sched_job_array_wire_e2e_test', + ['test_job_array_wire_e2e.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + # Register as meson tests so `meson test --suite scheduler` runs them all. # detect_leaks=0 matches the suite convention (fixtures are not torn down). test('scheduler test', scheduler_test, suite: 'scheduler', @@ -52,3 +66,5 @@ test('scheduler handlers test', sched_handlers_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) test('scheduler job-array expand test', sched_job_array_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler job-array wire e2e test', sched_job_array_wire_e2e_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_job_array_wire_e2e.c b/src/scheduler/testsuite/test_job_array_wire_e2e.c new file mode 100644 index 0000000..a24c9e7 --- /dev/null +++ b/src/scheduler/testsuite/test_job_array_wire_e2e.c @@ -0,0 +1,420 @@ +/* test_job_array_wire_e2e.c — job-array-single-wal-record task 5.5 + * + * Cross-daemon integration test for the Compact Scheduler/API-Server Push + * component's scheduler-side receive path. Proves the chain everything in + * tasks 5.1-5.4 was built to support, but that no single existing test + * composes end to end: + * + * xce_encode() (real wire bytes, same as xcesched_push_jobarray() sends) + * -> xce_decode() + * -> extract_data() (task 5.3b's fix) + * -> scheduler_req_handler() (the REAL top-level function vtask_entry() + * registers as ctx->handler -- xcectx.c; + * never exercised by any prior job-array + * test, which all call sched_handle_dataset() + * one layer below it) + * -> sched_handle_dataset() + * -> sched_on_job_array() (task 5.3) + * + * ...delivers exactly N correctly-addressed scheduler tasks and fires task + * 5.3's confirmation log line exactly once (Req 6.3, 6.4, 8.4); and that an + * identical byte-for-byte SECOND delivery (independently re-decoded from the + * same encoded bytes -- the wire-level equivalent of "the same push arrives + * twice", e.g. after a reconnect) creates zero new tasks and logs zero + * additional confirmation lines (Req 6.5, 8.4). + * + * Existing coverage this test does NOT duplicate: + * - test_job_array_sched_push.c (task 5.1) already proves the CONTROLLER + * encodes exactly one compact push per multi-element notification. + * - test_request.c's test_extract_data_job_array_passes_through_intact + * (task 5.3b) already proves extract_data() itself decodes the payload + * correctly via a real encode/decode round trip. + * - test_job_array_expand.c's test_wire_dispatch_chain_delivers_n_tasks + * (task 5.3b) already proves extract_data() -> sched_handle_dataset() + * compose, but stops one layer below the real registered handler and + * never exercises redelivery through the wire path or the log line. + * This test's unique contribution: the REAL top-level dispatcher + * (scheduler_req_handler(), not sched_handle_dataset() directly), the + * redelivery-under-real-wire-bytes case, and the confirmation log line's + * actual emission count. + * + * vtask_entry() (xcectx.c) is `static` and not exported for testing -- it is + * also the trigger normally driven by a live socket + iomgr epoll loop, which + * this task's status report judges disproportionate to stand up fresh for a + * single scenario (see the task 5.5 status report's TEST_STRATEGY). This + * test instead mirrors vtask_entry()'s few lines of push-handling glue + * verbatim (see deliver_via_real_dispatch_chain() below) -- every OTHER step + * (encode, decode, extract_data, the handler chain itself) is the real, + * unmodified production function, reached the same way production reaches + * it: via xce_decode()'s output, never a hand-built dataset_t. + * + * Bug-catching demonstration (this task's "RED phase"): temporarily + * reverting extract_data()'s `case Data_data_jobArray:` (task 5.3b's fix) + * back to falling through to `default: rc = -EINVAL` makes this test's first + * assertion (`callnext != 0` while dataset->rows stays NULL) fail loudly, + * and the subsequent job-count assertions fail too (zero tasks created) -- + * confirming this test has teeth against the exact class of bug 5.3b fixed. + */ +#define UNIT_TEST 1 +#include +#include +#include +#include +#include + +#include + +#include "kvmgr.h" +#include "scheduler.h" + +/* Pull the real handler-under-test into this TU, same white-box-but-real- + * symbols convention as test_job_array_expand.c (handler.c is otherwise + * self-contained; sched_init_config is its only unresolved extern). */ +#include "../src/handler.c" + +int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) +{ + (void)s; + (void)c; + return 0; +} + +/* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors + * test_job_array_expand.c's precedent), needed so sched_on_job_array()'s + * per-element dedup check (sched_get_job) is exercised for real. */ +int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) +{ + int rc; + lock(sched->jobs, RTTI_LOCK_WRITE); + rc = key_add(sched->jobs, (void*)j->jobid, j); + unlock(sched->jobs); + return rc; +} + +xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +{ + int rc; + xce_sched_job_t* job = NULL; + + lock(sched->jobs, RTTI_LOCK_READ); + rc = key_find(sched->jobs, (void*)jobid, (void**)&job); + unlock(sched->jobs); + + if (rc <= 0) { + return NULL; + } + + return job; +} + +void setUp(void) { return; } +void tearDown(void) { return; } + +static void fixture_init(xce_scheduler_t* sched) +{ + kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT, + .nslots = 59, + .delta = 16, + .datakiller = NULL }; + + memset(sched, 0, sizeof(*sched)); + mylock_init(&(sched->lock)); + sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->pend_updates[0], NULL); + sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->pend_updates[1], NULL); + sched->jobs = makeInstance(&xkvmgr_type, &jobs_params); +} + +/* Pop one pending update request, asserting it is an UPDATE_JOB op; hands + * back njobs/jobs (caller now owns them -- free via free_Job_ptr + free() to + * mirror sched_pend_update_killer). Returns -1 if the queue is empty. Mirrors + * test_job_array_expand.c's identical helper. */ +static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out) +{ + data_item_t* it = dataq_deq(sched->pend_updates[0], NULL); + xce_sched_update_request_t* req; + int njobs; + + if (it == NULL) { + return -1; + } + + req = ITPTR_TO_UPDATE_REQ(it); + TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_JOB, req->op); + + njobs = req->u.jobupdate.njobs; + if (jobs_out != NULL) { + *jobs_out = req->u.jobupdate.jobs; + } + free(req); + return njobs; +} + +/* Materializes a popped job array into sched->jobs (mirroring what the + * async consumer, scheduler.c's sched_update_jobs -> sched_add_job, not + * linked into this white-box TU, would do) -- so a SECOND, redelivered + * push's dedup check sees the first delivery's elements as already + * existing. Frees each popped xce_job_t and the array itself, same + * simulated-reconnect technique test_job_array_expand.c's + * test_job_array_redelivery_does_not_duplicate uses (there via + * seed_existing_job, here folded into one step since njobs is already + * known). */ +static void materialize_popped_jobs(xce_scheduler_t* sched, xce_job_t** jobs, + int njobs) +{ + int i; + + for (i = 0; i < njobs; i++) { + xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + job->jobid = jobs[i]->jobid; + sched_add_job(sched, job); + free_Job_ptr(&jobs[i]); + } + free(jobs); +} + +/* ---------------------------------------------------------------------- */ +/* Early-print log capture: sched_on_job_array()'s confirmation line uses */ +/* INFO() (tlog_info -> TLOG_INFO). No test binary in this codebase calls */ +/* tlog_init(), so every INFO() call routes through tlog's pre-init */ +/* "early print" fallback (tlog.c's _tlog_early_print(), confirmed via the */ +/* vendored tlog source). tlog_reg_early_printf_callback() lets a test */ +/* intercept that fallback synchronously, on the calling thread -- no */ +/* tlog_init()/tlog_exit() lifecycle, so it cannot disturb any other test */ +/* sharing this process. va_copy() is used so the caller's own va_list is */ +/* never consumed, in case tlog's own early-print path also reads it. */ +/* ---------------------------------------------------------------------- */ + +static int g_confirm_log_count = 0; + +static void wire_e2e_early_print_cb(struct tlog_loginfo* loginfo, + const char* format, va_list ap) +{ + char buf[512]; + va_list ap_copy; + + if (loginfo == NULL || loginfo->level != TLOG_INFO) { + return; + } + + va_copy(ap_copy, ap); + vsnprintf(buf, sizeof(buf), format, ap_copy); + va_end(ap_copy); + + if (strstr(buf, "job array push received and expanded") != NULL) { + g_confirm_log_count++; + } +} + +static void start_log_capture(void) +{ + g_confirm_log_count = 0; + tlog_set_early_printf(1, 1, 0); + tlog_reg_early_printf_callback(wire_e2e_early_print_cb); +} + +static void stop_log_capture(void) { tlog_reg_early_printf_callback(NULL); } + +/* ---------------------------------------------------------------------- */ +/* Real wire construction: the exact message xcesched_push_jobarray() */ +/* (sched.c) sends -- OPCode_push / Request_body_pushmsg carrying one */ +/* Data.jobArray descriptor. */ +/* ---------------------------------------------------------------------- */ + +static void build_encoded_push(io_packet_t** out_pkt, uint64_t base_jobid, + uint64_t array_size, + job_submit_data_t* submit_storage) +{ + xce_job_submit_t job_array; + xce_data_t data; + xce_request_t request; + io_packet_t* pkt; + + memset(submit_storage, 0, sizeof(*submit_storage)); + submit_storage->queue = "normal"; + submit_storage->user = "tester"; + submit_storage->mintasks = 1; + submit_storage->maxtasks = 1; + submit_storage->resreq = "type == local"; + + memset(&job_array, 0, sizeof(job_array)); + job_array.jobid = base_jobid; + job_array.array_size = array_size; + job_array.submit = submit_storage; + + memset(&data, 0, sizeof(data)); + data.table = Data_data_jobArray; + data.data.job_array = &job_array; + + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_push; + request.body_which = Request_body_pushmsg; + request.body.pushmsg = &data; + + pkt = (io_packet_t*)calloc(1, sizeof(*pkt)); + TEST_ASSERT_NOT_NULL(pkt); + TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt)); + + *out_pkt = pkt; +} + +/* Mirrors vtask_entry()'s (xcectx.c, static) push-handling glue exactly: + * 1. xce_decode() the packet -- fresh, independent decode every call, so + * a "second delivery" is genuinely re-derived from wire bytes, not a + * reused in-memory struct. + * 2. Allocate a dataset_t, call the REAL extract_data() (task 5.3b). + * 3. On success (vtask_entry(): `if (callnext)`, true for ANY nonzero + * return, negative included -- this is the exact amplifier that let + * 5.3b's bug through silently), free the now-hollow Data container via + * the same FREEUP() production macro vtask_entry() uses, then install + * the reassembled dataset into body.datamsg/body_which exactly as + * vtask_entry()'s MOVE_PTR does. + * 4. Call the REAL top-level scheduler_req_handler() -- never reached by + * any prior job-array test. + * Returns scheduler_req_handler()'s rc; *out_ds receives the dataset (caller + * owns dataset->rows cleanup via free_JobSubmit_ptr, matching + * sched_handle_dataset()'s jobArray case, which deliberately never nulls + * dataset->rows -- see that function's own comment). */ +static int deliver_via_real_dispatch_chain(io_packet_t* pkt, + xce_scheduler_t* sched, + dataset_t** out_ds) +{ + xce_request_t decoded; + dataset_t* ds; + int callnext; + int rc; + + memset(&decoded, 0, sizeof(decoded)); + TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &decoded)); + TEST_ASSERT_EQUAL_INT(Request_body_pushmsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.pushmsg); + + ds = (dataset_t*)calloc(1, sizeof(*ds)); + TEST_ASSERT_NOT_NULL(ds); + ds->table = decoded.body.pushmsg->table; + + callnext = extract_data(ds, decoded.body.pushmsg); + TEST_ASSERT_TRUE_MESSAGE( + callnext != 0, + "extract_data() must return nonzero (vtask_entry()'s `if (callnext)` " + "treats ANY nonzero -- including a negative errno -- as reassembly " + "complete); a zero return here would mean this test's own harness is " + "broken, not exercising the bug class"); + + FREEUP(decoded.body.pushmsg); + decoded.body_which = Request_body_datamsg; + decoded.body.datamsg = ds; + + rc = scheduler_req_handler(&decoded, sched); + + decoded.body.datamsg = NULL; /* caller owns *ds; avoid double free below */ + xce_free_request(&decoded); + + *out_ds = ds; + return rc; +} + +/* ── Test 1: one real wire push -> N correctly-addressed tasks, log fires */ +/* exactly once (Req 6.3, 6.4, 8.4) */ + +static void test_one_push_produces_n_tasks_and_logs_once(void) +{ + xce_scheduler_t sched; + job_submit_data_t submit_storage; + io_packet_t* pkt = NULL; + dataset_t* ds = NULL; + xce_job_t** jobs = NULL; + int njobs; + int i; + + fixture_init(&sched); + start_log_capture(); + + build_encoded_push(&pkt, 42000, 6, &submit_storage); + + TEST_ASSERT_TRUE( + deliver_via_real_dispatch_chain(pkt, &sched, &ds) >= 0); + + njobs = pop_job_update(&sched, &jobs); + TEST_ASSERT_EQUAL_INT(6, njobs); + for (i = 0; i < njobs; i++) { + TEST_ASSERT_EQUAL_UINT64(42000 + (uint64_t)i, jobs[i]->jobid); + TEST_ASSERT_EQUAL_STRING("normal", jobs[i]->submit->queue); + free_Job_ptr(&jobs[i]); + } + free(jobs); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, g_confirm_log_count, + "task 5.3's confirmation log line must fire exactly once for one " + "successful array expansion"); + + stop_log_capture(); + free_JobSubmit_ptr((xce_job_submit_t**)&ds->rows); + free(ds); + printf("test_one_push_produces_n_tasks_and_logs_once: PASS\n"); +} + +/* ── Test 2: redelivery of the IDENTICAL encoded bytes (independently */ +/* re-decoded, simulating a reconnect replay) creates zero new tasks and */ +/* zero additional log lines (Req 6.5, 8.4) */ + +static void test_redelivery_creates_no_new_tasks_and_logs_nothing(void) +{ + xce_scheduler_t sched; + job_submit_data_t submit_storage; + io_packet_t* pkt1 = NULL; + io_packet_t* pkt2 = NULL; + dataset_t* ds1 = NULL; + dataset_t* ds2 = NULL; + xce_job_t** jobs = NULL; + int njobs; + + fixture_init(&sched); + + /* Build the wire bytes twice (two independent xce_encode() calls with + * identical inputs) so each delivery genuinely goes through its own + * encode and decode -- not one decode fed to the handler twice. */ + build_encoded_push(&pkt1, 43000, 4, &submit_storage); + build_encoded_push(&pkt2, 43000, 4, &submit_storage); + + start_log_capture(); + + /* First delivery: 4 new tasks. */ + TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt1, &sched, &ds1) >= 0); + njobs = pop_job_update(&sched, &jobs); + TEST_ASSERT_EQUAL_INT(4, njobs); + materialize_popped_jobs(&sched, jobs, njobs); + TEST_ASSERT_EQUAL_INT(1, g_confirm_log_count); + free_JobSubmit_ptr((xce_job_submit_t**)&ds1->rows); + free(ds1); + + /* Second, byte-identical delivery (simulated redelivery/reconnect): + * every index (43000..43003) already exists in sched->jobs -- must be a + * pure no-op, and must NOT log a second confirmation line. */ + TEST_ASSERT_EQUAL_INT( + 0, deliver_via_real_dispatch_chain(pkt2, &sched, &ds2)); + TEST_ASSERT_EQUAL_INT_MESSAGE( + -1, pop_job_update(&sched, &jobs), + "redelivery must enqueue nothing new"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, g_confirm_log_count, + "redelivery must not add a second confirmation log line -- still " + "exactly 1 total, from the first delivery only"); + + stop_log_capture(); + free_JobSubmit_ptr((xce_job_submit_t**)&ds2->rows); + free(ds2); + printf("test_redelivery_creates_no_new_tasks_and_logs_nothing: PASS\n"); +} + +int main(int argc, char* argv[]) +{ + (void)argc; + (void)argv; + UNITY_BEGIN(); + RUN_TEST(test_one_push_produces_n_tasks_and_logs_once, 0); + RUN_TEST(test_redelivery_creates_no_new_tasks_and_logs_nothing, 0); + return UNITY_END(); +}