From 7146decc3c86575185ca8a1212e1df7b293bb5ae Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 6 Oct 2026 22:58:25 +0000 Subject: [PATCH 1/8] Write relay Notices in fd 2 order and simplify shim I/O A Notice now goes through fd 2's output pump, after the stderr queued before it. readFD, writeFD and waitFD take one stop flag, the Request frame is built in a bytes.Buffer, and the unused Conn.Unix is gone. --- .../daemon/internal/processshim/conn_linux.go | 19 ++----- .../internal/processshim/conn_linux_test.go | 5 +- .../internal/processshim/endpoint_linux.go | 41 +++++--------- .../internal/processshim/relay_linux.go | 25 +++++---- .../internal/processshim/relay_linux_test.go | 54 +++++++++++++++++++ 5 files changed, 90 insertions(+), 54 deletions(-) diff --git a/apps/daemon/internal/processshim/conn_linux.go b/apps/daemon/internal/processshim/conn_linux.go index a83d1a855..99e57f9d8 100644 --- a/apps/daemon/internal/processshim/conn_linux.go +++ b/apps/daemon/internal/processshim/conn_linux.go @@ -3,6 +3,7 @@ package processshim import ( + "bytes" "errors" "fmt" "io" @@ -29,26 +30,23 @@ func NewConn(c *net.UnixConn) *Conn { return &Conn{c: c, first: true} } // Close closes the socket. func (c *Conn) Close() error { return c.c.Close() } -// Unix returns the socket. -func (c *Conn) Unix() *net.UnixConn { return c.c } - // SendRequest sends r with fds attached to its first byte. func (c *Conn) SendRequest(r Request, fds [3]int) error { f := Frame(r) if len(f.Payload) > MaxRequestBytes { return fmt.Errorf("%w: request of %d bytes exceeds %d", ErrProtocol, len(f.Payload), MaxRequestBytes) } - var b frameBuffer + var b bytes.Buffer if err := sandboxwire.WriteFrame(&b, f); err != nil { return err } - n, _, err := c.c.WriteMsgUnix(b, unix.UnixRights(fds[:]...), nil) + n, _, err := c.c.WriteMsgUnix(b.Bytes(), unix.UnixRights(fds[:]...), nil) // An empty write fails with EPIPE once the relay has run the invocation // and closed the connection, so write only what remains. - if err != nil || n == len(b) { + if err != nil || n == b.Len() { return err } - _, err = c.c.Write(b[n:]) + _, err = c.c.Write(b.Bytes()[n:]) return err } @@ -149,10 +147,3 @@ func closeAll(fds []int) { unix.Close(fd) } } - -type frameBuffer []byte - -func (b *frameBuffer) Write(p []byte) (int, error) { - *b = append(*b, p...) - return len(p), nil -} diff --git a/apps/daemon/internal/processshim/conn_linux_test.go b/apps/daemon/internal/processshim/conn_linux_test.go index b2e654dee..3bf892fce 100644 --- a/apps/daemon/internal/processshim/conn_linux_test.go +++ b/apps/daemon/internal/processshim/conn_linux_test.go @@ -1,6 +1,7 @@ package processshim import ( + "bytes" "errors" "net" "os" @@ -95,9 +96,9 @@ func TestUnexpectedDescriptorsAreRejected(t *testing.T) { } closeAll(recv[:]) frame := Frame(Signal{Number: 2}) - var buf frameBuffer + var buf bytes.Buffer _ = sandboxwire.WriteFrame(&buf, frame) - if _, _, err := shim.c.WriteMsgUnix(buf, unix.UnixRights(fds[0]), nil); err != nil { + if _, _, err := shim.c.WriteMsgUnix(buf.Bytes(), unix.UnixRights(fds[0]), nil); err != nil { t.Fatal(err) } if _, err := broker.ReadMessage(); !errors.Is(err, ErrProtocol) { diff --git a/apps/daemon/internal/processshim/endpoint_linux.go b/apps/daemon/internal/processshim/endpoint_linux.go index 284ec3690..df713477a 100644 --- a/apps/daemon/internal/processshim/endpoint_linux.go +++ b/apps/daemon/internal/processshim/endpoint_linux.go @@ -172,13 +172,13 @@ func (e endpoint) write(data []byte) (int, error) { } // readFD reads once. It returns 0 and nil at end of file. -func readFD(e endpoint, buf []byte, stops ...*stopFlag) (int, error) { +func readFD(e endpoint, buf []byte, stop *stopFlag) (int, error) { for { - if stopped(stops) { + if stop.isSet() { return 0, errStopped } if e.kind == ioShared { - if err := waitFD(e.io, unix.POLLIN, stops...); err != nil { + if err := waitFD(e.io, unix.POLLIN, stop); err != nil { return 0, err } } @@ -187,7 +187,7 @@ func readFD(e endpoint, buf []byte, stops ...*stopFlag) (int, error) { case unix.EINTR: continue case unix.EAGAIN: - if err := waitFD(e.io, unix.POLLIN, stops...); err != nil { + if err := waitFD(e.io, unix.POLLIN, stop); err != nil { return 0, err } continue @@ -197,13 +197,13 @@ func readFD(e endpoint, buf []byte, stops ...*stopFlag) (int, error) { } // writeFD writes all of data, handling short writes. -func writeFD(e endpoint, data []byte, stops ...*stopFlag) error { +func writeFD(e endpoint, data []byte, stop *stopFlag) error { for len(data) > 0 { - if stopped(stops) { + if stop.isSet() { return errStopped } if e.kind == ioShared { - if err := waitFD(e.io, unix.POLLOUT, stops...); err != nil { + if err := waitFD(e.io, unix.POLLOUT, stop); err != nil { return err } } @@ -212,7 +212,7 @@ func writeFD(e endpoint, data []byte, stops ...*stopFlag) error { case err == unix.EINTR: continue case err == unix.EAGAIN: - if err := waitFD(e.io, unix.POLLOUT, stops...); err != nil { + if err := waitFD(e.io, unix.POLLOUT, stop); err != nil { return err } continue @@ -235,21 +235,9 @@ func tryWrite(e endpoint, data []byte) { e.write(data[:min(len(data), pipeBuf)]) } -func stopped(stops []*stopFlag) bool { - for _, s := range stops { - if s.isSet() { - return true - } - } - return false -} - -// waitFD waits until fd has events or a stop flag is set. -func waitFD(fd int, events int16, stops ...*stopFlag) error { - pfds := []unix.PollFd{{Fd: int32(fd), Events: events}} - for _, s := range stops { - pfds = append(pfds, unix.PollFd{Fd: int32(s.fd), Events: unix.POLLIN}) - } +// waitFD waits until fd has events or stop is set. +func waitFD(fd int, events int16, stop *stopFlag) error { + pfds := []unix.PollFd{{Fd: int32(fd), Events: events}, {Fd: int32(stop.fd), Events: unix.POLLIN}} for { if _, err := unix.Poll(pfds, -1); err != nil { if err == unix.EINTR { @@ -257,12 +245,9 @@ func waitFD(fd int, events int16, stops ...*stopFlag) error { } return err } - for _, p := range pfds[1:] { - if p.Revents != 0 { - return errStopped - } - } switch r := pfds[0].Revents; { + case pfds[1].Revents != 0: + return errStopped case r&unix.POLLNVAL != 0: return unix.EBADF case r != 0: diff --git a/apps/daemon/internal/processshim/relay_linux.go b/apps/daemon/internal/processshim/relay_linux.go index 0bb37296b..5600816aa 100644 --- a/apps/daemon/internal/processshim/relay_linux.go +++ b/apps/daemon/internal/processshim/relay_linux.go @@ -340,6 +340,8 @@ func (inv *invocation) receive(m BrokerMessage) { inv.out[2].push(item{hold: e.written}) } inv.control(e) + case Notice: + inv.out[2].push(item{notice: m.Message}) case End: inv.end.set() inv.control(m) @@ -393,8 +395,6 @@ func (inv *invocation) run() { inv.rawOnce.Do(func() { close(inv.raw) }) case exiting: inv.exit(m) - case Notice: - inv.out[2].message(m.Message) case End: inv.finish("") return @@ -553,15 +553,16 @@ func (in *input) run() { } } -// item is an Output or Close for an output pump, or a hold. A local Close -// closes the descriptor without a report. A hold keeps the items behind it -// until it closes. +// item is an Output, Close or Notice for an output pump, or a hold. A local +// Close closes the descriptor without a report. A hold keeps the items +// behind it until it closes. type item struct { - seq uint64 - data []byte - close bool - local bool - hold <-chan struct{} + seq uint64 + data []byte + close bool + local bool + hold <-chan struct{} + notice []byte } // output writes one descriptor's Output in order and reports each write. @@ -630,6 +631,10 @@ func (o *output) run() { } continue } + if it.notice != nil { + o.message(it.notice) + continue + } if it.close { o.closeEP() if it.local { diff --git a/apps/daemon/internal/processshim/relay_linux_test.go b/apps/daemon/internal/processshim/relay_linux_test.go index 16288c006..2407e00da 100644 --- a/apps/daemon/internal/processshim/relay_linux_test.go +++ b/apps/daemon/internal/processshim/relay_linux_test.go @@ -5,6 +5,7 @@ package processshim import ( "bufio" "fmt" + "io" "net" "os" "path/filepath" @@ -299,6 +300,59 @@ func TestTerminalOutputWaitsForRawMode(t *testing.T) { } } +// A Notice reaches fd 2 after the stderr queued before it, even while that +// stderr waits in its pump and the control goroutine has moved on. +func TestNoticeFollowsQueuedStderr(t *testing.T) { + answered := make(chan struct{}) + r := startTestRelay(t, func(r *relay) { + r.gating = func() { + select { + case <-answered: + case <-time.After(10 * time.Second): + t.Error("the shim never got its Result") + } + } + }) + go func() { + for { + m, err := r.read() + if err != nil { + return + } + switch m := m.(type) { + case Open: + id := m.ID + // The Exit has no Marks, so its Result shows that the + // control goroutine is past the Notice. + r.send(Accept{ID: id}, Started{ID: id}, Output{ID: id, FD: 2, Seq: 1, Data: []byte("err\n")}, + Notice{ID: id, Message: []byte("note")}, Exit{ID: id, Result: Result{Code: 0}}, Close{ID: id, FD: 2, Seq: 2}) + case Written: + if m.Seq == 2 { + r.send(End{ID: m.ID}) + } + } + } + }() + pr, pw, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + defer pr.Close() + fds := devNull(t) + fds[2] = int(pw.Fd()) + conn := r.shim(t, "sh", fds) + pw.Close() + if res, err := finished(conn); err != nil || res.Code != 0 { + t.Fatalf("Result %+v, %v", res, err) + } + close(answered) + pr.SetDeadline(time.Now().Add(10 * time.Second)) + got, err := io.ReadAll(pr) + if want := "err\noac-process-shim: note\n"; err != nil || string(got) != want { + t.Fatalf("fd 2 got %q, %v; want %q", got, err, want) + } +} + // readN reads n bytes from the terminal's master side. It fails the test // when they do not arrive within 10s. func readN(t *testing.T, fd, n int) []byte { From 1694a3ec500376e2e1f2291a83d37b4eba78bc98 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 6 Oct 2026 23:01:03 +0000 Subject: [PATCH 2/8] Keep view teardown out of the broker's warning logs The broker no longer logs its relay's end, which every view teardown causes; the agent host logs a relay lost while the view runs at ERROR. Failures, a lost shim and redials caused by the broker stopping log nothing above DEBUG. --- apps/daemon/internal/agenthost/launch_linux.go | 1 + apps/daemon/internal/processbroker/broker_linux.go | 6 +----- apps/daemon/internal/processbroker/invocation_linux.go | 8 +++++--- apps/daemon/internal/processbroker/link_linux.go | 4 +++- apps/daemon/internal/processbroker/pumps_linux.go | 2 +- 5 files changed, 11 insertions(+), 10 deletions(-) diff --git a/apps/daemon/internal/agenthost/launch_linux.go b/apps/daemon/internal/agenthost/launch_linux.go index c48beb4a5..f1e23d3e6 100644 --- a/apps/daemon/internal/agenthost/launch_linux.go +++ b/apps/daemon/internal/agenthost/launch_linux.go @@ -325,6 +325,7 @@ func (h *ownedView) watch() { // and the relay was lost while the Harness ran. A failed one // means that the view is ending, and Wait reports how. if h.v.Signal(0) == nil { + h.s.log.Error("process relay lost", "error", h.broker.Err()) h.s.brokerFailed("relay", h.broker.Err()) return } diff --git a/apps/daemon/internal/processbroker/broker_linux.go b/apps/daemon/internal/processbroker/broker_linux.go index 332c49fda..7c852d590 100644 --- a/apps/daemon/internal/processbroker/broker_linux.go +++ b/apps/daemon/internal/processbroker/broker_linux.go @@ -112,14 +112,10 @@ func (b *Broker) read() { } } b.mu.Lock() - lost := !b.closing - if lost { + if !b.closing { b.err = fmt.Errorf("%w: %w", ErrRelayLost, err) } b.mu.Unlock() - if lost { - b.log.Error("process relay lost", "error", err) - } b.cancel() b.conn.Close() close(b.done) diff --git a/apps/daemon/internal/processbroker/invocation_linux.go b/apps/daemon/internal/processbroker/invocation_linux.go index eea43903a..26955ff5b 100644 --- a/apps/daemon/internal/processbroker/invocation_linux.go +++ b/apps/daemon/internal/processbroker/invocation_linux.go @@ -585,7 +585,7 @@ func (inv *invocation) releaseOp(h handle) bool { if f.Code == sp.CodeBusy && inv.sleep(backoff) { continue } - if f.Code != sp.CodeReleased && f.Code != sp.CodeBusy { + if f.Code != sp.CodeReleased && f.Code != sp.CodeBusy && inv.b.ctx.Err() == nil { inv.log.Warn("operation release failed", "error", err) } h.op.Detach() @@ -653,7 +653,9 @@ func (inv *invocation) answer(r processshim.Result, marks []processshim.Mark, be // reason is written even after the shim exited, because the Harness may still // read the output the failure cut short. func (inv *invocation) fail(reason string) { - inv.log.Warn("process invocation failed", "reason", reason) + if inv.b.ctx.Err() == nil { // a stopped broker fails every invocation + inv.log.Warn("process invocation failed", "reason", reason) + } inv.halted() if !inv.reply(processshim.Result{Code: processshim.ExitLost, Message: message(reason)}, nil) { inv.send(processshim.Notice{ID: inv.rid, Message: message(reason)}) @@ -673,7 +675,7 @@ func (inv *invocation) shimGone() { inv.loseShim() cancel := !inv.exited inv.mu.Unlock() - inv.log.Info("process shim lost", "cancel", cancel) + inv.log.Debug("process shim lost", "cancel", cancel) inv.stopInput() if cancel { inv.cancelRemote() diff --git a/apps/daemon/internal/processbroker/link_linux.go b/apps/daemon/internal/processbroker/link_linux.go index b4abd56fe..652fd4738 100644 --- a/apps/daemon/internal/processbroker/link_linux.go +++ b/apps/daemon/internal/processbroker/link_linux.go @@ -67,7 +67,9 @@ func (l *link) get(ctx context.Context) (*stream, error) { l.mu.Unlock() return s, nil } - l.log.Warn("process stream unavailable", "error", err, "retry", backoff) + if ctx.Err() == nil { + l.log.Warn("process stream unavailable", "error", err, "retry", backoff) + } select { case <-time.After(backoff): case <-ctx.Done(): diff --git a/apps/daemon/internal/processbroker/pumps_linux.go b/apps/daemon/internal/processbroker/pumps_linux.go index 16e85cf94..3df3b61e3 100644 --- a/apps/daemon/internal/processbroker/pumps_linux.go +++ b/apps/daemon/internal/processbroker/pumps_linux.go @@ -436,7 +436,7 @@ func (inv *invocation) signal(s processshim.Signaled) { if err == nil { return } - if f := asFailure(err); f.Code != sp.CodeNotRunning && f.Code != sp.CodeReleased { + if f := asFailure(err); f.Code != sp.CodeNotRunning && f.Code != sp.CodeReleased && inv.b.ctx.Err() == nil { inv.log.Info("signal not delivered", "signal", n, "error", err) } } From ba24735c43cae795e1317df8575ecfad5ac6b7f5 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 6 Oct 2026 23:08:05 +0000 Subject: [PATCH 3/8] Script the broker's fault tests with a single-operation peer The fake process service gives way to a peer with one operation that scripts the four faults its tests need. The fixture takes a dial function, so those tests never start processserve, and a Busy acknowledgement comes from intercept instead of a rewritten response. --- .../processbroker/broker_linux_test.go | 122 +++--- .../processbroker/fakeservice_linux_test.go | 410 ------------------ .../internal/processbroker/peer_linux_test.go | 281 ++++++++++++ .../internal/processbroker/view_linux_test.go | 6 +- 4 files changed, 339 insertions(+), 480 deletions(-) delete mode 100644 apps/daemon/internal/processbroker/fakeservice_linux_test.go create mode 100644 apps/daemon/internal/processbroker/peer_linux_test.go diff --git a/apps/daemon/internal/processbroker/broker_linux_test.go b/apps/daemon/internal/processbroker/broker_linux_test.go index f7c076e9b..2ee9e164b 100644 --- a/apps/daemon/internal/processbroker/broker_linux_test.go +++ b/apps/daemon/internal/processbroker/broker_linux_test.go @@ -262,7 +262,13 @@ func TestEnvironmentIsDeclaredOnly(t *testing.T) { } func TestLinkLossKeepsOutputOrdered(t *testing.T) { - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { return &cutConn{Conn: c, left: 64 << 10} })) + var dials atomic.Int32 + f := newFixture(t, dialService(t, func(n int32, c net.Conn) io.ReadWriteCloser { + if dials.Store(n); n > 1 { + return c + } + return &cutConn{Conn: c, left: 64 << 10} + })) out, err := f.command("seq", "1", "2000000").Output() if err != nil { t.Fatalf("Output = %v", err) @@ -275,7 +281,7 @@ func TestLinkLossKeepsOutputOrdered(t *testing.T) { if !bytes.Equal(out, want) { t.Fatalf("output differs: %d bytes, want %d", len(out), len(want)) } - if n := f.dials.Load(); n < 2 { + if n := dials.Load(); n < 2 { t.Fatalf("%d dials; the link was never cut", n) } } @@ -325,7 +331,7 @@ func TestIncompleteRequestHoldsNothing(t *testing.T) { // A gone output reader must not hold the exit back while the stream that // would close the remote output is lost behind the settled operation. func TestGoneReaderDoesNotHoldExit(t *testing.T) { - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { return &holdEvents{Conn: c} })) + f := newFixture(t, dialService(t, first(func(c net.Conn) io.ReadWriteCloser { return &holdEvents{Conn: c} }))) r, w, err := os.Pipe() if err != nil { t.Fatal(err) @@ -555,7 +561,15 @@ func TestRelayDescriptorsAreDiscarded(t *testing.T) { // A Busy acknowledgement is retried, or output past the replay limit would // never arrive. func TestBusyAckIsRetried(t *testing.T) { - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { return &busyAck{Conn: c} })) + var acks atomic.Int32 + f := newFixture(t, dialService(t, first(func(c net.Conn) io.ReadWriteCloser { + return intercept(c, func(fr sandboxwire.Frame) verdict { + if fr.Type == sp.OpAckEvents && acks.Add(1) == 1 { + return refuseBusy + } + return pass + }) + }))) cmd := f.command("seq", "1", "2000000") var out countWriter cmd.Stdout = &out @@ -631,14 +645,14 @@ func TestSharedTerminalRestoredByLastUser(t *testing.T) { // accept, so the program reads every byte once. func TestBusyStdinIsRetried(t *testing.T) { var writes atomic.Int32 - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { + f := newFixture(t, dialService(t, first(func(c net.Conn) io.ReadWriteCloser { return intercept(c, func(fr sandboxwire.Frame) verdict { if fr.Type == sp.OpWriteStdin && writes.Add(1) == 2 { return refuseBusy } return pass }) - })) + }))) var in []byte for i := range 30000 { in = strconv.AppendInt(in, int64(i), 10) @@ -668,14 +682,14 @@ func TestBusyStdinIsRetried(t *testing.T) { // scope gets one TERM, and KILL after the grace. func TestUncertainCancelIsNotReplayed(t *testing.T) { var cancels atomic.Int32 - f := newFixture(t, func(_ int32, c net.Conn) io.ReadWriteCloser { + f := newFixture(t, dialService(t, func(_ int32, c net.Conn) io.ReadWriteCloser { return intercept(c, func(fr sandboxwire.Frame) verdict { if fr.Type == sp.OpCancel && cancels.Add(1) == 1 { return loseResponse } return pass }) - }) + })) r, w, err := os.Pipe() if err != nil { t.Fatal(err) @@ -713,7 +727,7 @@ func TestUncertainCancelIsNotReplayed(t *testing.T) { // invocation: the broker closes the shim's descriptors. func TestShimLossEndsWaitForStream(t *testing.T) { dialed := make(chan struct{}) - f := newFixture(t, func(n int32, c net.Conn) io.ReadWriteCloser { + f := newFixture(t, dialService(t, func(n int32, c net.Conn) io.ReadWriteCloser { if n > 1 { return c } @@ -723,7 +737,7 @@ func TestShimLossEndsWaitForStream(t *testing.T) { broker, svc := net.Pipe() go io.Copy(io.Discard, svc) return broker - }) + })) r, w, err := os.Pipe() if err != nil { t.Fatal(err) @@ -766,9 +780,9 @@ func TestShimLossEndsWaitForStream(t *testing.T) { func TestRetriedStartWaitsForStarted(t *testing.T) { for _, fails := range []bool{false, true} { t.Run(fmt.Sprintf("start fails %v", fails), func(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.startLate, svc.startFails = true, fails - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpStart })) + f := newFixture(t, svc.dial(sp.OpStart)) // Once the retried Start's Attach found the operation starting, // it starts when the broker observes it without forwarding stdin. // A broker that forwards stdin first gets the refusal it counts. @@ -819,14 +833,14 @@ func awaitsStarted() bool { // resumes from the offset Inspect reports: the program reads every byte // once, and stdin closes at the end of the input. func TestUncertainStdinWriteResumes(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.partialFirst = true - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpWriteStdin })) + f := newFixture(t, svc.dial(sp.OpWriteStdin)) const in = "stdin that crosses a lost stream\n" out, stderr, err := runFor(t, f.command("sh", "-c", "cat"), in) c := svc.counts() - if err != nil || out != in || string(c.stdin) != in || c.closedAt != uint64(len(in)) || c.writes < 2 { - t.Fatalf("Run = %v after %d writes; stdout %q, stderr %q; the service took %q and closed stdin at %d", err, c.writes, out, stderr, c.stdin, c.closedAt) + if err != nil || out != in || string(c.stdin) != in || !c.st.StdinClosed || c.writes < 2 { + t.Fatalf("Run = %v after %d writes; stdout %q, stderr %q; the service took %q, stdin closed %v", err, c.writes, out, stderr, c.stdin, c.st.StdinClosed) } } @@ -834,9 +848,9 @@ func TestUncertainStdinWriteResumes(t *testing.T) { // shim exits with 255 and the reason, and the program is cancelled, rather // than both waiting for input that never comes. func TestUnresolvedStdinEndsTheInvocation(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.partialFirst, svc.inspectFails = true, true - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpWriteStdin })) + f := newFixture(t, svc.dial(sp.OpWriteStdin)) _, stderr, err := runFor(t, f.command("sh", "-c", "cat"), "stdin\n") if exitCode(err) != processshim.ExitLost || !strings.Contains(stderr, "stdin could not be resumed: inspect failed") { t.Fatalf("Run = %v; stderr %q", err, stderr) @@ -849,17 +863,17 @@ func TestUnresolvedStdinEndsTheInvocation(t *testing.T) { // before its Exited event arrives: stdin closes after the bytes the service // took, the background process gets end of file, and the operation settles. func TestBackgroundReaderGetsEOFAfterUncertainWrite(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.leaderExits = true - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpWriteStdin })) + f := newFixture(t, svc.dial(sp.OpWriteStdin)) const in = "stdin for a background reader\n" out, stderr, err := runFor(t, f.command("sh", "-c", "cat"), in) if err != nil || out != in { t.Fatalf("Run = %v; stdout %q, stderr %q", err, out, stderr) } - await(t, "the operation's release", func() bool { return svc.counts().released }) - if c := svc.counts(); c.closedAt != uint64(len(in)) { - t.Fatalf("stdin closed at %d, not %d", c.closedAt, len(in)) + await(t, "the operation's release", func() bool { return svc.counts().st.Released }) + if st := svc.counts().st; !st.StdinClosed || st.StdinOffset != uint64(len(in)) { + t.Fatalf("stdin closed %v at %d, not %d", st.StdinClosed, st.StdinOffset, len(in)) } } @@ -890,19 +904,19 @@ func runFor(t *testing.T, cmd *exec.Cmd, in string) (string, string, error) { type fixture struct { dir, bin string uid, gid int // the relay's - service string - wrap func(int32, net.Conn) io.ReadWriteCloser - dials atomic.Int32 relay *exec.Cmd broker *Broker } // newFixture starts a relay and its broker, whose shims are this binary -// under the names bash, sh, env and seq. As root the relay runs as viewID. A -// non-nil wrap wraps each stream to the service, which it gets with the -// stream's number, counting from 1. -func newFixture(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) *fixture { +// under the names bash, sh, env and seq. As root the relay runs as viewID. +// The broker reaches the process service through dial, or the shared one +// when dial is nil. +func newFixture(t *testing.T, dial func(context.Context) (io.ReadWriteCloser, error)) *fixture { t.Helper() + if dial == nil { + dial = dialService(t, nil) + } self, err := os.Executable() if err != nil { t.Fatal(err) @@ -915,7 +929,7 @@ func newFixture(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) *fi if err := os.Chmod(dir, 0o755); err != nil { t.Fatal(err) } - f := &fixture{dir: dir, uid: os.Getuid(), gid: os.Getgid(), service: service.socket(t), wrap: wrap} + f := &fixture{dir: dir, uid: os.Getuid(), gid: os.Getgid()} if f.uid == 0 { f.uid, f.gid = viewID, viewID } @@ -941,7 +955,7 @@ func newFixture(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) *fi Tool: map[string]string{"TOOL": "1"}, }, Scope: sp.ScopePOSIXSession, - Dial: f.dial, + Dial: dial, CancelGrace: time.Second, }) end.Close() @@ -1024,15 +1038,18 @@ func (f *fixture) command(name string, args ...string) *exec.Cmd { return cmd } -func (f *fixture) dial(ctx context.Context) (io.ReadWriteCloser, error) { - c, err := new(net.Dialer).DialContext(ctx, "unix", f.service) - if err != nil { - return nil, err - } - if n := f.dials.Add(1); f.wrap != nil { - return f.wrap(n, c), nil +// dialService dials the shared process service. A non-nil wrap wraps each +// stream, which it gets with the stream's number, counting from 1. +func dialService(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) func(context.Context) (io.ReadWriteCloser, error) { + sock := service.socket(t) + var dials atomic.Int32 + return func(ctx context.Context) (io.ReadWriteCloser, error) { + c, err := new(net.Dialer).DialContext(ctx, "unix", sock) + if err != nil || wrap == nil { + return c, err + } + return wrap(dials.Add(1), c), nil } - return c, nil } // first applies wrap to the first stream alone. @@ -1155,31 +1172,6 @@ func (c *holdEvents) Read(p []byte) (int, error) { return c.out.Read(p) } -// busyAck answers the first acknowledgement with Busy after the service -// applied it. -type busyAck struct { - net.Conn - out bytes.Buffer - sent bool -} - -func (c *busyAck) Read(p []byte) (int, error) { - for c.out.Len() == 0 { - fr, err := sandboxwire.ReadFrame(c.Conn, sandboxwire.MaxPayload) - if err != nil { - return 0, err - } - if !c.sent && fr.Type == sandboxwire.ResponseType(sp.OpAckEvents) { - c.sent = true - fr.Payload = sp.Encode(sp.ResponseFailure{Request: sp.OpAckEvents, Failure: *sp.Fail(sp.CodeBusy, sandboxwire.EffectNone, "busy")}) - } - if err := sandboxwire.WriteFrame(&c.out, fr); err != nil { - return 0, err - } - } - return c.out.Read(p) -} - type countWriter struct{ n int } func (w *countWriter) Write(p []byte) (int, error) { diff --git a/apps/daemon/internal/processbroker/fakeservice_linux_test.go b/apps/daemon/internal/processbroker/fakeservice_linux_test.go deleted file mode 100644 index 0e1e1a95b..000000000 --- a/apps/daemon/internal/processbroker/fakeservice_linux_test.go +++ /dev/null @@ -1,410 +0,0 @@ -//go:build linux - -package processbroker - -import ( - "context" - "io" - "net" - "slices" - "sync" - - sp "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" - "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" -) - -// fakeService is a process service whose program copies stdin to stdout and -// exits 0 when stdin closes. Its knobs reproduce one service behavior each. -type fakeService struct { - instance sandboxwire.ID - caps sp.Capabilities - - // startLate keeps an operation Starting until runStarting, or until - // stdin is written, which it refuses. attachedStarting closes once an - // Attach finds the operation Starting. - startLate bool - attachedStarting chan struct{} - // startFails ends the start with StartFailed instead of Started. - startFails bool - // partialFirst accepts half of the first stdin write. - partialFirst bool - // inspectFails refuses Inspect for good. - inspectFails bool - // leaderExits exits the leader with 0 as it takes the first stdin - // write. A background process it leaves copies the rest of stdin to - // stdout until end of file. - leaderExits bool - - mu sync.Mutex - ops map[sandboxwire.ID]*fakeOp - writes int // stdin writes the service took - cancels int - refusedIn int // stdin requests refused while Starting -} - -type fakeOp struct { - id sandboxwire.ID - state sp.OperationState - events []sp.Event // events[i] has sequence i+1 - changed chan struct{} - stdin []byte - stdinClosed bool - closedAt uint64 // the CloseStdin offset - stdout uint64 // the stdout offset - exit *sp.ExitStatus - failure *sp.Failure - background bool // a background process outlives the leader - released bool -} - -func newFakeService() *fakeService { - return &fakeService{ - instance: sandboxwire.NewID(), - caps: sp.Capabilities{ - Platform: sp.PlatformLinux, - Scopes: []sp.Scope{sp.ScopePOSIXSession}, - IOModes: []sp.IOMode{sp.IOPipes}, - Signals: []sp.Signal{1, 2, 15}, - SignalTargets: []sp.SignalTarget{sp.TargetInitialProcessGroup}, - MaxStartBytes: sandboxwire.MaxPayload, - MaxDataBytes: sandboxwire.MaxChunk, - MaxActiveOperations: 8, - MaxOperationRecords: 8, - MaxReplayBytesPerOperation: 1 << 20, - OwnerLossGraceMillis: 60000, - CancelGraceLimitMillis: 60000, - }, - ops: map[sandboxwire.ID]*fakeOp{}, - attachedStarting: make(chan struct{}), - } -} - -// serve replaces each stream to the real service with one to s, and loses -// the first stream in place of the response to the first request that lose -// selects. -func (s *fakeService) serve(ctx context.Context, lose func(sandboxwire.Frame) bool) func(int32, net.Conn) io.ReadWriteCloser { - return func(n int32, c net.Conn) io.ReadWriteCloser { - c.Close() - broker, svc := net.Pipe() - go sp.Serve(ctx, svc, sp.Attachment{ID: s.instance}, s) - if n > 1 || lose == nil { - return broker - } - return intercept(broker, func(fr sandboxwire.Frame) verdict { - if lose(fr) { - return loseResponse - } - return pass - }) - } -} - -// emit appends an event; s.mu is held. -func (s *fakeService) emit(op *fakeOp, ev func(sp.EventHeader) sp.Event) { - op.events = append(op.events, ev(sp.EventHeader{OperationID: op.id, Sequence: uint64(len(op.events) + 1)})) - close(op.changed) - op.changed = make(chan struct{}) -} - -// subscribe sends op's events after seq on conn until the stream ends. -func (s *fakeService) subscribe(conn *sp.Conn, op *fakeOp, after uint64) { - go func() { - next := after - for { - s.mu.Lock() - evs := slices.Clone(op.events[min(next, uint64(len(op.events))):]) - changed := op.changed - s.mu.Unlock() - for _, ev := range evs { - if conn.Send(ev) != nil { - return - } - next++ - } - select { - case <-changed: - case <-conn.Context().Done(): - return - } - } - }() -} - -// run ends the start; s.mu is held. -func (s *fakeService) run(op *fakeOp) { - if op.state != sp.StateStarting { - return - } - if s.startFails { - op.state = sp.StateStartFailed - op.failure = sp.Fail(sp.CodeNotFound, sandboxwire.EffectNone, "no such file") - f := *op.failure - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.StartFailedEvent{EventHeader: h, Failure: f} }) - return - } - op.state = sp.StateRunning - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.StartedEvent{EventHeader: h} }) -} - -// runStarting ends every start in progress. -func (s *fakeService) runStarting() { - s.mu.Lock() - defer s.mu.Unlock() - for _, op := range s.ops { - s.run(op) - } -} - -// exited ends the program with status, or only its background process -// when the leader already exited; s.mu is held. -func (s *fakeService) exited(op *fakeOp, status sp.ExitStatus) { - switch { - case op.background: - op.background = false - s.drained(op, nil) - case op.state == sp.StateRunning: - op.state, op.exit = sp.StateExited, &status - s.drained(op, &status) - } -} - -// drained closes the output and the scope, with the leader's exit between -// them when it exits now; s.mu is held. -func (s *fakeService) drained(op *fakeOp, exit *sp.ExitStatus) { - out := op.stdout - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Disposition: sp.OutputDrained} - }) - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStderr, Disposition: sp.OutputDrained} - }) - if exit != nil { - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: *exit} }) - } - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.OutputClosedEvent{EventHeader: h, Disposition: sp.OutputDrained} - }) - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.ScopeClosedEvent{EventHeader: h} }) -} - -func (s *fakeService) status(op *fakeOp) sp.OperationStatus { - st := sp.OperationStatus{ - State: op.state, Exit: op.exit, StartFailure: op.failure, - StdinOffset: uint64(len(op.stdin)), StdinClosed: op.stdinClosed, - Scope: sp.ScopeStateActive, Released: op.released, - FirstRetained: 1, LastSequence: uint64(len(op.events)), - } - if op.state == sp.StateExited && !op.background || op.state == sp.StateStartFailed { - d := sp.OutputDrained - st.Output, st.Scope = &d, sp.ScopeStateClosed - } - return st -} - -// lookup returns the operation with s.mu held. -func (s *fakeService) lookup(ref sp.OperationRef) (*fakeOp, error) { - if ref.ServerInstanceID != s.instance { - return nil, sp.Fail(sp.CodeInstanceChanged, sandboxwire.EffectNone, "instance changed") - } - s.mu.Lock() - op := s.ops[ref.OperationID] - if op == nil { - s.mu.Unlock() - return nil, sp.Fail(sp.CodeNotFound, sandboxwire.EffectNone, "no operation") - } - if op.released { - s.mu.Unlock() - return nil, sp.Fail(sp.CodeReleased, sandboxwire.EffectNone, "released") - } - return op, nil -} - -func (s *fakeService) Describe(context.Context, *sp.Conn, sp.DescribeRequest) (sp.DescribeResponse, error) { - return sp.DescribeResponse{ServerInstanceID: s.instance, Capabilities: s.caps}, nil -} - -func (s *fakeService) Start(_ context.Context, conn *sp.Conn, req sp.StartRequest) (sp.StartResponse, error) { - if req.ServerInstanceID != s.instance { - return sp.StartResponse{}, sp.Fail(sp.CodeInstanceChanged, sandboxwire.EffectNone, "instance changed") - } - s.mu.Lock() - defer s.mu.Unlock() - if s.ops[req.OperationID] != nil { - return sp.StartResponse{Disposition: sp.StartExisting}, nil - } - op := &fakeOp{id: req.OperationID, state: sp.StateStarting, changed: make(chan struct{})} - s.ops[op.id] = op - s.subscribe(conn, op, 0) - if !s.startLate { - s.run(op) - } - return sp.StartResponse{Disposition: sp.StartCreated}, nil -} - -func (s *fakeService) Attach(_ context.Context, conn *sp.Conn, req sp.AttachRequest) (sp.AttachResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.AttachResponse{}, err - } - defer s.mu.Unlock() - s.subscribe(conn, op, req.AfterSequence) - if op.state == sp.StateStarting { - select { - case <-s.attachedStarting: - default: - close(s.attachedStarting) - } - } - return sp.AttachResponse{Status: s.status(op)}, nil -} - -func (s *fakeService) Inspect(_ context.Context, _ *sp.Conn, req sp.InspectRequest) (sp.InspectResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.InspectResponse{}, err - } - defer s.mu.Unlock() - if s.inspectFails { - return sp.InspectResponse{}, sp.Fail(sp.CodeIO, sandboxwire.EffectNone, "inspect failed") - } - return sp.InspectResponse{Status: s.status(op)}, nil -} - -// stdinRefused refuses stdin while the operation starts, which also ends a -// late start; s.mu is held. -func (s *fakeService) stdinRefused(op *fakeOp) error { - switch { - case op.state == sp.StateStarting: - s.refusedIn++ - s.run(op) - return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is starting") - case op.state != sp.StateRunning && !op.background: - return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is not running") - case op.stdinClosed: - return sp.Fail(sp.CodeStdinClosed, sandboxwire.EffectNone, "stdin is closed") - } - return nil -} - -func (s *fakeService) WriteStdin(_ context.Context, _ *sp.Conn, req sp.WriteStdinRequest) (sp.WriteStdinResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.WriteStdinResponse{}, err - } - defer s.mu.Unlock() - if err := s.stdinRefused(op); err != nil { - return sp.WriteStdinResponse{}, err - } - if req.Offset != uint64(len(op.stdin)) { - return sp.WriteStdinResponse{}, sp.Fail(sp.CodeInputOffsetConflict, sandboxwire.EffectNone, "offset %d is not %d", req.Offset, len(op.stdin)) - } - data := req.Data - if s.writes++; s.writes == 1 && s.partialFirst { - data = data[:len(data)/2] - } - op.stdin = append(op.stdin, data...) - if len(data) > 0 { - out, echo := op.stdout, slices.Clone(data) - op.stdout += uint64(len(data)) - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.OutputEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Data: echo} - }) - } - if s.leaderExits && op.state == sp.StateRunning { - status := sp.ExitStatus{Kind: sp.ExitCode} - op.state, op.exit, op.background = sp.StateExited, &status, true - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: status} }) - } - return sp.WriteStdinResponse{Accepted: uint32(len(data))}, nil -} - -func (s *fakeService) CloseStdin(_ context.Context, _ *sp.Conn, req sp.CloseStdinRequest) (sp.CloseStdinResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.CloseStdinResponse{}, err - } - defer s.mu.Unlock() - if op.stdinClosed && req.Offset == op.closedAt { - return sp.CloseStdinResponse{}, nil - } - if err := s.stdinRefused(op); err != nil { - return sp.CloseStdinResponse{}, err - } - if req.Offset != uint64(len(op.stdin)) { - return sp.CloseStdinResponse{}, sp.Fail(sp.CodeInputOffsetConflict, sandboxwire.EffectNone, "offset %d is not %d", req.Offset, len(op.stdin)) - } - op.stdinClosed, op.closedAt = true, req.Offset - s.exited(op, sp.ExitStatus{Kind: sp.ExitCode}) - return sp.CloseStdinResponse{}, nil -} - -func (s *fakeService) CloseOutput(_ context.Context, _ *sp.Conn, req sp.CloseOutputRequest) (sp.CloseOutputResponse, error) { - if _, err := s.lookup(req.OperationRef); err != nil { - return sp.CloseOutputResponse{}, err - } - s.mu.Unlock() - return sp.CloseOutputResponse{}, nil -} - -func (s *fakeService) ResizePTY(context.Context, *sp.Conn, sp.ResizePTYRequest) (sp.ResizePTYResponse, error) { - return sp.ResizePTYResponse{}, sp.Fail(sp.CodeUnsupported, sandboxwire.EffectNone, "no terminal") -} - -func (s *fakeService) Signal(_ context.Context, _ *sp.Conn, req sp.SignalRequest) (sp.SignalResponse, error) { - if _, err := s.lookup(req.OperationRef); err != nil { - return sp.SignalResponse{}, err - } - s.mu.Unlock() - return sp.SignalResponse{}, nil -} - -// Cancel ends the program as TERM does. -func (s *fakeService) Cancel(_ context.Context, _ *sp.Conn, req sp.CancelRequest) (sp.CancelResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.CancelResponse{}, err - } - defer s.mu.Unlock() - s.cancels++ - s.exited(op, sp.ExitStatus{Kind: sp.ExitSignal, Signal: 15}) - return sp.CancelResponse{}, nil -} - -func (s *fakeService) AckEvents(_ context.Context, _ *sp.Conn, req sp.AckEventsRequest) (sp.AckEventsResponse, error) { - if _, err := s.lookup(req.OperationRef); err != nil { - return sp.AckEventsResponse{}, err - } - s.mu.Unlock() - return sp.AckEventsResponse{}, nil -} - -func (s *fakeService) Release(_ context.Context, _ *sp.Conn, req sp.ReleaseRequest) (sp.ReleaseResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.ReleaseResponse{}, err - } - defer s.mu.Unlock() - if op.state != sp.StateExited && op.state != sp.StateStartFailed || op.background { - return sp.ReleaseResponse{}, sp.Fail(sp.CodeBusy, sandboxwire.EffectNone, "not settled") - } - op.released = true - return sp.ReleaseResponse{}, nil -} - -// fakeCounts is what a fakeService saw. -type fakeCounts struct { - writes, cancels, refusedIn int - stdin []byte // the only operation's accepted stdin - closedAt uint64 // and its CloseStdin offset - released bool // and whether it was released -} - -func (s *fakeService) counts() fakeCounts { - s.mu.Lock() - defer s.mu.Unlock() - c := fakeCounts{writes: s.writes, cancels: s.cancels, refusedIn: s.refusedIn} - for _, op := range s.ops { - c.stdin, c.closedAt, c.released = slices.Clone(op.stdin), op.closedAt, op.released - } - return c -} diff --git a/apps/daemon/internal/processbroker/peer_linux_test.go b/apps/daemon/internal/processbroker/peer_linux_test.go new file mode 100644 index 000000000..dd710444e --- /dev/null +++ b/apps/daemon/internal/processbroker/peer_linux_test.go @@ -0,0 +1,281 @@ +//go:build linux + +package processbroker + +import ( + "context" + "io" + "net" + "slices" + "sync" + "sync/atomic" + + sp "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// peer is a process service with a single operation, scripted for the +// faults the uncertain-request tests need. Its program copies stdin to +// stdout and exits 0 when stdin closes. +type peer struct { + // startLate keeps the operation Starting until runStarting, or until + // stdin is written, which it refuses. attachedStarting closes once an + // Attach finds the operation Starting. + startLate bool + attachedStarting chan struct{} + startFails bool // the start ends with StartFailed + partialFirst bool // the first stdin write is half accepted + inspectFails bool // Inspect fails for good + // leaderExits exits the leader with 0 as it takes the first stdin + // write. A background process it leaves copies the rest of stdin to + // stdout until end of file. + leaderExits bool + + instance sandboxwire.ID + + mu sync.Mutex + seen + id sandboxwire.ID + events []sp.Event // events[i] has sequence i+1 + out net.Conn // the stream subscribed to the events + background bool // a background process outlives the leader +} + +// seen is what the peer saw. +type seen struct { + st sp.OperationStatus + stdin []byte // the stdin the service took + writes, cancels, refusedIn int // refusedIn counts stdin refused while Starting +} + +func newPeer() *peer { + return &peer{instance: sandboxwire.NewID(), attachedStarting: make(chan struct{}), + seen: seen{st: sp.OperationStatus{State: sp.StateStarting, Scope: sp.ScopeStateActive, FirstRetained: 1}}} +} + +func (p *peer) counts() seen { + p.mu.Lock() + defer p.mu.Unlock() + s := p.seen + s.stdin = slices.Clone(s.stdin) + return s +} + +// dial serves each stream the broker dials. The first stream is lost in +// place of the response to its first request of type lose. +func (p *peer) dial(lose uint16) func(context.Context) (io.ReadWriteCloser, error) { + var dials atomic.Int32 + return func(context.Context) (io.ReadWriteCloser, error) { + broker, svc := net.Pipe() + go p.serve(svc, dials.Add(1) == 1, lose) + return broker, nil + } +} + +func (p *peer) serve(c net.Conn, first bool, lose uint16) { + defer c.Close() + for { + f, err := sandboxwire.ReadFrame(c, sandboxwire.MaxPayload) + if err != nil { + return + } + m, err := sp.Decode(f.Type, f.Payload) + if err != nil { + return + } + p.mu.Lock() + resp, failure := p.handle(m) + if first && f.Type == lose { + p.mu.Unlock() + return + } + if failure != nil { + resp = sp.ResponseFailure{Request: f.Type, Failure: *failure} + } + send(c, f.RequestID, resp) + switch m := m.(type) { + case sp.StartRequest: + if resp == (sp.StartResponse{Disposition: sp.StartCreated}) { + p.subscribe(c, 0) + } + case sp.AttachRequest: + p.subscribe(c, m.AfterSequence) + } + p.mu.Unlock() + } +} + +// subscribe sends c the events after seq, and every later one; p.mu is +// held. +func (p *peer) subscribe(c net.Conn, seq uint64) { + p.out = c + for _, ev := range p.events[seq:] { + send(c, 0, ev) + } +} + +func send(c net.Conn, requestID uint64, m sp.Message) { + sandboxwire.WriteFrame(c, sandboxwire.Frame{Type: m.MessageType(), RequestID: requestID, Payload: sp.Encode(m)}) +} + +// handle applies one request; p.mu is held. +func (p *peer) handle(m sp.Message) (sp.Message, *sp.Failure) { + switch m := m.(type) { + case sp.DescribeRequest: + return sp.DescribeResponse{ServerInstanceID: p.instance, Capabilities: sp.Capabilities{ + Platform: sp.PlatformLinux, + Scopes: []sp.Scope{sp.ScopePOSIXSession}, + IOModes: []sp.IOMode{sp.IOPipes}, + Signals: []sp.Signal{1, 2, 15}, + SignalTargets: []sp.SignalTarget{sp.TargetInitialProcessGroup}, + MaxStartBytes: sandboxwire.MaxPayload, + MaxDataBytes: sandboxwire.MaxChunk, + MaxActiveOperations: 8, + MaxOperationRecords: 8, + MaxReplayBytesPerOperation: 1 << 20, + OwnerLossGraceMillis: 60000, + CancelGraceLimitMillis: 60000, + }}, nil + case sp.StartRequest: + if !p.id.IsZero() { + return sp.StartResponse{Disposition: sp.StartExisting}, nil + } + p.id = m.OperationID + if !p.startLate { + p.run() + } + return sp.StartResponse{Disposition: sp.StartCreated}, nil + case sp.AttachRequest: + if p.st.State == sp.StateStarting { + select { + case <-p.attachedStarting: + default: + close(p.attachedStarting) + } + } + return sp.AttachResponse{Status: p.st}, nil + case sp.InspectRequest: + if p.inspectFails { + return nil, sp.Fail(sp.CodeIO, sandboxwire.EffectNone, "inspect failed") + } + return sp.InspectResponse{Status: p.st}, nil + case sp.WriteStdinRequest: + if f := p.refuseStdin(m.Offset); f != nil { + return nil, f + } + data := m.Data + if p.writes++; p.writes == 1 && p.partialFirst { + data = data[:len(data)/2] + } + p.stdin = append(p.stdin, data...) + if len(data) > 0 { + out, echo := p.st.StdinOffset, slices.Clone(data) + p.emit(func(h sp.EventHeader) sp.Event { + return sp.OutputEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Data: echo} + }) + } + p.st.StdinOffset += uint64(len(data)) + if p.leaderExits && p.st.State == sp.StateRunning { + status := sp.ExitStatus{Kind: sp.ExitCode} + p.st.State, p.st.Exit, p.background = sp.StateExited, &status, true + p.emit(func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: status} }) + } + return sp.WriteStdinResponse{Accepted: uint32(len(data))}, nil + case sp.CloseStdinRequest: + if f := p.refuseStdin(m.Offset); f != nil { + return nil, f + } + p.st.StdinClosed = true + p.exit(sp.ExitStatus{Kind: sp.ExitCode}) + return sp.CloseStdinResponse{}, nil + case sp.CancelRequest: + p.cancels++ + p.exit(sp.ExitStatus{Kind: sp.ExitSignal, Signal: 15}) + return sp.CancelResponse{}, nil + case sp.AckEventsRequest: + return sp.AckEventsResponse{}, nil + case sp.ReleaseRequest: + if p.st.Scope != sp.ScopeStateClosed { + return nil, sp.Fail(sp.CodeBusy, sandboxwire.EffectNone, "not settled") + } + p.st.Released = true + return sp.ReleaseResponse{}, nil + } + return nil, sp.Fail(sp.CodeUnsupported, sandboxwire.EffectNone, "not scripted") +} + +// refuseStdin refuses stdin while the operation starts, which also ends a +// late start; p.mu is held. +func (p *peer) refuseStdin(offset uint64) *sp.Failure { + switch { + case p.st.State == sp.StateStarting: + p.refusedIn++ + p.run() + return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is starting") + case p.st.State != sp.StateRunning && !p.background: + return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is not running") + case p.st.StdinClosed: + return sp.Fail(sp.CodeStdinClosed, sandboxwire.EffectNone, "stdin is closed") + case offset != p.st.StdinOffset: + return sp.Fail(sp.CodeInputOffsetConflict, sandboxwire.EffectNone, "offset %d is not %d", offset, p.st.StdinOffset) + } + return nil +} + +// runStarting ends a start in progress. +func (p *peer) runStarting() { + p.mu.Lock() + defer p.mu.Unlock() + p.run() +} + +// run ends the start; p.mu is held. +func (p *peer) run() { + switch { + case p.st.State != sp.StateStarting: + case p.startFails: + f := *sp.Fail(sp.CodeNotFound, sandboxwire.EffectNone, "no such file") + d := sp.OutputDrained + p.st.State, p.st.StartFailure, p.st.Output, p.st.Scope = sp.StateStartFailed, &f, &d, sp.ScopeStateClosed + p.emit(func(h sp.EventHeader) sp.Event { return sp.StartFailedEvent{EventHeader: h, Failure: f} }) + default: + p.st.State = sp.StateRunning + p.emit(func(h sp.EventHeader) sp.Event { return sp.StartedEvent{EventHeader: h} }) + } +} + +// exit ends the program with status, or only its background process when +// the leader already exited: the output and then the scope close, with +// the leader's exit between them when it exits now; p.mu is held. +func (p *peer) exit(status sp.ExitStatus) { + leader := p.st.State == sp.StateRunning + if !leader && !p.background { + return + } + p.background = false + out, d := p.st.StdinOffset, sp.OutputDrained + p.emit(func(h sp.EventHeader) sp.Event { + return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Disposition: d} + }) + p.emit(func(h sp.EventHeader) sp.Event { + return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStderr, Disposition: d} + }) + if leader { + p.st.State, p.st.Exit = sp.StateExited, &status + p.emit(func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: status} }) + } + p.emit(func(h sp.EventHeader) sp.Event { return sp.OutputClosedEvent{EventHeader: h, Disposition: d} }) + p.emit(func(h sp.EventHeader) sp.Event { return sp.ScopeClosedEvent{EventHeader: h} }) + p.st.Output, p.st.Scope = &d, sp.ScopeStateClosed +} + +// emit records an event and sends it to the subscribed stream; p.mu is +// held. +func (p *peer) emit(ev func(sp.EventHeader) sp.Event) { + e := ev(sp.EventHeader{OperationID: p.id, Sequence: uint64(len(p.events) + 1)}) + p.events = append(p.events, e) + p.st.LastSequence++ + if p.out != nil { + send(p.out, 0, e) + } +} diff --git a/apps/daemon/internal/processbroker/view_linux_test.go b/apps/daemon/internal/processbroker/view_linux_test.go index 294a41b58..f884a4b34 100644 --- a/apps/daemon/internal/processbroker/view_linux_test.go +++ b/apps/daemon/internal/processbroker/view_linux_test.go @@ -6,7 +6,6 @@ import ( "context" "fmt" "io" - "net" "os" "path/filepath" "sync" @@ -88,7 +87,6 @@ func startView(t *testing.T, w *hangWorld, p sessionview.Process) (*sessionview. if err := sessionview.Probe(); err != nil { t.Fatalf("Probe: %v", err) } - sock := service.socket(t) harness := t.TempDir() self, err := os.Executable() if err != nil { @@ -119,9 +117,7 @@ func startView(t *testing.T, w *hangWorld, p sessionview.Process) (*sessionview. Executables: Executables{Paths: map[string]string{"/bin/sh": "/bin/sh"}}, Environment: Environment{Sandbox: map[string]string{"PATH": "/usr/bin:/bin"}}, Scope: sp.ScopePOSIXSession, - Dial: func(ctx context.Context) (io.ReadWriteCloser, error) { - return new(net.Dialer).DialContext(ctx, "unix", sock) - }, + Dial: dialService(t, nil), CancelGrace: time.Second, }) if err != nil { From f271e1ccbcf2fd26bf1f4377cf0e2091f0e2d309 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 6 Oct 2026 23:08:10 +0000 Subject: [PATCH 4/8] Delete the non-Linux broker and view stubs Every caller of the broker and of a view is Linux-only, and agenthost.Open already reports the typed unsupported error elsewhere. Outside Linux, sessionview keeps Init and its types and processbroker its configuration. --- .../internal/processbroker/broker_other.go | 23 ---------------- apps/daemon/internal/sessionview/doc.go | 2 +- apps/daemon/internal/sessionview/errors.go | 1 - .../daemon/internal/sessionview/view_other.go | 27 ------------------- 4 files changed, 1 insertion(+), 52 deletions(-) delete mode 100644 apps/daemon/internal/processbroker/broker_other.go diff --git a/apps/daemon/internal/processbroker/broker_other.go b/apps/daemon/internal/processbroker/broker_other.go deleted file mode 100644 index 7c4971c86..000000000 --- a/apps/daemon/internal/processbroker/broker_other.go +++ /dev/null @@ -1,23 +0,0 @@ -//go:build !linux - -package processbroker - -import "errors" - -// ErrUnsupported is returned on platforms without the process broker. -var ErrUnsupported = errors.New("processbroker: unsupported on this platform") - -// Broker is unavailable on this platform. -type Broker struct{} - -// Start returns ErrUnsupported. -func Start(Config) (*Broker, error) { return nil, ErrUnsupported } - -// Close does nothing. -func (*Broker) Close() error { return nil } - -// Done returns nil. -func (*Broker) Done() <-chan struct{} { return nil } - -// Err returns ErrUnsupported. -func (*Broker) Err() error { return ErrUnsupported } diff --git a/apps/daemon/internal/sessionview/doc.go b/apps/daemon/internal/sessionview/doc.go index 8e1c63099..d198a111c 100644 --- a/apps/daemon/internal/sessionview/doc.go +++ b/apps/daemon/internal/sessionview/doc.go @@ -14,5 +14,5 @@ // // Teardown never waits unconditionally. Once the launcher has exited, the view shuts its end of the relay connection down, ends every process in its cgroup with cgroup.kill and stops the world server, which ends the requests still pending on the view's FUSE connection so that a process blocked on the world can exit. Only when the world server and the launcher have ended and the cgroup is empty, all within 30 seconds, does it remove the cgroup. Otherwise Wait and Close report [ErrCleanup] and the cgroup stays for Recover; nothing touches it after the teardown. When Start's context ends during the build, Start kills the launcher, stops waiting for it and tears the view down the same way, so a launcher blocked on the world cannot hold Start past this bound. // -// The package works only on Linux. Elsewhere [Start] returns [ErrUnsupported]. +// Views work only on Linux; elsewhere the package offers only [Init] and its types. package sessionview diff --git a/apps/daemon/internal/sessionview/errors.go b/apps/daemon/internal/sessionview/errors.go index 8c0466de9..a0f0ba559 100644 --- a/apps/daemon/internal/sessionview/errors.go +++ b/apps/daemon/internal/sessionview/errors.go @@ -8,7 +8,6 @@ import ( // Error kinds. Every error the package returns matches one of them with errors.Is. var ( - ErrUnsupported = errors.New("sessionview: unsupported platform") ErrInvalidSpec = errors.New("sessionview: invalid spec") ErrCapability = errors.New("sessionview: missing capability") ErrNoFUSE = errors.New("sessionview: /dev/fuse unavailable") diff --git a/apps/daemon/internal/sessionview/view_other.go b/apps/daemon/internal/sessionview/view_other.go index 2336ebdbf..793d0c9be 100644 --- a/apps/daemon/internal/sessionview/view_other.go +++ b/apps/daemon/internal/sessionview/view_other.go @@ -2,32 +2,5 @@ package sessionview -import ( - "context" - "os" - "syscall" -) - // Init does nothing outside Linux. func Init() {} - -// Probe reports that views need Linux. -func Probe() error { return ErrUnsupported } - -// Recover reports that view cgroups need Linux. -func Recover(string) error { return ErrUnsupported } - -// Start reports that views need Linux. -func Start(context.Context, Spec) (*View, error) { return nil, ErrUnsupported } - -// View is a running view. Outside Linux none exists. -type View struct{} - -func (*View) Presentation() Presentation { return Presentation{} } -func (*View) Wait() (Exit, error) { return Exit{}, ErrUnsupported } -func (*View) Signal(syscall.Signal) error { return ErrUnsupported } -func (*View) Close() error { return nil } -func (*View) Stdin() *os.File { return nil } -func (*View) Stdout() *os.File { return nil } -func (*View) Stderr() *os.File { return nil } -func (*View) Relay() *os.File { return nil } From d3022e156e55eceb3dbfe3b7ec85511f493d0370 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 6 Oct 2026 23:09:11 +0000 Subject: [PATCH 5/8] Delete the unused view Groups and CoreDumped fields No caller sets Process.Groups or reads Exit.CoreDumped. The view's processes keep an empty supplementary group list, which the identity test now asserts. --- .../internal/sessionview/control_linux.go | 1 - .../internal/sessionview/launcher_linux.go | 14 +++++--------- apps/daemon/internal/sessionview/spec.go | 18 ++++++++---------- apps/daemon/internal/sessionview/view_linux.go | 2 +- .../internal/sessionview/view_linux_test.go | 4 ++-- 5 files changed, 16 insertions(+), 23 deletions(-) diff --git a/apps/daemon/internal/sessionview/control_linux.go b/apps/daemon/internal/sessionview/control_linux.go index 2e28bf14f..cd84da03b 100644 --- a/apps/daemon/internal/sessionview/control_linux.go +++ b/apps/daemon/internal/sessionview/control_linux.go @@ -36,7 +36,6 @@ type launchSpec struct { Command command UID uint32 GID uint32 - Groups []uint32 Grace time.Duration } diff --git a/apps/daemon/internal/sessionview/launcher_linux.go b/apps/daemon/internal/sessionview/launcher_linux.go index 440be1d62..63e5e3029 100644 --- a/apps/daemon/internal/sessionview/launcher_linux.go +++ b/apps/daemon/internal/sessionview/launcher_linux.go @@ -362,7 +362,7 @@ func (l *launcher) reaped(pid int, ws unix.WaitStatus) { // exitOf describes how a process ended, with the code the launcher exits with for it. func exitOf(ws unix.WaitStatus) (Exit, int) { if ws.Signaled() { - return Exit{Signal: ws.Signal(), CoreDumped: ws.CoreDump()}, 128 + int(ws.Signal()) + return Exit{Signal: ws.Signal()}, 128 + int(ws.Signal()) } return Exit{Code: ws.ExitStatus()}, ws.ExitStatus() } @@ -417,7 +417,7 @@ func startRelay(spec *launchSpec, listener int) (int, error) { Files: files, Sys: &syscall.SysProcAttr{ Setsid: true, - Credential: &syscall.Credential{Uid: spec.UID, Gid: spec.GID, Groups: spec.Groups}, + Credential: &syscall.Credential{Uid: spec.UID, Gid: spec.GID}, // and no supplementary groups }, }) if err != nil { @@ -426,12 +426,8 @@ func startRelay(spec *launchSpec, listener int) (int, error) { return pid, nil } -// takeIdentity gives this thread a working directory of its own and the process's file system identity, with no capability but the two a fork needs to set the child's user, so that it enters a directory as the process would. The other threads keep theirs. +// takeIdentity gives this thread a working directory of its own and the process's file system identity, with no supplementary groups and no capability but the two a fork needs to set the child's user, so that it enters a directory as the process would. The other threads keep theirs. func takeIdentity(spec *launchSpec) error { - groups := make([]int, len(spec.Groups)) - for i, g := range spec.Groups { - groups[i] = int(g) - } const setID = 1< Date: Tue, 6 Oct 2026 23:16:58 +0000 Subject: [PATCH 6/8] Point the broker's package doc at processshim for the relay The shim, the relay and their IPC are processshim's to describe. The broker's doc keeps its own trust rules, acknowledgement and backpressure, background processes and qualification limits. --- apps/daemon/internal/processbroker/doc.go | 46 +++++++---------------- 1 file changed, 14 insertions(+), 32 deletions(-) diff --git a/apps/daemon/internal/processbroker/doc.go b/apps/daemon/internal/processbroker/doc.go index a511ea96b..faa566931 100644 --- a/apps/daemon/internal/processbroker/doc.go +++ b/apps/daemon/internal/processbroker/doc.go @@ -1,20 +1,14 @@ // Package processbroker runs a Session's shim invocations in its sandbox. // -// A Harness in the Session view executes oac-process-shim -// (apps/daemon/internal/processshim), which hands its invocation and its -// descriptors 0, 1 and 2 to the Session's process relay. The relay is the -// same binary in relay mode, which sessionview starts in the view as the -// Session user, with no capabilities, before the Harness. It is the only -// process that holds a descriptor from the view: it does every read, write -// and terminal ioctl on them with the Session's own authority. It hands the -// broker the request and the terminal's mode over a socketpair that -// sessionview creates. The broker does all process protocol work: it -// resolves the invocation against the declared executable table and -// environment policy, starts the program with the process protocol -// (internal/sandboxprocess), sends the relay the program's output and asks -// it for stdin, forwards the signals the shim reports, and has the relay -// send the shim the remote exit. Neither the shim nor the relay holds -// credentials. +// Package processshim (apps/daemon/internal/processshim) describes the shim, +// the Session's process relay and their IPC with the broker. The broker does +// all process protocol work: it resolves each invocation against the +// declared executable table and environment policy, starts the program with +// the process protocol (internal/sandboxprocess), sends the relay the +// program's output and asks it for stdin, forwards the signals the shim +// reports, and has the relay send the shim the remote exit. Neither the shim +// nor the relay holds credentials. The broker runs only on Linux; elsewhere +// the package declares only its configuration. // // The broker treats the relay as untrusted Session input. It reads the // socketpair without a control buffer, so the kernel closes any descriptor @@ -24,23 +18,11 @@ // ErrRelayLost. The broker never restarts the relay and never replays output // whose delivery is uncertain. Close never waits on the relay. // -// The relay pumps each descriptor on its own, and its dispatch never waits -// on one, so a descriptor nobody reads holds back only its own stream. The -// broker sends the relay at most processshim.OutputWindow bytes of a stream -// that the relay has not reported written, and acknowledges output to the -// process service only after that report, so the service in turn holds back -// the program. -// -// The relay never changes the flags of a passed descriptor, whose open file -// description the Harness shares. It reopens a pipe, FIFO or pty slave -// through /proc/self/fd as its own non-blocking description, uses a socket -// with MSG_DONTWAIT and a regular file or block device as it is, and polls -// any other descriptor before each call, so it waits on a peer only in a -// poll that ending the invocation interrupts. Output on an AF_UNIX socket -// carries the relay's own credentials. -// -// Invocations on one terminal share its saved mode in the relay: the first -// saves it, each runs the terminal raw, and the last to finish restores it. +// The relay pumps each descriptor on its own, so a descriptor nobody reads +// holds back only its own stream. The broker sends the relay at most +// processshim.OutputWindow bytes of a stream that the relay has not reported +// written, and acknowledges output to the process service only after that +// report, so the service in turn holds back the program. // // The broker forwards the exit without waiting for output, and the relay // answers the shim once the output the program wrote before exiting is From 0c20a26c70da425b7b1bf5e96885f4755ee13ed1 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 6 Oct 2026 23:26:34 +0000 Subject: [PATCH 7/8] Build view and spawn stdio with one sessionview helper sessionview.Stdio makes the stdio of a process that runs as a given user: the files the caller passes, a pipe for each missing one, or /dev/null for an unwanted stdin. View.Start, View.Spawn and the agent host's view process all use it, replacing three copies of the pipe making and its cleanup. Every pipe it makes belongs to the process's user, so the relay can open it as its own, as it already could for the view's own pipes. --- .../daemon/internal/agenthost/launch_linux.go | 71 ++++------------ .../internal/agenthost/session_linux_test.go | 8 +- .../daemon/internal/sessionview/view_linux.go | 83 ++++++++----------- 3 files changed, 56 insertions(+), 106 deletions(-) diff --git a/apps/daemon/internal/agenthost/launch_linux.go b/apps/daemon/internal/agenthost/launch_linux.go index f1e23d3e6..b6bc7c0d8 100644 --- a/apps/daemon/internal/agenthost/launch_linux.go +++ b/apps/daemon/internal/agenthost/launch_linux.go @@ -127,12 +127,8 @@ func (s *session) spawn(opts clirunner.StartOptions) (*clirunner.Process, error) } return nil, fmt.Errorf("%w: spawn: %w", ErrLaunch, err) } - var stdin io.WriteCloser - if p.Stdin != nil { - stdin = p.Stdin - } // FromHandle fails only without stdout and stderr, which a spawned process always has. - process, _ := clirunner.FromHandle(p, clirunner.HandleOptions{Parent: opts.Parent, Stdin: stdin, Stdout: p.Stdout, Stderr: p.Stderr, KillTimeout: opts.KillTimeout}) + process, _ := clirunner.FromHandle(p, clirunner.HandleOptions{Parent: opts.Parent, Stdin: writer(p.Stdin), Stdout: p.Stdout, Stderr: p.Stderr, KillTimeout: opts.KillTimeout}) return process, nil } @@ -169,13 +165,13 @@ func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.P s.release(lv) return nil, fmt.Errorf("%w: home: %w", ErrLaunch, err) } - ends, err := newStdio(opts.NeedStdin) + child, ends, err := sessionview.Stdio([3]*os.File{}, opts.NeedStdin, s.uid, s.uid) if err != nil { s.release(lv) return nil, fmt.Errorf("%w: stdio: %w", ErrLaunch, err) } world := worldfs.New(worldExport, s.openFile) - spec := s.spec(world, opts, ends) + spec := s.spec(world, opts, child) // The gateway serves from the view's network hook until the view has ended. var stopGateway func() spec.Network.Setup = func(netns *os.File) (err error) { @@ -183,12 +179,12 @@ func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.P return err } v, err := sessionview.Start(startCtx, spec) - ends.closeChild() + closeFiles(child[:]) if err != nil { if stopGateway != nil { stopGateway() } - ends.closeParent() + closeFiles(ends[:]) defer s.release(lv) // sessionview stops a world that served; Stop reports how that went. serr := world.Stop() @@ -241,7 +237,7 @@ func (s *session) brokerFailed(op string, err error) error { // own hands a started view to the clirunner.Process the adapter receives. A // view with a relay gets its own process broker, which serves the view's // shims in scope until the view ends. -func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway func(), opts clirunner.StartOptions, ends *stdio, scope sandboxprocess.Scope) (*clirunner.Process, error) { +func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway func(), opts clirunner.StartOptions, ends [3]*os.File, scope sandboxprocess.Scope) (*clirunner.Process, error) { h := &ownedView{s: s, lv: lv, v: v, world: world, stopGateway: stopGateway, ended: make(chan struct{}), watched: make(chan struct{})} var brokerErr error if relay := v.Relay(); relay != nil { @@ -269,8 +265,8 @@ func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway } var process *clirunner.Process if err == nil { - process, err = clirunner.FromHandle(h, clirunner.HandleOptions{Parent: opts.Parent, Stdin: ends.stdin(), - Stdout: ends.parent[1], Stderr: ends.parent[2], KillTimeout: opts.KillTimeout}) + process, err = clirunner.FromHandle(h, clirunner.HandleOptions{Parent: opts.Parent, Stdin: writer(ends[0]), + Stdout: ends[1], Stderr: ends[2], KillTimeout: opts.KillTimeout}) if err != nil { err = fmt.Errorf("%w: %w", ErrLaunch, err) } @@ -278,7 +274,7 @@ func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway if err != nil { v.Close() h.Wait() - ends.closeParent() + closeFiles(ends[:]) return nil, err } return process, nil @@ -379,7 +375,7 @@ func (h *ownedView) end(waitErr error) { // spec builds the view: the closure and home directories, the agent // host's /etc files and CA directory, the adapter's overlays and masks, and // the shim. -func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, ends *stdio) sessionview.Spec { +func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, stdio [3]*os.File) sessionview.Spec { view := s.plan.view var private []sessionview.PrivateDir for _, m := range view.Closure { @@ -407,57 +403,22 @@ func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, ends * Overlays: overlays, Shim: sessionview.Shim{Binary: s.cfg.Shim, Names: view.Shims, Paths: view.ShimPaths}, Process: sessionview.Process{Path: opts.Binary, Args: append([]string{opts.Binary}, opts.Args...), Env: opts.Env, - Dir: opts.Dir, UID: s.uid, GID: s.uid, Stdin: ends.child[0], Stdout: ends.child[1], Stderr: ends.child[2], + Dir: opts.Dir, UID: s.uid, GID: s.uid, Stdin: stdio[0], Stdout: stdio[1], Stderr: stdio[2], Grace: opts.KillTimeout}, StagingParent: s.dir.entry(stagingEntry), CgroupParent: s.cfg.ViewCgroups, } } -// stdio holds the view process's stdio: the child ends sessionview passes to -// the process and the parent ends the clirunner.Process owns. Without a stdin -// pipe the child's stdin is /dev/null and there is no parent end. -type stdio struct { - child, parent [3]*os.File -} - -func newStdio(needStdin bool) (*stdio, error) { - e := &stdio{} - if needStdin { - r, w, err := os.Pipe() - if err != nil { - return nil, err - } - e.child[0], e.parent[0] = r, w - } else { - null, err := os.Open(os.DevNull) - if err != nil { - return nil, err - } - e.child[0] = null - } - for i := 1; i < 3; i++ { - r, w, err := os.Pipe() - if err != nil { - e.closeChild() - e.closeParent() - return nil, err - } - e.child[i], e.parent[i] = w, r - } - return e, nil -} - -func (e *stdio) stdin() io.WriteCloser { - if e.parent[0] == nil { +// writer returns f, or nil without f: a nil *os.File would be a writer +// that fails. +func writer(f *os.File) io.WriteCloser { + if f == nil { return nil } - return e.parent[0] + return f } -func (e *stdio) closeChild() { closeFiles(e.child[:]) } -func (e *stdio) closeParent() { closeFiles(e.parent[:]) } - func closeFiles(files []*os.File) { for _, f := range files { if f != nil { diff --git a/apps/daemon/internal/agenthost/session_linux_test.go b/apps/daemon/internal/agenthost/session_linux_test.go index 06bcb99dc..c738be114 100644 --- a/apps/daemon/internal/agenthost/session_linux_test.go +++ b/apps/daemon/internal/agenthost/session_linux_test.go @@ -54,11 +54,11 @@ func TestViewEndReleasesTheSlotBeforeTheProcessEnds(t *testing.T) { lv := &liveView{} s.live = lv s.views.Add(1) - ends, err := newStdio(false) + child, ends, err := sessionview.Stdio([3]*os.File{}, false, uint32(os.Getuid()), uint32(os.Getgid())) if err != nil { t.Fatal(err) } - ends.closeChild() + closeFiles(child[:]) v := &fakeView{exit: make(chan struct{})} p, err := s.own(lv, v, fakeWorld{stop: stop}, func() {}, clirunner.StartOptions{Parent: context.Background(), KillTimeout: time.Second}, ends, 0) if err != nil { @@ -108,11 +108,11 @@ func TestViewEndStopsTheGateway(t *testing.T) { lv := &liveView{} s.live = lv s.views.Add(1) - ends, err := newStdio(false) + child, ends, err := sessionview.Stdio([3]*os.File{}, false, uint32(os.Getuid()), uint32(os.Getgid())) if err != nil { t.Fatal(err) } - ends.closeChild() + closeFiles(child[:]) v := &fakeView{exit: make(chan struct{})} p, err := s.own(lv, v, fakeWorld{}, stop, clirunner.StartOptions{Parent: context.Background(), KillTimeout: time.Second}, ends, 0) if err != nil { diff --git a/apps/daemon/internal/sessionview/view_linux.go b/apps/daemon/internal/sessionview/view_linux.go index af0f9e52e..d3e775ed8 100644 --- a/apps/daemon/internal/sessionview/view_linux.go +++ b/apps/daemon/internal/sessionview/view_linux.go @@ -48,6 +48,8 @@ type View struct { dev *os.File staging string cgroup string // the view's cgroup, once created + uid uint32 // the process's user and group + gid uint32 pipes [3]*os.File relay *os.File // the broker's end of the relay connection @@ -96,12 +98,14 @@ func Start(ctx context.Context, spec Spec) (*View, error) { } func (v *View) launch(spec *Spec) error { - child, err := v.stdio(&spec.Process) + p := &spec.Process + given := [3]*os.File{p.Stdin, p.Stdout, p.Stderr} + child, pipes, err := Stdio(given, true, p.UID, p.GID) if err != nil { return &Error{Kind: ErrLauncher, Op: "pipe", Err: err} } + v.pipes, v.uid, v.gid = pipes, p.UID, p.GID defer func() { - given := []*os.File{spec.Process.Stdin, spec.Process.Stdout, spec.Process.Stderr} for i, f := range child { if f != given[i] { f.Close() @@ -187,36 +191,36 @@ func (v *View) launch(spec *Spec) error { return nil } -// stdio returns the launcher's ends of the process's stdio, creating pipes where the spec gives no file. The pipes belong to the process's user, so that the relay can open them as its own. -func (v *View) stdio(p *Process) ([3]*os.File, error) { - child := [3]*os.File{p.Stdin, p.Stdout, p.Stderr} +// Stdio returns the stdio of a process that runs as uid and gid: each file of given, and for each nil one a pipe, or /dev/null for stdin when stdin is false. ends holds the other ends of the pipes. The pipes belong to uid and gid, so that the process relay can open them as its own. On error, nothing Stdio made stays open. +func Stdio(given [3]*os.File, stdin bool, uid, gid uint32) (child, ends [3]*os.File, err error) { + child = given for i := range child { - if child[i] != nil { + switch { + case child[i] != nil: continue - } - r, w, err := os.Pipe() - if err == nil { - if err = r.Chown(int(p.UID), int(p.GID)); err != nil { - r.Close() - w.Close() + case i == 0 && !stdin: + child[0], err = os.Open(os.DevNull) + default: + var r, w *os.File + if r, w, err = os.Pipe(); err == nil { + child[i], ends[i] = w, r + if i == 0 { + child[i], ends[i] = r, w + } + err = r.Chown(int(uid), int(gid)) } } if err != nil { - for j := range i { - if v.pipes[j] != nil { - v.pipes[j].Close() - child[j].Close() + for j, f := range child { + if f != given[j] { + f.Close() } } - return child, err - } - if i == 0 { - child[i], v.pipes[i] = r, w - } else { - child[i], v.pipes[i] = w, r + closeFiles(ends[:]) + return given, [3]*os.File{}, err } } - return child, nil + return child, ends, nil } func (v *View) handshake(ctx context.Context, spec *Spec) error { @@ -301,7 +305,7 @@ func (v *View) abort(err error) error { _ = v.cmd.Process.Kill() } _, _ = v.teardown() - v.closePipes() + closeFiles(v.pipes[:]) if v.cleanupErr != nil { return errors.Join(err, v.cleanupErr) } @@ -385,14 +389,6 @@ func shutdown(f *os.File) { } } -func (v *View) closePipes() { - for _, p := range v.pipes { - if p != nil { - p.Close() - } - } -} - // reply is the launcher's answer to a request, with the process a spawn started. type reply struct { message @@ -549,25 +545,18 @@ func (v *View) Spawn(ctx context.Context, path string, args, env []string, dir s return nil, &Error{Kind: ErrExited, Op: "spawn"} } // files are the launcher's: the command's read end and the process's stdio. ends are the caller's. - files := make([]*os.File, 4) - var cmdW *os.File - var ends [3]*os.File - var err error - files[0], cmdW, err = os.Pipe() - if err == nil && stdin { - files[1], ends[0], err = os.Pipe() - } else if err == nil { - files[1], err = os.Open(os.DevNull) - } - for i := 1; i < 3 && err == nil; i++ { - ends[i], files[i+1], err = os.Pipe() + child, ends, err := Stdio([3]*os.File{}, stdin, v.uid, v.gid) + var cmdR, cmdW *os.File + if err == nil { + if cmdR, cmdW, err = os.Pipe(); err != nil { + closeFiles(append(child[:], ends[:]...)) + } } if err != nil { - closeFiles(append(files, cmdW)) - closeFiles(ends[:]) <-v.spawning return nil, &Error{Kind: ErrLauncher, Op: "pipe", Err: err} } + files := append([]*os.File{cmdR}, child[:]...) // The command travels over a pipe, as the spec does, so its size is exec's to bound. The write ends once the launcher has read it, or once no read end remains or the spawn is cancelled. c := command{Path: path, Args: args, Env: env, Dir: dir} go func() { @@ -668,7 +657,7 @@ func (v *View) Close() error { // A spawn blocked on the world keeps the killed launcher, and its end, open until the teardown stops the world. v.ctl.interrupt() <-v.done - v.closePipes() + closeFiles(v.pipes[:]) }) return v.cleanupErr } From f5cf15a7b0bcace29be8f9bcb3a5a876769064ed Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 7 Oct 2026 00:06:46 +0000 Subject: [PATCH 8/8] Keep Notices past End and judge relay loss by the view The relay's output pump now writes the Notices still queued when End stops it, as far as fd 2 takes them, so a Notice behind pending stderr is no longer dropped and teardown stays bounded. The agent host decides a relay loss from the view's lifecycle instead of probing the Harness. The relay ignores the view's signals, so the launcher reaps it only when it ended before the launcher: a crash, a kill, or a broker that stopped serving it, which now ends the connection. The launcher reports that, View.RelayLost closes before Wait returns, and the agent host logs the ERROR and fails the Session on it, even when the Harness exits right after. A local reader that closes its end, as on a Cancel, now logs at DEBUG; other output write failures keep INFO. --- .../daemon/internal/agenthost/launch_linux.go | 40 +++++----- .../internal/agenthost/session_linux_test.go | 2 + .../internal/processbroker/broker_linux.go | 1 + .../processbroker/invocation_linux.go | 7 +- .../internal/processshim/relay_linux.go | 15 ++++ .../internal/processshim/relay_linux_test.go | 75 +++++++++++++++++++ .../internal/sessionview/control_linux.go | 1 + .../internal/sessionview/launcher_linux.go | 6 +- .../daemon/internal/sessionview/view_linux.go | 9 ++- .../internal/sessionview/view_linux_test.go | 26 +++++++ 10 files changed, 156 insertions(+), 26 deletions(-) diff --git a/apps/daemon/internal/agenthost/launch_linux.go b/apps/daemon/internal/agenthost/launch_linux.go index b6bc7c0d8..8f9d3bb78 100644 --- a/apps/daemon/internal/agenthost/launch_linux.go +++ b/apps/daemon/internal/agenthost/launch_linux.go @@ -38,6 +38,7 @@ type runningView interface { Wait() (sessionview.Exit, error) Close() error Relay() *os.File + RelayLost() <-chan struct{} Spawn(ctx context.Context, path string, args, env []string, dir string, stdin bool) (*sessionview.Spawned, error) } @@ -302,34 +303,27 @@ type ownedView struct { // while the view runs. func (h *ownedView) watch() { defer close(h.watched) - var relayEnded <-chan struct{} + var relayLost <-chan struct{} if h.broker != nil { - relayEnded = h.broker.Done() - } - for { + relayLost = h.v.RelayLost() + } + select { + case <-h.world.Lost(): + h.s.fail(fmt.Errorf("%w: world: %w", ErrWorld, h.world.Err())) + return + case <-relayLost: + case <-h.ended: + // A lost relay is reported before the view ends. select { - case <-h.world.Lost(): - h.s.fail(fmt.Errorf("%w: world: %w", ErrWorld, h.world.Err())) - return - case <-relayEnded: - // The broker stops serving on Close, which end calls only after - // watch returns, or when its relay connection ends. sessionview - // ends that connection itself only in its teardown, which starts - // once the launcher has stopped answering, and from then on - // Signal fails with ErrExited, ErrClosed or a lost launcher. So a - // delivered signal means that the view still runs its process - // and the relay was lost while the Harness ran. A failed one - // means that the view is ending, and Wait reports how. - if h.v.Signal(0) == nil { - h.s.log.Error("process relay lost", "error", h.broker.Err()) - h.s.brokerFailed("relay", h.broker.Err()) - return - } - relayEnded = nil - case <-h.ended: + case <-relayLost: + default: return } } + // The relay's end ends its connection, so the broker stops. + <-h.broker.Done() + h.s.log.Error("process relay lost", "error", h.broker.Err()) + h.s.brokerFailed("relay", h.broker.Err()) } func (h *ownedView) Signal(sig syscall.Signal) error { return h.v.Signal(sig) } diff --git a/apps/daemon/internal/agenthost/session_linux_test.go b/apps/daemon/internal/agenthost/session_linux_test.go index c738be114..261859b75 100644 --- a/apps/daemon/internal/agenthost/session_linux_test.go +++ b/apps/daemon/internal/agenthost/session_linux_test.go @@ -417,6 +417,8 @@ func (v *fakeView) Signal(syscall.Signal) error { return nil } func (v *fakeView) Relay() *os.File { return nil } +func (v *fakeView) RelayLost() <-chan struct{} { return nil } + func (v *fakeView) Spawn(context.Context, string, []string, []string, string, bool) (*sessionview.Spawned, error) { return nil, v.spawnErr } diff --git a/apps/daemon/internal/processbroker/broker_linux.go b/apps/daemon/internal/processbroker/broker_linux.go index 7c852d590..58810905c 100644 --- a/apps/daemon/internal/processbroker/broker_linux.go +++ b/apps/daemon/internal/processbroker/broker_linux.go @@ -117,6 +117,7 @@ func (b *Broker) read() { } b.mu.Unlock() b.cancel() + b.conn.CloseWrite() // a relay the broker stops serving ends too b.conn.Close() close(b.done) } diff --git a/apps/daemon/internal/processbroker/invocation_linux.go b/apps/daemon/internal/processbroker/invocation_linux.go index 26955ff5b..667004205 100644 --- a/apps/daemon/internal/processbroker/invocation_linux.go +++ b/apps/daemon/internal/processbroker/invocation_linux.go @@ -778,7 +778,12 @@ func (inv *invocation) receive(m processshim.RelayMessage) error { for _, seq := range sent { inv.acks.deliver(seq) } - inv.log.Info("output reader gone", "stream", w.stream, "error", unix.Errno(m.Errno)) + // A reader that closes its end, as on a Cancel, is expected. + if errno := unix.Errno(m.Errno); errno == unix.EPIPE || errno == unix.ECONNRESET { + inv.log.Debug("output reader gone", "stream", w.stream, "error", errno) + } else { + inv.log.Info("output write failed", "stream", w.stream, "error", errno) + } inv.helpers.Add(1) go func() { defer inv.helpers.Done() diff --git a/apps/daemon/internal/processshim/relay_linux.go b/apps/daemon/internal/processshim/relay_linux.go index 5600816aa..1bb642a47 100644 --- a/apps/daemon/internal/processshim/relay_linux.go +++ b/apps/daemon/internal/processshim/relay_linux.go @@ -617,6 +617,7 @@ func (o *output) next() (item, bool) { func (o *output) run() { defer o.inv.pumps.Done() + defer o.notices() id := o.inv.id for { it, ok := o.next() @@ -675,6 +676,20 @@ func (o *output) run() { } } +// notices writes the Notices still queued once the pump has stopped, each +// as far as the descriptor takes it now, so End drops none it can take. +func (o *output) notices() { + o.mu.Lock() + queue := o.queue + o.queue = nil + o.mu.Unlock() + for _, it := range queue { + if it.notice != nil { + o.message(it.notice) + } + } +} + // progress records seq as done, or the descriptor as broken, and reports // whether the broker is told. func (o *output) progress(seq uint64, failed bool) bool { diff --git a/apps/daemon/internal/processshim/relay_linux_test.go b/apps/daemon/internal/processshim/relay_linux_test.go index 2407e00da..bae42ca7b 100644 --- a/apps/daemon/internal/processshim/relay_linux_test.go +++ b/apps/daemon/internal/processshim/relay_linux_test.go @@ -353,6 +353,81 @@ func TestNoticeFollowsQueuedStderr(t *testing.T) { } } +// A Notice queued behind stderr still reaches fd 2 when End stops that +// stderr first, and a stderr nobody reads holds neither from ending. +func TestNoticeOutlivesEnd(t *testing.T) { + for _, full := range []bool{false, true} { + t.Run(fmt.Sprintf("full=%v", full), func(t *testing.T) { + var rl *relay + release := make(chan struct{}) + r := startTestRelay(t, func(r *relay) { + rl = r + if !full { + r.gating = func() { <-release } // the stderr waits until End has stopped it + } + }) + opened := make(chan uint64, 1) + go func() { + for { + m, err := r.read() + if err != nil { + return + } + if m, ok := m.(Open); ok { + id := m.ID + r.send(Accept{ID: id}, Started{ID: id}, Output{ID: id, FD: 2, Seq: 1, Data: []byte("err\n")}, + Notice{ID: id, Message: []byte("note")}, Exit{ID: id, Result: Result{Code: 0}}) + opened <- id + } + } + }() + pr, pw, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + defer pr.Close() + filled := 0 + if full { + fd := int(pw.Fd()) + unix.SetNonblock(fd, true) + for _, chunk := range []int{pipeBuf, 1} { + for n, err := 0, error(nil); err == nil; n, err = unix.Write(fd, make([]byte, chunk)) { + filled += n + } + } + } + fds := devNull(t) + fds[2] = int(pw.Fd()) + conn := r.shim(t, "sh", fds) + pw.Close() + if res, err := finished(conn); err != nil || res.Code != 0 { + t.Fatalf("Result %+v, %v", res, err) + } + r.send(End{ID: <-opened}) + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(time.Millisecond) { + rl.mu.Lock() + ended := len(rl.invs) == 0 + rl.mu.Unlock() + if ended { + break + } + if time.Now().After(deadline) { + t.Fatal("the invocation never took End") + } + } + close(release) + pr.SetDeadline(time.Now().Add(10 * time.Second)) + got, err := io.ReadAll(pr) + const note = "oac-process-shim: note\n" + rest, ok := strings.CutPrefix(string(got), string(make([]byte, filled))) + // A full stderr may still be full when the Notice is tried. + if err != nil || !ok || (rest != note && !(full && rest == "")) { + t.Fatalf("fd 2 got %d bytes ending in %q, %v; want %d filler bytes and %q", len(got), got[max(0, len(got)-32):], err, filled, note) + } + }) + } +} + // readN reads n bytes from the terminal's master side. It fails the test // when they do not arrive within 10s. func readN(t *testing.T, fd, n int) []byte { diff --git a/apps/daemon/internal/sessionview/control_linux.go b/apps/daemon/internal/sessionview/control_linux.go index cd84da03b..34dc420a8 100644 --- a/apps/daemon/internal/sessionview/control_linux.go +++ b/apps/daemon/internal/sessionview/control_linux.go @@ -59,6 +59,7 @@ const ( msgSignaled // launcher: whether msgSignal reached it msgSpawn // daemon: start another process; carries a pipe with its command, then its stdin, stdout and stderr msgSpawned // launcher: the spawned process's Pid, or why none started + msgNoRelay // launcher: the relay ended before the launcher ) type message struct { diff --git a/apps/daemon/internal/sessionview/launcher_linux.go b/apps/daemon/internal/sessionview/launcher_linux.go index 63e5e3029..2d2d26a60 100644 --- a/apps/daemon/internal/sessionview/launcher_linux.go +++ b/apps/daemon/internal/sessionview/launcher_linux.go @@ -345,8 +345,12 @@ func (l *launcher) reapChildren() bool { } } -// reaped retires the ID of pid, if it has one, and reports its exit. mu is held. +// reaped retires the ID of pid, if it has one, and reports its exit, or reports the relay's loss: the relay ignores the view's signals, so it ends before the launcher only when it is lost. mu is held. func (l *launcher) reaped(pid int, ws unix.WaitStatus) { + if pid == l.relay { + l.post(message{Kind: msgNoRelay}) + return + } id, ok := l.spawned[pid] if !ok { return diff --git a/apps/daemon/internal/sessionview/view_linux.go b/apps/daemon/internal/sessionview/view_linux.go index d3e775ed8..2a2793d6f 100644 --- a/apps/daemon/internal/sessionview/view_linux.go +++ b/apps/daemon/internal/sessionview/view_linux.go @@ -53,6 +53,8 @@ type View struct { pipes [3]*os.File relay *os.File // the broker's end of the relay connection + relayLost chan struct{} // closed when the relay ends before the launcher + waited chan struct{} // closed once the launcher has been reaped waitErr error cleanupErr error // set before done closes @@ -76,7 +78,7 @@ func Start(ctx context.Context, spec Spec) (*View, error) { if err := Probe(); err != nil { return nil, err } - v := &View{done: make(chan struct{}), requests: map[uint64]chan reply{}, spawning: make(chan struct{}, 1)} + v := &View{done: make(chan struct{}), relayLost: make(chan struct{}), requests: map[uint64]chan reply{}, spawning: make(chan struct{}, 1)} if err := v.launch(&spec); err != nil { return nil, v.abort(err) } @@ -434,6 +436,8 @@ func (v *View) watch() { v.requestMu.Unlock() case msgFailed: failed = m.Fail.err() + case msgNoRelay: + close(v.relayLost) } } v.requestMu.Lock() @@ -662,6 +666,9 @@ func (v *View) Close() error { return v.cleanupErr } +// RelayLost closes when the relay ends before the launcher, which the view's own end never causes: the relay crashed, was killed or lost its broker. It closes before Wait returns. +func (v *View) RelayLost() <-chan struct{} { return v.relayLost } + // Relay is the broker's end of the relay's connection, or nil when the spec declares no shim. processbroker.Start takes a duplicate of it. The view shuts the connection down as it ends, so a broker still running then sees the relay lost. func (v *View) Relay() *os.File { return v.relay } diff --git a/apps/daemon/internal/sessionview/view_linux_test.go b/apps/daemon/internal/sessionview/view_linux_test.go index 85182e381..be4f03df9 100644 --- a/apps/daemon/internal/sessionview/view_linux_test.go +++ b/apps/daemon/internal/sessionview/view_linux_test.go @@ -144,6 +144,32 @@ func TestViewSignalAndTeardown(t *testing.T) { } } +// TestRelayLossIsReported checks that a relay lost while the process runs is reported, even when the process exits right after, and that the view's own end reports no loss. +func TestRelayLossIsReported(t *testing.T) { + for _, lose := range []bool{false, true} { + v, _ := startStalled(t) + if lose { + if err := syscall.Kill(relayPID(), syscall.SIGKILL); err != nil { + t.Fatal(err) + } + eventually(t, "the relay's end", func() bool { return relayPID() == 0 }) + } + if err := v.Signal(syscall.SIGTERM); err != nil { + t.Fatalf("Signal: %v", err) + } + v.Wait() + reported := false + select { + case <-v.RelayLost(): + reported = true + default: + } + if reported != lose { + t.Errorf("relay killed: %v; RelayLost closed: %v", lose, reported) + } + } +} + // TestViewDescendantsKeepTheGrace checks that helpers still cleaning up when their parent exits, the Harness or a spawned process, get TERM and finish within the grace. func TestViewDescendantsKeepTheGrace(t *testing.T) { requireView(t)