From 141ac22c5451f7009018ca6098b44bfe589ec05a Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 30 Sep 2026 04:23:10 +0000 Subject: [PATCH] fix(apiserver): an accepted job is in the job store at once (bug-1894) The apiserver's job store learned of a new job only from the controller's asynchronous OnJobCreated / array push, so a 'jobs kill' (or query) straight after 'jobs submit' could answer 'job not found' for the job just accepted (2 of 4 immediate kills). services.SubmitJob now records the accepted job right after the controller's reply (which comes only after the job is committed and applied): one record per element, base+i for arrays, stored with the new DataStore.PutIfAbsent and marked JobInfo.Provisional (json:"-", not a column). An OnJobCreated that already arrived is never overwritten; the push that arrives later replaces the provisional record -- including in dispatchJobArray, whose duplicate check no longer treats a provisional element as already known. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/frontend/apiserver/backend/accepted.go | 43 +++++++++ .../apiserver/backend/accepted_test.go | 90 +++++++++++++++++++ src/frontend/apiserver/backend/handler.go | 3 +- src/frontend/apiserver/models/jobs.go | 5 ++ .../apiserver/services/job_service.go | 9 +- src/frontend/apiserver/store/store.go | 13 +++ src/frontend/apiserver/store/store_test.go | 16 ++++ 7 files changed, 177 insertions(+), 2 deletions(-) create mode 100644 src/frontend/apiserver/backend/accepted.go create mode 100644 src/frontend/apiserver/backend/accepted_test.go diff --git a/src/frontend/apiserver/backend/accepted.go b/src/frontend/apiserver/backend/accepted.go new file mode 100644 index 00000000..a17fa282 --- /dev/null +++ b/src/frontend/apiserver/backend/accepted.go @@ -0,0 +1,43 @@ +package backend + +import ( + "time" + + "apiserver/models" + "apiserver/store" +) + +// RecordAcceptedJob stores a job the controller has just accepted, so the +// apiserver can find it at once (bug-1894). Otherwise the job reached the +// store only with the controller's asynchronous OnJobCreated / array push, +// and a `jobs kill` straight after `jobs submit` answered "job not found". +// +// The controller replies to a submit only after the job is committed and +// applied, so the job it names does exist. Each element (base+i for an array, +// as the controller expands it) is stored with PutIfAbsent and marked +// Provisional: a push that has already arrived is never overwritten, and the +// push that arrives later replaces the provisional record. +func RecordAcceptedJob(job *models.JobInfo, arraySize int) { + if job == nil || job.Unique_JobID == 0 { + return + } + if arraySize < 1 { + arraySize = 1 + } + + now := time.Now().Unix() + base := uint64(job.Unique_JobID) + for i := 0; i < arraySize; i++ { + id := base + uint64(i) + rec := *job + rec.Provisional = true + rec.Unique_JobID = int64(id) + rec.JobID = int64(ExtractArrayBaseID(id)) + rec.ArrayIdx = ExtractArrayIdx(id) + rec.JobStatus = models.JOBSTAT["PEND"] + if rec.SubmitTime == 0 { + rec.SubmitTime = now + } + store.Jobs.PutIfAbsent(id, rec) + } +} diff --git a/src/frontend/apiserver/backend/accepted_test.go b/src/frontend/apiserver/backend/accepted_test.go new file mode 100644 index 00000000..02f89d44 --- /dev/null +++ b/src/frontend/apiserver/backend/accepted_test.go @@ -0,0 +1,90 @@ +package backend + +import ( + "testing" + + "apiserver/models" + "apiserver/store" +) + +// bug-1894: a job the controller has accepted must be visible to the job +// store right away, not only once the asynchronous OnJobCreated push lands -- +// `jobs kill` issued straight after `jobs submit` answered "job not found". + +func TestRecordAcceptedJob_VisibleImmediately(t *testing.T) { + resetStores() + job := models.JobInfo{Unique_JobID: 500, JobID: 500, SubmitUser: "alice", + QueueName: "normal", JobStatus: models.JOBSTAT["PEND"]} + + RecordAcceptedJob(&job, 1) + + got, ok := store.Jobs.Get(500) + if !ok { + t.Fatal("bug-1894: accepted job not in store.Jobs before its OnJobCreated push") + } + if got.SubmitUser != "alice" || got.JobStatus != models.JOBSTAT["PEND"] { + t.Errorf("provisional record: user %q status %d, want alice/PEND", got.SubmitUser, got.JobStatus) + } + if got.SubmitTime == 0 { + t.Error("provisional record should carry a submit time") + } + if !got.Provisional { + t.Error("provisional record should be marked Provisional") + } +} + +func TestRecordAcceptedJob_DoesNotOverwriteBackendRecord(t *testing.T) { + resetStores() + h := &DefaultHandler{} + h.OnJobCreated(&models.JobInfo{Unique_JobID: 501, JobID: 501, FromHost: "ctl", JobStatus: models.JOBSTAT["RUN"]}) + + RecordAcceptedJob(&models.JobInfo{Unique_JobID: 501, JobID: 501, JobStatus: models.JOBSTAT["PEND"]}, 1) + + got, _ := store.Jobs.Get(501) + if got.Provisional || got.FromHost != "ctl" || got.JobStatus != models.JOBSTAT["RUN"] { + t.Errorf("backend record overwritten by the provisional one: %+v", got) + } +} + +func TestRecordAcceptedJob_BackendPushReplacesProvisional(t *testing.T) { + resetStores() + h := &DefaultHandler{} + RecordAcceptedJob(&models.JobInfo{Unique_JobID: 502, JobID: 502}, 1) + + h.OnJobCreated(&models.JobInfo{Unique_JobID: 502, JobID: 502, FromHost: "ctl"}) + + got, _ := store.Jobs.Get(502) + if got.Provisional || got.FromHost != "ctl" { + t.Errorf("OnJobCreated did not replace the provisional record: %+v", got) + } +} + +// Array elements are base+i, as the controller's array push expands them; the +// push must replace provisional elements, not skip them as duplicates. +func TestRecordAcceptedJob_ArrayElementsReplacedByPush(t *testing.T) { + resetStores() + captureLog(t) + h := &DefaultHandler{} + RecordAcceptedJob(&models.JobInfo{Unique_JobID: 600, JobID: 600, QueueName: "normal"}, 3) + + for _, id := range []uint64{600, 601, 602} { + if got, ok := store.Jobs.Get(id); !ok || !got.Provisional { + t.Fatalf("element %d: provisional record missing (ok=%v)", id, ok) + } + } + + elements := []*models.JobInfo{ + {Unique_JobID: 600, JobID: 600, ArrayIdx: 0, FromHost: "ctl"}, + {Unique_JobID: 601, JobID: 600, ArrayIdx: 1, FromHost: "ctl"}, + {Unique_JobID: 602, JobID: 600, ArrayIdx: 2, FromHost: "ctl"}, + } + if n := dispatchJobArray(h, elements, 600, 3); n != 3 { + t.Errorf("array push stored %d elements, want 3 (provisional ones must not count as duplicates)", n) + } + for _, id := range []uint64{600, 601, 602} { + got, _ := store.Jobs.Get(id) + if got.Provisional || got.FromHost != "ctl" { + t.Errorf("element %d still provisional after the array push: %+v", id, got) + } + } +} diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index 0fcec604..935f153f 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -260,7 +260,8 @@ func dispatchJobArray(handler NotificationHandler, elements []*models.JobInfo, b nNew := 0 for _, job := range elements { jobid := uint64(job.Unique_JobID) - if store.Jobs.Has(jobid) { + // A provisional record (bug-1894) is not a duplicate: replace it. + if existing, ok := store.Jobs.Get(jobid); ok && !existing.Provisional { continue } handler.OnJobUpdated(job) diff --git a/src/frontend/apiserver/models/jobs.go b/src/frontend/apiserver/models/jobs.go index a2e3f902..0ffdd63b 100644 --- a/src/frontend/apiserver/models/jobs.go +++ b/src/frontend/apiserver/models/jobs.go @@ -484,6 +484,11 @@ var JobFieldMappings = map[string]string{ } type JobInfo struct { + // Provisional marks a record the apiserver stored itself when the + // controller accepted the submit, before the controller's own + // OnJobCreated/array push arrived (bug-1894); that push replaces it. + // Not a column: json "-" keeps it out of query results. + Provisional bool `json:"-"` Unique_JobID int64 `json:"unique_job_id"` JobID int64 `json:"job_id"` ArrayIdx int `json:"array_idx"` //new diff --git a/src/frontend/apiserver/services/job_service.go b/src/frontend/apiserver/services/job_service.go index c853f2cc..d5ead624 100644 --- a/src/frontend/apiserver/services/job_service.go +++ b/src/frontend/apiserver/services/job_service.go @@ -6,6 +6,13 @@ import ( ) // SubmitJob sends a job to the backend. arraySize=1 for normal, >1 for array. +// On success the accepted job is recorded in the job store right away +// (bug-1894), so it can be queried and controlled before the controller's +// asynchronous OnJobCreated push arrives. func SubmitJob(job *models.JobInfo, arraySize int) error { - return backend.SubmitJob(job, arraySize) + if err := backend.SubmitJob(job, arraySize); err != nil { + return err + } + backend.RecordAcceptedJob(job, arraySize) + return nil } diff --git a/src/frontend/apiserver/store/store.go b/src/frontend/apiserver/store/store.go index 3930eb4c..d8577d46 100644 --- a/src/frontend/apiserver/store/store.go +++ b/src/frontend/apiserver/store/store.go @@ -34,6 +34,19 @@ func (s *DataStore[K, V]) Put(key K, val V) { s.data[key] = val } +// PutIfAbsent stores val under key only if key is missing, atomically, and +// reports whether it did (bug-1894: a provisional job record must never +// overwrite one the backend has already pushed). +func (s *DataStore[K, V]) PutIfAbsent(key K, val V) bool { + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.data[key]; ok { + return false + } + s.data[key] = val + return true +} + func (s *DataStore[K, V]) Delete(key K) bool { s.mu.Lock() defer s.mu.Unlock() diff --git a/src/frontend/apiserver/store/store_test.go b/src/frontend/apiserver/store/store_test.go index d9052626..a59d73f8 100644 --- a/src/frontend/apiserver/store/store_test.go +++ b/src/frontend/apiserver/store/store_test.go @@ -135,3 +135,19 @@ func TestDataStore_ConcurrentAccess(t *testing.T) { } wg.Wait() } + +// bug-1894: PutIfAbsent inserts only when the key is missing, atomically, so a +// provisional record never overwrites one the backend already pushed. +func TestDataStore_PutIfAbsent(t *testing.T) { + s := NewDataStore[string, int]() + + if !s.PutIfAbsent("a", 1) { + t.Fatal("PutIfAbsent on a missing key should insert and return true") + } + if s.PutIfAbsent("a", 2) { + t.Error("PutIfAbsent on an existing key should return false") + } + if v, _ := s.Get("a"); v != 1 { + t.Errorf("existing value overwritten: got %d, want 1", v) + } +}