From 6ce7bdbe11f43f6fc90a07e659010c1102b8743f Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Thu, 15 Oct 2026 11:21:02 +0000 Subject: [PATCH] xcecli: backoff poll cadence and suspend tolerance for -I URL wait MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit waitForInteractiveURL polled PEND every 5s, adding ~4.4s of dead time to every interactive submit (the job typically reaches RUN with its session URL ~0.6s after submit once the apiserver applies the controller's jobPidInfo push), and the RUN-but-no-URL-yet round waited another 2s. Front-load fast 500ms polls (10 rounds) so a normally dispatched job attaches within ~1s, then back off to 2s so a job stuck in a long queue costs the server a query every couple of seconds instead of a permanent 2 qps; the RUN race polls 500ms too. checkJobStatus also failed the wait outright on any state other than PEND/RUN: treat PSUSP like PEND (still queued), and USUSP/SSUSP/ TRANSIENT as keep-waiting states — the allocation is held, a resume (or failover settle) may follow, and tearing down the attach on a transient state is wrong. Cluster-verified on t1: submit-to-attach-attempt 0.56-1.08s across three runs (previously ~6s even after the jobPidInfo fix), full submit/run/finish lifecycle clean, daemons healthy. Co-Authored-By: Claude Code --- src/frontend/xcecli/cmd/jobs/submit.go | 57 ++++++- src/frontend/xcecli/cmd/jobs/submit_test.go | 161 ++++++++++++++++++++ 2 files changed, 211 insertions(+), 7 deletions(-) diff --git a/src/frontend/xcecli/cmd/jobs/submit.go b/src/frontend/xcecli/cmd/jobs/submit.go index 46177111..a1577585 100755 --- a/src/frontend/xcecli/cmd/jobs/submit.go +++ b/src/frontend/xcecli/cmd/jobs/submit.go @@ -576,7 +576,7 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin } for _, row := range resp.Data.Rows { switch row.JobStatus { - case "PEND": + case "PEND", "PSUSP": return row.JobStatus, nil case "RUN": // Print the dispatch line once: waitForInteractiveURL keeps @@ -590,6 +590,17 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin } } return row.JobStatus, nil + case "USUSP", "SSUSP": + // A suspended job keeps its allocation and can be resumed + // (`xce jobs resume`), so the session may still come up -- + // wait instead of failing the attach the way a terminal + // state (DONE/EXIT) does. + return row.JobStatus, nil + case "TRANSIENT": + // State uncertain after a controller failover; the job may + // settle back to RUN. Keep waiting (the 24h cap still bounds + // this) rather than tearing down the attach on a hiccup. + return row.JobStatus, nil default: return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, row.JobStatus) } @@ -622,26 +633,57 @@ func (js *JobSubmitter) checkExtendData(apiClient *request.APIClient, token stri return "", nil } +// Interactive-wait poll cadence. A dispatched job produces its session URL +// well under a second after submit (scheduler round ~0.5s, start report +// right behind it), so pendingInterval fronts fast polls to make attach +// snappy; pendingBackoffInterval takes over after pendingFastPolls so a job +// stuck in a long queue (closed window, no free slots) costs the server one +// query every couple of seconds instead of a permanent 2 qps. runInterval +// covers the RUN-but-no-URL-yet race, which the start-report push (applied +// by the apiserver's OnJobPidInfo) normally closes in well under a second. +const ( + pendingFastPolls = 10 + pendingFastInterval = 500 * time.Millisecond + pendingBackoffInterval = 2 * time.Second + runInterval = 500 * time.Millisecond +) + func (js *JobSubmitter) waitForInteractiveURL(apiClient *request.APIClient, token string, jobID string) (string, error) { timeout := time.After(24 * time.Hour) timer := time.NewTimer(0) defer timer.Stop() + polls := 0 for { select { case <-timeout: return "", fmt.Errorf("timeout waiting for interactive URL") case <-timer.C: + polls++ status, err := js.checkJobStatus(apiClient, token, jobID) if err != nil { return "", err } - if status == "PEND" { - timer.Reset(5 * time.Second) + switch status { + case "PEND", "PSUSP", "TRANSIENT": + // Not running yet (or state uncertain after a controller + // failover): fast-poll the first few rounds so a normally + // dispatched job attaches with no perceptible delay, then + // back off while the wait drags on. + if polls <= pendingFastPolls { + timer.Reset(pendingFastInterval) + } else { + timer.Reset(pendingBackoffInterval) + } continue - } - if status == "RUN" { + case "USUSP", "SSUSP": + // Suspended after starting: the allocation is held and a + // resume may follow any moment, but there is nothing to + // poll eagerly -- poll at the backed-off cadence. + timer.Reset(pendingBackoffInterval) + continue + case "RUN": url, err := js.checkExtendData(apiClient, token, jobID) if err != nil { return "", err @@ -649,12 +691,13 @@ func (js *JobSubmitter) waitForInteractiveURL(apiClient *request.APIClient, toke if url == "" { // running but the worker's start report (carrying the // URL) has not arrived yet -- poll again shortly - timer.Reset(2 * time.Second) + timer.Reset(runInterval) continue } return url, nil + default: + return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, status) } - return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, status) } } } diff --git a/src/frontend/xcecli/cmd/jobs/submit_test.go b/src/frontend/xcecli/cmd/jobs/submit_test.go index 9aa63a55..e951546c 100644 --- a/src/frontend/xcecli/cmd/jobs/submit_test.go +++ b/src/frontend/xcecli/cmd/jobs/submit_test.go @@ -7,7 +7,9 @@ import ( "net/http/httptest" "os" "strings" + "sync" "testing" + "time" "go.uber.org/zap" @@ -130,3 +132,162 @@ func TestSubmitHelpShowsArrayForms(t *testing.T) { } } } + +// jobStateServer answers the per-job status query with the states fed in +// sequence, holding the last one forever; every request's arrival time is +// recorded so tests can assert the poll cadence. +type jobStateServer struct { + srv *httptest.Server + states chan string + mu sync.Mutex + pollAts []time.Time + // urlQueries counts requests carrying fields=interactive_job_url. + urlQueries int +} + +func newJobStateServer(t *testing.T, states ...string) *jobStateServer { + t.Helper() + js := &jobStateServer{states: make(chan string, len(states))} + for _, s := range states { + js.states <- s + } + held := "" + js.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + js.mu.Lock() + js.pollAts = append(js.pollAts, time.Now()) + if r.URL.Query().Get("fields") == "interactive_job_url" { + js.urlQueries++ + } + js.mu.Unlock() + select { + case held = <-js.states: + default: + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(models.JobsResponse{ + Code: 0, + Data: struct { + Count int `json:"count"` + Rows []models.QueryJob `json:"rows"` + Page_token string `json:"page_token"` + }{Rows: []models.QueryJob{{JobStatus: held, InteractiveJobURL: "http://h:9000"}}}, + }) + })) + t.Cleanup(js.srv.Close) + return js +} + +func (js *jobStateServer) pollCount() int { + js.mu.Lock() + defer js.mu.Unlock() + return len(js.pollAts) +} + +// waitForInteractiveURL fast-polls the PEND phase (500ms) so a normally +// dispatched job attaches within ~a second, then backs off to 2s so a job +// waiting on a long queue doesn't hold 2 qps against the server forever. +func TestWaitForInteractiveURL_PendFastThenBackoff(t *testing.T) { + if config.Log == nil { + config.Log = zap.NewNop() + t.Cleanup(func() { config.Log = nil }) + } + // PEND through the fast window (11 polls at t=0..5.0s) plus one backoff + // round, then RUN+URL: the total wait is 10*0.5 + 2 + ~0 = ~7s. + states := make([]string, 11) + for i := range states { + states[i] = "PEND" + } + states = append(states, "RUN") + js := newJobStateServer(t, states...) + cli := NewJobSubmitter(nil) + start := time.Now() + url, err := cli.waitForInteractiveURL(request.NewAPIClient(js.srv.URL), "tok", "123") + elapsed := time.Since(start) + if err != nil { + t.Fatalf("waitForInteractiveURL: %v", err) + } + if url != "http://h:9000" { + t.Errorf("url = %q, want http://h:9000", url) + } + // ~7s expected; under 6s means the backoff never happened (all fast), + // over 9s means the fast window was skipped (all 2s). + if elapsed < 6*time.Second || elapsed > 9*time.Second { + t.Errorf("elapsed = %v, want ~7s (10 fast intervals then a 2s backoff)", elapsed) + } + if n := js.pollCount(); n != 13 { + t.Errorf("polls = %d, want 13 (12 status + 1 url query)", n) + } +} + +// RUN-before-URL polls at the 500ms run cadence: the start-report push +// normally closes the race in well under a second, so a 2s wait here would +// add that much dead time to every interactive submit. +func TestWaitForInteractiveURL_RunRaceUsesFastInterval(t *testing.T) { + if config.Log == nil { + config.Log = zap.NewNop() + t.Cleanup(func() { config.Log = nil }) + } + // First poll sees RUN with the URL already missing (the state server + // only exposes a URL once the state flips), second sees it present. + // We emulate the missing-URL round by answering the first url query + // with an empty URL: swap in a custom server. + seen := 0 + emptyURL := true + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + url := "" + if !emptyURL { + url = "http://h:9000" + } + row := models.QueryJob{JobStatus: "RUN", InteractiveJobURL: url} + if r.URL.Query().Get("fields") == "interactive_job_url" { + seen++ + if seen == 1 { + row.InteractiveJobURL = "" + } + emptyURL = row.InteractiveJobURL != "" + } + _ = json.NewEncoder(w).Encode(models.JobsResponse{ + Code: 0, + Data: struct { + Count int `json:"count"` + Rows []models.QueryJob `json:"rows"` + Page_token string `json:"page_token"` + }{Rows: []models.QueryJob{row}}, + }) + })) + t.Cleanup(srv.Close) + cli := NewJobSubmitter(nil) + start := time.Now() + url, err := cli.waitForInteractiveURL(request.NewAPIClient(srv.URL), "tok", "123") + elapsed := time.Since(start) + if err != nil { + t.Fatalf("waitForInteractiveURL: %v", err) + } + if url != "http://h:9000" { + t.Errorf("url = %q, want http://h:9000", url) + } + if elapsed > 1200*time.Millisecond { + t.Errorf("elapsed = %v, want ~0.5s (run cadence, not the old 2s)", elapsed) + } +} + +// A suspended interactive job must not fail the wait the way a terminal +// state does: the allocation is held and `xce jobs resume` may follow, so +// the loop keeps waiting (backed off) and picks up the URL when the job is +// running again. +func TestWaitForInteractiveURL_SuspendedJobKeepsWaiting(t *testing.T) { + if config.Log == nil { + config.Log = zap.NewNop() + t.Cleanup(func() { config.Log = nil }) + } + js := newJobStateServer(t, "USUSP", "RUN") + cli := NewJobSubmitter(nil) + url, err := cli.waitForInteractiveURL(request.NewAPIClient(js.srv.URL), "tok", "123") + if err != nil { + t.Fatalf("waitForInteractiveURL: %v", err) + } + if url != "http://h:9000" { + t.Errorf("url = %q, want http://h:9000", url) + } +}