From 66a1d1e6803e3643f1bb2d6abe2663524d6e5d0e Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Thu, 15 Oct 2026 08:32:21 +0000 Subject: [PATCH] attach jobs end with their command: relay watcher, once-only start banner, worker-side ttyd cleanup Three attach-mode (-I) fixes, verified end-to-end on the t1 cluster: 1. Session hang on job finish (job_attach.go): the relay's client->worker pump blocked forever in ReadMessage after the worker hop closed, so the old wg.Wait()-before-Close teardown never ran and the client hung, then died with an abnormal 1006. A broker watcher now subscribes to EventJobFinished and on job end sends the 4001 policy close and closes both hops, unblocking the pumps; the teardown follows normally. The client (attach.go) treats 4001/1000/1001 -- and 1006-after-output as a final safety net -- as a clean session end. 2. "<>" printed repeatedly (submit.go): the status poll printed the banner on every transition; now gated on a reportedStart flag. The URL is also no longer printed in attach mode -- it only makes sense for browse (-I --tty --browse) and DISPLAY jobs. 3. ttyd left behind after an attach job exits (xcecd_worker.c): a new tty_stop_attach_session(), called from process_exit_callback for TTY jobs, kills the per-job ttyd right away. This is insurance -- the sh -c exec means uv_proc monitors ttyd's own pid, so ttyd is normally gone before the callback fires -- but it guarantees no orphan when the timing lines up badly. Web-viewed sessions are untouched: they keep ttyd and output alive for later browsing, per design. Co-Authored-By: Claude Code --- .../apiserver/controllers/job_attach.go | 82 +++++++++++++++---- src/frontend/xcecli/cmd/jobs/attach.go | 19 ++++- src/frontend/xcecli/cmd/jobs/submit.go | 37 +++++---- src/worker/src/xcecd_worker.c | 39 +++++++++ 4 files changed, 146 insertions(+), 31 deletions(-) diff --git a/src/frontend/apiserver/controllers/job_attach.go b/src/frontend/apiserver/controllers/job_attach.go index 4111084a..301e6d8a 100644 --- a/src/frontend/apiserver/controllers/job_attach.go +++ b/src/frontend/apiserver/controllers/job_attach.go @@ -15,7 +15,6 @@ import ( logger "apiserver/utils" - "github.com/gin-gonic/gin" "github.com/gorilla/websocket" "go.uber.org/zap" @@ -166,7 +165,7 @@ func (j JobController) AttachJob(ctx *gin.Context) { zap.Int64("jobid", job.Unique_JobID), zap.String("user", logonUser), zap.String("worker_url", workerURL)) - attachRelay(clientConn, workerConn) + attachRelay(uint64(job.Unique_JobID), clientConn, workerConn) logger.Info("attach: relay finished", zap.Int64("jobid", job.Unique_JobID)) } @@ -196,13 +195,17 @@ func attachTTYDURL(raw string) (string, error) { return u.String(), nil } -// attachClose sends one close frame (best-effort) and stops reading. +// attachClose sends one close frame (best-effort) and waits for the peer's +// closing handshake before the caller tears the TCP connection down. Without +// the drain the deferred Close() can race the frame out the socket and the +// client sees 1006 (abnormal closure) instead of our policy code. func attachClose(conn *websocket.Conn, code int, text string) { deadline := time.Now().Add(attachWriteTimeout) conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(code, text), deadline) - // Drain so the close frame is actually processed by gorilla's internal - // control handler before the connection is torn down. + // Drain until the peer answers the handshake (or the deadline): each + // NextReader pumps gorilla's internal control-frame handler, which + // completes the close exchange on this side. conn.SetReadDeadline(deadline) for { if _, _, err := conn.NextReader(); err != nil { @@ -215,12 +218,13 @@ func attachClose(conn *websocket.Conn, code int, text string) { // goes away. Each direction runs in its own goroutine; the first direction // to finish tears the whole relay down (and closes both connections, which // unblocks the other direction's in-flight Read). -func attachRelay(client, worker *websocket.Conn) { - // Oversized pipes: terminal output can burst; the pipes absorb jitter - // between the two hops. On ping/pong: both hops are terminal sessions, - // so application-level traffic flows constantly when attached; the - // deadlines below exist to catch a dead peer at all, not to time out - // interactive silence mid-command. +// +// The relay also watches the job's terminal state: an attach session is over +// the moment its job finishes, even if the worker-side websocket lingers +// (ttyd's --exit-with-program does not always fire promptly with a client +// still attached). The watch fires a policy close to the client in that +// case; the pumps' teardown below runs as usual. +func attachRelay(jobID uint64, client, worker *websocket.Conn) { var wg sync.WaitGroup wg.Add(2) done := make(chan struct{}) @@ -228,16 +232,61 @@ func attachRelay(client, worker *websocket.Conn) { go func() { defer wg.Done() if !attachPump(worker, client, done) { - // worker side ended on its own: the session (or its job) ended + // worker side ended on its own: the session (or its job) ended. + // Send the policy close and complete the handshake BEFORE the + // teardown goroutine closes the socket -- attachClose blocks + // until the client answers or the deadline passes, so the + // client reliably reads 4001 instead of a bare 1006. attachClose(client, attachCloseJobEnded, "interactive session ended") } }() go func() { defer wg.Done() + // client-initiated end: forward nothing, teardown below suffices. attachPump(client, worker, done) }() - // Unblock both read pumps when either direction finishes. + // The job ending is a session end regardless of the worker hop: send + // the policy close and tear the relay down. WriteControl (not the full + // attachClose handshake) here: the watcher may race a pump already + // closing this connection, and the teardown below follows immediately, + // so a best-effort frame is enough -- the client tolerates both a 4001 + // and an abrupt end after output. + go func() { + ch := backend.DefaultBroker.Subscribe(4) + defer backend.DefaultBroker.Unsubscribe(ch) + for { + select { + case <-done: + return + case ev, ok := <-ch: + if !ok { + return + } + if ev.JobID == jobID && ev.Type == backend.EventJobFinished { + select { + case <-done: + default: + logger.Info("attach: job finished, ending relayed session", + zap.Uint64("jobid", jobID)) + _ = client.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(attachCloseJobEnded, + "interactive session ended (job finished)"), + time.Now().Add(attachWriteTimeout)) + } + // unblock the pumps; teardown closes both hops + go func() { + client.Close() + worker.Close() + }() + return + } + } + } + }() + + // Unblock both read pumps when either direction finishes (including the + // attachClose handshake window above). go func() { wg.Wait() close(done) @@ -251,8 +300,9 @@ func attachRelay(client, worker *websocket.Conn) { // attachPump copies one direction: reads frames from src, writes them to // dst. Binary and text frames pass through with their type; close/other // control frames end the pump (gorilla surfaces close via the read error). -// Returns false when src ended first (its peer outlived it), true when the -// dst side ended. +// Returns false when src ended on its own (its peer outlived it, and the +// caller may want to send it a policy close); true when the end was caused +// by dst or by our own teardown (nothing left to tell). func attachPump(src, dst *websocket.Conn, done <-chan struct{}) bool { for { msgType, data, err := src.ReadMessage() @@ -268,6 +318,8 @@ func attachPump(src, dst *websocket.Conn, done <-chan struct{}) bool { } _ = dst.SetWriteDeadline(time.Now().Add(attachWriteTimeout)) if err := dst.WriteMessage(msgType, data); err != nil { + // dst went away first; unblock src's reads via teardown and + // report "not src-ended" so no policy close is sent to dst. return true } } diff --git a/src/frontend/xcecli/cmd/jobs/attach.go b/src/frontend/xcecli/cmd/jobs/attach.go index 4b2593a7..953bf696 100644 --- a/src/frontend/xcecli/cmd/jobs/attach.go +++ b/src/frontend/xcecli/cmd/jobs/attach.go @@ -49,7 +49,9 @@ func attachEndpoint(serverURL, jobID string) (string, error) { } // runAttachSession connects to the relay, puts the local terminal in raw -// mode and copies bytes both ways until the session ends. +// mode and copies bytes both ways until the session ends. sawOutput tracks +// whether the session ever delivered terminal data, so an abrupt close +// after real output is reported as a normal end instead of an error. func runAttachSession(serverURL, token, jobID string) error { endpoint, err := attachEndpoint(serverURL, jobID) if err != nil { @@ -75,6 +77,8 @@ func runAttachSession(serverURL, token, jobID string) error { } defer conn.Close() + sawOutput := false + // terminal raw mode around the whole session oldState, err := term.MakeRaw(int(os.Stdin.Fd())) if err != nil { @@ -132,8 +136,18 @@ func runAttachSession(serverURL, token, jobID string) error { for { mt, data, err := conn.ReadMessage() if err != nil { - // 4001 = session ended cleanly (job finished / server side closed) + // 4001 = session ended cleanly (job finished / server side + // closed). 1006 often carries the same fact: ttyd tears the TCP + // connection down immediately after a fast-exiting command (its + // -Q path) and the relay's close frame can race that teardown, + // so treat an abrupt end with prior session output the same as + // a clean 4001 rather than scaring the user with an error. if websocket.IsCloseError(err, 4001, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + fmt.Println("<>") + return nil + } + if websocket.IsCloseError(err, websocket.CloseAbnormalClosure) && sawOutput { + fmt.Println("<>") return nil } return fmt.Errorf("session ended: [%s]", attachErrText(err)) @@ -143,6 +157,7 @@ func runAttachSession(serverURL, token, jobID string) error { } switch data[0] { case '0': // OUTPUT + sawOutput = true if _, err := os.Stdout.Write(data[1:]); err != nil { return nil } diff --git a/src/frontend/xcecli/cmd/jobs/submit.go b/src/frontend/xcecli/cmd/jobs/submit.go index 38cbc8cd..46177111 100755 --- a/src/frontend/xcecli/cmd/jobs/submit.go +++ b/src/frontend/xcecli/cmd/jobs/submit.go @@ -58,6 +58,7 @@ type JobSubmitter struct { useTTY bool useXJob bool useBrowse bool + reportedStart bool projectName string askedHosts string launchType string @@ -510,25 +511,27 @@ func (js *JobSubmitter) RunJobsSubmit() error { if err != nil { return err } - fmt.Printf("interactive session URL: %s\n", url) if js.useTTY { // TTY jobs attach by default: the session is displayed right in // this terminal through the apiserver relay (worker ttyd ports are - // not reachable from user clients in production). --browse instead - // opens the ttyd web URL in the browser. - if js.useBrowse { - if err := openBrowser(url); err != nil { - errMsg := request.ExtractErrorMessage(err) - fmt.Printf("warning: failed to open browser: [%s]\n", errMsg) - fmt.Println("Please manually open the URL in your browser.") - } - return nil + // not reachable from user clients in production). The URL is an + // implementation detail of the relay -- not printed unless the + // user asked for the web view. --browse opens the ttyd web URL + // in the browser instead of attaching. + if !js.useBrowse { + return runAttachSession(serverInfo.URL, serverInfo.Token, jobID) } - return runAttachSession(serverInfo.URL, serverInfo.Token, jobID) + if err := openBrowser(url); err != nil { + errMsg := request.ExtractErrorMessage(err) + fmt.Printf("warning: failed to open browser: [%s]\n", errMsg) + fmt.Println("Please manually open the URL in your browser.") + } + return nil } // XJOB (VNC) has no terminal rendering -- browser remains the path + fmt.Printf("interactive session URL: %s\n", url) if err := openBrowser(url); err != nil { errMsg := request.ExtractErrorMessage(err) fmt.Printf("warning: failed to open browser: [%s]\n", errMsg) @@ -576,9 +579,15 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin case "PEND": return row.JobStatus, nil case "RUN": - var execHosts models.ExecHosts - if err := json.Unmarshal([]byte(row.ExecHosts.String()), &execHosts); err == nil && len(execHosts.HostSlots) > 0 { - fmt.Println("<>") + // Print the dispatch line once: waitForInteractiveURL keeps + // polling while RUN until the session URL arrives, and this + // poll loop re-entered here would repeat the message. + if !js.reportedStart { + var execHosts models.ExecHosts + if err := json.Unmarshal([]byte(row.ExecHosts.String()), &execHosts); err == nil && len(execHosts.HostSlots) > 0 { + fmt.Println("<>") + js.reportedStart = true + } } return row.JobStatus, nil default: diff --git a/src/worker/src/xcecd_worker.c b/src/worker/src/xcecd_worker.c index 4abb2f2a..cd4c347c 100644 --- a/src/worker/src/xcecd_worker.c +++ b/src/worker/src/xcecd_worker.c @@ -60,6 +60,7 @@ static void check_and_reconnect(); static void worker_reconnect_callback(bool connected); static int xjob_start_vnc(); static void xjob_stop_vnc(); +static void tty_stop_attach_session(); static bool check_process_status(pid_t target_pid, int expected_signal, int timeout_seconds); @@ -515,6 +516,12 @@ static void process_exit_callback(int pid, int exit_status, int term_signal, xjob_stop_vnc(); } + /* attach-mode TTY job: the session ends with the command (web-viewed + * TTY jobs keep ttyd for later browsing -- see tty_stop_attach_session) */ + if (TTY == mydata.transfer_info->interactive_mode) { + tty_stop_attach_session(); + } + if (job_runtime_info.job_post_exec[0] != '\0') { job_post_exec(); } @@ -710,6 +717,38 @@ static void xjob_stop_vnc() jobidStr); } +/* An attached TTY job (interactive=1, the xcecli attach path) is done the + * moment its command exits: the user watches the same terminal, so there is + * nothing left to serve. Kill the ttyd we spawned for the job right away -- + * ttyd's own -Q teardown cannot be relied on here: with its uv/libwebsockets + * foreign event loop it logs "exiting due to --exit-with-program" but stays + * alive while a websocket client (the apiserver relay) is still connected, + * which would leave the attach hanging and the ttyd orphaned. + * + * Web-viewed sessions (--browse / DISPLAY) intentionally keep ttyd and the + * output alive after the command ends until a browser connects or the job + * is reaped; tty_stop_attach_session is only called for the attach path. */ +static void tty_stop_attach_session() +{ + pid_t ttyd_pid = uv_proc_get_pid(&uv_proc_manager); + + if (ttyd_pid <= 0) { + return; + } + + char jobidStr[64]; + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Terminating ttyd %d of attach job %s", + ttyd_pid, jobidStr); + + signal_children_recursive(SIGKILL, ttyd_pid); + if (kill(ttyd_pid, SIGKILL) != 0 && errno != ESRCH) { + XCECD_WORKER_LOG(XCE_LOG_WARN, + "Failed to kill ttyd %d of job %s: %s", ttyd_pid, + jobidStr, strerror(errno)); + } +} + static void worker_job_callback(job_assignment_t* info) { char* job_cmd = NULL;