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) <noreply@anthropic.com>

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

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

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)

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

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
 }

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

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