diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index 08332347..be843e68 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -552,6 +552,21 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int { } gHandler.OnJobFinished(uint64(ff.jobid), int(ff.code)) + case C.NotifyData_data_jobPidInfo: + // The controller's start-report push (apiserver_job_start_fun, fired + // from HOOK_DETAIL_JOB_START right after WAL_JOB_START commits). + // Carries the worker-reported pids/pgids and exdata — whose + // {"url": ...} JSON is the interactive session URL. Without this + // case the push was silently dropped here and interactive_job_url + // stayed invisible until the next periodic rusage-driven job UPDATE + // re-delivered the full row — the dominant chunk of xcecli's + // submit→first-output latency for -I jobs (measured ~6-10s on t1). + pi := (*C.job_pidinfo_t)(C.xce_notify_data_ptr(nd)) + if pi == nil { + return 0 + } + gHandler.OnJobPidInfo(uint64(pi.jobid), cPidInfoToRaw(pi)) + case C.NotifyData_data_schedUpdateResvals: rvl := (*C.xce_resval_list_t)(C.xce_notify_data_ptr(nd)) if rvl == nil { diff --git a/src/frontend/apiserver/backend/convert.go b/src/frontend/apiserver/backend/convert.go index 9bff7ea0..b571e57f 100644 --- a/src/frontend/apiserver/backend/convert.go +++ b/src/frontend/apiserver/backend/convert.go @@ -93,6 +93,27 @@ func cJobArrayToModels(jobArray *C.xce_job_submit_t) []*models.JobInfo { return rawJobArrayToModels(uint64(jobArray.jobid), uint64(jobArray.array_size), rd) } +// cPidInfoToRaw decodes a C job_pidinfo_t — the worker's start report +// (pids/pgids plus exdata, whose {"url": ...} JSON is the interactive +// session URL for ttyd/XJOB jobs) — into the rawPidInfo the pure layer +// consumes. Shared by cJobToModel (the full xce_job_t conversion, where the +// capnp Job wire carries pidinfo at Job.pidinfo@9) and the standalone +// jobPidInfo notify path (NotifyData_data_jobPidInfo), which delivers the +// same struct the moment WAL_JOB_START commits. +func cPidInfoToRaw(pi *C.job_pidinfo_t) *rawPidInfo { + if pi == nil { + return nil + } + p := &rawPidInfo{exdata: goString(pi.exdata)} + for i := 0; i < int(pi.npids); i++ { + p.pids = append(p.pids, int(C.pidinfo_pid(pi, C.int(i)))) + } + for i := 0; i < int(pi.npgids); i++ { + p.pgids = append(p.pgids, int(C.pidinfo_pgid(pi, C.int(i)))) + } + return p +} + // cJobToModel extracts C xce_job_t fields into a rawJobData and maps it to JobInfo. func cJobToModel(job *C.xce_job_t) *models.JobInfo { r := rawJobData{ @@ -207,17 +228,7 @@ func cJobToModel(job *C.xce_job_t) *models.JobInfo { // capnp Job wire carries pidinfo (Job.pidinfo@9 / JobPIDInfo.exdata@3) // but the conversion dropped it, so interactive_job_url never reached // the API and xcecli's browser flow could never fire. - if job.pidinfo != nil { - pi := job.pidinfo - p := &rawPidInfo{exdata: goString(pi.exdata)} - for i := 0; i < int(pi.npids); i++ { - p.pids = append(p.pids, int(C.pidinfo_pid(pi, C.int(i)))) - } - for i := 0; i < int(pi.npgids); i++ { - p.pgids = append(p.pgids, int(C.pidinfo_pgid(pi, C.int(i)))) - } - r.pidInfo = p - } + r.pidInfo = cPidInfoToRaw(job.pidinfo) return rawJobToModel(r) } diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index c3d159be..8eb9469f 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -606,6 +606,23 @@ func interactiveURLFromExdata(exdata string) string { return obj.URL } +// pidInfoJSONB marshals a start report's pids/pgids into the PIDInfo Jsonb +// the API exposes (processJobData parses the JSONB string before the row +// leaves the API, so clients see an object with pids/pgids). Shared by +// rawJobToModel (the xce_job_t conversion, which carries the full +// start report) and OnJobPidInfo (the standalone jobPidInfo notify, whose +// only extra payload is the interactive URL). +func pidInfoJSONB(pids, pgids []int) (models.Jsonb, bool) { + data, err := json.Marshal(struct { + Pids []int `json:"pids,omitempty"` + Pgids []int `json:"pgids,omitempty"` + }{Pids: pids, Pgids: pgids}) + if err != nil { + return models.Jsonb{}, false + } + return models.Jsonb{String: string(data)}, true +} + // rawJobToModel maps a rawJobData to a models.JobInfo. // This pure-Go function contains the field-mapping logic and is tested without CGO. func rawJobToModel(r rawJobData) *models.JobInfo { @@ -714,11 +731,8 @@ func rawJobToModel(r rawJobData) *models.JobInfo { // with pids/pgids. The interactive URL is promoted to its own // top-level field — it is what xcecli's waitForInteractiveURL and // the --browse flow consume. - if data, err := json.Marshal(struct { - Pids []int `json:"pids,omitempty"` - Pgids []int `json:"pgids,omitempty"` - }{Pids: p.pids, Pgids: p.pgids}); err == nil { - info.PIDInfo = models.Jsonb{String: string(data)} + if jb, ok := pidInfoJSONB(p.pids, p.pgids); ok { + info.PIDInfo = jb } info.InteractiveJobURL = interactiveURLFromExdata(p.exdata) } diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index d4df0ff8..b72073b2 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -115,6 +115,44 @@ func (h *DefaultHandler) OnJobFinished(jobid uint64, exitCode int) { zap.Int("job_status", existing.JobStatus)) } +// OnJobPidInfo applies a controller-pushed job start report +// (NotifyData_data_jobPidInfo, fired from HOOK_DETAIL_JOB_START after +// WAL_JOB_START commits) onto the already-stored job record: the worker's +// pids/pgids become PIDInfo, and an interactive job's exdata {"url": ...} +// becomes InteractiveJobURL — the field xcecli's waitForInteractiveURL +// polls. Without this handler the push was silently dropped by +// handleNotifyData and the URL stayed invisible until the next periodic +// rusage-driven job UPDATE re-delivered the full row (up to 10s later, +// job_rusage_collect_interval). +// +// Mirrors OnJobFinished's get-mutate-put-publish shape: the full job row +// (submit spec, state, timestamps) always arrives ahead of or with the +// start report — SUBMIT via OnJobCreated/join push, RUN state itself via +// the dispatch-time update — so a jobid not in the store is logged and +// dropped rather than inserted half-empty. +func (h *DefaultHandler) OnJobPidInfo(jobid uint64, pidInfo *rawPidInfo) { + existing, ok := store.Jobs.Get(jobid) + if !ok { + logger.Info("OnJobPidInfo: jobid not found in store, ignoring", + zap.Uint64("jobid", jobid)) + return + } + if pidInfo != nil { + if jb, ok := pidInfoJSONB(pidInfo.pids, pidInfo.pgids); ok { + existing.PIDInfo = jb + } + existing.InteractiveJobURL = interactiveURLFromExdata(pidInfo.exdata) + } + store.Jobs.Put(jobid, existing) + DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: jobid, + Status: models.JobStateStr(existing.JobStatus), Job: &existing}) + + logger.Info("OnJobPidInfo: start report applied", + zap.Uint64("jobid", jobid), + zap.Int("pids", len(pidInfo.pids)), + zap.String("interactive_job_url", existing.InteractiveJobURL)) +} + func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) { store.Queues.Put(queue.QueueName, *queue) } diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go index ab81f131..ba6651e0 100644 --- a/src/frontend/apiserver/backend/handler_test.go +++ b/src/frontend/apiserver/backend/handler_test.go @@ -4,6 +4,7 @@ import ( "apiserver/models" "apiserver/store" "apiserver/utils" + "encoding/json" "testing" "go.uber.org/zap" @@ -60,6 +61,72 @@ func TestHandlerPublishesJobEventsToBroker(t *testing.T) { } } +// OnJobPidInfo: the controller's start-report push must land PIDInfo and the +// interactive URL on the already-stored job record the moment it arrives — +// not wait for the next periodic rusage-driven UPDATE to re-deliver the full +// row. Interactive-URL latency for -I jobs is exactly this gap. +func TestOnJobPidInfo_AppliesStartReportToStoredJob(t *testing.T) { + resetStores() + b := freshBroker(t) + captureLog(t) + ch := b.Subscribe(16) + defer b.Unsubscribe(ch) + + h := DefaultHandler{} + h.OnJobUpdated(&models.JobInfo{Unique_JobID: 7, JobStatus: models.JOBSTAT["RUN"]}) + + h.OnJobPidInfo(7, &rawPidInfo{ + pids: []int{4321}, + pgids: []int{4320}, + exdata: `{"url": "http://10.0.1.5:7681"}`, + }) + + got, ok := store.Jobs.Get(7) + if !ok { + t.Fatal("job 7 should still exist after the start report") + } + if want := "http://10.0.1.5:7681"; got.InteractiveJobURL != want { + t.Errorf("InteractiveJobURL: got %q, want %q", got.InteractiveJobURL, want) + } + var decoded struct { + Pids []int `json:"pids"` + Pgids []int `json:"pgids"` + } + if err := json.Unmarshal([]byte(got.PIDInfo.String), &decoded); err != nil { + t.Fatalf("PIDInfo should be valid JSON, got err: %v", err) + } + if len(decoded.Pids) != 1 || decoded.Pids[0] != 4321 { + t.Errorf("PIDInfo.pids: got %v, want [4321]", decoded.Pids) + } + + ev := <-ch + if ev.Type != EventJobUpdated || ev.JobID != 7 || ev.Job == nil { + t.Fatalf("event = %+v, want job.updated 7 with the updated job", ev) + } +} + +// A start report for an unknown jobid (store pruned the record, or the push +// raced ahead of the job-list push) must be dropped, not inserted half-empty. +func TestOnJobPidInfo_UnknownJobidIgnored(t *testing.T) { + resetStores() + b := freshBroker(t) + captureLog(t) + ch := b.Subscribe(16) + defer b.Unsubscribe(ch) + + h := DefaultHandler{} + h.OnJobPidInfo(999, &rawPidInfo{exdata: `{"url": "http://10.0.1.5:7681"}`}) + + if _, ok := store.Jobs.Get(999); ok { + t.Error("OnJobPidInfo must not create a job entry; the full-row push owns insertion") + } + select { + case ev := <-ch: + t.Fatalf("unexpected event published: %+v", ev) + default: + } +} + // 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 { diff --git a/src/frontend/apiserver/backend/notification.go b/src/frontend/apiserver/backend/notification.go index dc82dc84..6f48232e 100644 --- a/src/frontend/apiserver/backend/notification.go +++ b/src/frontend/apiserver/backend/notification.go @@ -12,6 +12,13 @@ type NotificationHandler interface { // OnJobFinished is called when the controller reports job completion or // dispatch failure. exitCode==0 → DONE, non-zero → EXIT. OnJobFinished(jobid uint64, exitCode int) + // OnJobPidInfo is called when the controller pushes a job's start report + // (worker-reported pids/pgids plus exdata, whose {"url": ...} JSON is the + // interactive session URL). The controller fires this the moment + // WAL_JOB_START commits — seconds before the next periodic job UPDATE + // would carry the same fields — so handling it is what makes + // interactive_job_url pollable immediately. + OnJobPidInfo(jobid uint64, pidInfo *rawPidInfo) OnQueueUpdated(queue *models.QueueInfo) OnHostUpdated(host *models.HostInfo) OnHostConfigUpdated(host *models.HostInfo)