From 978fc779460ae3175b59058b39712202f62088b6 Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 14:18:27 -0600 Subject: [PATCH 1/7] feat(system): expose shared terminals locally --- internal/system/terminal_sessions.go | 270 +++++++++++++++++++--- internal/system/terminal_sessions_test.go | 202 ++++++++++++++++ relay/contract.go | 10 + relay/contract_test.go | 7 +- 4 files changed, 451 insertions(+), 38 deletions(-) diff --git a/internal/system/terminal_sessions.go b/internal/system/terminal_sessions.go index 70c7d10..afe2326 100644 --- a/internal/system/terminal_sessions.go +++ b/internal/system/terminal_sessions.go @@ -5,6 +5,7 @@ package system import ( "bufio" "fmt" + "io" "net" "os" "os/exec" @@ -78,6 +79,9 @@ const ( // output to batch with. It is the entire latency cost of coalescing, and it // is paid only by output that already arrived in tty-chunk quantities. terminalCoalesceWindow = 2 * time.Millisecond + // terminalLocalActionTO gives an in-process attachment the same bounded + // action window that a relay request obtains from its accepted fence. + terminalLocalActionTO = 10 * time.Second ) // terminalKillGrace is how long close waits for the shell's process group to @@ -184,7 +188,7 @@ type terminalSession struct { mu sync.Mutex inputMu sync.Mutex - conns map[*sealedConn]struct{} + conns map[terminalConn]struct{} replay replayRing active bool closed bool @@ -193,6 +197,16 @@ type terminalSession struct { closeOnce sync.Once } +// terminalConn is the common outbound half of a browser or local attachment. +// It deliberately contains no read side: browser input remains sealed and +// local input is accepted only through TerminalClient. Both kinds use the same +// bounded fan-out queue and resynchronisation path. +type terminalConn interface { + enqueue([]byte) bool + resync([]byte) bool + kill() +} + // sealedConn is one browser connection attached to a terminal session. // // Sealing is per connection, not per session: a session the browser reattaches @@ -331,6 +345,180 @@ func (c *sealedConn) kill() { }) } +// TerminalClient is a local attachment to a terminal session. Output is raw +// terminal bytes, including the reset-prefixed bounded replay sent on attach +// and resynchronisation. Callers must treat output slices as read-only. +// +// Detach releases only this attachment; the session remains alive while any +// browser or local attachment remains, then follows terminalDetachTTL. +type TerminalClient struct { + session *terminalSession + conn *localTerminalConn +} + +// Output is the bounded stream of terminal output for this attachment. Done is +// closed if this attachment is detached because its consumer stopped draining. +func (c *TerminalClient) Output() <-chan []byte { return c.conn.output } +func (c *TerminalClient) Done() <-chan struct{} { return c.conn.dead } + +// Write sends input to the shared PTY. +func (c *TerminalClient) Write(p []byte) error { + if len(p) > relay.MaxFrameSize { + return fmt.Errorf("terminal input exceeds max frame size") + } + select { + case <-c.conn.dead: + return fmt.Errorf("terminal attachment is closed") + default: + } + c.session.inputMu.Lock() + defer c.session.inputMu.Unlock() + for len(p) > 0 { + n, err := c.session.ptmx.Write(p) + if err != nil { + return err + } + if n == 0 { + return io.ErrShortWrite + } + p = p[n:] + } + return nil +} + +// Resize updates the shared PTY's dimensions. +func (c *TerminalClient) Resize(cols, rows int) error { + if !relay.ValidTerminalSize(cols, rows) { + return fmt.Errorf("invalid terminal size") + } + select { + case <-c.conn.dead: + return fmt.Errorf("terminal attachment is closed") + default: + } + return pty.Setsize(c.session.ptmx, &pty.Winsize{Cols: uint16(cols), Rows: uint16(rows)}) +} + +// Detach releases this local attachment without ending its shared shell. +func (c *TerminalClient) Detach() { c.session.detach(c.conn) } + +// AttachTerminal creates or reattaches a local terminal client. projectRoot is +// passed through terminal so the same path resolution, policy checks and action +// fence check run on every local create and reattach as on a browser request. +func (d *system) AttachTerminal(projectRoot, sessionID string, cols, rows int) (*TerminalClient, error) { + if sessionID == "" { + return nil, fmt.Errorf("missing session id") + } + header := relay.StreamHeader{Kind: relay.KindTerminal, Cwd: projectRoot, SessionID: sessionID, Cols: cols, Rows: rows} + s, err := d.terminal(header, d.actionTime().Add(terminalLocalActionTO)) + if err != nil { + return nil, err + } + conn := newLocalTerminalConn() + if err := s.attachConn(conn); err != nil { + conn.kill() + return nil, err + } + go func() { + <-conn.dead + s.detach(conn) + }() + return &TerminalClient{session: s, conn: conn}, nil +} + +// localTerminalConn is a bounded in-memory counterpart to sealedConn. It does +// not model a network connection or sealing; it only consumes the same encoded +// fan-out frames, exposing data frames to the local TUI. A blocked consumer is +// bounded by terminalWriteTO, just as a blocked browser write is. +type localTerminalConn struct { + send chan []byte + output chan []byte + queued atomic.Int64 + dead chan struct{} + killOne sync.Once +} + +func newLocalTerminalConn() *localTerminalConn { + c := &localTerminalConn{ + send: make(chan []byte, terminalSendQueue), + // Hold one decoded chunk so AttachTerminal can publish the initial replay + // before returning without creating a second, unaccounted output backlog. + output: make(chan []byte, 1), + dead: make(chan struct{}), + } + go c.writeLoop() + return c +} + +func (c *localTerminalConn) enqueue(frame []byte) bool { + select { + case <-c.dead: + return false + default: + } + if c.queued.Load()+int64(len(frame)) > terminalSendBytes { + return false + } + select { + case c.send <- frame: + c.queued.Add(int64(len(frame))) + return true + default: + return false + } +} + +func (c *localTerminalConn) resync(frame []byte) bool { +drain: + for { + select { + case stale := <-c.send: + c.queued.Add(-int64(len(stale))) + default: + break drain + } + } + return c.enqueue(frame) +} + +func (c *localTerminalConn) writeLoop() { + for { + select { + case <-c.dead: + return + case frame := <-c.send: + c.queued.Add(-int64(len(frame))) + decoded, err := relay.DecodeFrame(frame) + if err != nil { + c.kill() + return + } + if decoded.Data == nil { + continue + } + timer := time.NewTimer(terminalWriteTO) + select { + case c.output <- decoded.Data: + if !timer.Stop() { + <-timer.C + } + case <-c.dead: + if !timer.Stop() { + <-timer.C + } + return + case <-timer.C: + c.kill() + return + } + } + } +} + +func (c *localTerminalConn) kill() { + c.killOne.Do(func() { close(c.dead) }) +} + func (d *system) handleTerminal(stream net.Conn, br *bufio.Reader, h relay.StreamHeader, actionDeadline time.Time) { if h.SessionID == "" { d.logf("terminal refused: missing session id") @@ -485,7 +673,7 @@ func (d *system) terminal(h relay.StreamHeader, actionDeadline time.Time) (*term // Unconditional: the header's dimensions were validated at the top of this // function, so there is no reachable case where they are zero. _ = pty.Setsize(ptmx, &pty.Winsize{Cols: uint16(h.Cols), Rows: uint16(h.Rows)}) - s := &terminalSession{id: h.SessionID, owner: d, cwd: cwd, ptmx: ptmx, cmd: cmd, pgid: pgid, conns: make(map[*sealedConn]struct{}), done: make(chan struct{})} + s := &terminalSession{id: h.SessionID, owner: d, cwd: cwd, ptmx: ptmx, cmd: cmd, pgid: pgid, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} d.terminals[h.SessionID] = s d.addSession(1) d.logf("terminal session started (%s)", d.cfg.Shell) @@ -523,20 +711,50 @@ func (d *system) enforcePolicy() { } func (s *terminalSession) attach(conn *sealedConn) { + if err := s.attachConn(conn); err != nil { + s.owner.logf("terminal refused: %v", err) + conn.kill() + return + } + + for { + frame, err := conn.stream.ReadFrame() + if err != nil { + break + } + switch { + case frame.Data != nil: + s.inputMu.Lock() + if _, err := s.ptmx.Write(frame.Data); err != nil { + s.inputMu.Unlock() + s.detach(conn) + return + } + s.inputMu.Unlock() + case frame.Resize != nil: + if !relay.ValidTerminalSize(frame.Resize.Cols, frame.Resize.Rows) { + continue + } + _ = pty.Setsize(s.ptmx, &pty.Winsize{Cols: uint16(frame.Resize.Cols), Rows: uint16(frame.Resize.Rows)}) + } + } + s.detach(conn) +} + +// attachConn adds either kind of terminal attachment and queues its initial +// replay. The caller handles its input path after this returns successfully. +func (s *terminalSession) attachConn(conn terminalConn) error { s.mu.Lock() if s.closed { s.mu.Unlock() - conn.kill() - return + return fmt.Errorf("terminal session is closed") } if s.conns == nil { - s.conns = make(map[*sealedConn]struct{}) + s.conns = make(map[terminalConn]struct{}) } if len(s.conns) >= maxTerminalConns { s.mu.Unlock() - s.owner.logf("terminal refused: session already has %d connections", maxTerminalConns) - conn.kill() - return + return fmt.Errorf("session already has %d connections", maxTerminalConns) } s.conns[conn] = struct{}{} if s.expires != nil { @@ -562,35 +780,13 @@ func (s *terminalSession) attach(conn *sealedConn) { } s.mu.Unlock() conn.kill() - return + return fmt.Errorf("terminal attachment queue unavailable") } s.mu.Unlock() - - for { - frame, err := conn.stream.ReadFrame() - if err != nil { - break - } - switch { - case frame.Data != nil: - s.inputMu.Lock() - if _, err := s.ptmx.Write(frame.Data); err != nil { - s.inputMu.Unlock() - s.detach(conn) - return - } - s.inputMu.Unlock() - case frame.Resize != nil: - if !relay.ValidTerminalSize(frame.Resize.Cols, frame.Resize.Rows) { - continue - } - _ = pty.Setsize(s.ptmx, &pty.Winsize{Cols: uint16(frame.Resize.Cols), Rows: uint16(frame.Resize.Rows)}) - } - } - s.detach(conn) + return nil } -func (s *terminalSession) detach(conn *sealedConn) { +func (s *terminalSession) detach(conn terminalConn) { conn.kill() s.mu.Lock() defer s.mu.Unlock() @@ -741,7 +937,7 @@ func (s *terminalSession) output(data []byte) { return } s.replay.append(data) - var dropped []*sealedConn + var dropped []terminalConn if len(s.conns) > 0 { // Encoded here, under the lock, rather than before it. encodeFrame // copies, and that copy is both what makes readPTY's reusable buffer @@ -785,8 +981,8 @@ func (s *terminalSession) catchUp() []byte { // // It also arms the detach TTL when the last connection goes, which is how a // session whose every client died eventually reaps itself. -func (s *terminalSession) broadcast(frame []byte, resendOnResync bool) []*sealedConn { - var dropped []*sealedConn +func (s *terminalSession) broadcast(frame []byte, resendOnResync bool) []terminalConn { + var dropped []terminalConn var catchUp []byte for conn := range s.conns { if conn.enqueue(frame) { @@ -811,7 +1007,7 @@ func (s *terminalSession) broadcast(frame []byte, resendOnResync bool) []*sealed return dropped } -func killAll(conns []*sealedConn) { +func killAll(conns []terminalConn) { for _, conn := range conns { conn.kill() } diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index 8110643..b94a0ea 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -426,6 +426,208 @@ func TestTerminalSessionBroadcastsToMultipleClients(t *testing.T) { } } +// Local attachments deliberately use the same session fan-out as their sealed +// browser siblings. This is the control comparison: both receive one output +// from one session, rather than a local attach starting a second shell. +func TestLocalTerminalClientSharesBrowserSession(t *testing.T) { + withTempConfigDir(t) + d := newSystem(systemConfig{Shell: "/bin/cat"}) + root := t.TempDir() + local, err := d.AttachTerminal(root, "local-shared", 80, 24) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { local.session.close() }) + if got := string(<-local.Output()); got != terminalResyncPrefix { + t.Fatalf("local initial replay = %q, want reset", got) + } + + browser, browserClient := sealedPair(t, "local-shared") + go local.session.attach(browser) + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatalf("read browser initial replay: %v", err) + } + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatalf("read browser initial activity: %v", err) + } + + local.session.output([]byte("shared with both")) + if got := string(<-local.Output()); got != "shared with both" { + t.Fatalf("local output = %q", got) + } + frame, err := browserClient.ReadFrame() + if err != nil { + t.Fatalf("read browser output: %v", err) + } + if got := string(frame.Data); got != "shared with both" { + t.Fatalf("browser output = %q", got) + } + + if err := local.Write([]byte("local input\n")); err != nil { + t.Fatalf("local input: %v", err) + } + deadline := time.After(5 * time.Second) + for { + select { + case output := <-local.Output(): + if strings.Contains(string(output), "local input") { + goto resized + } + case <-deadline: + t.Fatal("local input did not reach the shared PTY") + } + } + +resized: + if err := local.Resize(100, 40); err != nil { + t.Fatalf("local resize: %v", err) + } + rows, cols, err := pty.Getsize(local.session.ptmx) + if err != nil { + t.Fatal(err) + } + if cols != 100 || rows != 40 { + t.Fatalf("PTY size = %dx%d, want 100x40", cols, rows) + } + + local.Detach() + waitConns(t, local.session, 1) // browser remains; detach did not kill the shell. + if d.terminals["local-shared"] != local.session { + t.Fatal("local detach removed a shell still attached in the browser") + } +} + +func TestLocalTerminalClientReplayDetachAndConnectionLimit(t *testing.T) { + withTempConfigDir(t) + d := newSystem(systemConfig{Shell: "/bin/cat"}) + root := t.TempDir() + first, err := d.AttachTerminal(root, "local-limit", 80, 24) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { first.session.close() }) + <-first.Output() // initial reset + first.session.output([]byte("replayed")) + if got := string(<-first.Output()); got != "replayed" { + t.Fatalf("first output = %q", got) + } + first.Detach() + waitConns(t, first.session, 0) + first.session.mu.Lock() + armed := first.session.expires != nil + first.session.mu.Unlock() + if !armed { + t.Fatal("last local detach did not arm terminal TTL") + } + + reattached, err := d.AttachTerminal(root, "local-limit", 80, 24) + if err != nil { + t.Fatal(err) + } + if reattached.session != first.session { + t.Fatal("local reattach created a duplicate shell") + } + if got, want := string(<-reattached.Output()), terminalResyncPrefix+"replayed"; got != want { + t.Fatalf("local replay = %q, want %q", got, want) + } + clients := []*TerminalClient{reattached} + for len(clients) < maxTerminalConns { + client, err := d.AttachTerminal(root, "local-limit", 80, 24) + if err != nil { + t.Fatal(err) + } + <-client.Output() + clients = append(clients, client) + } + if _, err := d.AttachTerminal(root, "local-limit", 80, 24); err == nil || !strings.Contains(err.Error(), "connections") { + t.Fatalf("connection over limit error = %v, want connection-limit refusal", err) + } + for _, client := range clients { + client.Detach() + } +} + +// The browser sibling above proves the same bounded queue design over a sealed +// transport. This local control proves that an unread local output channel also +// cannot park the PTY read path and is detached at terminalWriteTO. +func TestLocalTerminalOutputNeverBlocksOnStalledClient(t *testing.T) { + s := &terminalSession{id: "local-stall", done: make(chan struct{})} + stalled := newLocalTerminalConn() + if got := cap(stalled.output); got != 1 { + t.Fatalf("local decoded output queue capacity = %d, want one bounded handoff", got) + } + if err := s.attachConn(stalled); err != nil { + t.Fatal(err) + } + go func() { <-stalled.dead; s.detach(stalled) }() + start := time.Now() + for range 150 { + s.output(make([]byte, 8<<10)) + } + if elapsed := time.Since(start); elapsed > terminalWriteTO/2 { + t.Fatalf("local stalled output took %s; PTY path waited on local client", elapsed) + } + // Fill the local consumer queue one writer hand-off at a time. This avoids + // treating a particular goroutine scheduling pattern during the output burst + // as evidence that the write-deadline path is reachable. + deadline := time.Now().Add(time.Second) + for len(stalled.output) < cap(stalled.output) { + if !stalled.enqueue(relay.EncodeData([]byte("x"))) { + t.Fatal("could not queue local output while filling its consumer buffer") + } + for len(stalled.send) != 0 { + if time.Now().After(deadline) { + t.Fatal("local writer did not consume queued output") + } + time.Sleep(time.Millisecond) + } + } + if !stalled.enqueue(relay.EncodeData([]byte("x"))) { + t.Fatal("could not queue output that should block the local writer") + } + select { + case <-stalled.dead: + case <-time.After(terminalWriteTO + time.Second): + t.Fatal("stalled local client was not detached at the write deadline") + } + waitConns(t, s, 0) + s.mu.Lock() + if s.expires != nil { + s.expires.Stop() + } + s.mu.Unlock() +} + +func TestLocalTerminalClientRechecksPolicyOnReattach(t *testing.T) { + dir := withTempConfigDir(t) + allowed := filepath.Join(dir, "allowed") + if err := os.Mkdir(allowed, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "policy.json"), []byte(fmt.Sprintf(`{"allowedRoots":[%q]}`, allowed)), 0o600); err != nil { + t.Fatal(err) + } + d := newSystem(systemConfig{Shell: "/bin/cat"}) + if _, err := d.AttachTerminal(t.TempDir(), "local-policy", 80, 24); err == nil { + t.Fatal("local create outside allowedRoots succeeded") + } + client, err := d.AttachTerminal(allowed, "local-policy", 80, 24) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { client.session.close() }) + <-client.Output() + if err := os.WriteFile(filepath.Join(dir, "policy.json"), []byte(`{"terminalsDisabled":true}`), 0o600); err != nil { + t.Fatal(err) + } + if _, err := d.AttachTerminal(allowed, "local-policy", 80, 24); err == nil || !strings.Contains(err.Error(), "disabled") { + t.Fatalf("policy-denied reattach error = %v", err) + } + if d.terminals["local-policy"] != client.session { + t.Fatal("policy denial removed the existing terminal") + } +} + // A session's id can be asked for with any cwd on the header: the requested // directory passes the policy check, then the lookup returns a shell rooted // somewhere the policy would never have allowed. Reattach has to re-decide diff --git a/relay/contract.go b/relay/contract.go index e30edc6..040cbfa 100644 --- a/relay/contract.go +++ b/relay/contract.go @@ -36,6 +36,16 @@ type ProjectInfo struct { Ports []PortEntry `json:"ports"` } +// TerminalSessionInfo identifies a terminal session and the project it belongs +// to. It is shared by the agent and relay control plane so their JSON field +// spelling cannot drift. +type TerminalSessionInfo struct { + ID string `json:"id"` + ProjectID string `json:"project_id"` + ProjectName string `json:"project_name"` + SessionID string `json:"session_id"` +} + // PortEntry is one exposed port with its record id. type PortEntry struct { ID string `json:"id"` diff --git a/relay/contract_test.go b/relay/contract_test.go index 925eb38..31494f4 100644 --- a/relay/contract_test.go +++ b/relay/contract_test.go @@ -21,13 +21,14 @@ func TestDeviceStatusValuesArePinnedToTheirWireLiterals(t *testing.T) { } } -// All 27 JSON tag occurrences in contract.go are shared HTTP control-plane +// All 31 JSON tag occurrences in contract.go are shared HTTP control-plane // contracts between the api and system binaries, so none are excluded. This is // the complete inventory, grouped by DTO: // // - SystemInfo (5): id, name, hostname, online, ip_addr // - PortInfo (3): project, port, label // - ProjectInfo (4): id, name, root_dir, ports +// - TerminalSessionInfo (4): id, project_id, project_name, session_id // - PortEntry (3): id, port, label // - DeviceStartRequest (2): client_id, hostname // - DeviceStartResponse (5): user_code, device_code, verification_url, @@ -52,6 +53,10 @@ func TestControlPlaneDTOTagsArePinnedToLiteralPayloads(t *testing.T) { assertLiteralJSON(t, `{"id":"proj-1","name":"ormos","root_dir":"/code/ormos","ports":[{"id":"port-1","port":8080,"label":"web"}]}`, ProjectInfo{ID: "proj-1", Name: "ormos", RootDir: "/code/ormos", Ports: []PortEntry{{ID: "port-1", Port: 8080, Label: "web"}}}) }) + t.Run("TerminalSessionInfo", func(t *testing.T) { + assertLiteralJSON(t, `{"id":"terminal-1","project_id":"proj-1","project_name":"ormos","session_id":"session-1"}`, + TerminalSessionInfo{ID: "terminal-1", ProjectID: "proj-1", ProjectName: "ormos", SessionID: "session-1"}) + }) t.Run("PortEntry", func(t *testing.T) { assertLiteralJSON(t, `{"id":"port-1","port":8080,"label":"web"}`, PortEntry{ID: "port-1", Port: 8080, Label: "web"}) From 3a84e2d9ca99ee1f35678d2d9e0d1e601c45a948 Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 14:58:03 -0600 Subject: [PATCH 2/7] fix(system): bound local terminal input --- internal/system/terminal_sessions.go | 184 ++++++++++++++++++--- internal/system/terminal_sessions_test.go | 191 +++++++++++++++++++++- 2 files changed, 352 insertions(+), 23 deletions(-) diff --git a/internal/system/terminal_sessions.go b/internal/system/terminal_sessions.go index afe2326..614e9a8 100644 --- a/internal/system/terminal_sessions.go +++ b/internal/system/terminal_sessions.go @@ -4,8 +4,8 @@ package system import ( "bufio" + "errors" "fmt" - "io" "net" "os" "os/exec" @@ -82,6 +82,22 @@ const ( // terminalLocalActionTO gives an in-process attachment the same bounded // action window that a relay request obtains from its accepted fence. terminalLocalActionTO = 10 * time.Second + // terminalInputQueue bounds both queued local calls and their copied bytes. + // It is intentionally the same byte budget as outbound terminal traffic: + // four local attachments cannot turn pasted input into unbounded agent memory. + terminalInputQueue = terminalSendQueue + terminalInputBytes = terminalSendBytes + terminalInputPollWindow = 25 * time.Millisecond + terminalInputChunk = 4 << 10 +) + +var ( + // ErrTerminalInputBackpressure means a local Write was not accepted. Callers + // may retry after consuming output or yielding; bytes are never dropped. + ErrTerminalInputBackpressure = errors.New("terminal input backpressure") + // ErrTerminalClientClosed means this attachment has detached or its session + // ended. It never means the shared shell was killed by the attachment. + ErrTerminalClientClosed = errors.New("terminal attachment is closed") ) // terminalKillGrace is how long close waits for the shell's process group to @@ -356,34 +372,45 @@ type TerminalClient struct { conn *localTerminalConn } -// Output is the bounded stream of terminal output for this attachment. Done is -// closed if this attachment is detached because its consumer stopped draining. +// Output is the bounded stream of terminal output for this attachment. It is +// not closed; receive it with Done. Done closes on Detach, slow-reader eviction, +// session/process exit, or policy revocation. func (c *TerminalClient) Output() <-chan []byte { return c.conn.output } func (c *TerminalClient) Done() <-chan struct{} { return c.conn.dead } -// Write sends input to the shared PTY. +// Write copies and queues input for the shared PTY without blocking on a +// non-reading slave. Local writes keep their acceptance order. It returns +// ErrTerminalInputBackpressure when the bounded queue cannot accept the bytes, +// and ErrTerminalClientClosed after this attachment ends. func (c *TerminalClient) Write(p []byte) error { - if len(p) > relay.MaxFrameSize { - return fmt.Errorf("terminal input exceeds max frame size") + if len(p) > relay.MaxFrameSize || len(p) > terminalInputBytes { + return ErrTerminalInputBackpressure + } + if len(p) == 0 { + return nil } + copy := append([]byte(nil), p...) + c.conn.inputMu.Lock() + defer c.conn.inputMu.Unlock() select { case <-c.conn.dead: - return fmt.Errorf("terminal attachment is closed") + return ErrTerminalClientClosed default: } - c.session.inputMu.Lock() - defer c.session.inputMu.Unlock() - for len(p) > 0 { - n, err := c.session.ptmx.Write(p) - if err != nil { - return err - } - if n == 0 { - return io.ErrShortWrite - } - p = p[n:] + if c.conn.inputBytes.Load()+int64(len(copy)) > terminalInputBytes { + return ErrTerminalInputBackpressure + } + c.conn.inputBytes.Add(int64(len(copy))) + select { + case <-c.conn.dead: + c.conn.inputBytes.Add(-int64(len(copy))) + return ErrTerminalClientClosed + case c.conn.input <- copy: + return nil + default: + c.conn.inputBytes.Add(-int64(len(copy))) + return ErrTerminalInputBackpressure } - return nil } // Resize updates the shared PTY's dimensions. @@ -415,10 +442,11 @@ func (d *system) AttachTerminal(projectRoot, sessionID string, cols, rows int) ( return nil, err } conn := newLocalTerminalConn() - if err := s.attachConn(conn); err != nil { + if err := s.attachLocal(conn, cols, rows); err != nil { conn.kill() return nil, err } + go conn.inputLoop(s) go func() { <-conn.dead s.detach(conn) @@ -436,6 +464,11 @@ type localTerminalConn struct { queued atomic.Int64 dead chan struct{} killOne sync.Once + + input chan []byte + inputBytes atomic.Int64 + inputMu sync.Mutex // serialises local Write acceptance order + inputDone chan struct{} } func newLocalTerminalConn() *localTerminalConn { @@ -443,8 +476,10 @@ func newLocalTerminalConn() *localTerminalConn { send: make(chan []byte, terminalSendQueue), // Hold one decoded chunk so AttachTerminal can publish the initial replay // before returning without creating a second, unaccounted output backlog. - output: make(chan []byte, 1), - dead: make(chan struct{}), + output: make(chan []byte, 1), + dead: make(chan struct{}), + input: make(chan []byte, terminalInputQueue), + inputDone: make(chan struct{}), } go c.writeLoop() return c @@ -519,6 +554,94 @@ func (c *localTerminalConn) kill() { c.killOne.Do(func() { close(c.dead) }) } +// inputLoop is one worker per attachment, never one per Write. It waits for +// POLLOUT before entering the session's shared input critical section, so a +// full local PTY cannot pin inputMu and starve a browser. poll is available on +// both supported Unix targets; its short timeout makes waiting detach-aware +// without claiming that an already-entered kernel Write can be cancelled. +func (c *localTerminalConn) inputLoop(s *terminalSession) { + defer func() { + for { + select { + case p := <-c.input: + c.inputBytes.Add(-int64(len(p))) + default: + close(c.inputDone) + return + } + } + }() + for { + select { + case <-c.dead: + return + case p := <-c.input: + if !c.writeInput(s, p) { + c.inputBytes.Add(-int64(len(p))) + return + } + c.inputBytes.Add(-int64(len(p))) + } + } +} + +func (c *localTerminalConn) writeInput(s *terminalSession, p []byte) bool { + for len(p) > 0 { + if !c.waitWritable(s.ptmx) { + return false + } + select { + case <-c.dead: + return false + default: + } + n := min(len(p), terminalInputChunk) + s.inputMu.Lock() + select { + case <-c.dead: + s.inputMu.Unlock() + return false + default: + } + written, err := s.ptmx.Write(p[:n]) + s.inputMu.Unlock() + if err != nil || written == 0 { + c.kill() + return false + } + p = p[written:] + } + return true +} + +func (c *localTerminalConn) waitWritable(f *os.File) bool { + rc, err := f.SyscallConn() + if err != nil { + c.kill() + return false + } + for { + select { + case <-c.dead: + return false + default: + } + ready := false + err := rc.Control(func(fd uintptr) { + pfd := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLOUT}} + n, pollErr := unix.Poll(pfd, int(terminalInputPollWindow/time.Millisecond)) + ready = pollErr == nil && n > 0 && pfd[0].Revents&unix.POLLOUT != 0 + }) + if err != nil { + c.kill() + return false + } + if ready { + return true + } + } +} + func (d *system) handleTerminal(stream net.Conn, br *bufio.Reader, h relay.StreamHeader, actionDeadline time.Time) { if h.SessionID == "" { d.logf("terminal refused: missing session id") @@ -741,9 +864,20 @@ func (s *terminalSession) attach(conn *sealedConn) { s.detach(conn) } +// attachLocal applies the dimensions promised by AttachTerminal while holding +// the same session lock that admits the attachment, so a successful reattach +// cannot return with the previous client's size still installed. +func (s *terminalSession) attachLocal(conn *localTerminalConn, cols, rows int) error { + return s.attachConnSize(conn, &pty.Winsize{Cols: uint16(cols), Rows: uint16(rows)}) +} + // attachConn adds either kind of terminal attachment and queues its initial // replay. The caller handles its input path after this returns successfully. func (s *terminalSession) attachConn(conn terminalConn) error { + return s.attachConnSize(conn, nil) +} + +func (s *terminalSession) attachConnSize(conn terminalConn, size *pty.Winsize) error { s.mu.Lock() if s.closed { s.mu.Unlock() @@ -756,6 +890,12 @@ func (s *terminalSession) attachConn(conn terminalConn) error { s.mu.Unlock() return fmt.Errorf("session already has %d connections", maxTerminalConns) } + if size != nil { + if err := pty.Setsize(s.ptmx, size); err != nil { + s.mu.Unlock() + return fmt.Errorf("set terminal size: %w", err) + } + } s.conns[conn] = struct{}{} if s.expires != nil { s.expires.Stop() diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index b94a0ea..db416ee 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -4,6 +4,7 @@ package system import ( "bytes" + "errors" "fmt" "io" "net" @@ -479,6 +480,22 @@ func TestLocalTerminalClientSharesBrowserSession(t *testing.T) { } resized: + if err := browserClient.WriteFrame(relay.EncodeData([]byte("browser input\n"))); err != nil { + t.Fatalf("browser input: %v", err) + } + deadline = time.After(5 * time.Second) + for { + select { + case output := <-local.Output(): + if strings.Contains(string(output), "browser input") { + goto resizedByBrowser + } + case <-deadline: + t.Fatal("browser input did not reach the same PTY as local input") + } + } + +resizedByBrowser: if err := local.Resize(100, 40); err != nil { t.Fatalf("local resize: %v", err) } @@ -497,6 +514,37 @@ resized: } } +func TestLocalTerminalCreationUsesTerminalSessionLimit(t *testing.T) { + withTempConfigDir(t) + d := newSystem(systemConfig{Shell: "/bin/cat"}) + root := t.TempDir() + clients := make([]*TerminalClient, 0, maxTerminalSessions) + for i := range maxTerminalSessions { + client, err := d.AttachTerminal(root, fmt.Sprintf("local-session-%d", i), 80, 24) + if err != nil { + t.Fatalf("local session %d: %v", i, err) + } + <-client.Output() + clients = append(clients, client) + } + t.Cleanup(func() { + for _, client := range clients { + client.session.close() + } + }) + if _, err := d.AttachTerminal(root, "local-session-over-limit", 80, 24); err == nil || !strings.Contains(err.Error(), "session limit") { + t.Fatalf("local session limit error = %v", err) + } + + // Browser headers enter the same terminal authority. Calling it directly is + // the control for the already-sealed browser path: sealing happens before + // terminal(), but neither can bypass its shared session limit. + header := fencedHeader(relay.StreamHeader{Kind: relay.KindTerminal, Cwd: root, SessionID: "browser-session-over-limit", Cols: 80, Rows: 24}) + if _, err := d.terminal(header, acceptedFenceDeadline(t, header)); err == nil || !strings.Contains(err.Error(), "session limit") { + t.Fatalf("browser terminal authority limit error = %v", err) + } +} + func TestLocalTerminalClientReplayDetachAndConnectionLimit(t *testing.T) { withTempConfigDir(t) d := newSystem(systemConfig{Shell: "/bin/cat"}) @@ -520,13 +568,20 @@ func TestLocalTerminalClientReplayDetachAndConnectionLimit(t *testing.T) { t.Fatal("last local detach did not arm terminal TTL") } - reattached, err := d.AttachTerminal(root, "local-limit", 80, 24) + reattached, err := d.AttachTerminal(root, "local-limit", 100, 40) if err != nil { t.Fatal(err) } if reattached.session != first.session { t.Fatal("local reattach created a duplicate shell") } + rows, cols, err := pty.Getsize(reattached.session.ptmx) + if err != nil { + t.Fatal(err) + } + if cols != 100 || rows != 40 { + t.Fatalf("reattached PTY size = %dx%d, want 100x40", cols, rows) + } if got, want := string(<-reattached.Output()), terminalResyncPrefix+"replayed"; got != want { t.Fatalf("local replay = %q, want %q", got, want) } @@ -547,6 +602,140 @@ func TestLocalTerminalClientReplayDetachAndConnectionLimit(t *testing.T) { } } +// A local TUI must never be held hostage by a PTY whose slave has stopped +// reading. The browser sibling remains the control: once the local attachment +// is gone, a browser can still write to this one PTY. +func TestLocalTerminalDetachReleasesInputAndDoesNotStarveBrowser(t *testing.T) { + inputRead, inputWrite, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) + fd := int(inputWrite.Fd()) + if err := unix.SetNonblock(fd, true); err != nil { + t.Fatal(err) + } + filled := 0 + fill := make([]byte, terminalInputChunk) + for { + n, err := unix.Write(fd, fill) + filled += n + if err == unix.EAGAIN { + break + } + if err != nil { + t.Fatal(err) + } + } + if err := unix.SetNonblock(fd, false); err != nil { + t.Fatal(err) + } + s := &terminalSession{id: "local-input", ptmx: inputWrite, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} + localConn := newLocalTerminalConn() + if err := s.attachConn(localConn); err != nil { + t.Fatal(err) + } + go localConn.inputLoop(s) + local := &TerminalClient{session: s, conn: localConn} + + writeDone := make(chan error, 1) + go func() { writeDone <- local.Write(make([]byte, terminalSendBytes)) }() + select { + case err := <-writeDone: + if err != nil { + t.Fatalf("local Write returned %v before detach", err) + } + case <-time.After(time.Second): + t.Fatal("local Write blocked while PTY slave was not reading") + } + + local.Detach() + select { + case <-local.Done(): + case <-time.After(time.Second): + t.Fatal("local detach did not close Done") + } + select { + case <-localConn.inputDone: + case <-time.After(time.Second): + t.Fatal("local input worker did not release queued input after detach") + } + readFD := int(inputRead.Fd()) + if err := unix.SetNonblock(readFD, true); err != nil { + t.Fatal(err) + } + for { + _, err := unix.Read(readFD, make([]byte, filled)) + if err == unix.EAGAIN { + break + } + if err != nil { + t.Fatal(err) + } + } + if err := unix.SetNonblock(readFD, false); err != nil { + t.Fatal(err) + } + + browser, browserClient := sealedPair(t, "local-input") + go s.attach(browser) + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatalf("read browser replay: %v", err) + } + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatalf("read browser activity: %v", err) + } + if err := browserClient.WriteFrame(relay.EncodeData([]byte("browser input"))); err != nil { + t.Fatalf("write browser input: %v", err) + } + got := make(chan []byte, 1) + go func() { + buf := make([]byte, len("browser input")) + _, err := io.ReadFull(inputRead, buf) + if err != nil { + got <- nil + return + } + got <- buf + }() + select { + case data := <-got: + if string(data) != "browser input" { + t.Fatalf("browser input after local detach = %q", data) + } + case <-time.After(time.Second): + t.Fatal("browser input was starved after local detach") + } + browser.kill() +} + +func TestLocalTerminalWriteBoundsInputByBytes(t *testing.T) { + inputRead, inputWrite, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) + s := &terminalSession{id: "local-input-bound", ptmx: inputWrite, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} + conn := newLocalTerminalConn() + if err := s.attachConn(conn); err != nil { + t.Fatal(err) + } + go conn.inputLoop(s) + client := &TerminalClient{session: s, conn: conn} + if err := client.Write(make([]byte, terminalInputBytes)); err != nil { + t.Fatalf("fill bounded local input queue: %v", err) + } + if err := client.Write([]byte("x")); !errors.Is(err, ErrTerminalInputBackpressure) { + t.Fatalf("input beyond byte budget error = %v, want ErrTerminalInputBackpressure", err) + } + client.Detach() + select { + case <-conn.inputDone: + case <-time.After(time.Second): + t.Fatal("detach did not release bounded local input worker") + } +} + // The browser sibling above proves the same bounded queue design over a sealed // transport. This local control proves that an unread local output channel also // cannot park the PTY read path and is detached at terminalWriteTO. From 7647c364691f4abf6d9ae534044ba512dde8c411 Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 15:17:49 -0600 Subject: [PATCH 3/7] fix(system): harden terminal input admission --- internal/system/terminal_sessions.go | 110 +++++++++---- internal/system/terminal_sessions_test.go | 182 ++++++++++++++++++++++ 2 files changed, 259 insertions(+), 33 deletions(-) diff --git a/internal/system/terminal_sessions.go b/internal/system/terminal_sessions.go index 614e9a8..5a3b623 100644 --- a/internal/system/terminal_sessions.go +++ b/internal/system/terminal_sessions.go @@ -82,19 +82,26 @@ const ( // terminalLocalActionTO gives an in-process attachment the same bounded // action window that a relay request obtains from its accepted fence. terminalLocalActionTO = 10 * time.Second - // terminalInputQueue bounds both queued local calls and their copied bytes. - // It is intentionally the same byte budget as outbound terminal traffic: - // four local attachments cannot turn pasted input into unbounded agent memory. + // terminalInputQueue bounds accepted local Write calls, including the one the + // input worker is currently processing. terminalInputBytes bounds the copied + // bytes belonging to those calls, including that in-flight call. Caller-owned + // input is never counted or retained when admission rejects it. terminalInputQueue = terminalSendQueue terminalInputBytes = terminalSendBytes terminalInputPollWindow = 25 * time.Millisecond - terminalInputChunk = 4 << 10 + // PIPE_BUF is 4096 on Linux and 512 on Darwin. Keeping a write within the + // portable 512-byte floor lets the locked POLLOUT recheck establish that the + // subsequent write will not wait for a sibling master writer. + terminalInputChunk = 512 ) var ( // ErrTerminalInputBackpressure means a local Write was not accepted. Callers // may retry after consuming output or yielding; bytes are never dropped. ErrTerminalInputBackpressure = errors.New("terminal input backpressure") + // ErrTerminalInputTooLarge means one input exceeds the fixed per-call limit + // and cannot become acceptable by retrying. + ErrTerminalInputTooLarge = errors.New("terminal input is too large") // ErrTerminalClientClosed means this attachment has detached or its session // ended. It never means the shared shell was killed by the attachment. ErrTerminalClientClosed = errors.New("terminal attachment is closed") @@ -379,17 +386,14 @@ func (c *TerminalClient) Output() <-chan []byte { return c.conn.output } func (c *TerminalClient) Done() <-chan struct{} { return c.conn.dead } // Write copies and queues input for the shared PTY without blocking on a -// non-reading slave. Local writes keep their acceptance order. It returns -// ErrTerminalInputBackpressure when the bounded queue cannot accept the bytes, -// and ErrTerminalClientClosed after this attachment ends. +// non-reading slave. Local writes keep their acceptance order. Each live local +// attachment holds at most 256 accepted queued-or-in-flight calls and 1 MiB of +// implementation-owned copied bytes; rejected calls retain no copy. Success +// means admission completed before Detach; teardown may discard those accepted +// bytes. It returns ErrTerminalInputBackpressure when retrying may work, +// ErrTerminalInputTooLarge for a permanently oversized call, and +// ErrTerminalClientClosed after this attachment ends. func (c *TerminalClient) Write(p []byte) error { - if len(p) > relay.MaxFrameSize || len(p) > terminalInputBytes { - return ErrTerminalInputBackpressure - } - if len(p) == 0 { - return nil - } - copy := append([]byte(nil), p...) c.conn.inputMu.Lock() defer c.conn.inputMu.Unlock() select { @@ -397,17 +401,32 @@ func (c *TerminalClient) Write(p []byte) error { return ErrTerminalClientClosed default: } - if c.conn.inputBytes.Load()+int64(len(copy)) > terminalInputBytes { + if len(p) > min(relay.MaxFrameSize, terminalInputBytes) { + return ErrTerminalInputTooLarge + } + if len(p) == 0 { + return nil + } + if c.conn.inputCalls.Load() >= terminalInputQueue || c.conn.inputBytes.Load()+int64(len(p)) > terminalInputBytes { return ErrTerminalInputBackpressure } - c.conn.inputBytes.Add(int64(len(copy))) + // Reserve before allocation. inputMu also makes admission and kill one + // transaction, so a successful Write cannot be admitted after Detach. + c.conn.inputCalls.Add(1) + c.conn.inputBytes.Add(int64(len(p))) + if c.conn.beforeInputCopy != nil { + c.conn.beforeInputCopy() + } + copy := c.conn.copyInput(p) select { case <-c.conn.dead: + c.conn.inputCalls.Add(-1) c.conn.inputBytes.Add(-int64(len(copy))) return ErrTerminalClientClosed case c.conn.input <- copy: return nil default: + c.conn.inputCalls.Add(-1) c.conn.inputBytes.Add(-int64(len(copy))) return ErrTerminalInputBackpressure } @@ -466,9 +485,14 @@ type localTerminalConn struct { killOne sync.Once input chan []byte + inputCalls atomic.Int64 inputBytes atomic.Int64 inputMu sync.Mutex // serialises local Write acceptance order inputDone chan struct{} + copyInput func([]byte) []byte + // Test seams; nil in production. + beforeInputCopy func() + beforeInputRecheck func() } func newLocalTerminalConn() *localTerminalConn { @@ -480,6 +504,7 @@ func newLocalTerminalConn() *localTerminalConn { dead: make(chan struct{}), input: make(chan []byte, terminalInputQueue), inputDone: make(chan struct{}), + copyInput: func(p []byte) []byte { return append([]byte(nil), p...) }, } go c.writeLoop() return c @@ -551,7 +576,9 @@ func (c *localTerminalConn) writeLoop() { } func (c *localTerminalConn) kill() { + c.inputMu.Lock() c.killOne.Do(func() { close(c.dead) }) + c.inputMu.Unlock() } // inputLoop is one worker per attachment, never one per Write. It waits for @@ -564,6 +591,7 @@ func (c *localTerminalConn) inputLoop(s *terminalSession) { for { select { case p := <-c.input: + c.inputCalls.Add(-1) c.inputBytes.Add(-int64(len(p))) default: close(c.inputDone) @@ -577,9 +605,11 @@ func (c *localTerminalConn) inputLoop(s *terminalSession) { return case p := <-c.input: if !c.writeInput(s, p) { + c.inputCalls.Add(-1) c.inputBytes.Add(-int64(len(p))) return } + c.inputCalls.Add(-1) c.inputBytes.Add(-int64(len(p))) } } @@ -595,7 +625,6 @@ func (c *localTerminalConn) writeInput(s *terminalSession, p []byte) bool { return false default: } - n := min(len(p), terminalInputChunk) s.inputMu.Lock() select { case <-c.dead: @@ -603,6 +632,14 @@ func (c *localTerminalConn) writeInput(s *terminalSession, p []byte) bool { return false default: } + if c.beforeInputRecheck != nil { + c.beforeInputRecheck() + } + if !c.pollWritable(s.ptmx, terminalInputPollWindow) { + s.inputMu.Unlock() + continue + } + n := min(len(p), terminalInputChunk) written, err := s.ptmx.Write(p[:n]) s.inputMu.Unlock() if err != nil || written == 0 { @@ -615,33 +652,40 @@ func (c *localTerminalConn) writeInput(s *terminalSession, p []byte) bool { } func (c *localTerminalConn) waitWritable(f *os.File) bool { - rc, err := f.SyscallConn() - if err != nil { - c.kill() - return false - } for { select { case <-c.dead: return false default: } - ready := false - err := rc.Control(func(fd uintptr) { - pfd := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLOUT}} - n, pollErr := unix.Poll(pfd, int(terminalInputPollWindow/time.Millisecond)) - ready = pollErr == nil && n > 0 && pfd[0].Revents&unix.POLLOUT != 0 - }) - if err != nil { - c.kill() - return false - } - if ready { + if c.pollWritable(f, terminalInputPollWindow) { return true } } } +// pollWritable performs exactly one bounded readiness check. It is called once +// before inputMu and again while holding it; the latter closes the interval in +// which a browser's normal master write could otherwise consume readiness. +func (c *localTerminalConn) pollWritable(f *os.File, timeout time.Duration) bool { + rc, err := f.SyscallConn() + if err != nil { + c.kill() + return false + } + ready := false + err = rc.Control(func(fd uintptr) { + pfd := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLOUT}} + n, pollErr := unix.Poll(pfd, int(timeout/time.Millisecond)) + ready = pollErr == nil && n > 0 && pfd[0].Revents&unix.POLLOUT != 0 + }) + if err != nil { + c.kill() + return false + } + return ready +} + func (d *system) handleTerminal(stream net.Conn, br *bufio.Reader, h relay.StreamHeader, actionDeadline time.Time) { if h.SessionID == "" { d.logf("terminal refused: missing session id") diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index db416ee..de110da 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -13,6 +13,8 @@ import ( "path/filepath" "strconv" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -736,6 +738,186 @@ func TestLocalTerminalWriteBoundsInputByBytes(t *testing.T) { } } +func TestLocalTerminalWriteReservesCallsAndBytesBeforeCopy(t *testing.T) { + inputRead, inputWrite, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) + fd := int(inputWrite.Fd()) + if err := unix.SetNonblock(fd, true); err != nil { + t.Fatal(err) + } + for { + _, err := unix.Write(fd, make([]byte, 4<<10)) + if err == unix.EAGAIN { + break + } + if err != nil { + t.Fatal(err) + } + } + if err := unix.SetNonblock(fd, false); err != nil { + t.Fatal(err) + } + s := &terminalSession{ptmx: inputWrite, done: make(chan struct{})} + conn := newLocalTerminalConn() + go conn.inputLoop(s) + client := &TerminalClient{session: s, conn: conn} + var copied atomic.Int64 + conn.copyInput = func(p []byte) []byte { + copied.Add(1) + return append([]byte(nil), p...) + } + if err := client.Write([]byte("x")); err != nil { + t.Fatalf("initial in-flight input: %v", err) + } + deadline := time.Now().Add(time.Second) + for len(conn.input) != 0 { + if time.Now().After(deadline) { + t.Fatal("input worker did not take the in-flight call") + } + time.Sleep(time.Millisecond) + } + var accepted atomic.Int64 + accepted.Add(1) // the worker holds this first call in the exact bound. + var group sync.WaitGroup + for range 2 * terminalInputQueue { + group.Add(1) + go func() { + defer group.Done() + if client.Write([]byte("x")) == nil { + accepted.Add(1) + } + }() + } + group.Wait() + if got := accepted.Load(); got != terminalInputQueue { + t.Fatalf("accepted local input calls = %d, want exact bound %d", got, terminalInputQueue) + } + if got := copied.Load(); got != accepted.Load() { + t.Fatalf("implementation copies = %d, accepted calls = %d; rejected calls allocated", got, accepted.Load()) + } + if got := conn.inputCalls.Load(); got != terminalInputQueue { + t.Fatalf("reserved local input calls = %d, want %d", got, terminalInputQueue) + } + if got := conn.inputBytes.Load(); got != terminalInputQueue { + t.Fatalf("reserved local input bytes = %d, want %d", got, terminalInputQueue) + } + conn.kill() + <-conn.inputDone +} + +func TestLocalTerminalWriteRejectsOversizedInputWithoutCopy(t *testing.T) { + conn := newLocalTerminalConn() + client := &TerminalClient{conn: conn} + var copied atomic.Int64 + conn.copyInput = func(p []byte) []byte { + copied.Add(1) + return append([]byte(nil), p...) + } + err := client.Write(make([]byte, min(relay.MaxFrameSize, terminalInputBytes)+1)) + if !errors.Is(err, ErrTerminalInputTooLarge) { + t.Fatalf("oversized local input error = %v, want ErrTerminalInputTooLarge", err) + } + if errors.Is(err, ErrTerminalInputBackpressure) { + t.Fatal("oversized input matched retryable ErrTerminalInputBackpressure") + } + if got := copied.Load(); got != 0 { + t.Fatalf("oversized input made %d implementation copies", got) + } + conn.kill() +} + +func TestLocalTerminalWriteAdmissionIsOrderedWithDetach(t *testing.T) { + conn := newLocalTerminalConn() + client := &TerminalClient{conn: conn} + entered := make(chan struct{}) + release := make(chan struct{}) + conn.beforeInputCopy = func() { + close(entered) + <-release + } + writeDone := make(chan error, 1) + go func() { writeDone <- client.Write([]byte("accepted before detach")) }() + <-entered // reservation and the inputMu ownership are now established. + detached := make(chan struct{}) + go func() { conn.kill(); close(detached) }() + select { + case <-detached: + t.Fatal("detach passed a Write already holding admission") + case <-time.After(25 * time.Millisecond): + } + close(release) + if err := <-writeDone; err != nil { + t.Fatalf("Write admitted before detach = %v", err) + } + <-detached + if err := client.Write([]byte("after detach")); !errors.Is(err, ErrTerminalClientClosed) { + t.Fatalf("Write after detach = %v, want ErrTerminalClientClosed", err) + } +} + +// A browser can consume POLLOUT after the local worker's first poll. The second +// poll must run while inputMu is held, or the following <= PIPE_BUF write can +// block under that lock. A real pipe supplies the readiness semantics; the +// callback represents the sibling master writer in the stale interval. +func TestLocalTerminalRechecksWritableWhileHoldingInputLock(t *testing.T) { + inputRead, inputWrite, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) + fd := int(inputWrite.Fd()) + if err := unix.SetNonblock(fd, true); err != nil { + t.Fatal(err) + } + for { + _, err := unix.Write(fd, make([]byte, 4<<10)) + if err == unix.EAGAIN { + break + } + if err != nil { + t.Fatal(err) + } + } + if err := unix.SetNonblock(fd, false); err != nil { + t.Fatal(err) + } + if _, err := io.ReadFull(inputRead, make([]byte, 4<<10)); err != nil { + t.Fatalf("make pipe writable for initial poll: %v", err) + } + s := &terminalSession{ptmx: inputWrite, done: make(chan struct{})} + conn := newLocalTerminalConn() + consumed := make(chan struct{}) + var once sync.Once + conn.beforeInputRecheck = func() { + once.Do(func() { + if _, err := unix.Write(fd, make([]byte, 4<<10)); err != nil { + t.Errorf("consume stale POLLOUT: %v", err) + } + close(consumed) + }) + } + go conn.inputLoop(s) + client := &TerminalClient{session: s, conn: conn} + if err := client.Write([]byte("x")); err != nil { + t.Fatal(err) + } + <-consumed + time.Sleep(2 * terminalInputPollWindow) + if !s.inputMu.TryLock() { + t.Fatal("stale POLLOUT let local input hold inputMu in a blocking write") + } + s.inputMu.Unlock() + conn.kill() + select { + case <-conn.inputDone: + case <-time.After(time.Second): + t.Fatal("detach did not release stale-readiness input worker") + } +} + // The browser sibling above proves the same bounded queue design over a sealed // transport. This local control proves that an unread local output channel also // cannot park the PTY read path and is detached at terminalWriteTO. From a950de6ffe69f110e75fb637d417bc88df78f505 Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 16:31:33 -0600 Subject: [PATCH 4/7] fix(system): serialize nonblocking PTY input --- internal/system/terminal_input_linux_test.go | 239 +++++++++ internal/system/terminal_sessions.go | 521 +++++++++++++------ internal/system/terminal_sessions_test.go | 397 +++++++++++--- 3 files changed, 937 insertions(+), 220 deletions(-) create mode 100644 internal/system/terminal_input_linux_test.go diff --git a/internal/system/terminal_input_linux_test.go b/internal/system/terminal_input_linux_test.go new file mode 100644 index 0000000..0dc8b95 --- /dev/null +++ b/internal/system/terminal_input_linux_test.go @@ -0,0 +1,239 @@ +//go:build linux && !android + +package system + +import ( + "bytes" + "errors" + "os" + "testing" + "time" + + "github.com/creack/pty" + "github.com/nicodes/ormos/relay" + "golang.org/x/sys/unix" +) + +const terminalInputSchedulerAllowance = 100 * time.Millisecond + +// This is the kernel-level shutdown/progress case: the slave is raw and does +// not echo, the nonblocking master is filled to EAGAIN, and no shell drains it. +// Detach gets one 25ms POLLOUT window plus 100ms of explicit scheduler allowance +// to release the active 1MiB local request; the session itself must stay alive. +func TestRealPTYDetachReleasesMaximalLocalInputAndBrowserProgresses(t *testing.T) { + ptmx, tty := rawNonblockingPTYPair(t) + fillPTYToEAGAIN(t, ptmx) + s := newInputTestSession(ptmx) + localConn := newLocalTerminalConn() + if err := s.attachConn(localConn); err != nil { + t.Fatal(err) + } + local := &TerminalClient{session: s, conn: localConn} + admitted := make(chan error, 1) + go func() { admitted <- local.Write(make([]byte, terminalInputBytes)) }() + select { + case err := <-admitted: + if err != nil { + t.Fatalf("admit maximal local request: %v", err) + } + case <-time.After(time.Second): + t.Fatal("maximal local admission blocked on a full PTY") + } + waitSessionInput(t, s, 1, 0, time.Second) + + local.Detach() + waitInputAccounting(t, s, localConn, 0, 0, terminalInputPollWindow+terminalInputSchedulerAllowance) + select { + case <-s.done: + t.Fatal("local detach closed the shared terminal session") + default: + } + + browser, browserClient := sealedPair(t, "real-pty-detach") + go s.attach(browser) + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatal(err) + } + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatal(err) + } + marker := []byte("[[browser-after-detach-7d3f]]") + if err := browserClient.WriteFrame(relay.EncodeData(marker)); err != nil { + t.Fatal(err) + } + readPTYUntil(t, tty, marker, 2*time.Second) + waitSessionInput(t, s, 0, 0, time.Second) + browser.kill() + s.close() +} + +// A request is returned to the queue tail after no more than 4096 bytes. The +// permit-driven 512-byte slave reads provide capacity without sleep timing: the +// browser marker admitted behind the active local request must therefore occur +// after its prefix but no more than one exact fairness quantum into it, and +// before its unique tail. +func TestRealPTYInputFairnessAt4096ByteQuantum(t *testing.T) { + ptmx, tty := rawNonblockingPTYPair(t) + fillPTYToEAGAIN(t, ptmx) + s := newInputTestSession(ptmx) + localConn := newLocalTerminalConn() + local := &TerminalClient{session: s, conn: localConn} + localBegin := []byte("[[local-begin-29ac]]") + localTail := []byte("[[local-tail-e841]]") + localInput := append(append(append([]byte(nil), localBegin...), bytes.Repeat([]byte{'L'}, 64<<10)...), localTail...) + if err := local.Write(localInput); err != nil { + t.Fatal(err) + } + waitSessionInput(t, s, 1, 0, time.Second) + + browser, browserClient := sealedPair(t, "real-pty-fairness") + go s.attach(browser) + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatal(err) + } + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatal(err) + } + browserMarker := []byte("[[browser-fair-54b2]]") + if err := browserClient.WriteFrame(relay.EncodeData(browserMarker)); err != nil { + t.Fatal(err) + } + waitSessionInput(t, s, 2, 1, time.Second) + + permits := make(chan struct{}) + chunks := make(chan []byte) + go func() { + buf := make([]byte, 512) + for range permits { + n, err := tty.Read(buf) + if err != nil { + close(chunks) + return + } + chunks <- append([]byte(nil), buf[:n]...) + } + }() + defer close(permits) + deadline := time.Now().Add(3 * time.Second) + var seen []byte + for !bytes.Contains(seen, localTail) { + select { + case permits <- struct{}{}: + case <-time.After(time.Until(deadline)): + t.Fatal("deadline granting controlled PTY drain permit") + } + select { + case chunk, ok := <-chunks: + if !ok { + t.Fatal("PTY slave read ended before local tail") + } + seen = append(seen, chunk...) + case <-time.After(time.Until(deadline)): + t.Fatal("deadline draining PTY input") + } + } + beginAt := bytes.Index(seen, localBegin) + browserAt := bytes.Index(seen, browserMarker) + tailAt := bytes.Index(seen, localTail) + if beginAt < 0 || browserAt < 0 || tailAt < 0 { + t.Fatalf("missing delimiters: begin=%d browser=%d tail=%d", beginAt, browserAt, tailAt) + } + if browserAt <= beginAt || browserAt >= tailAt { + t.Fatalf("browser marker order begin/browser/tail = %d/%d/%d", beginAt, browserAt, tailAt) + } + if beforeBrowser := browserAt - beginAt; beforeBrowser > 4<<10 { + t.Fatalf("browser marker followed %d local bytes, exceeds exact 4096-byte quantum", beforeBrowser) + } + waitInputAccounting(t, s, localConn, 0, 0, time.Second) + browser.kill() + localConn.kill() + s.close() +} + +func rawNonblockingPTYPair(t *testing.T) (*os.File, *os.File) { + t.Helper() + ptmx, tty, err := pty.Open() + if err != nil { + t.Skipf("no PTY available: %v", err) + } + termios, err := unix.IoctlGetTermios(int(tty.Fd()), unix.TCGETS) + if err != nil { + t.Fatal(err) + } + raw := *termios + raw.Iflag &^= unix.IGNBRK | unix.BRKINT | unix.PARMRK | unix.ISTRIP | unix.INLCR | unix.IGNCR | unix.ICRNL | unix.IXON + raw.Oflag &^= unix.OPOST + raw.Lflag &^= unix.ECHO | unix.ECHONL | unix.ICANON | unix.ISIG | unix.IEXTEN + raw.Cflag &^= unix.CSIZE | unix.PARENB + raw.Cflag |= unix.CS8 + raw.Cc[unix.VMIN] = 1 + raw.Cc[unix.VTIME] = 0 + if err := unix.IoctlSetTermios(int(tty.Fd()), unix.TCSETS, &raw); err != nil { + t.Fatal(err) + } + ptmx, err = normalizePTY(ptmx) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = tty.Close(); _ = ptmx.Close() }) + return ptmx, tty +} + +func fillPTYToEAGAIN(t *testing.T, ptmx *os.File) int { + t.Helper() + total := 0 + block := bytes.Repeat([]byte{'F'}, 4<<10) + for { + n, err := ptyWrite(ptmx, block) + total += n + if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EWOULDBLOCK) { + if total == 0 { + t.Fatal("PTY reached EAGAIN before accepting fill data") + } + return total + } + if err != nil { + t.Fatal(err) + } + } +} + +func newInputTestSession(ptmx *os.File) *terminalSession { + s := &terminalSession{ptmx: ptmx, conns: make(map[terminalConn]struct{}), input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s.startInput() + return s +} + +func readPTYUntil(t *testing.T, tty *os.File, marker []byte, timeout time.Duration) { + t.Helper() + chunks := make(chan []byte) + go func() { + buf := make([]byte, 4<<10) + for { + n, err := tty.Read(buf) + if n > 0 { + chunks <- append([]byte(nil), buf[:n]...) + } + if err != nil { + close(chunks) + return + } + } + }() + deadline := time.After(timeout) + var seen []byte + for { + select { + case chunk, ok := <-chunks: + if !ok { + t.Fatal("PTY slave closed before browser marker") + } + seen = append(seen, chunk...) + if bytes.Contains(seen, marker) { + return + } + case <-deadline: + t.Fatalf("browser marker not observed within %s", timeout) + } + } +} diff --git a/internal/system/terminal_sessions.go b/internal/system/terminal_sessions.go index 5a3b623..8b56d64 100644 --- a/internal/system/terminal_sessions.go +++ b/internal/system/terminal_sessions.go @@ -89,10 +89,7 @@ const ( terminalInputQueue = terminalSendQueue terminalInputBytes = terminalSendBytes terminalInputPollWindow = 25 * time.Millisecond - // PIPE_BUF is 4096 on Linux and 512 on Darwin. Keeping a write within the - // portable 512-byte floor lets the locked POLLOUT recheck establish that the - // subsequent write will not wait for a sibling master writer. - terminalInputChunk = 512 + terminalInputQuantum = 4 << 10 ) var ( @@ -209,15 +206,32 @@ type terminalSession struct { cmd *exec.Cmd pgid int // the shell's process group id (it leads it: pty.Start sets Setsid) - mu sync.Mutex - inputMu sync.Mutex - conns map[terminalConn]struct{} - replay replayRing - active bool - closed bool - expires *time.Timer - done chan struct{} - closeOnce sync.Once + mu sync.Mutex + inputMu sync.Mutex + input chan *terminalInput + inputBytes int + inputCalls int + inputClosed bool + inputOnce sync.Once + inputStarted chan struct{} + inputStopped chan struct{} + conns map[terminalConn]struct{} + replay replayRing + active bool + closed bool + expires *time.Timer + done chan struct{} + closeOnce sync.Once +} + +// terminalInput is one accepted browser or local request. The scheduler writes +// at most terminalInputQuantum bytes before returning an unfinished request to +// the tail, so FIFO admission cannot let a large local paste monopolise a +// slowly-draining PTY ahead of browser keystrokes. +type terminalInput struct { + p []byte + off int + local *localTerminalConn } // terminalConn is the common outbound half of a browser or local attachment. @@ -418,18 +432,12 @@ func (c *TerminalClient) Write(p []byte) error { c.conn.beforeInputCopy() } copy := c.conn.copyInput(p) - select { - case <-c.conn.dead: - c.conn.inputCalls.Add(-1) - c.conn.inputBytes.Add(-int64(len(copy))) - return ErrTerminalClientClosed - case c.conn.input <- copy: - return nil - default: + if !c.session.submitInput(&terminalInput{p: copy, local: c.conn}) { c.conn.inputCalls.Add(-1) c.conn.inputBytes.Add(-int64(len(copy))) return ErrTerminalInputBackpressure } + return nil } // Resize updates the shared PTY's dimensions. @@ -442,7 +450,7 @@ func (c *TerminalClient) Resize(cols, rows int) error { return fmt.Errorf("terminal attachment is closed") default: } - return pty.Setsize(c.session.ptmx, &pty.Winsize{Cols: uint16(cols), Rows: uint16(rows)}) + return setPTYSize(c.session.ptmx, cols, rows) } // Detach releases this local attachment without ending its shared shell. @@ -465,7 +473,6 @@ func (d *system) AttachTerminal(projectRoot, sessionID string, cols, rows int) ( conn.kill() return nil, err } - go conn.inputLoop(s) go func() { <-conn.dead s.detach(conn) @@ -484,15 +491,12 @@ type localTerminalConn struct { dead chan struct{} killOne sync.Once - input chan []byte inputCalls atomic.Int64 inputBytes atomic.Int64 inputMu sync.Mutex // serialises local Write acceptance order - inputDone chan struct{} copyInput func([]byte) []byte // Test seams; nil in production. - beforeInputCopy func() - beforeInputRecheck func() + beforeInputCopy func() } func newLocalTerminalConn() *localTerminalConn { @@ -502,8 +506,6 @@ func newLocalTerminalConn() *localTerminalConn { // before returning without creating a second, unaccounted output backlog. output: make(chan []byte, 1), dead: make(chan struct{}), - input: make(chan []byte, terminalInputQueue), - inputDone: make(chan struct{}), copyInput: func(p []byte) []byte { return append([]byte(nil), p...) }, } go c.writeLoop() @@ -581,111 +583,6 @@ func (c *localTerminalConn) kill() { c.inputMu.Unlock() } -// inputLoop is one worker per attachment, never one per Write. It waits for -// POLLOUT before entering the session's shared input critical section, so a -// full local PTY cannot pin inputMu and starve a browser. poll is available on -// both supported Unix targets; its short timeout makes waiting detach-aware -// without claiming that an already-entered kernel Write can be cancelled. -func (c *localTerminalConn) inputLoop(s *terminalSession) { - defer func() { - for { - select { - case p := <-c.input: - c.inputCalls.Add(-1) - c.inputBytes.Add(-int64(len(p))) - default: - close(c.inputDone) - return - } - } - }() - for { - select { - case <-c.dead: - return - case p := <-c.input: - if !c.writeInput(s, p) { - c.inputCalls.Add(-1) - c.inputBytes.Add(-int64(len(p))) - return - } - c.inputCalls.Add(-1) - c.inputBytes.Add(-int64(len(p))) - } - } -} - -func (c *localTerminalConn) writeInput(s *terminalSession, p []byte) bool { - for len(p) > 0 { - if !c.waitWritable(s.ptmx) { - return false - } - select { - case <-c.dead: - return false - default: - } - s.inputMu.Lock() - select { - case <-c.dead: - s.inputMu.Unlock() - return false - default: - } - if c.beforeInputRecheck != nil { - c.beforeInputRecheck() - } - if !c.pollWritable(s.ptmx, terminalInputPollWindow) { - s.inputMu.Unlock() - continue - } - n := min(len(p), terminalInputChunk) - written, err := s.ptmx.Write(p[:n]) - s.inputMu.Unlock() - if err != nil || written == 0 { - c.kill() - return false - } - p = p[written:] - } - return true -} - -func (c *localTerminalConn) waitWritable(f *os.File) bool { - for { - select { - case <-c.dead: - return false - default: - } - if c.pollWritable(f, terminalInputPollWindow) { - return true - } - } -} - -// pollWritable performs exactly one bounded readiness check. It is called once -// before inputMu and again while holding it; the latter closes the interval in -// which a browser's normal master write could otherwise consume readiness. -func (c *localTerminalConn) pollWritable(f *os.File, timeout time.Duration) bool { - rc, err := f.SyscallConn() - if err != nil { - c.kill() - return false - } - ready := false - err = rc.Control(func(fd uintptr) { - pfd := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLOUT}} - n, pollErr := unix.Poll(pfd, int(timeout/time.Millisecond)) - ready = pollErr == nil && n > 0 && pfd[0].Revents&unix.POLLOUT != 0 - }) - if err != nil { - c.kill() - return false - } - return ready -} - func (d *system) handleTerminal(stream net.Conn, br *bufio.Reader, h relay.StreamHeader, actionDeadline time.Time) { if h.SessionID == "" { d.logf("terminal refused: missing session id") @@ -826,6 +723,18 @@ func (d *system) terminal(h relay.StreamHeader, actionDeadline time.Time) (*term if err != nil { return nil, fmt.Errorf("start pty: %w", err) } + // pty.Start's master is not uniformly pollable: creack/pty opens it through + // os.OpenFile on Linux but wraps a blocking descriptor with os.NewFile on + // Darwin, and its startup ioctls call File.Fd. Duplicate it, make the new + // descriptor nonblocking before os.NewFile sees it, and retain only that + // runtime-pollable wrapper. No later operation calls File.Fd. + ptmx, err = normalizePTY(ptmx) + if err != nil { + _ = ptmx.Close() + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + return nil, fmt.Errorf("normalise pty master: %w", err) + } // pty.Start runs the shell with Setsid, so it leads a new session and its // process group id equals its pid. Capture the pgid now — close needs it // to signal the whole group, and by the time close runs the shell may @@ -839,13 +748,19 @@ func (d *system) terminal(h relay.StreamHeader, actionDeadline time.Time) (*term } // Unconditional: the header's dimensions were validated at the top of this // function, so there is no reachable case where they are zero. - _ = pty.Setsize(ptmx, &pty.Winsize{Cols: uint16(h.Cols), Rows: uint16(h.Rows)}) - s := &terminalSession{id: h.SessionID, owner: d, cwd: cwd, ptmx: ptmx, cmd: cmd, pgid: pgid, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} + if err := setPTYSize(ptmx, h.Cols, h.Rows); err != nil { + _ = ptmx.Close() + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + return nil, fmt.Errorf("set pty size: %w", err) + } + s := &terminalSession{id: h.SessionID, owner: d, cwd: cwd, ptmx: ptmx, cmd: cmd, pgid: pgid, conns: make(map[terminalConn]struct{}), input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} d.terminals[h.SessionID] = s d.addSession(1) d.logf("terminal session started (%s)", d.cfg.Shell) go s.readPTY() go s.pollActivity() + s.startInput() return s, nil } @@ -891,18 +806,15 @@ func (s *terminalSession) attach(conn *sealedConn) { } switch { case frame.Data != nil: - s.inputMu.Lock() - if _, err := s.ptmx.Write(frame.Data); err != nil { - s.inputMu.Unlock() + if !s.submitInput(&terminalInput{p: frame.Data}) { s.detach(conn) return } - s.inputMu.Unlock() case frame.Resize != nil: if !relay.ValidTerminalSize(frame.Resize.Cols, frame.Resize.Rows) { continue } - _ = pty.Setsize(s.ptmx, &pty.Winsize{Cols: uint16(frame.Resize.Cols), Rows: uint16(frame.Resize.Rows)}) + _ = setPTYSize(s.ptmx, frame.Resize.Cols, frame.Resize.Rows) } } s.detach(conn) @@ -935,7 +847,7 @@ func (s *terminalSession) attachConnSize(conn terminalConn, size *pty.Winsize) e return fmt.Errorf("session already has %d connections", maxTerminalConns) } if size != nil { - if err := pty.Setsize(s.ptmx, size); err != nil { + if err := setPTYSize(s.ptmx, int(size.Cols), int(size.Rows)); err != nil { s.mu.Unlock() return fmt.Errorf("set terminal size: %w", err) } @@ -983,6 +895,142 @@ func (s *terminalSession) detach(conn terminalConn) { } } +// submitInput is nonblocking. Browser readers detach on a full session queue; +// local callers roll back their per-client admission and return backpressure. +func (s *terminalSession) submitInput(in *terminalInput) bool { + s.startInput() + s.inputMu.Lock() + defer s.inputMu.Unlock() + if s.inputClosed || s.inputCalls >= terminalInputQueue || s.inputBytes+len(in.p) > terminalInputBytes { + return false + } + select { + case s.input <- in: + s.inputCalls++ + s.inputBytes += len(in.p) + return true + default: + return false + } +} + +func (s *terminalSession) startInput() { + s.inputOnce.Do(func() { + if s.input == nil { + s.input = make(chan *terminalInput, terminalInputQueue) + } + s.inputStarted = make(chan struct{}) + s.inputStopped = make(chan struct{}) + go s.inputLoop() + }) +} + +func (s *terminalSession) releaseInput(in *terminalInput) { + n := len(in.p) + s.inputMu.Lock() + s.inputCalls-- + s.inputBytes -= n + s.inputMu.Unlock() + if in.local != nil { + in.local.inputCalls.Add(-1) + in.local.inputBytes.Add(-int64(n)) + } +} + +func (s *terminalSession) inputLoop() { + close(s.inputStarted) + defer close(s.inputStopped) + for { + select { + case <-s.done: + s.drainInput() + return + case in := <-s.input: + if in.local != nil { + select { + case <-in.local.dead: + s.releaseInput(in) + continue + default: + } + } + complete, err := s.writeInputQuantum(in) + if err != nil { + s.releaseInput(in) + s.stopInput() + s.drainInput() + go s.close() + return + } + if complete { + s.releaseInput(in) + continue + } + if in.off == len(in.p) { + s.releaseInput(in) + continue + } + select { + case <-s.done: + s.releaseInput(in) + case s.input <- in: + // Keep accounting: the request remains admitted at the queue tail. + // terminalInputQueue counts the active request, so at most 255 + // others can be queued while this one is out of the channel. The + // active request therefore always owns this requeue slot. + } + } + } +} + +func (s *terminalSession) drainInput() { + for { + select { + case in := <-s.input: + s.releaseInput(in) + default: + return + } + } +} + +// writeInputQuantum is the sole PTY write path. It uses the permanently +// nonblocking descriptor directly, then waits no longer than 25ms for POLLOUT. +func (s *terminalSession) writeInputQuantum(in *terminalInput) (complete bool, err error) { + limit := min(len(in.p), in.off+terminalInputQuantum) + deadline := time.Now().Add(terminalInputPollWindow) + for in.off < limit { + select { + case <-s.done: + return true, nil + default: + } + if in.local != nil { + select { + case <-in.local.dead: + return true, nil + default: + } + } + written, err := ptyWrite(s.ptmx, in.p[in.off:limit]) + if written > 0 { + in.off += written + continue + } + if err != unix.EAGAIN && err != unix.EWOULDBLOCK { + return true, fmt.Errorf("write pty: %w", err) + } + ready, err := ptyWritable(s.ptmx, deadline) + if err != nil { + return true, err + } + if !ready { + return false, nil + } + } + return in.off == len(in.p), nil +} + func (s *terminalSession) readPTY() { r, err := newPTYReader(s.ptmx) if err != nil { @@ -1003,6 +1051,129 @@ func (s *terminalSession) readPTY() { } } +func setPTYNonblock(f *os.File) error { + rc, err := f.SyscallConn() + if err != nil { + return err + } + var opErr error + if err := rc.Control(func(fd uintptr) { opErr = unix.SetNonblock(int(fd), true) }); err != nil { + return err + } + return opErr +} + +func setPTYSize(f *os.File, cols, rows int) error { + rc, err := f.SyscallConn() + if err != nil { + return err + } + var opErr error + if err := rc.Control(func(fd uintptr) { + opErr = unix.IoctlSetWinsize(int(fd), unix.TIOCSWINSZ, &unix.Winsize{Col: uint16(cols), Row: uint16(rows)}) + }); err != nil { + return err + } + return opErr +} + +// normalizePTY returns a separately owned wrapper whose descriptor was already +// O_NONBLOCK when os.NewFile registered it with Go's poller. Wrapping the same +// descriptor would create two owners; duplicating first makes closing the +// creack/pty wrapper safe and keeps RawConn.Read deadline-capable on both Linux +// and Darwin. +func normalizePTY(f *os.File) (*os.File, error) { + rc, err := f.SyscallConn() + if err != nil { + return f, err + } + dup := -1 + var opErr error + if err := rc.Control(func(fd uintptr) { + dup, opErr = unix.FcntlInt(fd, unix.F_DUPFD_CLOEXEC, 0) + if opErr == nil { + opErr = unix.SetNonblock(dup, true) + } + }); err != nil { + return f, err + } + if opErr != nil { + if dup >= 0 { + _ = unix.Close(dup) + } + return f, opErr + } + normalized := os.NewFile(uintptr(dup), f.Name()) + if normalized == nil { + _ = unix.Close(dup) + return f, fmt.Errorf("wrap duplicated pty descriptor") + } + if err := normalized.SetReadDeadline(time.Time{}); err != nil { + _ = normalized.Close() + return f, fmt.Errorf("register pty with runtime poller: %w", err) + } + if err := f.Close(); err != nil { + _ = normalized.Close() + return f, err + } + return normalized, nil +} + +func ptyWrite(f *os.File, p []byte) (n int, err error) { + rc, err := f.SyscallConn() + if err != nil { + return 0, err + } + var opErr error + if err := rc.Control(func(fd uintptr) { + for { + n, opErr = unix.Write(int(fd), p) + if opErr != unix.EINTR { + return + } + } + }); err != nil { + return 0, err + } + return n, opErr +} + +func ptyWritable(f *os.File, deadline time.Time) (bool, error) { + rc, err := f.SyscallConn() + if err != nil { + return false, err + } + ready := false + var opErr error + err = rc.Control(func(fd uintptr) { + pfd := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLOUT}} + for { + left := time.Until(deadline) + if left <= 0 { + return + } + n, pollErr := unix.Poll(pfd, int((left+time.Millisecond-1)/time.Millisecond)) + if pollErr == unix.EINTR { + continue + } + if pollErr != nil { + opErr = pollErr + return + } + if n > 0 && pfd[0].Revents&(unix.POLLERR|unix.POLLHUP|unix.POLLNVAL) != 0 { + opErr = fmt.Errorf("pty poll terminal revents %#x", pfd[0].Revents) + return + } + ready = n > 0 && pfd[0].Revents&unix.POLLOUT != 0 + return + } + }) + if err != nil { + return false, err + } + return ready, opErr +} + // ptyReader reads coalesced chunks from a PTY master. // // A Linux PTY master hands back at most one tty chunk per read, and one read @@ -1046,7 +1217,7 @@ func newPTYReader(f *os.File) (*ptyReader, error) { // leaves without waiting a nanosecond, and bulk output gets the full window // whatever the host is doing. func (p *ptyReader) read(buf []byte) (n, reads int, err error) { - n, err = p.f.Read(buf) + n, err = p.readOnce(buf) reads = 1 if err != nil || n < terminalCoalesceMinChunk || p.blind { return n, reads, err @@ -1056,7 +1227,7 @@ func (p *ptyReader) read(buf []byte) (n, reads int, err error) { if !p.readable(deadline) { return n, reads, nil } - m, rerr := p.f.Read(buf[n:]) + m, rerr := p.readOnce(buf[n:]) reads++ n += m if rerr != nil { @@ -1066,6 +1237,23 @@ func (p *ptyReader) read(buf []byte) (n, reads int, err error) { return n, reads, nil } +func (p *ptyReader) readOnce(buf []byte) (n int, err error) { + var opErr error + err = p.rc.Read(func(fd uintptr) bool { + for { + n, opErr = unix.Read(int(fd), buf) + if opErr != unix.EINTR { + break + } + } + return opErr != unix.EAGAIN && opErr != unix.EWOULDBLOCK + }) + if err != nil { + return n, err + } + return n, opErr +} + // readable reports whether the master has bytes ready, waiting no later than // deadline. // @@ -1205,7 +1393,7 @@ func (s *terminalSession) pollActivity() { case <-s.done: return case <-ticker.C: - fg, err := unix.IoctlGetInt(int(s.ptmx.Fd()), unix.TIOCGPGRP) + fg, err := ptyForegroundPgrp(s.ptmx) if err != nil { return } @@ -1214,6 +1402,18 @@ func (s *terminalSession) pollActivity() { } } +func ptyForegroundPgrp(f *os.File) (fg int, err error) { + rc, err := f.SyscallConn() + if err != nil { + return 0, err + } + var opErr error + if err := rc.Control(func(fd uintptr) { fg, opErr = unix.IoctlGetInt(int(fd), unix.TIOCGPGRP) }); err != nil { + return 0, err + } + return fg, opErr +} + func (s *terminalSession) setActivity(active bool) { s.mu.Lock() // Before the encode, not after: pollActivity ticks every 700 ms for the @@ -1245,6 +1445,9 @@ func (s *terminalSession) setActivity(active bool) { // closed anything that matters. func (s *terminalSession) killProcessGroup() { if s.cmd == nil || s.cmd.Process == nil { + if s.ptmx != nil { + _ = s.ptmx.Close() + } return } if s.pgid > 0 { @@ -1279,6 +1482,7 @@ func (s *terminalSession) killProcessGroup() { func (s *terminalSession) close() { s.closeOnce.Do(func() { + s.stopInput() s.mu.Lock() s.closed = true conns := s.conns @@ -1295,14 +1499,27 @@ func (s *terminalSession) close() { conn.kill() } s.killProcessGroup() - close(s.done) - s.owner.terminalMu.Lock() - if s.owner.terminals[s.id] == s { - delete(s.owner.terminals, s.id) + if s.owner != nil { + s.owner.terminalMu.Lock() + if s.owner.terminals[s.id] == s { + delete(s.owner.terminals, s.id) + } + s.owner.terminalMu.Unlock() + s.owner.addSession(-1) + s.owner.logf("terminal session ended") } - s.owner.terminalMu.Unlock() - s.owner.addSession(-1) - s.owner.logf("terminal session ended") }) } + +// stopInput is the admission/worker half of close. Closing done before process +// teardown makes queued and currently polled requests release within one 25ms +// poll window, independently of the process-group grace period. +func (s *terminalSession) stopInput() { + s.inputMu.Lock() + if !s.inputClosed { + s.inputClosed = true + close(s.done) + } + s.inputMu.Unlock() +} diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index de110da..c9f84ac 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -6,6 +6,9 @@ import ( "bytes" "errors" "fmt" + "go/ast" + "go/parser" + "go/token" "io" "net" "os" @@ -618,7 +621,7 @@ func TestLocalTerminalDetachReleasesInputAndDoesNotStarveBrowser(t *testing.T) { t.Fatal(err) } filled := 0 - fill := make([]byte, terminalInputChunk) + fill := make([]byte, terminalInputQuantum) for { n, err := unix.Write(fd, fill) filled += n @@ -629,15 +632,12 @@ func TestLocalTerminalDetachReleasesInputAndDoesNotStarveBrowser(t *testing.T) { t.Fatal(err) } } - if err := unix.SetNonblock(fd, false); err != nil { - t.Fatal(err) - } - s := &terminalSession{id: "local-input", ptmx: inputWrite, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} + s := &terminalSession{id: "local-input", ptmx: inputWrite, conns: make(map[terminalConn]struct{}), input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s.startInput() localConn := newLocalTerminalConn() if err := s.attachConn(localConn); err != nil { t.Fatal(err) } - go localConn.inputLoop(s) local := &TerminalClient{session: s, conn: localConn} writeDone := make(chan error, 1) @@ -657,11 +657,7 @@ func TestLocalTerminalDetachReleasesInputAndDoesNotStarveBrowser(t *testing.T) { case <-time.After(time.Second): t.Fatal("local detach did not close Done") } - select { - case <-localConn.inputDone: - case <-time.After(time.Second): - t.Fatal("local input worker did not release queued input after detach") - } + waitInputAccounting(t, s, localConn, 0, 0, time.Second) readFD := int(inputRead.Fd()) if err := unix.SetNonblock(readFD, true); err != nil { t.Fatal(err) @@ -675,9 +671,6 @@ func TestLocalTerminalDetachReleasesInputAndDoesNotStarveBrowser(t *testing.T) { t.Fatal(err) } } - if err := unix.SetNonblock(readFD, false); err != nil { - t.Fatal(err) - } browser, browserClient := sealedPair(t, "local-input") go s.attach(browser) @@ -717,12 +710,12 @@ func TestLocalTerminalWriteBoundsInputByBytes(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) - s := &terminalSession{id: "local-input-bound", ptmx: inputWrite, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} + s := &terminalSession{id: "local-input-bound", ptmx: inputWrite, conns: make(map[terminalConn]struct{}), input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s.startInput() conn := newLocalTerminalConn() if err := s.attachConn(conn); err != nil { t.Fatal(err) } - go conn.inputLoop(s) client := &TerminalClient{session: s, conn: conn} if err := client.Write(make([]byte, terminalInputBytes)); err != nil { t.Fatalf("fill bounded local input queue: %v", err) @@ -731,11 +724,7 @@ func TestLocalTerminalWriteBoundsInputByBytes(t *testing.T) { t.Fatalf("input beyond byte budget error = %v, want ErrTerminalInputBackpressure", err) } client.Detach() - select { - case <-conn.inputDone: - case <-time.After(time.Second): - t.Fatal("detach did not release bounded local input worker") - } + waitInputAccounting(t, s, conn, 0, 0, time.Second) } func TestLocalTerminalWriteReservesCallsAndBytesBeforeCopy(t *testing.T) { @@ -757,12 +746,9 @@ func TestLocalTerminalWriteReservesCallsAndBytesBeforeCopy(t *testing.T) { t.Fatal(err) } } - if err := unix.SetNonblock(fd, false); err != nil { - t.Fatal(err) - } - s := &terminalSession{ptmx: inputWrite, done: make(chan struct{})} + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s.startInput() conn := newLocalTerminalConn() - go conn.inputLoop(s) client := &TerminalClient{session: s, conn: conn} var copied atomic.Int64 conn.copyInput = func(p []byte) []byte { @@ -773,7 +759,7 @@ func TestLocalTerminalWriteReservesCallsAndBytesBeforeCopy(t *testing.T) { t.Fatalf("initial in-flight input: %v", err) } deadline := time.Now().Add(time.Second) - for len(conn.input) != 0 { + for len(s.input) != 0 { if time.Now().After(deadline) { t.Fatal("input worker did not take the in-flight call") } @@ -805,7 +791,7 @@ func TestLocalTerminalWriteReservesCallsAndBytesBeforeCopy(t *testing.T) { t.Fatalf("reserved local input bytes = %d, want %d", got, terminalInputQueue) } conn.kill() - <-conn.inputDone + waitInputAccounting(t, s, conn, 0, 0, time.Second) } func TestLocalTerminalWriteRejectsOversizedInputWithoutCopy(t *testing.T) { @@ -830,8 +816,11 @@ func TestLocalTerminalWriteRejectsOversizedInputWithoutCopy(t *testing.T) { } func TestLocalTerminalWriteAdmissionIsOrderedWithDetach(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) conn := newLocalTerminalConn() - client := &TerminalClient{conn: conn} + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s.startInput() + client := &TerminalClient{session: s, conn: conn} entered := make(chan struct{}) release := make(chan struct{}) conn.beforeInputCopy = func() { @@ -858,64 +847,336 @@ func TestLocalTerminalWriteAdmissionIsOrderedWithDetach(t *testing.T) { } } -// A browser can consume POLLOUT after the local worker's first poll. The second -// poll must run while inputMu is held, or the following <= PIPE_BUF write can -// block under that lock. A real pipe supplies the readiness semantics; the -// callback represents the sibling master writer in the stale interval. -func TestLocalTerminalRechecksWritableWhileHoldingInputLock(t *testing.T) { - inputRead, inputWrite, err := os.Pipe() +func TestPTYMasterRemainsNonblockingAcrossTerminalIOCTLS(t *testing.T) { + ptmx, tty := ptyPair(t) + var err error + ptmx, err = normalizePTY(ptmx) if err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) - fd := int(inputWrite.Fd()) - if err := unix.SetNonblock(fd, true); err != nil { + t.Cleanup(func() { _ = ptmx.Close() }) + assertPTYNonblocking(t, ptmx, "normalization") + if err := setPTYSize(ptmx, 100, 40); err != nil { + t.Fatal(err) + } + assertPTYNonblocking(t, ptmx, "resize ioctl") + _, _ = ptyForegroundPgrp(ptmx) + assertPTYNonblocking(t, ptmx, "activity ioctl") + r, err := newPTYReader(ptmx) + if err != nil { + t.Fatal(err) + } + if err := ptmx.SetReadDeadline(time.Now().Add(25 * time.Millisecond)); err != nil { + t.Fatalf("normalized PTY does not support read deadlines: %v", err) + } + _, err = r.readOnce(make([]byte, 1)) + if !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("idle RawConn.Read error = %v, want os.ErrDeadlineExceeded (not ErrNoDeadline)", err) + } + _ = ptmx.SetReadDeadline(time.Time{}) + _ = tty +} + +func TestCreatedPTYRemainsNonblockingAcrossLocalAndBrowserResize(t *testing.T) { + withTempConfigDir(t) + d := newSystem(systemConfig{Shell: "/bin/cat"}) + local, err := d.AttachTerminal(t.TempDir(), "nonblock-resize", 80, 24) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { local.session.close() }) + <-local.Output() + assertPTYNonblocking(t, local.session.ptmx, "terminal creation and initial local resize") + if err := local.Resize(100, 40); err != nil { + t.Fatal(err) + } + assertPTYNonblocking(t, local.session.ptmx, "local Resize") + + browser, browserClient := sealedPair(t, "nonblock-resize") + go local.session.attach(browser) + if _, err := browserClient.ReadFrame(); err != nil { + t.Fatal(err) + } + if _, err := browserClient.ReadFrame(); err != nil { t.Fatal(err) } + if err := browserClient.WriteFrame(relay.EncodeResize(120, 50)); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(time.Second) for { - _, err := unix.Write(fd, make([]byte, 4<<10)) - if err == unix.EAGAIN { - break - } + rows, cols, err := rawPTYSize(local.session.ptmx) if err != nil { t.Fatal(err) } + if rows == 50 && cols == 120 { + break + } + if time.Now().After(deadline) { + t.Fatalf("browser resize left PTY at %dx%d", cols, rows) + } + time.Sleep(time.Millisecond) } - if err := unix.SetNonblock(fd, false); err != nil { - t.Fatal(err) - } - if _, err := io.ReadFull(inputRead, make([]byte, 4<<10)); err != nil { - t.Fatalf("make pipe writable for initial poll: %v", err) - } - s := &terminalSession{ptmx: inputWrite, done: make(chan struct{})} - conn := newLocalTerminalConn() - consumed := make(chan struct{}) - var once sync.Once - conn.beforeInputRecheck = func() { - once.Do(func() { - if _, err := unix.Write(fd, make([]byte, 4<<10)); err != nil { - t.Errorf("consume stale POLLOUT: %v", err) + assertPTYNonblocking(t, local.session.ptmx, "browser Resize") + _, _ = ptyForegroundPgrp(local.session.ptmx) + assertPTYNonblocking(t, local.session.ptmx, "activity ioctl") + browser.kill() +} + +func TestTerminalInputStartsExactlyOneWorkerAndCloseDrainsAccounting(t *testing.T) { + for _, tc := range []struct { + name string + currentBrowser bool + }{{name: "current-local-queued-browser"}, {name: "current-browser-queued-local", currentBrowser: true}} { + t.Run(tc.name, func(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + local := newLocalTerminalConn() + client := &TerminalClient{session: s, conn: local} + s.startInput() + <-s.inputStarted + for range 32 { + s.startInput() + } + select { + case <-s.inputStopped: + t.Fatal("the sole input worker exited before close") + default: + } + if tc.currentBrowser { + if !s.submitInput(&terminalInput{p: []byte("browser-current")}) { + t.Fatal("admit current browser input") + } + waitSessionInput(t, s, 1, 0, time.Second) + if err := client.Write([]byte("local-queued")); err != nil { + t.Fatal(err) + } + } else { + if err := client.Write([]byte("local-current")); err != nil { + t.Fatal(err) + } + waitSessionInput(t, s, 1, 0, time.Second) + if !s.submitInput(&terminalInput{p: []byte("browser-queued")}) { + t.Fatal("admit queued browser input") + } + } + // Keep a real backlog behind both kinds. This makes shutdown prove + // draining rather than winning a select between done and one item. + for i := range 32 { + if !s.submitInput(&terminalInput{p: []byte{byte(i)}}) { + t.Fatalf("admit queued browser request %d", i) + } + } + waitSessionInput(t, s, 34, 33, time.Second) + go s.close() + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("session close did not stop the input worker") } - close(consumed) + waitInputAccounting(t, s, local, 0, 0, time.Second) + s.close() + waitInputAccounting(t, s, local, 0, 0, time.Second) + local.kill() }) } - go conn.inputLoop(s) - client := &TerminalClient{session: s, conn: conn} - if err := client.Write([]byte("x")); err != nil { - t.Fatal(err) +} + +func TestTerminalInputCallBoundLeavesActiveRequestARequeueSlot(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + local := newLocalTerminalConn() + client := &TerminalClient{session: s, conn: local} + s.startInput() + <-s.inputStarted + for i := range terminalInputQueue { + if err := client.Write([]byte{byte(i)}); err != nil { + t.Fatalf("admit call %d of exact %d-call bound: %v", i+1, terminalInputQueue, err) + } + } + waitSessionInput(t, s, terminalInputQueue, terminalInputQueue-1, time.Second) + if got, want := cap(s.input), terminalInputQueue; got != want { + t.Fatalf("scheduler channel capacity = %d, want %d so the active request retains one requeue slot", got, want) } - <-consumed + // Let the active request time out and exercise its tail requeue before + // detach. A channel sized only for the other 255 calls deadlocks here. time.Sleep(2 * terminalInputPollWindow) - if !s.inputMu.TryLock() { - t.Fatal("stale POLLOUT let local input hold inputMu in a blocking write") + local.kill() + waitInputAccounting(t, s, local, 0, 0, terminalInputPollWindow+100*time.Millisecond) + s.close() +} + +func TestTerminalInputFatalWriteClosesOwnerlessSessionAndReleasesAccounting(t *testing.T) { + inputRead, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + local := newLocalTerminalConn() + client := &TerminalClient{session: s, conn: local} + s.startInput() + if err := client.Write([]byte("current")); err != nil { + t.Fatal(err) + } + waitSessionInput(t, s, 1, 0, time.Second) + if err := client.Write([]byte("queued")); err != nil { + t.Fatal(err) + } + if err := inputRead.Close(); err != nil { + t.Fatal(err) } - s.inputMu.Unlock() - conn.kill() select { - case <-conn.inputDone: + case <-s.inputStopped: case <-time.After(time.Second): - t.Fatal("detach did not release stale-readiness input worker") + t.Fatal("fatal PTY write/POLLERR did not stop ownerless session scheduler") + } + waitInputAccounting(t, s, local, 0, 0, time.Second) + select { + case <-s.done: + default: + t.Fatal("fatal PTY write/POLLERR did not converge ownerless session close") + } + local.kill() +} + +func TestTerminalInputHasOneRawPTYWriterAndOneScheduler(t *testing.T) { + fset := token.NewFileSet() + f, err := parser.ParseFile(fset, "terminal_sessions.go", nil, 0) + if err != nil { + t.Fatal(err) + } + rawWrites, schedulerStarts, onceStarts := 0, 0, 0 + ast.Inspect(f, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return true + } + if sel.Sel.Name == "Write" { + pkg, unixWrite := sel.X.(*ast.Ident) + if !unixWrite || pkg.Name != "unix" { + t.Errorf("direct terminal_sessions.go Write call at %s; PTY input must use ptyWrite", fset.Position(call.Pos())) + } else { + rawWrites++ + } + } + if sel.Sel.Name == "Fd" { + t.Errorf("terminal_sessions.go File.Fd call at %s; raw PTY operations must preserve runtime pollability", fset.Position(call.Pos())) + } + if sel.Sel.Name == "inputLoop" { + schedulerStarts++ + } + if sel.Sel.Name == "Do" { + if x, ok := sel.X.(*ast.SelectorExpr); ok && x.Sel.Name == "inputOnce" { + onceStarts++ + } + } + return true + }) + if rawWrites != 1 { + t.Fatalf("unix.Write PTY helpers = %d, want exactly one", rawWrites) + } + if schedulerStarts != 1 || onceStarts != 1 { + t.Fatalf("scheduler structure: inputLoop calls=%d inputOnce.Do calls=%d, want 1/1", schedulerStarts, onceStarts) + } + source, err := os.ReadFile("terminal_sessions.go") + if err != nil { + t.Fatal(err) + } + if !bytes.Contains(source, []byte("s.submitInput(&terminalInput{p: frame.Data})")) { + t.Fatal("browser attach does not submit through the terminal input scheduler") + } +} + +func fullNonblockingPipe(t *testing.T) (*os.File, *os.File) { + t.Helper() + r, w, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = r.Close(); _ = w.Close() }) + fd := int(w.Fd()) + if err := unix.SetNonblock(fd, true); err != nil { + t.Fatal(err) + } + block := make([]byte, 4<<10) + for { + if _, err := unix.Write(fd, block); err == unix.EAGAIN { + return r, w + } else if err != nil { + t.Fatal(err) + } + } +} + +func waitSessionInput(t *testing.T, s *terminalSession, calls, queued int, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for { + s.inputMu.Lock() + gotCalls := s.inputCalls + s.inputMu.Unlock() + if gotCalls == calls && (queued < 0 || len(s.input) == queued) { + return + } + if time.Now().After(deadline) { + t.Fatalf("session input calls/queued = %d/%d, want %d/%d", gotCalls, len(s.input), calls, queued) + } + time.Sleep(time.Millisecond) + } +} + +func waitInputAccounting(t *testing.T, s *terminalSession, local *localTerminalConn, calls, inputBytes int, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for { + s.inputMu.Lock() + sc, sb := s.inputCalls, s.inputBytes + s.inputMu.Unlock() + lc, lb := local.inputCalls.Load(), local.inputBytes.Load() + if sc == calls && sb == inputBytes && lc == int64(calls) && lb == int64(inputBytes) { + return + } + if time.Now().After(deadline) { + t.Fatalf("input accounting session=%d/%d local=%d/%d, want %d/%d", sc, sb, lc, lb, calls, inputBytes) + } + time.Sleep(time.Millisecond) + } +} + +func assertPTYNonblocking(t *testing.T, f *os.File, after string) { + t.Helper() + rc, err := f.SyscallConn() + if err != nil { + t.Fatal(err) + } + var flags int + var opErr error + if err := rc.Control(func(fd uintptr) { flags, opErr = unix.FcntlInt(fd, unix.F_GETFL, 0) }); err != nil { + t.Fatal(err) + } + if opErr != nil { + t.Fatal(opErr) + } + if flags&unix.O_NONBLOCK == 0 { + t.Fatalf("PTY master lost O_NONBLOCK after %s", after) + } +} + +func rawPTYSize(f *os.File) (rows, cols int, err error) { + rc, err := f.SyscallConn() + if err != nil { + return 0, 0, err + } + var size *unix.Winsize + var opErr error + if err := rc.Control(func(fd uintptr) { size, opErr = unix.IoctlGetWinsize(int(fd), unix.TIOCGWINSZ) }); err != nil { + return 0, 0, err + } + if opErr != nil { + return 0, 0, opErr } + return int(size.Row), int(size.Col), nil } // The browser sibling above proves the same bounded queue design over a sealed From 38b55d629acd706bde9529c147dd1db6587dcdfa Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 16:52:21 -0600 Subject: [PATCH 5/7] fix(system): backpressure browser terminal input --- internal/system/terminal_input_linux_test.go | 7 +- internal/system/terminal_sessions.go | 146 +++++++++++++--- internal/system/terminal_sessions_test.go | 167 ++++++++++++++++++- 3 files changed, 290 insertions(+), 30 deletions(-) diff --git a/internal/system/terminal_input_linux_test.go b/internal/system/terminal_input_linux_test.go index 0dc8b95..22daae3 100644 --- a/internal/system/terminal_input_linux_test.go +++ b/internal/system/terminal_input_linux_test.go @@ -80,7 +80,10 @@ func TestRealPTYInputFairnessAt4096ByteQuantum(t *testing.T) { local := &TerminalClient{session: s, conn: localConn} localBegin := []byte("[[local-begin-29ac]]") localTail := []byte("[[local-tail-e841]]") - localInput := append(append(append([]byte(nil), localBegin...), bytes.Repeat([]byte{'L'}, 64<<10)...), localTail...) + localInput := append(append(append([]byte(nil), localBegin...), bytes.Repeat([]byte{'L'}, terminalInputBytes-len(localBegin)-len(localTail))...), localTail...) + if len(localInput) != terminalInputBytes { + t.Fatalf("local fixture = %d bytes, want exact %d-byte bound", len(localInput), terminalInputBytes) + } if err := local.Write(localInput); err != nil { t.Fatal(err) } @@ -199,7 +202,7 @@ func fillPTYToEAGAIN(t *testing.T, ptmx *os.File) int { } func newInputTestSession(ptmx *os.File) *terminalSession { - s := &terminalSession{ptmx: ptmx, conns: make(map[terminalConn]struct{}), input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s := &terminalSession{ptmx: ptmx, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} s.startInput() return s } diff --git a/internal/system/terminal_sessions.go b/internal/system/terminal_sessions.go index 8b56d64..ecf69f2 100644 --- a/internal/system/terminal_sessions.go +++ b/internal/system/terminal_sessions.go @@ -86,10 +86,18 @@ const ( // input worker is currently processing. terminalInputBytes bounds the copied // bytes belonging to those calls, including that in-flight call. Caller-owned // input is never counted or retained when admission rejects it. - terminalInputQueue = terminalSendQueue - terminalInputBytes = terminalSendBytes - terminalInputPollWindow = 25 * time.Millisecond - terminalInputQuantum = 4 << 10 + terminalInputQueue = terminalSendQueue + terminalInputBytes = terminalSendBytes + // Browser input has its own reserve beyond the entire local-input budget. + // Thus a maximal accepted local workload (256 calls/1 MiB) still leaves 64 + // calls/256 KiB for browser keystrokes in the same scheduler. Browser frames + // larger than the byte reserve are admitted in bounded chunks. + terminalBrowserInputQueue = 64 + terminalBrowserInputBytes = 256 << 10 + terminalSchedulerQueue = terminalInputQueue + terminalBrowserInputQueue + terminalSchedulerBytes = terminalInputBytes + terminalBrowserInputBytes + terminalInputPollWindow = 25 * time.Millisecond + terminalInputQuantum = 4 << 10 ) var ( @@ -206,22 +214,27 @@ type terminalSession struct { cmd *exec.Cmd pgid int // the shell's process group id (it leads it: pty.Start sets Setsid) - mu sync.Mutex - inputMu sync.Mutex - input chan *terminalInput - inputBytes int - inputCalls int - inputClosed bool - inputOnce sync.Once - inputStarted chan struct{} - inputStopped chan struct{} - conns map[terminalConn]struct{} - replay replayRing - active bool - closed bool - expires *time.Timer - done chan struct{} - closeOnce sync.Once + mu sync.Mutex + inputMu sync.Mutex + input chan *terminalInput + inputBytes int + inputCalls int + localInputBytes int + localInputCalls int + browserInputBytes int + browserInputCalls int + inputClosed bool + inputCapacity chan struct{} // closed and replaced under inputMu on accounting release + inputOnce sync.Once + inputStarted chan struct{} + inputStopped chan struct{} + conns map[terminalConn]struct{} + replay replayRing + active bool + closed bool + expires *time.Timer + done chan struct{} + closeOnce sync.Once } // terminalInput is one accepted browser or local request. The scheduler writes @@ -754,7 +767,7 @@ func (d *system) terminal(h relay.StreamHeader, actionDeadline time.Time) (*term _, _ = cmd.Process.Wait() return nil, fmt.Errorf("set pty size: %w", err) } - s := &terminalSession{id: h.SessionID, owner: d, cwd: cwd, ptmx: ptmx, cmd: cmd, pgid: pgid, conns: make(map[terminalConn]struct{}), input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} + s := &terminalSession{id: h.SessionID, owner: d, cwd: cwd, ptmx: ptmx, cmd: cmd, pgid: pgid, conns: make(map[terminalConn]struct{}), done: make(chan struct{})} d.terminals[h.SessionID] = s d.addSession(1) d.logf("terminal session started (%s)", d.cfg.Shell) @@ -806,7 +819,7 @@ func (s *terminalSession) attach(conn *sealedConn) { } switch { case frame.Data != nil: - if !s.submitInput(&terminalInput{p: frame.Data}) { + if !s.submitBrowserInput(frame.Data, conn.dead) { s.detach(conn) return } @@ -895,29 +908,92 @@ func (s *terminalSession) detach(conn terminalConn) { } } -// submitInput is nonblocking. Browser readers detach on a full session queue; -// local callers roll back their per-client admission and return backpressure. +// submitInput is the common nonblocking admission primitive. Local callers use +// it directly so Write remains immediate. Browser readers use +// submitBrowserInput, which waits for this same scheduler's bounded reserve. func (s *terminalSession) submitInput(in *terminalInput) bool { s.startInput() s.inputMu.Lock() defer s.inputMu.Unlock() - if s.inputClosed || s.inputCalls >= terminalInputQueue || s.inputBytes+len(in.p) > terminalInputBytes { + return s.submitInputLocked(in) +} + +func (s *terminalSession) submitInputLocked(in *terminalInput) bool { + if s.inputClosed || s.inputCalls >= terminalSchedulerQueue || s.inputBytes+len(in.p) > terminalSchedulerBytes { + return false + } + if in.local != nil { + if s.localInputCalls >= terminalInputQueue || s.localInputBytes+len(in.p) > terminalInputBytes { + return false + } + } else if s.browserInputCalls >= terminalBrowserInputQueue || s.browserInputBytes+len(in.p) > terminalBrowserInputBytes { return false } select { case s.input <- in: s.inputCalls++ s.inputBytes += len(in.p) + if in.local != nil { + s.localInputCalls++ + s.localInputBytes += len(in.p) + } else { + s.browserInputCalls++ + s.browserInputBytes += len(in.p) + } return true default: return false } } +// submitBrowserInput backpressures one browser reader without disconnecting it. +// The generation channel is captured while inputMu protects the failed +// admission, so a release cannot be missed between checking capacity and +// waiting. A connection retains only its already-decoded frame while parked. +func (s *terminalSession) submitBrowserInput(p []byte, dead <-chan struct{}) bool { + for len(p) > 0 { + n := min(len(p), terminalBrowserInputBytes) + in := &terminalInput{p: p[:n]} + for { + select { + case <-s.done: + return false + case <-dead: + return false + default: + } + s.startInput() + s.inputMu.Lock() + if s.submitInputLocked(in) { + s.inputMu.Unlock() + break + } + if s.inputClosed { + s.inputMu.Unlock() + return false + } + if s.inputCapacity == nil { + s.inputCapacity = make(chan struct{}) + } + capacity := s.inputCapacity + s.inputMu.Unlock() + select { + case <-capacity: + case <-s.done: + return false + case <-dead: + return false + } + } + p = p[n:] + } + return true +} + func (s *terminalSession) startInput() { s.inputOnce.Do(func() { if s.input == nil { - s.input = make(chan *terminalInput, terminalInputQueue) + s.input = make(chan *terminalInput, terminalSchedulerQueue) } s.inputStarted = make(chan struct{}) s.inputStopped = make(chan struct{}) @@ -930,6 +1006,14 @@ func (s *terminalSession) releaseInput(in *terminalInput) { s.inputMu.Lock() s.inputCalls-- s.inputBytes -= n + if in.local != nil { + s.localInputCalls-- + s.localInputBytes -= n + } else { + s.browserInputCalls-- + s.browserInputBytes -= n + } + s.notifyInputCapacityLocked() s.inputMu.Unlock() if in.local != nil { in.local.inputCalls.Add(-1) @@ -937,6 +1021,13 @@ func (s *terminalSession) releaseInput(in *terminalInput) { } } +func (s *terminalSession) notifyInputCapacityLocked() { + if s.inputCapacity != nil { + close(s.inputCapacity) + s.inputCapacity = nil + } +} + func (s *terminalSession) inputLoop() { close(s.inputStarted) defer close(s.inputStopped) @@ -975,7 +1066,7 @@ func (s *terminalSession) inputLoop() { s.releaseInput(in) case s.input <- in: // Keep accounting: the request remains admitted at the queue tail. - // terminalInputQueue counts the active request, so at most 255 + // terminalSchedulerQueue counts the active request, so at most 319 // others can be queued while this one is out of the channel. The // active request therefore always owns this requeue slot. } @@ -1520,6 +1611,7 @@ func (s *terminalSession) stopInput() { if !s.inputClosed { s.inputClosed = true close(s.done) + s.notifyInputCapacityLocked() } s.inputMu.Unlock() } diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index c9f84ac..6d2ada4 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -982,6 +982,154 @@ func TestTerminalInputStartsExactlyOneWorkerAndCloseDrainsAccounting(t *testing. } } +func TestTerminalInputMaximalLocalLeavesBrowserReserve(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + local := newLocalTerminalConn() + client := &TerminalClient{session: s, conn: local} + if err := client.Write(make([]byte, terminalInputBytes)); err != nil { + t.Fatalf("admit exact maximal local input: %v", err) + } + waitSessionInput(t, s, 1, 0, time.Second) + + dead := make(chan struct{}) + admitted := make(chan bool, 1) + go func() { admitted <- s.submitBrowserInput([]byte("browser-marker"), dead) }() + select { + case ok := <-admitted: + if !ok { + t.Fatal("browser marker rejected behind maximal local input") + } + case <-time.After(time.Second): + t.Fatal("browser marker waited for maximal local input to complete") + } + waitSessionInput(t, s, 2, 1, time.Second) + s.inputMu.Lock() + if s.localInputCalls != 1 || s.localInputBytes != terminalInputBytes || s.browserInputCalls != 1 || s.browserInputBytes != len("browser-marker") { + t.Fatalf("split accounting local=%d/%d browser=%d/%d", s.localInputCalls, s.localInputBytes, s.browserInputCalls, s.browserInputBytes) + } + s.inputMu.Unlock() + + local.kill() + s.close() + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("close did not drain maximal local plus browser reserve") + } + waitInputAccounting(t, s, local, 0, 0, time.Second) +} + +func TestBrowserInputFullCapacityWaitsThenResumes(t *testing.T) { + inputRead, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + s.startInput() + for i := range terminalBrowserInputQueue { + if !s.submitInput(&terminalInput{p: bytes.Repeat([]byte{byte(i)}, terminalInputQuantum)}) { + t.Fatalf("admit browser call %d of exact %d-call reserve", i+1, terminalBrowserInputQueue) + } + } + waitSessionInput(t, s, terminalBrowserInputQueue, terminalBrowserInputQueue-1, time.Second) + dead := make(chan struct{}) + admitted := make(chan bool, 1) + go func() { admitted <- s.submitBrowserInput([]byte("next"), dead) }() + waitInputCapacityWaiter(t, s) + select { + case got := <-admitted: + t.Fatalf("browser submission completed at full capacity: %v", got) + default: + } + select { + case <-dead: + t.Fatal("capacity pressure disconnected the browser") + default: + } + + buf := make([]byte, terminalInputQuantum) + if _, err := inputRead.Read(buf); err != nil { + t.Fatal(err) + } + select { + case ok := <-admitted: + if !ok { + t.Fatal("waiting browser did not resume after accounting release") + } + case <-time.After(time.Second): + t.Fatal("waiting browser did not observe accounting release") + } + waitSessionInput(t, s, terminalBrowserInputQueue, terminalBrowserInputQueue-1, time.Second) + s.close() +} + +func TestBrowserInputCapacityWaitCancellation(t *testing.T) { + for _, tc := range []struct { + name string + cancel func(*terminalSession, chan struct{}) + }{ + {name: "connection", cancel: func(_ *terminalSession, dead chan struct{}) { close(dead) }}, + {name: "session", cancel: func(s *terminalSession, _ chan struct{}) { s.stopInput() }}, + } { + t.Run(tc.name, func(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + s.startInput() + for i := range terminalBrowserInputQueue { + if !s.submitInput(&terminalInput{p: bytes.Repeat([]byte{byte(i)}, terminalInputQuantum)}) { + t.Fatalf("fill browser reserve at call %d", i+1) + } + } + waitSessionInput(t, s, terminalBrowserInputQueue, terminalBrowserInputQueue-1, time.Second) + dead := make(chan struct{}) + result := make(chan bool, 1) + go func() { result <- s.submitBrowserInput([]byte("blocked"), dead) }() + waitInputCapacityWaiter(t, s) + tc.cancel(s, dead) + select { + case ok := <-result: + if ok { + t.Fatal("cancelled browser submission was admitted") + } + case <-time.After(time.Second): + t.Fatal("cancellation did not unblock browser submission") + } + s.close() + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("scheduler did not stop and drain after cancellation") + } + s.inputMu.Lock() + calls, inputBytes := s.inputCalls, s.inputBytes + s.inputMu.Unlock() + if calls != 0 || inputBytes != 0 { + t.Fatalf("accounting after cancellation and close = %d/%d, want 0/0", calls, inputBytes) + } + }) + } +} + +func TestTerminalInputExactSchedulerBounds(t *testing.T) { + if terminalSchedulerQueue != 320 || terminalSchedulerBytes != 1280<<10 { + t.Fatalf("scheduler bounds = %d calls/%d bytes, want 320 calls/1.25 MiB", terminalSchedulerQueue, terminalSchedulerBytes) + } + if terminalInputQueue != 256 || terminalInputBytes != 1<<20 || terminalBrowserInputQueue != 64 || terminalBrowserInputBytes != 256<<10 { + t.Fatalf("component bounds local=%d/%d browser=%d/%d", terminalInputQueue, terminalInputBytes, terminalBrowserInputQueue, terminalBrowserInputBytes) + } + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, done: make(chan struct{})} + if !s.submitInput(&terminalInput{p: make([]byte, terminalBrowserInputBytes)}) { + t.Fatal("exact browser byte reserve was not reachable") + } + if got := cap(s.input); got != terminalSchedulerQueue { + t.Fatalf("initialized scheduler channel capacity = %d, want %d", got, terminalSchedulerQueue) + } + waitSessionInput(t, s, 1, 0, time.Second) + if s.submitInput(&terminalInput{p: []byte{1}}) { + t.Fatal("browser byte reserve accepted one byte beyond its exact bound") + } + s.close() +} + func TestTerminalInputCallBoundLeavesActiveRequestARequeueSlot(t *testing.T) { _, inputWrite := fullNonblockingPipe(t) s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} @@ -1083,7 +1231,7 @@ func TestTerminalInputHasOneRawPTYWriterAndOneScheduler(t *testing.T) { if err != nil { t.Fatal(err) } - if !bytes.Contains(source, []byte("s.submitInput(&terminalInput{p: frame.Data})")) { + if !bytes.Contains(source, []byte("s.submitBrowserInput(frame.Data, conn.dead)")) { t.Fatal("browser attach does not submit through the terminal input scheduler") } } @@ -1126,6 +1274,23 @@ func waitSessionInput(t *testing.T, s *terminalSession, calls, queued int, timeo } } +func waitInputCapacityWaiter(t *testing.T, s *terminalSession) { + t.Helper() + deadline := time.Now().Add(time.Second) + for { + s.inputMu.Lock() + waiting := s.inputCapacity != nil + s.inputMu.Unlock() + if waiting { + return + } + if time.Now().After(deadline) { + t.Fatal("browser submission did not enter capacity wait") + } + time.Sleep(time.Millisecond) + } +} + func waitInputAccounting(t *testing.T, s *terminalSession, local *localTerminalConn, calls, inputBytes int, timeout time.Duration) { t.Helper() deadline := time.Now().Add(timeout) From cec1550decbac7e78f7b37e0b0b75a7a1ca09e45 Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 17:10:29 -0600 Subject: [PATCH 6/7] fix(system): reserve terminal input before copy --- internal/system/terminal_sessions.go | 86 +++++++-- internal/system/terminal_sessions_test.go | 206 ++++++++++++++++++++++ 2 files changed, 276 insertions(+), 16 deletions(-) diff --git a/internal/system/terminal_sessions.go b/internal/system/terminal_sessions.go index ecf69f2..4ea3a1b 100644 --- a/internal/system/terminal_sessions.go +++ b/internal/system/terminal_sessions.go @@ -242,9 +242,10 @@ type terminalSession struct { // the tail, so FIFO admission cannot let a large local paste monopolise a // slowly-draining PTY ahead of browser keystrokes. type terminalInput struct { - p []byte - off int - local *localTerminalConn + p []byte + off int + local *localTerminalConn + reservedBytes int // original local Write length; zero for browser/test submissions } // terminalConn is the common outbound half of a browser or local attachment. @@ -434,20 +435,18 @@ func (c *TerminalClient) Write(p []byte) error { if len(p) == 0 { return nil } - if c.conn.inputCalls.Load() >= terminalInputQueue || c.conn.inputBytes.Load()+int64(len(p)) > terminalInputBytes { + // Local inputMu -> session inputMu is the only nested input lock order. + // releaseInput never takes local inputMu. This makes per-attachment and + // session-local admission one reserve-before-allocation transaction while + // retaining the existing ordering with kill/Detach. + if !c.session.reserveLocalInput(c.conn, len(p)) { return ErrTerminalInputBackpressure } - // Reserve before allocation. inputMu also makes admission and kill one - // transaction, so a successful Write cannot be admitted after Detach. - c.conn.inputCalls.Add(1) - c.conn.inputBytes.Add(int64(len(p))) if c.conn.beforeInputCopy != nil { c.conn.beforeInputCopy() } copy := c.conn.copyInput(p) - if !c.session.submitInput(&terminalInput{p: copy, local: c.conn}) { - c.conn.inputCalls.Add(-1) - c.conn.inputBytes.Add(-int64(len(copy))) + if !c.session.enqueueReservedLocalInput(&terminalInput{p: copy, local: c.conn, reservedBytes: len(p)}) { return ErrTerminalInputBackpressure } return nil @@ -918,6 +917,50 @@ func (s *terminalSession) submitInput(in *terminalInput) bool { return s.submitInputLocked(in) } +// reserveLocalInput is called with local.inputMu held. Accounting the request +// before its copy also reserves its scheduler capacity: the production channel +// has terminalSchedulerQueue slots and inputCalls includes active, queued, and +// not-yet-enqueued reserved calls, so later admissions cannot consume this +// request's future slot. +func (s *terminalSession) reserveLocalInput(local *localTerminalConn, n int) bool { + s.startInput() + s.inputMu.Lock() + defer s.inputMu.Unlock() + if s.inputClosed || + local.inputCalls.Load() >= terminalInputQueue || local.inputBytes.Load()+int64(n) > terminalInputBytes || + s.localInputCalls >= terminalInputQueue || s.localInputBytes+n > terminalInputBytes || + s.inputCalls >= terminalSchedulerQueue || s.inputBytes+n > terminalSchedulerBytes { + return false + } + s.inputCalls++ + s.inputBytes += n + s.localInputCalls++ + s.localInputBytes += n + local.inputCalls.Add(1) + local.inputBytes.Add(int64(n)) + return true +} + +// enqueueReservedLocalInput publishes a request whose accounting and channel +// slot were reserved before its copy. Session close is the ordinary failure +// path; the default is a defensive check for an incorrectly sized channel. +func (s *terminalSession) enqueueReservedLocalInput(in *terminalInput) bool { + s.inputMu.Lock() + if !s.inputClosed { + select { + case s.input <- in: + s.inputMu.Unlock() + return true + default: + } + } + s.releaseInputLocked(in) + s.inputMu.Unlock() + in.local.inputCalls.Add(-1) + in.local.inputBytes.Add(-int64(in.accountedBytes())) + return false +} + func (s *terminalSession) submitInputLocked(in *terminalInput) bool { if s.inputClosed || s.inputCalls >= terminalSchedulerQueue || s.inputBytes+len(in.p) > terminalSchedulerBytes { return false @@ -1002,8 +1045,17 @@ func (s *terminalSession) startInput() { } func (s *terminalSession) releaseInput(in *terminalInput) { - n := len(in.p) s.inputMu.Lock() + s.releaseInputLocked(in) + s.inputMu.Unlock() + if in.local != nil { + in.local.inputCalls.Add(-1) + in.local.inputBytes.Add(-int64(in.accountedBytes())) + } +} + +func (s *terminalSession) releaseInputLocked(in *terminalInput) { + n := in.accountedBytes() s.inputCalls-- s.inputBytes -= n if in.local != nil { @@ -1014,11 +1066,13 @@ func (s *terminalSession) releaseInput(in *terminalInput) { s.browserInputBytes -= n } s.notifyInputCapacityLocked() - s.inputMu.Unlock() - if in.local != nil { - in.local.inputCalls.Add(-1) - in.local.inputBytes.Add(-int64(n)) +} + +func (in *terminalInput) accountedBytes() int { + if in.reservedBytes != 0 { + return in.reservedBytes } + return len(in.p) } func (s *terminalSession) notifyInputCapacityLocked() { diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index 6d2ada4..2e04845 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -794,6 +794,62 @@ func TestLocalTerminalWriteReservesCallsAndBytesBeforeCopy(t *testing.T) { waitInputAccounting(t, s, conn, 0, 0, time.Second) } +func TestLocalTerminalWriteReservesAggregateBeforeCopyAcrossClients(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + firstConn := newLocalTerminalConn() + secondConn := newLocalTerminalConn() + if err := s.attachConn(firstConn); err != nil { + t.Fatal(err) + } + if err := s.attachConn(secondConn); err != nil { + t.Fatal(err) + } + first := &TerminalClient{session: s, conn: firstConn} + second := &TerminalClient{session: s, conn: secondConn} + entered := make(chan struct{}) + release := make(chan struct{}) + firstConn.beforeInputCopy = func() { + close(entered) + <-release + } + var secondCopies atomic.Int64 + secondConn.copyInput = func(p []byte) []byte { + secondCopies.Add(1) + return append([]byte(nil), p...) + } + firstDone := make(chan error, 1) + go func() { firstDone <- first.Write(make([]byte, terminalInputBytes)) }() + <-entered + s.inputMu.Lock() + if s.inputCalls != 1 || s.inputBytes != terminalInputBytes || s.localInputCalls != 1 || s.localInputBytes != terminalInputBytes { + t.Fatalf("aggregate reservation before copy total=%d/%d local=%d/%d", s.inputCalls, s.inputBytes, s.localInputCalls, s.localInputBytes) + } + s.inputMu.Unlock() + if err := second.Write([]byte("overflow")); !errors.Is(err, ErrTerminalInputBackpressure) { + t.Fatalf("second client aggregate overflow = %v, want ErrTerminalInputBackpressure", err) + } + if got := secondCopies.Load(); got != 0 { + t.Fatalf("aggregate-rejected second client made %d copies", got) + } + close(release) + if err := <-firstDone; err != nil { + t.Fatalf("first reserved Write: %v", err) + } + s.close() + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("close did not drain aggregate reservation") + } + waitInputAccounting(t, s, firstConn, 0, 0, time.Second) + if secondConn.inputCalls.Load() != 0 || secondConn.inputBytes.Load() != 0 { + t.Fatalf("rejected second ledger = %d/%d, want 0/0", secondConn.inputCalls.Load(), secondConn.inputBytes.Load()) + } + firstConn.kill() + secondConn.kill() +} + func TestLocalTerminalWriteRejectsOversizedInputWithoutCopy(t *testing.T) { conn := newLocalTerminalConn() client := &TerminalClient{conn: conn} @@ -1130,6 +1186,156 @@ func TestTerminalInputExactSchedulerBounds(t *testing.T) { s.close() } +func TestTerminalInputExactCombinedCallBoundAndCloseDrain(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + local := newLocalTerminalConn() + client := &TerminalClient{session: s, conn: local} + block := make([]byte, terminalInputQuantum) + for i := range terminalInputQueue { + if err := client.Write(block); err != nil { + t.Fatalf("admit local call %d of %d: %v", i+1, terminalInputQueue, err) + } + } + for i := range terminalBrowserInputQueue { + if !s.submitInput(&terminalInput{p: block}) { + t.Fatalf("admit browser call %d of %d", i+1, terminalBrowserInputQueue) + } + } + waitSessionInput(t, s, terminalSchedulerQueue, terminalSchedulerQueue-1, time.Second) + s.inputMu.Lock() + got := [6]int{s.inputCalls, s.inputBytes, s.localInputCalls, s.localInputBytes, s.browserInputCalls, s.browserInputBytes} + s.inputMu.Unlock() + want := [6]int{terminalSchedulerQueue, terminalSchedulerBytes, terminalInputQueue, terminalInputBytes, terminalBrowserInputQueue, terminalBrowserInputBytes} + if got != want { + t.Fatalf("combined ledgers = %v, want %v", got, want) + } + if err := client.Write([]byte("local-overflow")); !errors.Is(err, ErrTerminalInputBackpressure) { + t.Fatalf("local call beyond combined bound = %v, want backpressure", err) + } + dead := make(chan struct{}) + browserResult := make(chan bool, 1) + go func() { browserResult <- s.submitBrowserInput([]byte("browser-overflow"), dead) }() + waitInputCapacityWaiter(t, s) + select { + case result := <-browserResult: + t.Fatalf("browser call did not wait at combined bound: %v", result) + default: + } + s.close() + select { + case result := <-browserResult: + if result { + t.Fatal("browser call was admitted during close") + } + case <-time.After(time.Second): + t.Fatal("close did not unblock browser at combined bound") + } + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("close did not drain combined bound") + } + waitInputAccounting(t, s, local, 0, 0, time.Second) + local.kill() +} + +func TestBrowserLargeFrameUsesOrderedBoundedSchedulerChunks(t *testing.T) { + inputRead, inputWrite, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) + if err := unix.SetNonblock(int(inputWrite.Fd()), true); err != nil { + t.Fatal(err) + } + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + first := bytes.Repeat([]byte{'A'}, terminalBrowserInputBytes) + tail := []byte("[[ordered-second-browser-chunk]]") + frame := append(append([]byte(nil), first...), tail...) + dead := make(chan struct{}) + result := make(chan bool, 1) + go func() { result <- s.submitBrowserInput(frame, dead) }() + waitInputCapacityWaiter(t, s) + s.inputMu.Lock() + if s.browserInputCalls != 1 || s.browserInputBytes != terminalBrowserInputBytes { + t.Fatalf("between chunks browser ledger = %d/%d, want 1/%d", s.browserInputCalls, s.browserInputBytes, terminalBrowserInputBytes) + } + s.inputMu.Unlock() + got := make([]byte, len(frame)) + readDone := make(chan error, 1) + go func() { _, err := io.ReadFull(inputRead, got); readDone <- err }() + select { + case ok := <-result: + if !ok { + t.Fatal("large browser frame submission failed") + } + case <-time.After(2 * time.Second): + t.Fatal("large browser frame did not admit its second chunk") + } + select { + case err := <-readDone: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("large browser frame did not reach writer") + } + if !bytes.Equal(got, frame) { + t.Fatal("large browser frame chunks were not written in order") + } + waitSessionInput(t, s, 0, 0, time.Second) + s.close() +} + +func TestBrowserLargeFrameChunkWaitCancellationDrainsAccounting(t *testing.T) { + for _, tc := range []struct { + name string + cancel func(*terminalSession, chan struct{}) + }{ + {name: "connection", cancel: func(_ *terminalSession, dead chan struct{}) { close(dead) }}, + {name: "session", cancel: func(s *terminalSession, _ chan struct{}) { s.stopInput() }}, + } { + t.Run(tc.name, func(t *testing.T) { + inputRead, inputWrite, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = inputRead.Close(); _ = inputWrite.Close() }) + if err := unix.SetNonblock(int(inputWrite.Fd()), true); err != nil { + t.Fatal(err) + } + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + frame := make([]byte, terminalBrowserInputBytes+1) + dead := make(chan struct{}) + result := make(chan bool, 1) + go func() { result <- s.submitBrowserInput(frame, dead) }() + waitInputCapacityWaiter(t, s) + tc.cancel(s, dead) + select { + case ok := <-result: + if ok { + t.Fatal("cancelled between-chunk submission succeeded") + } + case <-time.After(time.Second): + t.Fatal("between-chunk cancellation did not unblock submission") + } + s.close() + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("scheduler did not stop after chunk cancellation") + } + s.inputMu.Lock() + ledgers := [4]int{s.inputCalls, s.inputBytes, s.browserInputCalls, s.browserInputBytes} + s.inputMu.Unlock() + if ledgers != [4]int{} { + t.Fatalf("chunk cancellation leaked ledgers: %v", ledgers) + } + }) + } +} + func TestTerminalInputCallBoundLeavesActiveRequestARequeueSlot(t *testing.T) { _, inputWrite := fullNonblockingPipe(t) s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalInputQueue), done: make(chan struct{})} From 5cfd21288457f30b8a68be1398b0f509a0980faa Mon Sep 17 00:00:00 2001 From: nicodes Date: Mon, 24 Aug 2026 17:23:56 -0600 Subject: [PATCH 7/7] test(system): cover reserved terminal input races --- internal/system/terminal_sessions_test.go | 158 ++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/internal/system/terminal_sessions_test.go b/internal/system/terminal_sessions_test.go index 2e04845..d90f8d7 100644 --- a/internal/system/terminal_sessions_test.go +++ b/internal/system/terminal_sessions_test.go @@ -850,6 +850,164 @@ func TestLocalTerminalWriteReservesAggregateBeforeCopyAcrossClients(t *testing.T secondConn.kill() } +func TestReservedLocalPublicationRollsBackWhenSessionCloses(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, input: make(chan *terminalInput, terminalSchedulerQueue), done: make(chan struct{})} + local := newLocalTerminalConn() + if err := s.attachConn(local); err != nil { + t.Fatal(err) + } + client := &TerminalClient{session: s, conn: local} + reserved := make(chan struct{}) + release := make(chan struct{}) + local.beforeInputCopy = func() { + close(reserved) + <-release + } + writeResult := make(chan error, 1) + go func() { writeResult <- client.Write(make([]byte, terminalInputBytes)) }() + <-reserved + + browserBlock := make([]byte, terminalInputQuantum) + for i := range terminalBrowserInputQueue { + if !s.submitInput(&terminalInput{p: browserBlock}) { + t.Fatalf("fill browser reserve at call %d", i+1) + } + } + s.inputMu.Lock() + got := [6]int{s.inputCalls, s.inputBytes, s.localInputCalls, s.localInputBytes, s.browserInputCalls, s.browserInputBytes} + s.inputMu.Unlock() + want := [6]int{1 + terminalBrowserInputQueue, terminalSchedulerBytes, 1, terminalInputBytes, terminalBrowserInputQueue, terminalBrowserInputBytes} + if got != want { + t.Fatalf("reserved pre-publication ledgers = %v, want %v", got, want) + } + + browserResult := make(chan bool, 1) + browserDead := make(chan struct{}) + go func() { browserResult <- s.submitBrowserInput([]byte("waiting-browser"), browserDead) }() + waitInputCapacityWaiter(t, s) + closeResult := make(chan struct{}) + go func() { + s.close() + close(closeResult) + }() + select { + case ok := <-browserResult: + if ok { + t.Fatal("browser waiter was admitted during close") + } + case <-time.After(time.Second): + t.Fatal("session close did not wake browser capacity waiter") + } + select { + case <-closeResult: + t.Fatal("session close passed a reserved Write holding local inputMu") + default: + } + + close(release) + if err := <-writeResult; !errors.Is(err, ErrTerminalInputBackpressure) { + t.Fatalf("reserved publication after close = %v, want ErrTerminalInputBackpressure", err) + } + select { + case <-closeResult: + case <-time.After(time.Second): + t.Fatal("session close deadlocked after reserved Write released local inputMu") + } + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("input worker did not stop after reserved publication rollback") + } + s.inputMu.Lock() + final := [6]int{s.inputCalls, s.inputBytes, s.localInputCalls, s.localInputBytes, s.browserInputCalls, s.browserInputBytes} + s.inputMu.Unlock() + if final != [6]int{} { + t.Fatalf("session ledgers after close = %v, want all zero", final) + } + if calls, inputBytes := local.inputCalls.Load(), local.inputBytes.Load(); calls != 0 || inputBytes != 0 { + t.Fatalf("local ledgers after close = %d/%d, want 0/0", calls, inputBytes) + } +} + +func TestReservedLocalInputOwnsFutureSchedulerChannelSlot(t *testing.T) { + _, inputWrite := fullNonblockingPipe(t) + s := &terminalSession{ptmx: inputWrite, done: make(chan struct{})} + primary := newLocalTerminalConn() + overflow := newLocalTerminalConn() + if err := s.attachConn(primary); err != nil { + t.Fatal(err) + } + if err := s.attachConn(overflow); err != nil { + t.Fatal(err) + } + primaryClient := &TerminalClient{session: s, conn: primary} + overflowClient := &TerminalClient{session: s, conn: overflow} + block := make([]byte, terminalInputQuantum) + for i := range terminalInputQueue - 1 { + if err := primaryClient.Write(block); err != nil { + t.Fatalf("publish local call %d of %d: %v", i+1, terminalInputQueue-1, err) + } + } + for i := range terminalBrowserInputQueue { + if !s.submitInput(&terminalInput{p: block}) { + t.Fatalf("publish browser call %d of %d", i+1, terminalBrowserInputQueue) + } + } + if got := cap(s.input); got != terminalSchedulerQueue { + t.Fatalf("production scheduler channel capacity = %d, want %d future-slot invariant", got, terminalSchedulerQueue) + } + reserved := make(chan struct{}) + release := make(chan struct{}) + primary.beforeInputCopy = func() { + close(reserved) + <-release + } + lastResult := make(chan error, 1) + go func() { lastResult <- primaryClient.Write(block) }() + <-reserved + s.inputMu.Lock() + got := [5]int{s.inputCalls, s.localInputCalls, s.browserInputCalls, s.localInputBytes, s.browserInputBytes} + s.inputMu.Unlock() + want := [5]int{terminalSchedulerQueue, terminalInputQueue, terminalBrowserInputQueue, terminalInputBytes, terminalBrowserInputBytes} + if got != want { + t.Fatalf("320th reserved ledgers = %v, want %v", got, want) + } + var overflowCopies atomic.Int64 + overflow.copyInput = func(p []byte) []byte { + overflowCopies.Add(1) + return append([]byte(nil), p...) + } + if err := overflowClient.Write([]byte("321st")); !errors.Is(err, ErrTerminalInputBackpressure) { + t.Fatalf("321st local Write = %v, want ErrTerminalInputBackpressure", err) + } + if copies := overflowCopies.Load(); copies != 0 { + t.Fatalf("321st rejected Write made %d copies", copies) + } + close(release) + if err := <-lastResult; err != nil { + t.Fatalf("320th Write did not publish into its reserved slot: %v", err) + } + s.close() + select { + case <-s.inputStopped: + case <-time.After(time.Second): + t.Fatal("close did not drain exact 320-call scheduler bound") + } + s.inputMu.Lock() + final := [6]int{s.inputCalls, s.inputBytes, s.localInputCalls, s.localInputBytes, s.browserInputCalls, s.browserInputBytes} + s.inputMu.Unlock() + if final != [6]int{} { + t.Fatalf("session ledgers after exact-bound close = %v, want all zero", final) + } + if calls, inputBytes := primary.inputCalls.Load(), primary.inputBytes.Load(); calls != 0 || inputBytes != 0 { + t.Fatalf("primary ledgers after close = %d/%d, want 0/0", calls, inputBytes) + } + if calls, inputBytes := overflow.inputCalls.Load(), overflow.inputBytes.Load(); calls != 0 || inputBytes != 0 { + t.Fatalf("overflow ledgers after close = %d/%d, want 0/0", calls, inputBytes) + } +} + func TestLocalTerminalWriteRejectsOversizedInputWithoutCopy(t *testing.T) { conn := newLocalTerminalConn() client := &TerminalClient{conn: conn}