diff --git a/docs/sdk.md b/docs/sdk.md index 303b6687..6f4fee36 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -645,7 +645,7 @@ error pointing at `Push`. ## Terminals ```go -pty, err := sb.OpenPty(ctx, sandbox.PtyOpts{Cols: 120, Rows: 40}) +pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 120, Rows: 40}) defer pty.Close() pty.Write([]byte("make test\n")) io.Copy(os.Stdout, pty) // EOF when the shell exits diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index daba46e1..ea77258d 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -417,7 +417,7 @@ func smokePromote(ctx context.Context, client *sandbox.Client, sb *sandbox.Sandb } func smokePty(ctx context.Context, sb *sandbox.Sandbox) error { - pty, err := sb.OpenPty(ctx, sandbox.PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { return err } diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index a8539f1d..eaf0a635 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -143,9 +143,11 @@ type Response interface{ RespType() string } type respDecoder func([]byte) (Response, error) +type bulkMaker func([]byte) Response + type bulkResponse struct { head []byte - mk func([]byte) Response + mk bulkMaker } // reqPtr and respPtr turn a mistyped decoder table entry into a compile error. @@ -159,15 +161,13 @@ type respPtr[T any] interface { Response } -// Exec starts a process; with Session set it runs inside that persistent -// shell instead. Detach is emitted even when false — it is part of the -// fixture corpus shape. +// Exec starts a process; with Session set it runs inside that persistent shell instead. type Exec struct { Argv []string `json:"argv"` Cwd string `json:"cwd,omitempty"` Env map[string]string `json:"env,omitempty"` User string `json:"user,omitempty"` - Detach bool `json:"detach"` + Detach bool `json:"detach"` // emitted even when false: the fixture corpus carries it Session string `json:"session,omitempty"` } @@ -679,12 +679,25 @@ func DecodeRequest(line []byte) (Request, error) { // DecodeResponse parses one frame into its type's concrete Go type. Byte // fields are freshly allocated per frame, so callers may retain them. func DecodeResponse(line []byte) (Response, error) { - return decodeResponse(line, nil) + return DecodeResponseInto(line, nil) } // DecodeResponseInto is DecodeResponse for a streaming consumer: a bulk frame's bytes land in *scratch, grown as needed, and stay valid only until the next call with it. func DecodeResponseInto(line []byte, scratch *[]byte) (Response, error) { - return decodeResponse(line, scratch) + typ, err := frameTag(line, respTagHead, "type") + if err != nil { + return nil, fmt.Errorf("parse response frame: %w", err) + } + if bulk, ok := bulkResponses[string(typ)]; ok { + if data, ok := decodeBulk(line, bulk.head, scratch); ok { + return bulk.mk(data), nil + } + } + dec, ok := responseDecoders[string(typ)] + if !ok { + return nil, fmt.Errorf("unknown response type %q", typ) + } + return dec(line) } // NewFrameScanner wraps r for newline-delimited frames capped at MaxFrame. @@ -716,7 +729,7 @@ func IsContinuation(line []byte) bool { return false } -func bulkDecoders(mks ...func([]byte) Response) map[string]bulkResponse { +func bulkDecoders(mks ...bulkMaker) map[string]bulkResponse { out := make(map[string]bulkResponse, len(mks)) for _, mk := range mks { tag := mk(nil).RespType() @@ -725,23 +738,6 @@ func bulkDecoders(mks ...func([]byte) Response) map[string]bulkResponse { return out } -func decodeResponse(line []byte, scratch *[]byte) (Response, error) { - typ, err := frameTag(line, respTagHead, "type") - if err != nil { - return nil, fmt.Errorf("parse response frame: %w", err) - } - if bulk, ok := bulkResponses[string(typ)]; ok { - if data, ok := decodeBulk(line, bulk.head, scratch); ok { - return bulk.mk(data), nil - } - } - dec, ok := responseDecoders[string(typ)] - if !ok { - return nil, fmt.Errorf("unknown response type %q", typ) - } - return dec(line) -} - // decodeBulk decodes a canonical bulk frame's payload into scratch, or a fresh buffer when scratch is nil; ok is false for any other shape. func decodeBulk(line, head []byte, scratch *[]byte) ([]byte, bool) { after, ok := bytes.CutPrefix(line, head) @@ -826,26 +822,18 @@ func scanTag(line []byte, key string) (string, error) { return "", nil } -func decodeAs[T any](line []byte) (*T, error) { - v := new(T) +func decodeReq[T any, PT reqPtr[T]](line []byte) (Request, error) { + v := PT(new(T)) if err := json.Unmarshal(line, v); err != nil { return nil, err } return v, nil } -func decodeReq[T any, PT reqPtr[T]](line []byte) (Request, error) { - v, err := decodeAs[T](line) - if err != nil { - return nil, err - } - return PT(v), nil -} - func decodeResp[T any, PT respPtr[T]](line []byte) (Response, error) { - v, err := decodeAs[T](line) - if err != nil { + v := PT(new(T)) + if err := json.Unmarshal(line, v); err != nil { return nil, err } - return PT(v), nil + return v, nil } diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index 4e3e097c..7a5f5d68 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -101,7 +101,7 @@ func (e *Engine) VersionWarning(ctx context.Context) (version, warning string) { if err != nil { return "", fmt.Sprintf("cannot determine cocoon version (need >= %s): %v", RequiredCocoon, err) } - if below, comparable := belowFloor(v); comparable && below { + if below, known := belowFloor(v); known && below { return v, fmt.Sprintf("cocoon %s is below the required %s — upgrade cocoon", v, RequiredCocoon) } return v, "" @@ -493,8 +493,8 @@ func respFail(resp wire.Response) string { return "unexpected frame " + resp.RespType() } -// belowFloor reports whether v is below RequiredCocoon; comparable is false for a dev build. -func belowFloor(v string) (below, comparable bool) { +// belowFloor reports whether v is below RequiredCocoon; known is false for a dev build. +func belowFloor(v string) (below, known bool) { v = "v" + strings.TrimPrefix(v, "v") if semver.Canonical(v) != v { return false, false diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 451a6146..32123833 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -33,10 +33,6 @@ func (m *Manager) archiveDeleteFor(key types.PoolKey) time.Duration { return m.archiveDeleteDefault } -func (m *Manager) archiveEnabledFor(key types.PoolKey) bool { - return m.archiveAfterFor(key) > 0 -} - func (m *Manager) archiveOnce(ctx context.Context) { if !m.archiveSweep.CompareAndSwap(false, true) { return // the previous sweep's archives are still draining diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index a9726943..8b1b2c94 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -562,7 +562,7 @@ func TestArchiveRemovalCommitPinsCheckpoint(t *testing.T) { if !ckExists(t, m, ck) { t.Error("retry deleted the checkpoint before the claims commit") } - m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json") + breakStore(t, m) m.store.writeMu.Unlock() locked = false if err := <-done; action == testArchiveRelease && err == nil { @@ -667,7 +667,7 @@ func TestArchiveDeleteRetryRechecksWakeRollback(t *testing.T) { defer m.recLocksMu.Unlock() return m.recRefs[ck] >= 2 }) - m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json") + breakStore(t, m) m.store.writeMu.Unlock() locked = false if err := <-woke; err == nil { diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index cde7c084..a3c6dd12 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -502,7 +502,7 @@ func (m *Manager) reapOnce(ctx context.Context) { } m.mu.Lock() var expired []victim - var dropped []string + var dropped, purges []string for id, sb := range m.claimed { // a zero deadline means no expiry (an archived claim kept forever). if sb.Deadline.IsZero() || !now.After(sb.Deadline) { @@ -511,6 +511,7 @@ func (m *Manager) reapOnce(ctx context.Context) { switch { case sb.ArchiveCk != "": expired = append(expired, victim{action: reapPurge, id: id, ck: sb.ArchiveCk, tenant: sb.Tenant, sb: sb}) + purges = append(purges, sb.ArchiveCk) case m.archivesAtDeadline(sb): // archive instead of destroy, kept in m.claimed for archive() to move if _, busy := m.archiving[id]; busy { @@ -538,19 +539,13 @@ func (m *Manager) reapOnce(ctx context.Context) { } logger := log.WithFunc("pool.reapOnce") - var purges []string - for _, v := range expired { - if v.action == reapPurge { - purges = append(purges, v.ck) - } - } failed := m.markArchiveCks(purges) expired = slices.DeleteFunc(expired, func(v victim) bool { markErr, bad := failed[v.ck] - if bad && v.action == reapPurge { + if bad { logger.Errorf(ctx, markErr, "mark archive ck %s; keeping %s", v.ck, v.id) } - return bad && v.action == reapPurge + return bad }) if len(expired) == 0 { return @@ -629,7 +624,7 @@ func (m *Manager) reapOnce(ctx context.Context) { } func (m *Manager) archivesAtDeadline(sb *types.Sandbox) bool { - return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveEnabledFor(sb.Key)) + return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveAfterFor(sb.Key) > 0) } func (m *Manager) pauseExpired(ctx context.Context, sb *types.Sandbox) error { diff --git a/sandboxd/pool/liveness_test.go b/sandboxd/pool/liveness_test.go index ff0d61de..5a1a4747 100644 --- a/sandboxd/pool/liveness_test.go +++ b/sandboxd/pool/liveness_test.go @@ -524,7 +524,6 @@ func TestReconcileKeepsRunningClaimsTheHostLost(t *testing.T) { } } -// assertRestartFailsUntilBudget runs recoveries against a guest setup step that always fails. func assertRestartFailsUntilBudget(t *testing.T, m *Manager, eng *fakeEngine, sb *types.Sandbox) { t.Helper() for attempt := range vmmRestartBudget { diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index c2c519a8..68a0dbe9 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -426,14 +426,7 @@ func TestRefillServicesEveryPoolUnderTightBudget(t *testing.T) { infos, _ := m.Info() return len(infos) == 2 && infos[0].Warm == 3 && infos[1].Warm == 3 } - deadline := time.Now().Add(3 * time.Second) - for !both() && time.Now().Before(deadline) { - time.Sleep(5 * time.Millisecond) - } - if !both() { - infos, _ := m.Info() - t.Fatalf("one refill tick did not chain both pools to target: %+v", infos) - } + waitFor(t, both) }) } diff --git a/sandboxd/server/relay.go b/sandboxd/server/relay.go index ab8178aa..05916f1a 100644 --- a/sandboxd/server/relay.go +++ b/sandboxd/server/relay.go @@ -34,6 +34,11 @@ var ( silkdEnds = relayEnds{guest: func(c net.Conn) { _ = c.Close() }, client: drainClient} ) +// relayEnds is what a finished direction means to the peer that did not end it. +type relayEnds struct { + guest, client func(net.Conn) +} + // CloseRelays force-closes in-flight relays; http.Server.Shutdown does not track hijacked conns. func (s *Server) CloseRelays() { s.relayMu.Lock() @@ -94,12 +99,6 @@ func (s *Server) relay(ctx context.Context, id string, client net.Conn, clientBu s.splice(client, clientR, guest, switchingProtocols, silkdEnds) } -// relayEnds is what a finished direction means to the peer that did not end it. -type relayEnds struct { - guest func(net.Conn) - client func(net.Conn) -} - func (s *Server) splice(client net.Conn, clientR io.Reader, guest net.Conn, hello []byte, ends relayEnds) { release, ok := s.trackRelay(client, guest) if !ok { @@ -157,33 +156,6 @@ func (s *Server) trackRelay(client, guest net.Conn) (func(), bool) { }, true } -func keepAlive(client net.Conn) { - tcp, ok := client.(*net.TCPConn) - if !ok { - return - } - _ = tcp.SetKeepAliveConfig(net.KeepAliveConfig{ - Enable: true, - Idle: keepAliveIdle, - Interval: keepAliveInterval, - Count: keepAliveCount, - }) -} - -// drainClient half-closes; the read deadline is what unblocks the splice goroutine. -func drainClient(client net.Conn) { - utils.CloseWrite(client) - _ = client.SetReadDeadline(time.Now().Add(drainGrace)) -} - -// clientReader replays what the http server already buffered; reading the bufio past Buffered() would race the direct conn reads. -func clientReader(client net.Conn, clientBuf *bufio.Reader) io.Reader { - if n := clientBuf.Buffered(); n > 0 { - return io.MultiReader(io.LimitReader(clientBuf, int64(n)), client) - } - return client -} - // auditTee records each request line it relays; input frames and the bytes past the cap pass unrecorded. type auditTee struct { r io.Reader @@ -230,3 +202,30 @@ func (t *auditTee) endLine() { t.buf = t.buf[:0] t.skip = false } + +func keepAlive(client net.Conn) { + tcp, ok := client.(*net.TCPConn) + if !ok { + return + } + _ = tcp.SetKeepAliveConfig(net.KeepAliveConfig{ + Enable: true, + Idle: keepAliveIdle, + Interval: keepAliveInterval, + Count: keepAliveCount, + }) +} + +// drainClient half-closes; the read deadline is what unblocks the splice goroutine. +func drainClient(client net.Conn) { + utils.CloseWrite(client) + _ = client.SetReadDeadline(time.Now().Add(drainGrace)) +} + +// clientReader replays what the http server already buffered; reading the bufio past Buffered() would race the direct conn reads. +func clientReader(client net.Conn, clientBuf *bufio.Reader) io.Reader { + if n := clientBuf.Buffered(); n > 0 { + return io.MultiReader(io.LimitReader(clientBuf, int64(n)), client) + } + return client +} diff --git a/sdk/go/files.go b/sdk/go/files.go index b1ed4236..bd98ac4d 100644 --- a/sdk/go/files.go +++ b/sdk/go/files.go @@ -8,23 +8,18 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -// WriteFile writes data to path in the sandbox, atomically (silkd renames a -// temp file into place). mode, when non-nil, sets the file's permission bits. +// WriteFile atomically writes data to path in the sandbox; a non-nil mode sets the file's permission bits. func (s *Sandbox) WriteFile(ctx context.Context, path string, data []byte, mode *uint32) error { return s.uploadRPC(ctx, &wire.FsWrite{Path: path, Mode: mode}, bytes.NewReader(data)) } // ReadFile returns the contents of path. func (s *Sandbox) ReadFile(ctx context.Context, path string) ([]byte, error) { - var out []byte - err := s.downloadRPC(ctx, &wire.FsRead{Path: path}, func(b []byte) error { - out = append(out, b...) - return nil - }) - if err != nil { + var buf bytes.Buffer + if err := s.ReadFileTo(ctx, path, &buf); err != nil { return nil, err } - return out, nil + return buf.Bytes(), nil } // ReadFileTo streams path into w; an error from w ends the read there. diff --git a/sdk/go/files_test.go b/sdk/go/files_test.go index ac4240ca..b2082da2 100644 --- a/sdk/go/files_test.go +++ b/sdk/go/files_test.go @@ -61,32 +61,7 @@ func TestFilesRoundTrip(t *testing.T) { } func TestAnUploadFillsEachFrameFromSmallReads(t *testing.T) { - frames := make(chan int, 1) - sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { - defer conn.Close() - sc := wire.NewFrameScanner(conn) - data := 0 - for sc.Scan() { - req, err := wire.DecodeRequest(sc.Bytes()) - if err != nil { - return - } - var resp wire.Response - switch req.(type) { - case *wire.Info: - resp = &wire.InfoResp{Proto: wire.KeepAliveProto} - case *wire.Data: - data++ - case *wire.DataEnd: - frames <- data - resp = wire.Done{} - } - if resp != nil { - line, _ := wire.EncodeResponse(resp) - _, _ = conn.Write(append(line, '\n')) - } - } - })) + sb, frames := uploadAgent(t) if err := sb.Push(t.Context(), "/d", iotest.OneByteReader(bytes.NewReader(make([]byte, 4096)))); err != nil { t.Fatalf("push: %v", err) } @@ -96,32 +71,13 @@ func TestAnUploadFillsEachFrameFromSmallReads(t *testing.T) { } func TestAnUploadStopsOnItsReadersError(t *testing.T) { - ended := make(chan bool, 1) - sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { - defer conn.Close() - sc := wire.NewFrameScanner(conn) - for sc.Scan() { - req, err := wire.DecodeRequest(sc.Bytes()) - if err != nil { - break - } - switch req.(type) { - case *wire.Info: - line, _ := wire.EncodeResponse(&wire.InfoResp{Proto: wire.KeepAliveProto}) - _, _ = conn.Write(append(line, '\n')) - case *wire.DataEnd: - ended <- true - return - } - } - ended <- false - })) + sb, frames := uploadAgent(t) truncated := io.MultiReader(bytes.NewReader(make([]byte, 4096)), iotest.ErrReader(io.ErrUnexpectedEOF)) if err := sb.Push(t.Context(), "/d", truncated); !errors.Is(err, io.ErrUnexpectedEOF) { t.Errorf("push = %v, want the reader's io.ErrUnexpectedEOF", err) } - if <-ended { - t.Error("a truncated stream was committed with data_end") + if got := <-frames; got != -1 { + t.Errorf("a truncated stream was committed with data_end after %d data frames", got) } } @@ -223,6 +179,41 @@ func TestUploadToAnOldDaemonReturnsItsRejection(t *testing.T) { } } +func uploadAgent(t *testing.T) (*Sandbox, <-chan int) { + t.Helper() + frames := make(chan int, 1) + sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { + defer conn.Close() + sc := wire.NewFrameScanner(conn) + data := 0 + for sc.Scan() { + req, err := wire.DecodeRequest(sc.Bytes()) + if err != nil { + break + } + var resp wire.Response + switch req.(type) { + case *wire.Info: + resp = &wire.InfoResp{Proto: wire.KeepAliveProto} + case *wire.Data: + data++ + case *wire.DataEnd: + frames <- data + resp = wire.Done{} + } + if resp != nil { + line, _ := wire.EncodeResponse(resp) + _, _ = conn.Write(append(line, '\n')) + } + if _, end := req.(*wire.DataEnd); end { + return + } + } + frames <- -1 + })) + return sb, frames +} + func fakeSandbox(t testing.TB) *Sandbox { t.Helper() fake := silkdtest.NewFake(t.TempDir()) diff --git a/sdk/go/pty.go b/sdk/go/pty.go index a46a05e5..9079df0c 100644 --- a/sdk/go/pty.go +++ b/sdk/go/pty.go @@ -9,15 +9,6 @@ import ( "github.com/cocoonstack/sandbox/sdk/go/silkd" ) -// PtyOpts configures OpenPty. -type PtyOpts struct { - Cols uint16 - Rows uint16 - Cwd string - Env map[string]string - User string -} - // Pty is an open pseudo-terminal in the sandbox. type Pty struct { PID uint32 @@ -68,7 +59,7 @@ func (p *Pty) ExitCode() (code int, ok bool) { func (p *Pty) drain(ctx context.Context, pw *io.PipeWriter, stop func()) { defer stop() for { - resp, err := recvTransient(ctx, p.conn) + resp, err := recv(ctx, p.conn) if err != nil { _ = pw.CloseWithError(err) return @@ -96,9 +87,8 @@ func (p *Pty) drain(ctx context.Context, pw *io.PipeWriter, stop func()) { } // OpenPty starts a shell whose lifetime is governed by ctx or Close. -func (s *Sandbox) OpenPty(ctx context.Context, opts PtyOpts) (*Pty, error) { - req := &wire.PtyOpen{Cols: opts.Cols, Rows: opts.Rows, Cwd: opts.Cwd, Env: opts.Env, User: opts.User} - conn, l, err := s.call(ctx, req) +func (s *Sandbox) OpenPty(ctx context.Context, opts wire.PtyOpen) (*Pty, error) { + conn, l, err := s.call(ctx, &opts) if err != nil { return nil, err } diff --git a/sdk/go/pty_test.go b/sdk/go/pty_test.go index 66920caf..08c4ec90 100644 --- a/sdk/go/pty_test.go +++ b/sdk/go/pty_test.go @@ -17,7 +17,7 @@ import ( func TestPtyEchoAndExit(t *testing.T) { sb := fakeSandbox(t) - pty, err := sb.OpenPty(t.Context(), PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(t.Context(), wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } @@ -67,7 +67,7 @@ func TestPtyReleasesTheRelayWhenTheShellExits(t *testing.T) { _, _ = r.ReadByte() close(released) }) - pty, err := legacySandbox(t, ts).OpenPty(t.Context(), PtyOpts{Cols: 80, Rows: 24}) + pty, err := legacySandbox(t, ts).OpenPty(t.Context(), wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } @@ -88,7 +88,7 @@ func TestPtyReleasesTheRelayWhenTheShellExits(t *testing.T) { func TestPtyCtxCancelIsTyped(t *testing.T) { sb := fakeSandbox(t) ctx, cancel := context.WithCancel(t.Context()) - pty, err := sb.OpenPty(ctx, PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } @@ -103,7 +103,7 @@ func TestPtyCtxCancelIsTyped(t *testing.T) { func TestPtyCloseUnblocksUnreadOutput(t *testing.T) { sb := fakeSandbox(t) - pty, err := sb.OpenPty(t.Context(), PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(t.Context(), wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } diff --git a/sdk/go/utils.go b/sdk/go/utils.go index b05e8db1..7d414b98 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -18,6 +18,8 @@ type respPtr[T any] interface { wire.Response } +type chunkSink func([]byte) error + func (s *Sandbox) doneRPC(ctx context.Context, req wire.Request) error { conn, l, err := s.call(ctx, req) if err != nil { @@ -63,7 +65,7 @@ func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) return l.done(<-terminal) } -func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink func([]byte) error) error { +func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink chunkSink) error { conn, l, err := s.call(ctx, req) if err != nil { return err @@ -134,7 +136,7 @@ func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) stdout = cmp.Or(stdout, io.Discard) stderr = cmp.Or(stderr, io.Discard) for { - resp, err := recvTransient(ctx, conn) + resp, err := recv(ctx, conn) if err != nil { return pid, 0, false, err } @@ -162,9 +164,9 @@ func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) } // drainData consumes Data frames into sink until Done; sink must not retain its argument. -func drainData(ctx context.Context, conn *silkd.Conn, sink func([]byte) error) error { +func drainData(ctx context.Context, conn *silkd.Conn, sink chunkSink) error { for { - resp, err := recvTransient(ctx, conn) + resp, err := recv(ctx, conn) if err != nil { return err } @@ -203,18 +205,9 @@ func expect[T any, PT respPtr[T]](ctx context.Context, conn *silkd.Conn) (*T, er return nil, unexpected(resp) } +// recv reads one frame, translating a canceled ctx and an early EOF; a bulk frame's Data is valid until the next read. func recv(ctx context.Context, conn *silkd.Conn) (wire.Response, error) { - return recvWith(ctx, conn.Recv) -} - -// recvTransient is recv for a consumer done with a bulk frame's Data before its next read. -func recvTransient(ctx context.Context, conn *silkd.Conn) (wire.Response, error) { - return recvWith(ctx, conn.RecvTransient) -} - -// recvWith reads one frame, translating a canceled ctx and an early EOF. -func recvWith(ctx context.Context, next func() (wire.Response, error)) (wire.Response, error) { - resp, err := next() + resp, err := conn.RecvTransient() if err != nil { if ctxErr := ctx.Err(); ctxErr != nil { return nil, ctxErr diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py index 99244157..c8d1d892 100644 --- a/sdk/python/tests/conftest.py +++ b/sdk/python/tests/conftest.py @@ -1,5 +1,4 @@ -"""Fixtures and helpers shared by the suites: in-process fake nodes, addresses -that refuse a connection, and the relay-side upgrade handshake.""" +"""Shared fixtures: in-process fake nodes, refusing addresses, and the relay-side upgrade handshake.""" import json import os diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 723ae7b8..90b799dc 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -10,6 +10,20 @@ from cocoonsandbox import APIError, Client, Template +class KeptNode(FakeNode): + protocol_version = "HTTP/1.1" + conns = 0 + drop_after_reply = False + + def setup(self): + super().setup() + KeptNode.conns += 1 + + def _reply(self, code, payload): + super()._reply(code, payload) + self.close_connection = self.drop_after_reply + + @pytest.fixture def node(): server = HTTPServer(("127.0.0.1", 0), FakeNode) @@ -368,17 +382,3 @@ def kept_node(): threading.Thread(target=server.serve_forever, daemon=True).start() yield f"127.0.0.1:{server.server_port}" server.shutdown() - - -class KeptNode(FakeNode): - protocol_version = "HTTP/1.1" - conns = 0 - drop_after_reply = False - - def setup(self): - super().setup() - KeptNode.conns += 1 - - def _reply(self, code, payload): - super()._reply(code, payload) - self.close_connection = self.drop_after_reply diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 72f2fa14..05830729 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -16,6 +16,65 @@ INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") +class FakeAgent: + """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains; slam closes as a failed relay does.""" + + def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0, slam: bool = False) -> None: + self.proto = proto + self.hang_up_after = hang_up_after + self.slam = slam + self.upgrades = 0 + self.hangups = 0 + self.closed = 0 + self._server = socket.create_server(("127.0.0.1", 0)) + self.addr = f"127.0.0.1:{self._server.getsockname()[1]}" + threading.Thread(target=self._accept, daemon=True).start() + + def stop(self) -> None: + self._server.close() + + def _accept(self) -> None: + with contextlib.suppress(OSError): + while True: + conn, _ = self._server.accept() + self.upgrades += 1 + threading.Thread(target=self._serve, args=(conn,), daemon=True).start() + + def _serve(self, conn: socket.socket) -> None: + with conn, contextlib.suppress(OSError): + reader = accept_upgrade(conn) + replies = 0 + while True: + line = reader.readline() + if not line: + break + op = json.loads(line)["op"] + if op in INPUT_OPS: + continue + for frame in self._answer(op): + conn.sendall(json.dumps(frame).encode() + b"\n") + replies += 1 + if self.slam: + break + if self.proto < KEEP_ALIVE_PROTO or replies == self.hang_up_after: + conn.shutdown(socket.SHUT_WR) + self.hangups += 1 + while reader.readline(): + pass + break + self.closed += 1 + + def _answer(self, op: str) -> list: + if op == "info": + return [{"type": "info", "version": "fake", "proto": self.proto, "uptime_secs": 0, "procs": 0}] + if op == "fs_stat": + return [{"type": "stat", "info": {"kind": "dir", "size": 0, "mode": 0o755, "mtime_epoch_secs": 0}}] + if op == "exec": + out = base64.b64encode(b"42\n").decode() + return [{"type": "started", "pid": 1}, {"type": "stdout", "data": out}, {"type": "exit", "code": 0}] + return [{"type": "error", "kind": "unimplemented", "message": op}] + + def test_calls_share_one_connection(fd_floor): agent = FakeAgent() sb = sandbox_at(agent.addr) @@ -140,62 +199,3 @@ def test_an_upload_an_old_daemon_rejects_raises_its_error_frame(): sb.write_file("/missing/f", bytes(64 << 20)) finally: agent.stop() - - -class FakeAgent: - """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains; slam closes as a failed relay does.""" - - def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0, slam: bool = False) -> None: - self.proto = proto - self.hang_up_after = hang_up_after - self.slam = slam - self.upgrades = 0 - self.hangups = 0 - self.closed = 0 - self._server = socket.create_server(("127.0.0.1", 0)) - self.addr = f"127.0.0.1:{self._server.getsockname()[1]}" - threading.Thread(target=self._accept, daemon=True).start() - - def stop(self) -> None: - self._server.close() - - def _accept(self) -> None: - with contextlib.suppress(OSError): - while True: - conn, _ = self._server.accept() - self.upgrades += 1 - threading.Thread(target=self._serve, args=(conn,), daemon=True).start() - - def _serve(self, conn: socket.socket) -> None: - with conn, contextlib.suppress(OSError): - reader = accept_upgrade(conn) - replies = 0 - while True: - line = reader.readline() - if not line: - break - op = json.loads(line)["op"] - if op in INPUT_OPS: - continue - for frame in self._answer(op): - conn.sendall(json.dumps(frame).encode() + b"\n") - replies += 1 - if self.slam: - break - if self.proto < KEEP_ALIVE_PROTO or replies == self.hang_up_after: - conn.shutdown(socket.SHUT_WR) - self.hangups += 1 - while reader.readline(): - pass - break - self.closed += 1 - - def _answer(self, op: str) -> list: - if op == "info": - return [{"type": "info", "version": "fake", "proto": self.proto, "uptime_secs": 0, "procs": 0}] - if op == "fs_stat": - return [{"type": "stat", "info": {"kind": "dir", "size": 0, "mode": 0o755, "mtime_epoch_secs": 0}}] - if op == "exec": - out = base64.b64encode(b"42\n").decode() - return [{"type": "started", "pid": 1}, {"type": "stdout", "data": out}, {"type": "exit", "code": 0}] - return [{"type": "error", "kind": "unimplemented", "message": op}] diff --git a/sdk/python/tests/test_stream.py b/sdk/python/tests/test_stream.py index 83332410..8b723009 100644 --- a/sdk/python/tests/test_stream.py +++ b/sdk/python/tests/test_stream.py @@ -1,5 +1,4 @@ -"""The client timeout bounds the dial and the upgrade, not the frames that -follow: a guest stream outlives it.""" +"""The client timeout bounds the dial and the upgrade, not the guest stream that follows.""" import contextlib import json diff --git a/sdk/python/tests/test_upgrade.py b/sdk/python/tests/test_upgrade.py index 235f80a6..65bc8e65 100644 --- a/sdk/python/tests/test_upgrade.py +++ b/sdk/python/tests/test_upgrade.py @@ -1,5 +1,4 @@ -"""The relay upgrade: over plain HTTP the request leaves with the first frame and -the 101 is read on the first recv; a refusal surfaces as an APIError.""" +"""Over plain HTTP the upgrade request leaves with the first frame; a refusal surfaces as an APIError.""" import contextlib import json @@ -75,18 +74,9 @@ def test_rejected_upgrade_is_an_api_error(): assert agent.early == ["fs_mkdir"], agent.early -def test_run_timeout_cuts_an_unanswered_upgrade(): - server = socket.create_server(("127.0.0.1", 0)) - addr = f"127.0.0.1:{server.getsockname()[1]}" - held: list[socket.socket] = [] - threading.Thread(target=lambda: held.append(server.accept()[0]), daemon=True).start() - sb = sandbox_at(addr, timeout=30, keep_alive=0) +def test_run_timeout_cuts_an_unanswered_upgrade(black_hole): + sb = sandbox_at(black_hole, timeout=30, keep_alive=0) started = time.monotonic() - try: - with pytest.raises(SandboxTimeout): - sb.run(["true"], timeout=0.2) - finally: - server.close() - for conn in held: - conn.close() + with pytest.raises(SandboxTimeout): + sb.run(["true"], timeout=0.2) assert time.monotonic() - started < 2, "the run deadline must cut a hung upgrade under a long client timeout" diff --git a/silkd/src/exec.rs b/silkd/src/exec.rs index abe2ee27..12ee4cd8 100644 --- a/silkd/src/exec.rs +++ b/silkd/src/exec.rs @@ -252,7 +252,7 @@ fn effective_path<'a>( }) } -/// Finds a bare program on path as execvp would, so std can posix_spawn an absolute path instead of forking; a relative entry leaves the search to execvp. +/// Finds a bare program on path like execvp, minus its fall-through past an entry whose exec fails, so std can posix_spawn an absolute path; a relative entry leaves the search to execvp. fn resolve_program(name: &str, path: Option<&str>) -> Option { if name.is_empty() || name.contains('/') { return None;