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).

src/frontend/apiserver/backend/wire_dispatch_support.go

@@ -0,0 +1,193 @@
+//go:build cgo
+
+package backend
+
+/*
+#include <stdlib.h>
+#include <stdint.h>
+#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
+}

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)
+	}
+}

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'])

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 <errno.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#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();
+}