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).
This commit is contained in:
Rongsong Shen 2026-07-15 15:56:48 +00:00
parent 1cbd7df1a8
commit 9e0d87fe50
7 changed files with 345 additions and 9 deletions

View file

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