From 48505d991a3e80e60cea021b70cc476114822aff Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 14:20:35 +0800 Subject: [PATCH 1/4] sdk: follow a sandbox that moved to another node instead of failing until its lease ends A handle kept the owner it was claimed on, so after a live move, an evacuation or the old node's retirement every call answered 404 or failed to dial until the lease ran out; only Close looked the claim up. Both SDKs now run Lookup once on a 404 or a refused dial from the owner and adopt a different owner, dropping parked connections to the old one. A call whose request is its whole input is sent again there: the control calls and the one-reply and streaming verbs in Go, and every call in Python, whose inputs are bytes. Calls that already streamed input (Go Run with Stdin, WriteFile, Push) or handed back a stream fail once, and the next call reaches the new owner. The MCP server and the LangChain and OpenAI adapters inherit it. --- docs/cluster.md | 9 +- docs/sdk-python.md | 9 +- docs/sdk.md | 11 +- sdk/go/checkpoint.go | 4 +- sdk/go/client_test.go | 81 +++++++++++++ sdk/go/env.go | 7 +- sdk/go/port.go | 6 +- sdk/go/proc.go | 3 +- sdk/go/pty.go | 3 +- sdk/go/sandbox.go | 156 ++++++++++++++++++-------- sdk/go/utils.go | 110 ++++++++++-------- sdk/go/volumes.go | 21 ++-- sdk/go/watch.go | 3 +- sdk/python/cocoonsandbox/client.py | 2 +- sdk/python/cocoonsandbox/sandbox.py | 65 +++++++++-- sdk/python/tests/test_fault_matrix.py | 21 ++++ sdk/python/tests/test_keepalive.py | 31 ++++- 17 files changed, 409 insertions(+), 133 deletions(-) diff --git a/docs/cluster.md b/docs/cluster.md index 1726b0da..5dd75b28 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -305,8 +305,9 @@ retried every 5 seconds. Claims that cannot move are skipped and keep serving here; once `evacuating` is gone `claimed` counts them, and the log names each one with its reason. A client that holds a handle gets 404 from this node after the move, the same answer as for a released claim, or a connection error once -the node is retired, and only a `Lookup` tells these apart: the SDKs' `Close` -does this itself. A reader-served connection +the node is retired, and only a `Lookup` tells these apart: the SDKs run it on +either answer and send the call again to the new owner when nothing of it +reached the old one. A reader-served connection gets the retry as an error frame, not a 503, and a preview URL minted before the move stops working until it is minted again, except behind a shared `preview_advertise`, where it answers when it reaches the new owner. `DELETE /v1/drain` stops the evacuation: claims not yet started @@ -332,8 +333,8 @@ A move needs `meta_store` with a `cell`, which holds its record, and Only a none-lane claim without volumes moves; a hibernated, archived, failed or waking claim, an egress-lane claim and a claim with volumes answer 409. Open -exec and pty streams, port forwards and egress connections end; the client -meets a 404 on the source and runs `Lookup`. +exec and pty streams, port forwards and egress connections end; the next call +meets a 404 on the source, and the SDKs look the claim up and reach the target. **The record.** Each move has a record of its own in the meta store (table `sandboxd_moves`), separate from the archive rows: `id`, `from_node`, diff --git a/docs/sdk-python.md b/docs/sdk-python.md index a69bf78e..ee631ea8 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -142,9 +142,12 @@ branched). `sb.template_digest` is the exact content identity when the claim cloned a promoted template; it is empty for other sources. `Sandbox` is a context manager; `sb.close()` releases it (it follows a sandbox that moved to another node with a lookup, also when that node is gone; releasing one already -gone is not an error — double-release and reap races stay silent). Any other -call on a handle raises a 404 `APIError` once its sandbox has moved; run -`client.lookup` to tell that from a released sandbox. +gone is not an error — double-release and reap races stay silent). Every other +call follows a move too: a 404 or a refused connection from the owner runs +`client.lookup` once, and the call is sent again to a different owner, since +every argument (stdin, file data, tar streams) is bytes; only `find_iter` +fails once before the next call reaches the new owner. A released sandbox +still raises a 404 `APIError`. Volume sandboxes cannot hibernate, fork, checkpoint, or promote. Egress-lane sandboxes cannot either, and a sandbox whose tun is turned off cannot promote; each answers 409. Checkpoint diff --git a/docs/sdk.md b/docs/sdk.md index c596f3f0..5738c434 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -212,9 +212,14 @@ promoted template; it is empty for other sources. `Owner()` names the owning node, and `Token()` returns the per-sandbox bearer to persist with `ID` for a later `Lookup`; `Close()` releases the sandbox (it follows a sandbox that moved to another node with a `Lookup`, also when that node is gone, releasing -one already gone is not an error, and `Close` is bounded internally so it stays defer-friendly). Any other -call on a handle answers 404 once its sandbox has moved; run `Lookup` to tell -that from a released sandbox. +one already gone is not an error, and `Close` is bounded internally so it stays defer-friendly). Every +other call follows a move too: a 404 or a refused connection from the owner +runs `Lookup` once, and a different owner replaces the old one. A call whose +request is its whole input is sent again there: the control calls, `Run` +without `Stdin`, and the file, git, process and session calls that send no +data. `Run` with `Stdin`, `WriteFile`, `Push`, `OpenPty`, `Watch`, `DialPort`, `Lsp.Request`, +`Logs` and `Attach` fail once and the next call reaches the new owner. A +released sandbox still answers 404. Volume sandboxes and egress-lane sandboxes cannot hibernate, fork, checkpoint, or promote, nor take `WithArchiveOnExpire`; each answers 409, as does a promote of a sandbox whose tun is turned off. Passing diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index e49669a7..5ca79186 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -85,11 +85,11 @@ func (s *Sandbox) Checkpoint(ctx context.Context, name string) (*Checkpoint, err if err != nil { return nil, err } - cr, err := s.c.doJSON[checkpointResponse](ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/checkpoint", bytes.NewReader(body), s.c.apiToken, "checkpoint") + cr, err := s.ownerJSON[checkpointResponse](ctx, http.MethodPost, "/checkpoint", body, s.c.apiToken, "checkpoint") if err != nil { return nil, err } - return checkpointHandle(s.c, s.owner, cr.Checkpoint), nil + return checkpointHandle(s.c, s.Owner(), cr.Checkpoint), nil } // Checkpoints lists the connected node's checkpoints, newest first, as handles bound to it. diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 9e07cde4..30da5d3e 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -12,6 +12,8 @@ import ( "sync/atomic" "testing" "time" + + "github.com/cocoonstack/sandbox/sdk/go/silkd/silkdtest" ) func TestConnectAddress(t *testing.T) { @@ -160,6 +162,52 @@ func TestCloseReleasesASandboxThatMoved(t *testing.T) { } } +func TestAMovedSandboxFollowsItsNewOwnerOnTheNextCall(t *testing.T) { + sb, adopter := movedSandbox(t) + if _, err := sb.Stat(t.Context(), "/"); err != nil { + t.Fatalf("stat after the move: %v", err) + } + if got := sb.Owner(); got != adopter { + t.Errorf("owner %s, want the adopter %s", got, adopter) + } +} + +func TestAMovedSandboxRetriesAControlCallAtItsNewOwner(t *testing.T) { + sb, adopter := movedSandbox(t) + if _, err := sb.Renew(t.Context(), time.Minute); err != nil { + t.Fatalf("renew after the move: %v", err) + } + if got := sb.Owner(); got != adopter { + t.Errorf("owner %s, want the adopter %s", got, adopter) + } +} + +func TestAMovedSandboxDoesNotReplayARunThatStreamedStdin(t *testing.T) { + sb, adopter := movedSandbox(t) + if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"cat"}, Stdin: strings.NewReader("once")}); err == nil { + t.Fatal("a run whose stdin went to the old owner succeeded") + } + if got := sb.Owner(); got != adopter { + t.Errorf("owner %s, want the adopter %s for the next call", got, adopter) + } + if out, err := sb.Exec(t.Context(), "echo", "42"); err != nil || out != "42\n" { + t.Errorf("exec at the adopter: %q, %v", out, err) + } +} + +func TestAHandleFollowsItsSandboxOffARetiredNode(t *testing.T) { + sb, adopter := movedSandbox(t) + sb.mu.Lock() + sb.owner = retiredAddr(t) + sb.mu.Unlock() + if _, err := sb.Stat(t.Context(), "/"); err != nil { + t.Fatalf("stat after the owner retired: %v", err) + } + if got := sb.Owner(); got != adopter { + t.Errorf("owner %s, want the adopter %s", got, adopter) + } +} + func TestCloseFollowsAMoveOffARetiredNode(t *testing.T) { var released atomic.Bool var adopter *httptest.Server @@ -600,6 +648,39 @@ func TestAPIErrorDrainsOversizedBody(t *testing.T) { } } +func movedSandbox(t *testing.T) (*Sandbox, string) { + t.Helper() + fake := silkdtest.NewFake(t.TempDir()) + mux := http.NewServeMux() + var adopter *httptest.Server + mux.HandleFunc("GET /v1/sandboxes/sb_1/agent", func(w http.ResponseWriter, _ *http.Request) { + conn, err := silkdtest.Upgrade(w) + if err != nil { + t.Errorf("upgrade: %v", err) + return + } + fake.ServeConn(conn) + }) + mux.HandleFunc("GET /v1/sandboxes/sb_1/owner", func(w http.ResponseWriter, _ *http.Request) { + _ = json.MarshalWrite(w, map[string]string{"owner_addr": strings.TrimPrefix(adopter.URL, "http://")}) + }) + mux.HandleFunc("POST /v1/sandboxes/sb_1/renew", func(w http.ResponseWriter, _ *http.Request) { + _ = json.MarshalWrite(w, map[string]time.Time{"deadline": time.Now().Add(time.Minute)}) + }) + adopter = httptest.NewServer(mux) + t.Cleanup(adopter.Close) + old := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/peers" { + _ = json.MarshalWrite(w, map[string]any{"peers": []string{strings.TrimPrefix(adopter.URL, "http://")}}) + return + } + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(old.Close) + sb := testSandbox(t, old) + return sb, strings.TrimPrefix(adopter.URL, "http://") +} + func retiredAddr(t *testing.T) string { t.Helper() l, err := net.Listen("tcp", "127.0.0.1:0") diff --git a/sdk/go/env.go b/sdk/go/env.go index 9e76dc57..49c88a5b 100644 --- a/sdk/go/env.go +++ b/sdk/go/env.go @@ -1,7 +1,6 @@ package sandbox import ( - "bytes" "context" "net/http" ) @@ -16,7 +15,7 @@ type envPatchBody struct { // Env reads the claim's env with the node API token; a host-only entry comes back with an empty Value. func (s *Sandbox) Env(ctx context.Context) (map[string]EnvVar, error) { - resp, err := s.c.doJSON[envBody](ctx, http.MethodGet, s.owner, "/v1/sandboxes/"+s.ID+"/env", nil, s.c.apiToken, "read env") + resp, err := s.ownerJSON[envBody](ctx, http.MethodGet, "/env", nil, s.c.apiToken, "read env") if err != nil { return nil, err } @@ -29,7 +28,7 @@ func (s *Sandbox) SetEnv(ctx context.Context, env map[string]EnvVar) error { if err != nil { return err } - return s.c.doNoContent(ctx, http.MethodPut, s.owner, "/v1/sandboxes/"+s.ID+"/env", bytes.NewReader(body), s.c.apiToken, "set env") + return s.ownerNoContent(ctx, http.MethodPut, "/env", body, s.c.apiToken, "set env") } // PatchEnv sets each entry of patch, removes each nil one and keeps the rest as stored, host-only values included. @@ -38,5 +37,5 @@ func (s *Sandbox) PatchEnv(ctx context.Context, patch map[string]*EnvVar) error if err != nil { return err } - return s.c.doNoContent(ctx, http.MethodPatch, s.owner, "/v1/sandboxes/"+s.ID+"/env", bytes.NewReader(body), s.c.apiToken, "patch env") + return s.ownerNoContent(ctx, http.MethodPatch, "/env", body, s.c.apiToken, "patch env") } diff --git a/sdk/go/port.go b/sdk/go/port.go index 62980bb4..d28f4931 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -1,7 +1,6 @@ package sandbox import ( - "bytes" "context" "errors" "io" @@ -86,7 +85,7 @@ func (s *Sandbox) PreviewURL(ctx context.Context, port uint16, ttl time.Duration if err != nil { return "", err } - pr, err := s.c.doJSON[previewResponse](ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/preview", bytes.NewReader(body), s.c.apiToken, "preview") + pr, err := s.ownerJSON[previewResponse](ctx, http.MethodPost, "/preview", body, s.c.apiToken, "preview") if err != nil { return "", err } @@ -112,7 +111,8 @@ func (s *Sandbox) ProxyPort(ctx context.Context, localAddr string, port uint16) } // openStream turns a call into a raw byte stream, shared by DialPort and Lsp.Request. -func (s *Sandbox) openStream(ctx context.Context, req wire.Request) (*PortConn, error) { +func (s *Sandbox) openStream(ctx context.Context, req wire.Request) (_ *PortConn, err error) { + defer func() { s.rebind(ctx, err) }() conn, l, err := s.call(ctx, req) if err != nil { return nil, err diff --git a/sdk/go/proc.go b/sdk/go/proc.go index 8f4c6963..17b4c7a0 100644 --- a/sdk/go/proc.go +++ b/sdk/go/proc.go @@ -52,7 +52,8 @@ func (s *Sandbox) Attach(ctx context.Context, pid uint32, stdout, stderr io.Writ return s.drainProc(ctx, &wire.Attach{PID: pid}, stdout, stderr) } -func (s *Sandbox) drainProc(ctx context.Context, req wire.Request, stdout, stderr io.Writer) (int32, bool, error) { +func (s *Sandbox) drainProc(ctx context.Context, req wire.Request, stdout, stderr io.Writer) (_ int32, _ bool, err error) { + defer func() { s.rebind(ctx, err) }() conn, l, err := s.call(ctx, req) if err != nil { return 0, false, err diff --git a/sdk/go/pty.go b/sdk/go/pty.go index 9079df0c..f037bd27 100644 --- a/sdk/go/pty.go +++ b/sdk/go/pty.go @@ -87,7 +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 wire.PtyOpen) (*Pty, error) { +func (s *Sandbox) OpenPty(ctx context.Context, opts wire.PtyOpen) (_ *Pty, err error) { + defer func() { s.rebind(ctx, err) }() conn, l, err := s.call(ctx, &opts) if err != nil { return nil, err diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index f4e2c8af..2272af31 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -1,13 +1,14 @@ package sandbox import ( - "bytes" "context" "errors" "fmt" "io" "net/http" "strings" + "sync" + "syscall" "time" "github.com/cocoonstack/sandbox/protocol/wire" @@ -67,13 +68,17 @@ type Sandbox struct { c *Client token string - owner string pool agentPool + + mu sync.Mutex + owner string } // Owner returns the data-plane address of the node that owns the sandbox — // on a cluster the node a redirected claim landed on. func (s *Sandbox) Owner() string { + s.mu.Lock() + defer s.mu.Unlock() return s.owner } @@ -103,43 +108,11 @@ func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { if len(cmd.Argv) == 0 { return 0, fmt.Errorf("empty argv") } - conn, l, err := s.call(ctx, &wire.Exec{Argv: cmd.Argv, Cwd: cmd.Cwd, Env: cmd.Env, User: cmd.User, Session: cmd.Session}) - if err != nil { - return 0, err + code, err := s.run(ctx, cmd) + if s.rebind(ctx, err) && cmd.Stdin == nil { + return s.run(ctx, cmd) } - defer l.close() - - stdinDone := make(chan struct{}) - if cmd.Stdin == nil { - // A guest that already answered and closed fails this send; the frames it sent still come back below - _ = conn.Send(wire.StdinClose{}) - close(stdinDone) - } else { - go func() { - defer close(stdinDone) - pumpStdin(conn, cmd.Stdin) - }() - } - - pid, code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) - // A pump still parked in Read would write into the next RPC, so its connection is not kept - select { - case <-stdinDone: - err = l.done(err) - default: - } - if err != nil { - if ctx.Err() != nil && pid != 0 { - kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), killWait) - _ = s.Kill(kctx, pid, 0) - cancel() - } - return 0, err - } - if !exited { - return 0, unexpected(&wire.Done{}) - } - return int(code), nil + return code, err } // Fork clones the sandbox into count children with their own leases; on error no child survives. @@ -148,13 +121,13 @@ func (s *Sandbox) Fork(ctx context.Context, count int, ttl time.Duration) ([]*Sa if err != nil { return nil, err } - fr, err := s.c.doJSON[forkResponse](ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/fork", bytes.NewReader(body), s.c.apiToken, "fork") + fr, err := s.ownerJSON[forkResponse](ctx, http.MethodPost, "/fork", body, s.c.apiToken, "fork") if err != nil { return nil, err } children := make([]*Sandbox, len(fr.Children)) for i, c := range fr.Children { - children[i] = s.c.handleFrom(s.owner, c) + children[i] = s.c.handleFrom(s.Owner(), c) } return children, nil } @@ -165,20 +138,20 @@ func (s *Sandbox) Promote(ctx context.Context, template string) (*Template, erro if err != nil { return nil, err } - pr, err := s.c.doJSON[promoteResponse](ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/promote", bytes.NewReader(body), s.c.apiToken, "promote") + pr, err := s.ownerJSON[promoteResponse](ctx, http.MethodPost, "/promote", body, s.c.apiToken, "promote") if err != nil { return nil, err } return &Template{ Name: pr.Key.Template, ContentDigest: pr.ContentDigest, - c: s.c, addr: s.owner, net: pr.Key.Net, size: pr.Key.Size, + c: s.c, addr: s.Owner(), net: pr.Key.Net, size: pr.Key.Size, }, nil } // Hibernate snapshots the sandbox and stops its VM; the next guest call wakes it, and the TTL keeps running. func (s *Sandbox) Hibernate(ctx context.Context) error { s.pool.drain() - return s.c.doNoContent(ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/hibernate", nil, s.token, "hibernate") + return s.ownerNoContent(ctx, http.MethodPost, "/hibernate", nil, s.token, "hibernate") } // Renew resets the lease to ttl from now (zero means the server default) and records the granted deadline in Deadline. @@ -187,7 +160,7 @@ func (s *Sandbox) Renew(ctx context.Context, ttl time.Duration) (time.Time, erro if err != nil { return time.Time{}, err } - rr, err := s.c.doJSON[renewResponse](ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/renew", bytes.NewReader(body), s.token, "renew") + rr, err := s.ownerJSON[renewResponse](ctx, http.MethodPost, "/renew", body, s.token, "renew") if err != nil { return time.Time{}, err } @@ -201,7 +174,7 @@ func (s *Sandbox) Close() error { s.pool.drain() ctx, cancel := context.WithTimeout(context.Background(), releaseTimeout) defer cancel() - found, err := s.releaseAt(ctx, s.owner) + found, err := s.releaseAt(ctx, s.Owner()) if _, refused := errors.AsType[*APIError](err); refused || found { return err } @@ -213,6 +186,46 @@ func (s *Sandbox) Close() error { return err } +func (s *Sandbox) run(ctx context.Context, cmd Cmd) (int, error) { + conn, l, err := s.call(ctx, &wire.Exec{Argv: cmd.Argv, Cwd: cmd.Cwd, Env: cmd.Env, User: cmd.User, Session: cmd.Session}) + if err != nil { + return 0, err + } + defer l.close() + + stdinDone := make(chan struct{}) + if cmd.Stdin == nil { + // A guest that already answered and closed fails this send; the frames it sent still come back below + _ = conn.Send(wire.StdinClose{}) + close(stdinDone) + } else { + go func() { + defer close(stdinDone) + pumpStdin(conn, cmd.Stdin) + }() + } + + pid, code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) + // A pump still parked in Read would write into the next RPC, so its connection is not kept + select { + case <-stdinDone: + err = l.done(err) + default: + } + if err != nil { + if ctx.Err() != nil && pid != 0 { + kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), killWait) + _ = s.Kill(kctx, pid, 0) + cancel() + } + return 0, err + } + if !exited { + return 0, unexpected(&wire.Done{}) + } + return int(code), nil +} + func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err error) { resp, err := s.c.roundTrip(ctx, http.MethodPost, addr, "/v1/sandboxes/"+s.ID+"/release", nil, s.token) if err != nil { @@ -228,6 +241,49 @@ func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err e return false, apiError("release", resp) } +// rebind follows a moved sandbox: a 404 or a refused dial from its owner asks Lookup, and a different owner replaces it and its parked connections. +func (s *Sandbox) rebind(ctx context.Context, err error) bool { + if !ownerGone(err) { + return false + } + moved, lookupErr := s.c.Lookup(ctx, s.ID, s.token) + if lookupErr != nil { + return false + } + s.mu.Lock() + changed := moved.owner != s.owner + s.owner = moved.owner + s.mu.Unlock() + if changed { + s.pool.drain() + } + return changed +} + +// retryMoved runs rpc once more on the new owner when nothing of it reached the old one; rpc's request must be its whole input. +func (s *Sandbox) retryMoved(ctx context.Context, rpc func() error) error { + err := rpc() + if s.rebind(ctx, err) { + return rpc() + } + return err +} + +func (s *Sandbox) ownerJSON[T any](ctx context.Context, method, suffix string, body []byte, bearer, verb string) (T, error) { + var out T + err := s.retryMoved(ctx, func() (err error) { + out, err = s.c.doJSON[T](ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) + return err + }) + return out, err +} + +func (s *Sandbox) ownerNoContent(ctx context.Context, method, suffix string, body []byte, bearer, verb string) error { + return s.retryMoved(ctx, func() error { + return s.c.doNoContent(ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) + }) +} + // call sends req on a relay connection and hands back the lease that ends the RPC; ctx cancellation closes the connection under it. func (s *Sandbox) call(ctx context.Context, req wire.Request) (*silkd.Conn, *lease, error) { c, err := s.connect(ctx, req) @@ -253,13 +309,21 @@ func (s *Sandbox) connect(ctx context.Context, req wire.Request) (*agentConn, er } func (s *Sandbox) dial(ctx context.Context) (*agentConn, error) { - raw, err := s.c.dialAgent(ctx, s.owner, s.ID, s.token) + raw, err := s.c.dialAgent(ctx, s.Owner(), s.ID, s.token) if err != nil { return nil, err } return newAgentConn(raw), nil } +// ownerGone reports an error after which nothing reached the owner: a 404, or a refused dial. +func ownerGone(err error) bool { + if ae, ok := errors.AsType[*APIError](err); ok { + return ae.Status == http.StatusNotFound + } + return errors.Is(err, syscall.ECONNREFUSED) +} + // pumpStdin chunks the reader into stdin frames. func pumpStdin(conn *silkd.Conn, r io.Reader) { buf := make([]byte, stdinChunk) diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 7d414b98..9359f5b9 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -1,6 +1,7 @@ package sandbox import ( + "bytes" "cmp" "context" "errors" @@ -21,16 +22,19 @@ type respPtr[T any] interface { 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 { - return err - } - defer l.close() - return l.done(terminalErr(ctx, conn)) + return s.retryMoved(ctx, func() error { + conn, l, err := s.call(ctx, req) + if err != nil { + return err + } + defer l.close() + return l.done(terminalErr(ctx, conn)) + }) } // uploadRPC streams r as full Data frames after req and expects Done; an early terminal frame is the guest rejecting the upload and stops the stream. -func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) error { +func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) (err error) { + defer func() { s.rebind(ctx, err) }() conn, l, err := s.call(ctx, req) if err != nil { return err @@ -66,22 +70,28 @@ func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) } 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 - } - defer l.close() - return l.done(drainData(ctx, conn, sink)) + return s.retryMoved(ctx, func() error { + conn, l, err := s.call(ctx, req) + if err != nil { + return err + } + defer l.close() + return l.done(drainData(ctx, conn, sink)) + }) } func (s *Sandbox) oneShotRPC[T any, PT respPtr[T]](ctx context.Context, req wire.Request) (*T, error) { - conn, l, err := s.call(ctx, req) - if err != nil { - return nil, err - } - defer l.close() - v, err := expect[T, PT](ctx, conn) - return v, l.done(err) + var v *T + err := s.retryMoved(ctx, func() error { + conn, l, err := s.call(ctx, req) + if err != nil { + return err + } + defer l.close() + v, err = expect[T, PT](ctx, conn) + return l.done(err) + }) + return v, err } func (s *Sandbox) collectRPC[T any, PT respPtr[T]](ctx context.Context, req wire.Request) ([]T, error) { @@ -98,35 +108,38 @@ func (s *Sandbox) collectRPC[T any, PT respPtr[T]](ctx context.Context, req wire // streamRPC sends req and yields each streamed frame of type T until Done; breaking out closes the connection, which ends the guest-side producer. func (s *Sandbox) streamRPC[T any, PT respPtr[T]](ctx context.Context, req wire.Request) iter.Seq2[T, error] { return func(yield func(T, error) bool) { - var zero T - conn, l, err := s.call(ctx, req) - if err != nil { - yield(zero, err) - return - } - defer l.close() - for { - resp, err := recv(ctx, conn) + err := s.retryMoved(ctx, func() error { + conn, l, err := s.call(ctx, req) if err != nil { - yield(zero, err) - return + return err } - if v, ok := resp.(PT); ok { - if !yield(*v, nil) { - return + defer l.close() + for { + resp, err := recv(ctx, conn) + if err != nil { + return err + } + if v, ok := resp.(PT); ok { + if !yield(*v, nil) { + return nil + } + continue + } + switch r := resp.(type) { + case *wire.Done: + l.reuse() + return nil + case *wire.ErrorResp: + l.reuse() + return r + default: + return unexpected(resp) } - continue - } - switch r := resp.(type) { - case *wire.Done: - l.reuse() - case *wire.ErrorResp: - l.reuse() - yield(zero, r) - default: - yield(zero, unexpected(resp)) } - return + }) + if err != nil { + var zero T + yield(zero, err) } } } @@ -253,3 +266,10 @@ func readChunk(r io.Reader, buf []byte) (int, bool, error) { } return n, false, nil } + +func bodyReader(b []byte) io.Reader { + if b == nil { + return nil + } + return bytes.NewReader(b) +} diff --git a/sdk/go/volumes.go b/sdk/go/volumes.go index 4d7e3910..f6bc7313 100644 --- a/sdk/go/volumes.go +++ b/sdk/go/volumes.go @@ -1,12 +1,10 @@ package sandbox import ( - "bytes" "context" "encoding/json/v2" "errors" "fmt" - "io" "net/http" "slices" ) @@ -71,16 +69,25 @@ func (s *Sandbox) PatchVolumes(ctx context.Context, patch VolumePatch) (SandboxV } func (s *Sandbox) requestVolumes(ctx context.Context, method string, request any, verb string) (SandboxVolumesResponse, error) { - var out SandboxVolumesResponse - var body io.Reader + var body []byte if request != nil { encoded, err := encodeBody(verb, request) if err != nil { - return out, err + return SandboxVolumesResponse{}, err } - body = bytes.NewReader(encoded) + body = encoded } - resp, err := s.c.roundTrip(ctx, method, s.owner, "/v1/sandboxes/"+s.ID+"/volumes", body, s.c.apiToken) + var out SandboxVolumesResponse + err := s.retryMoved(ctx, func() (err error) { + out, err = s.volumesAt(ctx, method, body, verb) + return err + }) + return out, err +} + +func (s *Sandbox) volumesAt(ctx context.Context, method string, body []byte, verb string) (SandboxVolumesResponse, error) { + var out SandboxVolumesResponse + resp, err := s.c.roundTrip(ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+"/volumes", bodyReader(body), s.c.apiToken) if err != nil { return out, err } diff --git a/sdk/go/watch.go b/sdk/go/watch.go index cc732977..4f5495ff 100644 --- a/sdk/go/watch.go +++ b/sdk/go/watch.go @@ -76,7 +76,8 @@ func (w *Watcher) setErr(err error) { } // Watch streams filesystem events under path; events caused after it returns are guaranteed captured. -func (s *Sandbox) Watch(ctx context.Context, path string, recursive bool) (*Watcher, error) { +func (s *Sandbox) Watch(ctx context.Context, path string, recursive bool) (_ *Watcher, err error) { + defer func() { s.rebind(ctx, err) }() conn, l, err := s.call(ctx, &wire.FsWatch{Path: path, Recursive: recursive}) if err != nil { return nil, err diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 24532184..7fdf4f87 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -214,7 +214,7 @@ def _request( except (OSError, http.client.HTTPException) as exc: if deadline is not None and time.monotonic() >= deadline: raise SandboxTimeout(f"{verb} timed out") from None - raise APIError(verb, 0, str(exc)) from None + raise APIError(verb, 0, str(exc)) from exc if status >= 400: try: reply = json.loads(raw) diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index cc34f614..5f301fe7 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -3,12 +3,13 @@ from __future__ import annotations import contextlib +import functools import socket import threading import time from collections import deque from collections.abc import Callable, Iterator -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, TypeVar, cast from .checkpoint import Checkpoint from .conn import Conn, ConnPool, _Closeable, remaining_timeout @@ -32,6 +33,23 @@ _ACCEPT_POLL_SECONDS = 1.0 _KILL_WAIT_SECONDS = 5.0 +_F = TypeVar("_F", bound=Callable[..., Any]) + + +def _follows_move(method: _F) -> _F: + """Runs a call once more on the owner a 404 points to; its arguments must be re-sendable.""" + + @functools.wraps(method) + def wrapper(self: Sandbox, *args: Any, **kwargs: Any) -> Any: + try: + return method(self, *args, **kwargs) + except (APIError, ProtocolError) as exc: + if not self._rebind(exc): + raise + return method(self, *args, **kwargs) + + return cast(_F, wrapper) + class Sandbox: """One claimed microVM.""" @@ -124,6 +142,7 @@ def exec( raise ExitError(code, err.decode(errors="replace"), out.decode(errors="replace")) return out.decode(errors="replace") + @_follows_move def run( self, argv: list[str], @@ -223,15 +242,18 @@ def attach( """Replays then follows output; returns None if the process record disappears.""" return self._drain_proc("attach", pid, on_stdout, on_stderr) + @_follows_move def write_file(self, path: str, data: bytes, mode: int | None = None) -> None: """Writes data to path atomically (temp + rename on the guest).""" with self._lease(self._connect("fs_write", path=path, mode=mode)) as conn: _upload(conn, data, BULK_CHUNK) + @_follows_move def read_file(self, path: str) -> bytes: with self._lease(self._connect("fs_read", path=path)) as conn: return _drain_data(conn) + @_follows_move def list_dir(self, path: str) -> list[dict[str, Any]]: with self._lease(self._connect("fs_list", path=path)) as conn: entries: list[dict[str, Any]] = [] @@ -252,16 +274,19 @@ def remove(self, path: str, recursive: bool = False) -> None: def rename(self, src: str, dst: str) -> None: self._done_rpc("fs_rename", **{"from": src, "to": dst}) + @_follows_move def push(self, dest: str, tar_stream: bytes) -> None: """Extracts a tar stream into dest; a truncated stream leaves dest untouched.""" with self._lease(self._connect("fs_push", dest=dest)) as conn: _upload(conn, tar_stream, STREAM_CHUNK) + @_follows_move def pull(self, path: str) -> bytes: """Returns path (file or tree) as a tar archive.""" with self._lease(self._connect("fs_pull", path=path)) as conn: return _drain_data(conn) + @_follows_move def find(self, path: str, pattern: str, glob: str = "") -> list[dict[str, Any]]: return list(self.find_iter(path, pattern, glob)) @@ -272,6 +297,7 @@ def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict[st if f["type"] == "match": yield f + @_follows_move def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict[str, Any]]: with self._lease(self._connect("fs_replace", files=files, pattern=pattern, replacement=replacement)) as conn: return [f for f in conn.recv_until("done") if f["type"] == "replaced"] @@ -329,22 +355,18 @@ def fork(self, count: int, ttl_seconds: int = 0) -> list[Sandbox]: body = {"token": self.token, "count": count} if ttl_seconds: body["ttl_seconds"] = ttl_seconds - reply = self._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/fork", body, "fork") + reply = self._owner_request("fork", body, "fork") return [self._client._handle_from(self.owner, child, "fork") for child in reply.get("children") or []] def hibernate(self) -> None: """Snapshots and stops the VM; the next guest call restores its state.""" self._pool.drain() - self._client._request( - self.owner, "POST", f"/v1/sandboxes/{self.id}/hibernate", None, "hibernate", bearer=self.token - ) + self._owner_request("hibernate", None, "hibernate", bearer=self.token) def renew(self, ttl_seconds: int = 0) -> str: """Resets the lease to ttl_seconds from now (0 means the node default) and records the granted deadline.""" body = {"ttl_seconds": ttl_seconds} if ttl_seconds else None - reply = self._client._request( - self.owner, "POST", f"/v1/sandboxes/{self.id}/renew", body, "renew", bearer=self.token - ) + reply = self._owner_request("renew", body, "renew", bearer=self.token) self.deadline = cast(str, _require_field(reply, "deadline", "renew")) return self.deadline @@ -353,13 +375,13 @@ def checkpoint(self, name: str = "") -> Checkpoint: body = {"token": self.token} if name: body["name"] = name - reply = self._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/checkpoint", body, "checkpoint") + reply = self._owner_request("checkpoint", body, "checkpoint") return Checkpoint(self._client, self.owner, _require_field(reply, "checkpoint", "checkpoint")) def promote(self, template: str) -> Template: """Publishes a claimable template and returns an owner-bound handle.""" body = {"token": self.token, "template": template} - reply = self._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/promote", body, "promote") + reply = self._owner_request("promote", body, "promote") key = _require_field(reply, "key", "promote") return Template( self._client, @@ -398,7 +420,7 @@ def preview_url(self, port: int, ttl_seconds: int = 0) -> str: body = {"token": self.token, "port": port} if ttl_seconds: body["ttl_seconds"] = ttl_seconds - reply = self._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/preview", body, "preview") + reply = self._owner_request("preview", body, "preview") return cast(str, _require_field(reply, "url", "preview")) def dial_port(self, port: int) -> PortConn: @@ -431,6 +453,7 @@ def close(self) -> None: def _release_at(self, addr: str) -> None: self._client._request(addr, "POST", f"/v1/sandboxes/{self.id}/release", None, "release", bearer=self.token) + @_follows_move def _volumes_request( self, method: str, body: dict[str, Any] | None, verb: str, timeout: float | None ) -> SandboxVolumesResponse: @@ -462,6 +485,23 @@ def _record_volumes(self, view: SandboxVolumesResponse) -> None: self.volumes_want = [volume.copy() for volume in view["volumes_want"]] self.volumes = [volume.copy() for volume in view["volumes"]] + def _rebind(self, exc: SandboxError) -> bool: + if not (isinstance(exc, APIError) and exc.status == 404 or isinstance(exc.__cause__, ConnectionRefusedError)): + return False + try: + moved = self._client.lookup(self.id, self.token) + except APIError: + return False + if moved.owner == self.owner: + return False + self.owner = moved.owner + self._pool.drain() + return True + + @_follows_move + def _owner_request(self, suffix: str, body: dict[str, Any] | None, verb: str, bearer: str = "") -> dict[str, Any]: + return self._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/{suffix}", body, verb, bearer=bearer) + def _dial(self, deadline: float | None = None) -> Conn: return self._client._dial(self.owner, self.id, self.token, deadline) @@ -492,6 +532,7 @@ def _lease(self, conn: Conn) -> Iterator[Conn]: raise self._pool.park(conn) + @_follows_move def _open_stream(self, op: str, expect: str = "ready", **fields: object) -> tuple[Conn, dict[str, Any]]: conn = self._connect(op, **fields) try: @@ -549,6 +590,7 @@ def pump_out() -> None: with contextlib.suppress(OSError): local.close() + @_follows_move def _call(self, op: str, expect: str, **fields: object) -> dict[str, Any]: with self._lease(self._connect(op, **fields)) as conn: return _expect(conn, expect) @@ -556,6 +598,7 @@ def _call(self, op: str, expect: str, **fields: object) -> dict[str, Any]: def _done_rpc(self, op: str, **fields: object) -> None: self._call(op, "done", **fields) + @_follows_move def _drain_proc( self, op: str, diff --git a/sdk/python/tests/test_fault_matrix.py b/sdk/python/tests/test_fault_matrix.py index 0825e36b..8ae16e48 100644 --- a/sdk/python/tests/test_fault_matrix.py +++ b/sdk/python/tests/test_fault_matrix.py @@ -221,6 +221,27 @@ def release(body, path): assert released +def test_a_moved_sandbox_retries_a_control_call_at_its_new_owner(spawn_node): + adopter = spawn_node( + { + ("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {}), + ("POST", "/v1/sandboxes/sb_1/renew"): lambda body, path: (200, {"deadline": "2026-11-06T12:00:00Z"}), + } + ) + old = spawn_node({("GET", "/v1/peers"): lambda body, path: (200, {"peers": [adopter]})}) + sb = Client(old).attach(old, "sb_1", "tok") + assert sb.renew(60) == "2026-11-06T12:00:00Z" + assert sb.owner == adopter + + +def test_a_released_sandbox_still_answers_404(spawn_node): + old = spawn_node({("GET", "/v1/peers"): lambda body, path: (200, {"peers": []})}) + sb = Client(old).attach(old, "sb_1", "tok") + with pytest.raises(APIError) as exc: + sb.renew(60) + assert exc.value.status == 404 and sb.owner == old + + def test_close_follows_a_move_off_a_retired_node(spawn_node, dead_addr): released = [] diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index b70c2897..1db3c49e 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -9,7 +9,7 @@ import pytest from conftest import accept_upgrade, sandbox_at, wait_until -from cocoonsandbox import SilkdError +from cocoonsandbox import Client, SilkdError INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") @@ -181,6 +181,35 @@ def test_an_upload_the_daemon_rejects_at_open_raises_its_error_frame(relay): sb.write_file("/missing/f", bytes(64 << 20)) +def test_a_moved_sandbox_follows_its_new_owner_on_the_next_call(spawn_node): + agent = FakeAgent() + try: + directory = spawn_node( + {("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {"owner_addr": agent.addr})} + ) + old = spawn_node({("GET", "/v1/peers"): lambda body, path: (200, {"peers": [directory]})}) + sb = Client(old).attach(old, "sb_1", "tok") + assert sb.stat("/")["kind"] == "dir" + assert sb.owner == agent.addr + assert sb.exec("echo", "42") == "42\n" + finally: + agent.stop() + + +def test_a_handle_follows_its_sandbox_off_a_retired_node(spawn_node, dead_addr): + agent = FakeAgent() + try: + directory = spawn_node( + {("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {"owner_addr": agent.addr})} + ) + entry = spawn_node({("GET", "/v1/peers"): lambda body, path: (200, {"peers": [directory]})}) + sb = Client(entry).attach(dead_addr, "sb_1", "tok") + assert sb.stat("/")["kind"] == "dir" + assert sb.owner == agent.addr + finally: + agent.stop() + + @pytest.fixture def relay(): started = [] From ff30ed0ad47052e13471eda8f44d60b584275ec5 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 14:37:17 +0800 Subject: [PATCH 2/4] sdk: rebind only on an answer the owner gave, and bind derived handles to the node that answered Codex review. rebind judged errors by type, so a caller's own stdout writer failing with ECONNREFUSED, or a Python output callback raising a 404 from another sandbox call, made the SDK look the claim up and run the command again on another node. The owner's refused dial, its agent-upgrade 404 and its control-call 404 are now tagged where they arise, and only a tagged error rebinds. Fork, Checkpoint and Promote bound their new handles to the handle's current owner, which a concurrent rebind could have moved off the node that answered; they now take the address that answered. --- sdk/go/checkpoint.go | 4 +-- sdk/go/client_test.go | 28 ++++++++++++++++++ sdk/go/env.go | 2 +- sdk/go/port.go | 2 +- sdk/go/sandbox.go | 45 ++++++++++++++++++----------- sdk/go/upgrade.go | 4 +-- sdk/go/volumes.go | 2 +- sdk/python/cocoonsandbox/conn.py | 8 +++-- sdk/python/cocoonsandbox/errors.py | 12 +++++++- sdk/python/cocoonsandbox/sandbox.py | 43 ++++++++++++++++++--------- sdk/python/tests/test_keepalive.py | 21 +++++++++++++- 11 files changed, 129 insertions(+), 42 deletions(-) diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index 5ca79186..9c09ba01 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -85,11 +85,11 @@ func (s *Sandbox) Checkpoint(ctx context.Context, name string) (*Checkpoint, err if err != nil { return nil, err } - cr, err := s.ownerJSON[checkpointResponse](ctx, http.MethodPost, "/checkpoint", body, s.c.apiToken, "checkpoint") + cr, addr, err := s.ownerJSON[checkpointResponse](ctx, http.MethodPost, "/checkpoint", body, s.c.apiToken, "checkpoint") if err != nil { return nil, err } - return checkpointHandle(s.c, s.Owner(), cr.Checkpoint), nil + return checkpointHandle(s.c, addr, cr.Checkpoint), nil } // Checkpoints lists the connected node's checkpoints, newest first, as handles bound to it. diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 30da5d3e..52f78265 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -10,6 +10,7 @@ import ( "reflect" "strings" "sync/atomic" + "syscall" "testing" "time" @@ -208,6 +209,29 @@ func TestAHandleFollowsItsSandboxOffARetiredNode(t *testing.T) { } } +func TestARunWhoseWriterFailsIsNotReplayedElsewhere(t *testing.T) { + fake := silkdtest.NewFake(t.TempDir()) + var elsewhere atomic.Int32 + other := newAgentServer(t, func(c net.Conn) { + elsewhere.Add(1) + fake.ServeConn(c) + }) + owner := newAgentServer(t, fake.ServeConn) + entry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _ = json.MarshalWrite(w, map[string]string{"owner_addr": strings.TrimPrefix(other.URL, "http://")}) + })) + t.Cleanup(entry.Close) + sb := &Sandbox{ID: "sb_1", c: testClient(t, entry), token: "tok", owner: strings.TrimPrefix(owner.URL, "http://")} + t.Cleanup(sb.pool.drain) + refused := failingWriter{syscall.ECONNREFUSED} + if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"echo", "42"}, Stdout: refused}); !errors.Is(err, syscall.ECONNREFUSED) { + t.Fatalf("run: %v, want the writer's error", err) + } + if n := elsewhere.Load(); n != 0 { + t.Errorf("the command was sent again to another node %d times", n) + } +} + func TestCloseFollowsAMoveOffARetiredNode(t *testing.T) { var released atomic.Bool var adopter *httptest.Server @@ -700,3 +724,7 @@ func testClient(t testing.TB, ts *httptest.Server, opts ...ClientOption) *Client } return c } + +type failingWriter struct{ err error } + +func (w failingWriter) Write([]byte) (int, error) { return 0, w.err } diff --git a/sdk/go/env.go b/sdk/go/env.go index 49c88a5b..899d201a 100644 --- a/sdk/go/env.go +++ b/sdk/go/env.go @@ -15,7 +15,7 @@ type envPatchBody struct { // Env reads the claim's env with the node API token; a host-only entry comes back with an empty Value. func (s *Sandbox) Env(ctx context.Context) (map[string]EnvVar, error) { - resp, err := s.ownerJSON[envBody](ctx, http.MethodGet, "/env", nil, s.c.apiToken, "read env") + resp, _, err := s.ownerJSON[envBody](ctx, http.MethodGet, "/env", nil, s.c.apiToken, "read env") if err != nil { return nil, err } diff --git a/sdk/go/port.go b/sdk/go/port.go index d28f4931..35f3cb0e 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -85,7 +85,7 @@ func (s *Sandbox) PreviewURL(ctx context.Context, port uint16, ttl time.Duration if err != nil { return "", err } - pr, err := s.ownerJSON[previewResponse](ctx, http.MethodPost, "/preview", body, s.c.apiToken, "preview") + pr, _, err := s.ownerJSON[previewResponse](ctx, http.MethodPost, "/preview", body, s.c.apiToken, "preview") if err != nil { return "", err } diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index 2272af31..e0608a6d 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -48,6 +48,13 @@ func (e *ExitError) Error() string { return fmt.Sprintf("exit status %d", e.Code) } +// ownerGoneError tags an owner answer after which nothing reached it, so the call may go to the sandbox's new owner. +type ownerGoneError struct{ err error } + +func (e *ownerGoneError) Error() string { return e.err.Error() } + +func (e *ownerGoneError) Unwrap() error { return e.err } + // Sandbox is a claimed sandbox handle; all data-plane calls dial its owning // node directly. type Sandbox struct { @@ -121,13 +128,13 @@ func (s *Sandbox) Fork(ctx context.Context, count int, ttl time.Duration) ([]*Sa if err != nil { return nil, err } - fr, err := s.ownerJSON[forkResponse](ctx, http.MethodPost, "/fork", body, s.c.apiToken, "fork") + fr, addr, err := s.ownerJSON[forkResponse](ctx, http.MethodPost, "/fork", body, s.c.apiToken, "fork") if err != nil { return nil, err } children := make([]*Sandbox, len(fr.Children)) for i, c := range fr.Children { - children[i] = s.c.handleFrom(s.Owner(), c) + children[i] = s.c.handleFrom(addr, c) } return children, nil } @@ -138,13 +145,13 @@ func (s *Sandbox) Promote(ctx context.Context, template string) (*Template, erro if err != nil { return nil, err } - pr, err := s.ownerJSON[promoteResponse](ctx, http.MethodPost, "/promote", body, s.c.apiToken, "promote") + pr, addr, err := s.ownerJSON[promoteResponse](ctx, http.MethodPost, "/promote", body, s.c.apiToken, "promote") if err != nil { return nil, err } return &Template{ Name: pr.Key.Template, ContentDigest: pr.ContentDigest, - c: s.c, addr: s.Owner(), net: pr.Key.Net, size: pr.Key.Size, + c: s.c, addr: addr, net: pr.Key.Net, size: pr.Key.Size, }, nil } @@ -160,7 +167,7 @@ func (s *Sandbox) Renew(ctx context.Context, ttl time.Duration) (time.Time, erro if err != nil { return time.Time{}, err } - rr, err := s.ownerJSON[renewResponse](ctx, http.MethodPost, "/renew", body, s.token, "renew") + rr, _, err := s.ownerJSON[renewResponse](ctx, http.MethodPost, "/renew", body, s.token, "renew") if err != nil { return time.Time{}, err } @@ -241,9 +248,9 @@ func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err e return false, apiError("release", resp) } -// rebind follows a moved sandbox: a 404 or a refused dial from its owner asks Lookup, and a different owner replaces it and its parked connections. +// rebind follows a moved sandbox: a tagged owner error asks Lookup, and a different owner replaces it and its parked connections. func (s *Sandbox) rebind(ctx context.Context, err error) bool { - if !ownerGone(err) { + if _, gone := errors.AsType[*ownerGoneError](err); !gone { return false } moved, lookupErr := s.c.Lookup(ctx, s.ID, s.token) @@ -269,18 +276,21 @@ func (s *Sandbox) retryMoved(ctx context.Context, rpc func() error) error { return err } -func (s *Sandbox) ownerJSON[T any](ctx context.Context, method, suffix string, body []byte, bearer, verb string) (T, error) { +// ownerJSON returns the reply and the address that gave it, which derived handles bind to. +func (s *Sandbox) ownerJSON[T any](ctx context.Context, method, suffix string, body []byte, bearer, verb string) (T, string, error) { var out T + var addr string err := s.retryMoved(ctx, func() (err error) { - out, err = s.c.doJSON[T](ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) - return err + addr = s.Owner() + out, err = s.c.doJSON[T](ctx, method, addr, "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) + return markGone(err) }) - return out, err + return out, addr, err } func (s *Sandbox) ownerNoContent(ctx context.Context, method, suffix string, body []byte, bearer, verb string) error { return s.retryMoved(ctx, func() error { - return s.c.doNoContent(ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) + return markGone(s.c.doNoContent(ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb)) }) } @@ -316,12 +326,13 @@ func (s *Sandbox) dial(ctx context.Context) (*agentConn, error) { return newAgentConn(raw), nil } -// ownerGone reports an error after which nothing reached the owner: a 404, or a refused dial. -func ownerGone(err error) bool { - if ae, ok := errors.AsType[*APIError](err); ok { - return ae.Status == http.StatusNotFound +// markGone tags an owner's 404 or refused dial, after which nothing reached it; errors from the caller's own writers stay untagged. +func markGone(err error) error { + ae, ok := errors.AsType[*APIError](err) + if ok && ae.Status == http.StatusNotFound || !ok && errors.Is(err, syscall.ECONNREFUSED) { + return &ownerGoneError{err} } - return errors.Is(err, syscall.ECONNREFUSED) + return err } // pumpStdin chunks the reader into stdin frames. diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 117a69e9..5c9c88d4 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -48,7 +48,7 @@ func (u *upgradedConn) handshake() error { return fmt.Errorf("read upgrade response: %w", err) } if resp.StatusCode != http.StatusSwitchingProtocols { - err = apiError("agent upgrade", resp) + err = markGone(apiError("agent upgrade", resp)) _ = resp.Body.Close() _ = u.tcp.Close() return err @@ -64,7 +64,7 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (*upgrad var d net.Dialer raw, err := d.DialContext(ctx, "tcp", target) if err != nil { - return nil, err + return nil, markGone(err) } stop := context.AfterFunc(ctx, func() { _ = raw.Close() }) defer stop() diff --git a/sdk/go/volumes.go b/sdk/go/volumes.go index f6bc7313..eb9c9709 100644 --- a/sdk/go/volumes.go +++ b/sdk/go/volumes.go @@ -80,7 +80,7 @@ func (s *Sandbox) requestVolumes(ctx context.Context, method string, request any var out SandboxVolumesResponse err := s.retryMoved(ctx, func() (err error) { out, err = s.volumesAt(ctx, method, body, verb) - return err + return markGone(err) }) return out, err } diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index b37cc17e..a6cd216c 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -12,7 +12,7 @@ from collections.abc import Callable, Iterator from typing import Any, BinaryIO, Protocol, TypeVar, cast -from .errors import APIError, ProtocolError, SandboxError, SandboxTimeout, SilkdError, _error_message +from .errors import APIError, ProtocolError, SandboxError, SandboxTimeout, SilkdError, _error_message, _mark_gone from .frames import MAX_FRAME, decode_response, encode_request KEEP_ALIVE_CONNS = 8 @@ -192,7 +192,8 @@ def dial_agent( try: sock = socket.create_connection((host, port), timeout=remaining_timeout(timeout, deadline)) except OSError as exc: - raise _handshake_error(f"dial {addr}", exc) from exc + err = _handshake_error(f"dial {addr}", exc) + raise (_mark_gone(err) if isinstance(exc, ConnectionRefusedError) else err) from exc sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) for opt, setting in _TCP_KEEPALIVE.items(): @@ -293,4 +294,5 @@ def _read_upgrade(reader: BinaryIO, arm: Callable[[], None]) -> None: if code != 101: arm() body = reader.read(min(body_len, MAX_FRAME)) if body_len else b"" - raise APIError("agent upgrade", code, _error_message(body) or status.strip()) + err = APIError("agent upgrade", code, _error_message(body) or status.strip()) + raise _mark_gone(err) if code == 404 else err diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index ef50fdfd..03a383b8 100644 --- a/sdk/python/cocoonsandbox/errors.py +++ b/sdk/python/cocoonsandbox/errors.py @@ -3,15 +3,19 @@ from __future__ import annotations import json -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, TypeVar, cast if TYPE_CHECKING: from .volumes import SandboxVolumesResponse +_E = TypeVar("_E", bound="SandboxError") + class SandboxError(Exception): """Base class for all SDK errors.""" + _owner_gone = False + class SandboxTimeout(SandboxError, TimeoutError): """A caller's timeout or deadline expired before the call finished.""" @@ -71,3 +75,9 @@ def _error_message(raw: bytes) -> str: return cast(str, json.loads(raw)["error"]) except (ValueError, KeyError, TypeError): return raw.decode(errors="replace").strip() + + +def _mark_gone(exc: _E) -> _E: + """Tags an owner's 404 or refused dial, after which nothing reached it; caller callback errors stay untagged.""" + exc._owner_gone = True + return exc diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 5f301fe7..39b016e5 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -21,6 +21,7 @@ SandboxTimeout, SilkdError, VolumeError, + _mark_gone, _require_field, ) from .frames import BULK_CHUNK, STREAM_CHUNK @@ -43,7 +44,7 @@ def _follows_move(method: _F) -> _F: def wrapper(self: Sandbox, *args: Any, **kwargs: Any) -> Any: try: return method(self, *args, **kwargs) - except (APIError, ProtocolError) as exc: + except SandboxError as exc: if not self._rebind(exc): raise return method(self, *args, **kwargs) @@ -355,8 +356,8 @@ def fork(self, count: int, ttl_seconds: int = 0) -> list[Sandbox]: body = {"token": self.token, "count": count} if ttl_seconds: body["ttl_seconds"] = ttl_seconds - reply = self._owner_request("fork", body, "fork") - return [self._client._handle_from(self.owner, child, "fork") for child in reply.get("children") or []] + reply, addr = self._owner_request("fork", body, "fork") + return [self._client._handle_from(addr, child, "fork") for child in reply.get("children") or []] def hibernate(self) -> None: """Snapshots and stops the VM; the next guest call restores its state.""" @@ -366,7 +367,7 @@ def hibernate(self) -> None: def renew(self, ttl_seconds: int = 0) -> str: """Resets the lease to ttl_seconds from now (0 means the node default) and records the granted deadline.""" body = {"ttl_seconds": ttl_seconds} if ttl_seconds else None - reply = self._owner_request("renew", body, "renew", bearer=self.token) + reply, _ = self._owner_request("renew", body, "renew", bearer=self.token) self.deadline = cast(str, _require_field(reply, "deadline", "renew")) return self.deadline @@ -375,17 +376,17 @@ def checkpoint(self, name: str = "") -> Checkpoint: body = {"token": self.token} if name: body["name"] = name - reply = self._owner_request("checkpoint", body, "checkpoint") - return Checkpoint(self._client, self.owner, _require_field(reply, "checkpoint", "checkpoint")) + reply, addr = self._owner_request("checkpoint", body, "checkpoint") + return Checkpoint(self._client, addr, _require_field(reply, "checkpoint", "checkpoint")) def promote(self, template: str) -> Template: """Publishes a claimable template and returns an owner-bound handle.""" body = {"token": self.token, "template": template} - reply = self._owner_request("promote", body, "promote") + reply, addr = self._owner_request("promote", body, "promote") key = _require_field(reply, "key", "promote") return Template( self._client, - self.owner, + addr, key["template"], key.get("net", ""), key.get("size", ""), @@ -420,7 +421,7 @@ def preview_url(self, port: int, ttl_seconds: int = 0) -> str: body = {"token": self.token, "port": port} if ttl_seconds: body["ttl_seconds"] = ttl_seconds - reply = self._owner_request("preview", body, "preview") + reply, _ = self._owner_request("preview", body, "preview") return cast(str, _require_field(reply, "url", "preview")) def dial_port(self, port: int) -> PortConn: @@ -470,11 +471,12 @@ def _volumes_request( ) except APIError as exc: if exc.status == 0: + _owner_answer(exc) raise view = _volume_response(exc.reply) if view is not None: self._record_volumes(view) - raise VolumeError(exc, view) from exc + raise _owner_answer(VolumeError(exc, view)) from exc view = _volume_response(reply) if view is None: raise APIError(verb, 0, "reply without complete volume view") @@ -486,7 +488,7 @@ def _record_volumes(self, view: SandboxVolumesResponse) -> None: self.volumes = [volume.copy() for volume in view["volumes"]] def _rebind(self, exc: SandboxError) -> bool: - if not (isinstance(exc, APIError) and exc.status == 404 or isinstance(exc.__cause__, ConnectionRefusedError)): + if not exc._owner_gone: return False try: moved = self._client.lookup(self.id, self.token) @@ -499,8 +501,17 @@ def _rebind(self, exc: SandboxError) -> bool: return True @_follows_move - def _owner_request(self, suffix: str, body: dict[str, Any] | None, verb: str, bearer: str = "") -> dict[str, Any]: - return self._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/{suffix}", body, verb, bearer=bearer) + def _owner_request( + self, suffix: str, body: dict[str, Any] | None, verb: str, bearer: str = "" + ) -> tuple[dict[str, Any], str]: + """Returns the reply and the address that gave it, which derived handles bind to.""" + addr = self.owner + try: + reply = self._client._request(addr, "POST", f"/v1/sandboxes/{self.id}/{suffix}", body, verb, bearer=bearer) + except APIError as exc: + _owner_answer(exc) + raise + return reply, addr def _dial(self, deadline: float | None = None) -> Conn: return self._client._dial(self.owner, self.id, self.token, deadline) @@ -821,3 +832,9 @@ def _drain_data(conn: Conn) -> bytes: if frame["type"] == "data": chunks.append(_need(frame, "data")) return b"".join(chunks) + + +def _owner_answer(exc: APIError) -> APIError: + if exc.status == 404 or isinstance(exc.__cause__, ConnectionRefusedError): + return _mark_gone(exc) + return exc diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 1db3c49e..8d45d9d9 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -9,7 +9,7 @@ import pytest from conftest import accept_upgrade, sandbox_at, wait_until -from cocoonsandbox import Client, SilkdError +from cocoonsandbox import APIError, Client, SilkdError INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") @@ -210,6 +210,25 @@ def test_a_handle_follows_its_sandbox_off_a_retired_node(spawn_node, dead_addr): agent.stop() +def test_a_run_whose_callback_fails_is_not_replayed_elsewhere(spawn_node): + owner, other = FakeAgent(), FakeAgent() + try: + directory = spawn_node( + {("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {"owner_addr": other.addr})} + ) + sb = Client(directory).attach(owner.addr, "sb_1", "tok") + + def refuse(_: bytes) -> None: + raise APIError("another sandbox", 404, "gone") + + with pytest.raises(APIError, match="another sandbox"): + sb.run(["echo", "42"], on_stdout=refuse) + assert other.upgrades == 0 + finally: + owner.stop() + other.stop() + + @pytest.fixture def relay(): started = [] From c6cd6a4244b116e4260ef880585eb889d6abf804 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 14:49:17 +0800 Subject: [PATCH 3/4] sdk: scope the rebind tag to the sandbox and owner address it came from Codex review round 2. An output callback or writer that itself called another sandbox could surface that call's tagged 404 or refused dial, and the running command would accept it and replay on its own new owner. The tag now names the sandbox id and the owner address it was raised for, and a handle rebinds only on its own tag at its current owner. --- sdk/go/client_test.go | 38 ++++++++++++++++++++++++++--- sdk/go/sandbox.go | 18 ++++++++------ sdk/go/upgrade.go | 8 +++--- sdk/go/volumes.go | 9 ++++--- sdk/python/cocoonsandbox/client.py | 4 ++- sdk/python/cocoonsandbox/conn.py | 23 +++++++++++------ sdk/python/cocoonsandbox/errors.py | 8 +++--- sdk/python/cocoonsandbox/sandbox.py | 15 ++++++------ sdk/python/tests/test_keepalive.py | 22 ++++++++++++++++- 9 files changed, 108 insertions(+), 37 deletions(-) diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 52f78265..710735f4 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -223,7 +223,7 @@ func TestARunWhoseWriterFailsIsNotReplayedElsewhere(t *testing.T) { t.Cleanup(entry.Close) sb := &Sandbox{ID: "sb_1", c: testClient(t, entry), token: "tok", owner: strings.TrimPrefix(owner.URL, "http://")} t.Cleanup(sb.pool.drain) - refused := failingWriter{syscall.ECONNREFUSED} + refused := writerFunc(func([]byte) (int, error) { return 0, syscall.ECONNREFUSED }) if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"echo", "42"}, Stdout: refused}); !errors.Is(err, syscall.ECONNREFUSED) { t.Fatalf("run: %v, want the writer's error", err) } @@ -232,6 +232,38 @@ func TestARunWhoseWriterFailsIsNotReplayedElsewhere(t *testing.T) { } } +func TestAnotherSandboxsErrorInAWriterIsNotReplayed(t *testing.T) { + fake := silkdtest.NewFake(t.TempDir()) + var elsewhere atomic.Int32 + other := newAgentServer(t, func(c net.Conn) { + elsewhere.Add(1) + fake.ServeConn(c) + }) + owner := newAgentServer(t, fake.ServeConn) + entry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/sandboxes/sb_1/owner" { + w.WriteHeader(http.StatusNotFound) + return + } + _ = json.MarshalWrite(w, map[string]string{"owner_addr": strings.TrimPrefix(other.URL, "http://")}) + })) + t.Cleanup(entry.Close) + c := testClient(t, entry) + sb := &Sandbox{ID: "sb_1", c: c, token: "tok", owner: strings.TrimPrefix(owner.URL, "http://")} + gone := &Sandbox{ID: "sb_2", c: c, token: "tok", owner: retiredAddr(t)} + t.Cleanup(sb.pool.drain) + writeElsewhere := writerFunc(func([]byte) (int, error) { + _, err := gone.Stat(t.Context(), "/") + return 0, err + }) + if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"echo", "42"}, Stdout: writeElsewhere}); err == nil { + t.Fatal("run succeeded although its writer failed") + } + if n := elsewhere.Load(); n != 0 { + t.Errorf("the command was sent again to another node %d times", n) + } +} + func TestCloseFollowsAMoveOffARetiredNode(t *testing.T) { var released atomic.Bool var adopter *httptest.Server @@ -725,6 +757,6 @@ func testClient(t testing.TB, ts *httptest.Server, opts ...ClientOption) *Client return c } -type failingWriter struct{ err error } +type writerFunc func([]byte) (int, error) -func (w failingWriter) Write([]byte) (int, error) { return 0, w.err } +func (f writerFunc) Write(p []byte) (int, error) { return f(p) } diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index e0608a6d..c6ce2a80 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -48,8 +48,11 @@ func (e *ExitError) Error() string { return fmt.Sprintf("exit status %d", e.Code) } -// ownerGoneError tags an owner answer after which nothing reached it, so the call may go to the sandbox's new owner. -type ownerGoneError struct{ err error } +// ownerGoneError tags an owner answer after which nothing reached it, naming the sandbox and the address it came from. +type ownerGoneError struct { + err error + id, addr string +} func (e *ownerGoneError) Error() string { return e.err.Error() } @@ -250,7 +253,7 @@ func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err e // rebind follows a moved sandbox: a tagged owner error asks Lookup, and a different owner replaces it and its parked connections. func (s *Sandbox) rebind(ctx context.Context, err error) bool { - if _, gone := errors.AsType[*ownerGoneError](err); !gone { + if g, gone := errors.AsType[*ownerGoneError](err); !gone || g.id != s.ID || g.addr != s.Owner() { return false } moved, lookupErr := s.c.Lookup(ctx, s.ID, s.token) @@ -283,14 +286,15 @@ func (s *Sandbox) ownerJSON[T any](ctx context.Context, method, suffix string, b err := s.retryMoved(ctx, func() (err error) { addr = s.Owner() out, err = s.c.doJSON[T](ctx, method, addr, "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) - return markGone(err) + return markGone(err, s.ID, addr) }) return out, addr, err } func (s *Sandbox) ownerNoContent(ctx context.Context, method, suffix string, body []byte, bearer, verb string) error { return s.retryMoved(ctx, func() error { - return markGone(s.c.doNoContent(ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb)) + addr := s.Owner() + return markGone(s.c.doNoContent(ctx, method, addr, "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb), s.ID, addr) }) } @@ -327,10 +331,10 @@ func (s *Sandbox) dial(ctx context.Context) (*agentConn, error) { } // markGone tags an owner's 404 or refused dial, after which nothing reached it; errors from the caller's own writers stay untagged. -func markGone(err error) error { +func markGone(err error, id, addr string) error { ae, ok := errors.AsType[*APIError](err) if ok && ae.Status == http.StatusNotFound || !ok && errors.Is(err, syscall.ECONNREFUSED) { - return &ownerGoneError{err} + return &ownerGoneError{err, id, addr} } return err } diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 5c9c88d4..5252f585 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -18,6 +18,8 @@ type upgradedConn struct { r *bufio.Reader req []byte // unsent upgrade request; writes arrive serialized by silkd.Conn await bool + + id, addr string } func (u *upgradedConn) Read(p []byte) (int, error) { @@ -48,7 +50,7 @@ func (u *upgradedConn) handshake() error { return fmt.Errorf("read upgrade response: %w", err) } if resp.StatusCode != http.StatusSwitchingProtocols { - err = markGone(apiError("agent upgrade", resp)) + err = markGone(apiError("agent upgrade", resp), u.id, u.addr) _ = resp.Body.Close() _ = u.tcp.Close() return err @@ -64,7 +66,7 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (*upgrad var d net.Dialer raw, err := d.DialContext(ctx, "tcp", target) if err != nil { - return nil, markGone(err) + return nil, markGone(err, id, addr) } stop := context.AfterFunc(ctx, func() { _ = raw.Close() }) defer stop() @@ -85,7 +87,7 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (*upgrad req := "GET /v1/sandboxes/" + id + "/agent HTTP/1.1\r\nHost: " + authority + "\r\nConnection: Upgrade\r\nUpgrade: silkd\r\nAuthorization: Bearer " + token + "\r\n\r\n" - u := &upgradedConn{Conn: conn, tcp: raw, r: bufio.NewReader(conn), req: []byte(req), await: true} + u := &upgradedConn{Conn: conn, tcp: raw, r: bufio.NewReader(conn), req: []byte(req), await: true, id: id, addr: addr} if scheme != httpsScheme { return u, nil } diff --git a/sdk/go/volumes.go b/sdk/go/volumes.go index eb9c9709..bd9f9070 100644 --- a/sdk/go/volumes.go +++ b/sdk/go/volumes.go @@ -79,15 +79,16 @@ func (s *Sandbox) requestVolumes(ctx context.Context, method string, request any } var out SandboxVolumesResponse err := s.retryMoved(ctx, func() (err error) { - out, err = s.volumesAt(ctx, method, body, verb) - return markGone(err) + addr := s.Owner() + out, err = s.volumesAt(ctx, method, addr, body, verb) + return markGone(err, s.ID, addr) }) return out, err } -func (s *Sandbox) volumesAt(ctx context.Context, method string, body []byte, verb string) (SandboxVolumesResponse, error) { +func (s *Sandbox) volumesAt(ctx context.Context, method, addr string, body []byte, verb string) (SandboxVolumesResponse, error) { var out SandboxVolumesResponse - resp, err := s.c.roundTrip(ctx, method, s.Owner(), "/v1/sandboxes/"+s.ID+"/volumes", bodyReader(body), s.c.apiToken) + resp, err := s.c.roundTrip(ctx, method, addr, "/v1/sandboxes/"+s.ID+"/volumes", bodyReader(body), s.c.apiToken) if err != nil { return out, err } diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 7fdf4f87..0dfebfb8 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -231,7 +231,9 @@ def _request( def _dial(self, addr: str, sandbox_id: str, token: str, deadline: float | None = None) -> Conn: origin = addr if "://" in addr else f"{self._scheme}://{addr}" context = self._tls() if origin.startswith("https://") else None - return dial_agent(origin, sandbox_id, token, self.timeout, deadline, ssl_context=context) + return dial_agent( + origin, sandbox_id, token, self.timeout, deadline, ssl_context=context, tag=(sandbox_id, addr) + ) def _roundtrip( self, url: urllib.parse.SplitResult, method: str, data: bytes | None, headers: dict[str, str], timeout: float diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index a6cd216c..e8d7a3ca 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -37,12 +37,18 @@ class Conn(_Closeable): """A frame stream to one sandbox's silkd; a plain-HTTP upgrade rides the first send, its 101 the first recv.""" def __init__( - self, sock: socket.socket, reader: BinaryIO, request: bytes = b"", upgrade_timeout: float | None = None + self, + sock: socket.socket, + reader: BinaryIO, + request: bytes = b"", + upgrade_timeout: float | None = None, + tag: tuple[str, str] = ("", ""), ) -> None: self._sock = sock self._reader = reader self._request = request self._upgrade_timeout = upgrade_timeout + self._tag = tag def send(self, op: str, **fields: object) -> None: frame = encode_request(op, **fields) @@ -107,7 +113,9 @@ def _finish_upgrade(self, timeout: float) -> None: deadline = time.monotonic() + limit try: _read_upgrade( - self._reader, lambda: self._sock.settimeout(remaining_timeout(limit, deadline, "agent upgrade")) + self._reader, + lambda: self._sock.settimeout(remaining_timeout(limit, deadline, "agent upgrade")), + self._tag, ) self._sock.settimeout(caller) except Exception as exc: @@ -179,6 +187,7 @@ def dial_agent( deadline: float | None = None, *, ssl_context: ssl.SSLContext | None = None, + tag: tuple[str, str] = ("", ""), ) -> Conn: """Opens one TCP/HTTP Upgrade relay within timeout and the optional deadline.""" budget = time.monotonic() + timeout @@ -193,7 +202,7 @@ def dial_agent( sock = socket.create_connection((host, port), timeout=remaining_timeout(timeout, deadline)) except OSError as exc: err = _handshake_error(f"dial {addr}", exc) - raise (_mark_gone(err) if isinstance(exc, ConnectionRefusedError) else err) from exc + raise (_mark_gone(err, tag) if isinstance(exc, ConnectionRefusedError) else err) from exc sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) for opt, setting in _TCP_KEEPALIVE.items(): @@ -211,7 +220,7 @@ def dial_agent( try: if endpoint.scheme != "https": sock.settimeout(None) - return Conn(sock, _buffered_reader(sock), request, timeout) + return Conn(sock, _buffered_reader(sock), request, timeout, tag) context = ssl_context or ssl.create_default_context() sock = context.wrap_socket(sock, server_hostname=host, do_handshake_on_connect=False) sock.settimeout(remaining_timeout(timeout, deadline)) @@ -223,7 +232,7 @@ def dial_agent( sock.settimeout(remaining_timeout(timeout, deadline)) sock.sendall(request) reader = _buffered_reader(sock) - _read_upgrade(reader, lambda: sock.settimeout(remaining_timeout(timeout, deadline))) + _read_upgrade(reader, lambda: sock.settimeout(remaining_timeout(timeout, deadline)), tag) remaining_timeout(timeout, deadline) sock.settimeout(None) return Conn(sock, reader) @@ -270,7 +279,7 @@ def _handshake_error(what: str, exc: OSError) -> SandboxError: return ProtocolError(f"{what}: {exc}") -def _read_upgrade(reader: BinaryIO, arm: Callable[[], None]) -> None: +def _read_upgrade(reader: BinaryIO, arm: Callable[[], None], tag: tuple[str, str]) -> None: arm() status = reader.readline(1024).decode(errors="replace") if not status: @@ -295,4 +304,4 @@ def _read_upgrade(reader: BinaryIO, arm: Callable[[], None]) -> None: arm() body = reader.read(min(body_len, MAX_FRAME)) if body_len else b"" err = APIError("agent upgrade", code, _error_message(body) or status.strip()) - raise _mark_gone(err) if code == 404 else err + raise _mark_gone(err, tag) if code == 404 else err diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index 03a383b8..0112610d 100644 --- a/sdk/python/cocoonsandbox/errors.py +++ b/sdk/python/cocoonsandbox/errors.py @@ -14,7 +14,7 @@ class SandboxError(Exception): """Base class for all SDK errors.""" - _owner_gone = False + _owner_gone: tuple[str, str] | None = None class SandboxTimeout(SandboxError, TimeoutError): @@ -77,7 +77,7 @@ def _error_message(raw: bytes) -> str: return raw.decode(errors="replace").strip() -def _mark_gone(exc: _E) -> _E: - """Tags an owner's 404 or refused dial, after which nothing reached it; caller callback errors stay untagged.""" - exc._owner_gone = True +def _mark_gone(exc: _E, tag: tuple[str, str]) -> _E: + """Tags an owner's 404 or refused dial with its (sandbox id, owner address); callback errors stay untagged.""" + exc._owner_gone = tag return exc diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 39b016e5..74a109d5 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -460,9 +460,10 @@ def _volumes_request( ) -> SandboxVolumesResponse: if timeout is not None and timeout <= 0: raise ValueError("timeout must be positive") + addr = self.owner try: reply = self._client._request( - self.owner, + addr, method, f"/v1/sandboxes/{self.id}/volumes", body, @@ -471,12 +472,12 @@ def _volumes_request( ) except APIError as exc: if exc.status == 0: - _owner_answer(exc) + _owner_answer(exc, (self.id, addr)) raise view = _volume_response(exc.reply) if view is not None: self._record_volumes(view) - raise _owner_answer(VolumeError(exc, view)) from exc + raise _owner_answer(VolumeError(exc, view), (self.id, addr)) from exc view = _volume_response(reply) if view is None: raise APIError(verb, 0, "reply without complete volume view") @@ -488,7 +489,7 @@ def _record_volumes(self, view: SandboxVolumesResponse) -> None: self.volumes = [volume.copy() for volume in view["volumes"]] def _rebind(self, exc: SandboxError) -> bool: - if not exc._owner_gone: + if exc._owner_gone != (self.id, self.owner): return False try: moved = self._client.lookup(self.id, self.token) @@ -509,7 +510,7 @@ def _owner_request( try: reply = self._client._request(addr, "POST", f"/v1/sandboxes/{self.id}/{suffix}", body, verb, bearer=bearer) except APIError as exc: - _owner_answer(exc) + _owner_answer(exc, (self.id, addr)) raise return reply, addr @@ -834,7 +835,7 @@ def _drain_data(conn: Conn) -> bytes: return b"".join(chunks) -def _owner_answer(exc: APIError) -> APIError: +def _owner_answer(exc: APIError, tag: tuple[str, str]) -> APIError: if exc.status == 404 or isinstance(exc.__cause__, ConnectionRefusedError): - return _mark_gone(exc) + return _mark_gone(exc, tag) return exc diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 8d45d9d9..6ab7df5a 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -9,7 +9,7 @@ import pytest from conftest import accept_upgrade, sandbox_at, wait_until -from cocoonsandbox import APIError, Client, SilkdError +from cocoonsandbox import APIError, Client, SandboxError, SilkdError INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") @@ -229,6 +229,26 @@ def refuse(_: bytes) -> None: other.stop() +def test_another_sandboxs_error_in_a_callback_is_not_replayed(spawn_node, dead_addr): + owner, other = FakeAgent(), FakeAgent() + try: + directory = spawn_node( + {("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {"owner_addr": other.addr})} + ) + sb = Client(directory).attach(owner.addr, "sb_1", "tok") + gone = Client(directory).attach(dead_addr, "sb_2", "tok") + + def write_elsewhere(_: bytes) -> None: + gone.write_file("/f", b"x") + + with pytest.raises(SandboxError): + sb.run(["echo", "42"], on_stdout=write_elsewhere) + assert other.upgrades == 0 + finally: + owner.stop() + other.stop() + + @pytest.fixture def relay(): started = [] From 6d53150abdaaa892d23c4fcc486a2b820b1a6d46 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 14:58:44 +0800 Subject: [PATCH 4/4] sdk: never retry an error that came out of the caller's own writer or callback, and let a call that lost a race to a move retry at the owner found Codex review round 3. A Python output callback that used the same handle's find_iter could surface a 404 tagged for this very sandbox and owner, and the running command replayed on the new owner. Errors from the caller's code are now cut off where it is called: Go wraps a Run writer's or a ReadFileTo/Pull sink's error in callerError, Python clears the tag of an SDK error an output callback raises, and rebind ignores both. A tagged call whose old owner a concurrent call has already replaced retries at the new owner without a second Lookup. --- docs/sdk-python.md | 5 +++-- docs/sdk.md | 3 ++- sdk/go/client_test.go | 20 +++++++++++++++++--- sdk/go/sandbox.go | 16 +++++++++++++++- sdk/go/utils.go | 6 +++--- sdk/python/cocoonsandbox/sandbox.py | 18 +++++++++++++++--- sdk/python/tests/test_keepalive.py | 7 ++++--- 7 files changed, 59 insertions(+), 16 deletions(-) diff --git a/docs/sdk-python.md b/docs/sdk-python.md index ee631ea8..f71b6dd4 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -146,8 +146,9 @@ gone is not an error — double-release and reap races stay silent). Every other call follows a move too: a 404 or a refused connection from the owner runs `client.lookup` once, and the call is sent again to a different owner, since every argument (stdin, file data, tar streams) is bytes; only `find_iter` -fails once before the next call reaches the new owner. A released sandbox -still raises a 404 `APIError`. +fails once before the next call reaches the new owner. An exception from the +caller's own `on_stdout`/`on_stderr` is raised as it is and never retried, and +a released sandbox still raises a 404 `APIError`. Volume sandboxes cannot hibernate, fork, checkpoint, or promote. Egress-lane sandboxes cannot either, and a sandbox whose tun is turned off cannot promote; each answers 409. Checkpoint diff --git a/docs/sdk.md b/docs/sdk.md index 5738c434..f89020ed 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -218,7 +218,8 @@ runs `Lookup` once, and a different owner replaces the old one. A call whose request is its whole input is sent again there: the control calls, `Run` without `Stdin`, and the file, git, process and session calls that send no data. `Run` with `Stdin`, `WriteFile`, `Push`, `OpenPty`, `Watch`, `DialPort`, `Lsp.Request`, -`Logs` and `Attach` fail once and the next call reaches the new owner. A +`Logs` and `Attach` fail once and the next call reaches the new owner. An +error from the caller's own writer is returned as it is and never retried, and a released sandbox still answers 404. Volume sandboxes and egress-lane sandboxes cannot hibernate, fork, checkpoint, or promote, nor take `WithArchiveOnExpire`; each answers 409, as does a promote diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 710735f4..6e14f191 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -209,7 +209,7 @@ func TestAHandleFollowsItsSandboxOffARetiredNode(t *testing.T) { } } -func TestARunWhoseWriterFailsIsNotReplayedElsewhere(t *testing.T) { +func TestAWriterErrorTaggedForThisSandboxIsNotReplayed(t *testing.T) { fake := silkdtest.NewFake(t.TempDir()) var elsewhere atomic.Int32 other := newAgentServer(t, func(c net.Conn) { @@ -223,8 +223,10 @@ func TestARunWhoseWriterFailsIsNotReplayedElsewhere(t *testing.T) { t.Cleanup(entry.Close) sb := &Sandbox{ID: "sb_1", c: testClient(t, entry), token: "tok", owner: strings.TrimPrefix(owner.URL, "http://")} t.Cleanup(sb.pool.drain) - refused := writerFunc(func([]byte) (int, error) { return 0, syscall.ECONNREFUSED }) - if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"echo", "42"}, Stdout: refused}); !errors.Is(err, syscall.ECONNREFUSED) { + tagged := writerFunc(func([]byte) (int, error) { + return 0, &ownerGoneError{err: syscall.ECONNREFUSED, id: sb.ID, addr: sb.Owner()} + }) + if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"echo", "42"}, Stdout: tagged}); !errors.Is(err, syscall.ECONNREFUSED) { t.Fatalf("run: %v, want the writer's error", err) } if n := elsewhere.Load(); n != 0 { @@ -264,6 +266,18 @@ func TestAnotherSandboxsErrorInAWriterIsNotReplayed(t *testing.T) { } } +func TestACallThatLostARaceToAMoveRetriesAtTheOwnerFound(t *testing.T) { + entry := httptest.NewServer(http.NotFoundHandler()) + t.Cleanup(entry.Close) + sb := &Sandbox{ID: "sb_1", c: testClient(t, entry), token: "tok", owner: "new:7777"} + if !sb.rebind(t.Context(), &ownerGoneError{err: errors.New("gone"), id: "sb_1", addr: "old:7777"}) { + t.Error("a call tagged by the old owner did not retry at the owner another call found") + } + if sb.rebind(t.Context(), &ownerGoneError{err: errors.New("gone"), id: "sb_2", addr: "old:7777"}) { + t.Error("another sandbox's tag rebound this handle") + } +} + func TestCloseFollowsAMoveOffARetiredNode(t *testing.T) { var released atomic.Bool var adopter *httptest.Server diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index c6ce2a80..48bae30d 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -48,6 +48,13 @@ func (e *ExitError) Error() string { return fmt.Sprintf("exit status %d", e.Code) } +// callerError carries an error from the caller's own writer, which never sends a call to another node. +type callerError struct{ err error } + +func (e *callerError) Error() string { return e.err.Error() } + +func (e *callerError) Unwrap() error { return e.err } + // ownerGoneError tags an owner answer after which nothing reached it, naming the sandbox and the address it came from. type ownerGoneError struct { err error @@ -253,9 +260,16 @@ func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err e // rebind follows a moved sandbox: a tagged owner error asks Lookup, and a different owner replaces it and its parked connections. func (s *Sandbox) rebind(ctx context.Context, err error) bool { - if g, gone := errors.AsType[*ownerGoneError](err); !gone || g.id != s.ID || g.addr != s.Owner() { + if _, caller := errors.AsType[*callerError](err); caller { return false } + g, gone := errors.AsType[*ownerGoneError](err) + if !gone || g.id != s.ID { + return false + } + if g.addr != s.Owner() { + return true + } moved, lookupErr := s.c.Lookup(ctx, s.ID, s.token) if lookupErr != nil { return false diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 9359f5b9..eb7f053b 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -158,11 +158,11 @@ func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) pid = resp.PID case *wire.Stdout: if _, err := stdout.Write(resp.Data); err != nil { - return pid, 0, false, err + return pid, 0, false, &callerError{err} } case *wire.Stderr: if _, err := stderr.Write(resp.Data); err != nil { - return pid, 0, false, err + return pid, 0, false, &callerError{err} } case *wire.Exit: return pid, resp.Code, true, nil @@ -186,7 +186,7 @@ func drainData(ctx context.Context, conn *silkd.Conn, sink chunkSink) error { switch r := resp.(type) { case *wire.DataResp: if err := sink(r.Data); err != nil { - return err + return &callerError{err} } case *wire.Done: return nil diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 74a109d5..fed19f15 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -489,8 +489,11 @@ def _record_volumes(self, view: SandboxVolumesResponse) -> None: self.volumes = [volume.copy() for volume in view["volumes"]] def _rebind(self, exc: SandboxError) -> bool: - if exc._owner_gone != (self.id, self.owner): + tag = exc._owner_gone + if tag is None or tag[0] != self.id: return False + if tag[1] != self.owner: + return True try: moved = self._client.lookup(self.id, self.token) except APIError: @@ -798,14 +801,23 @@ def _pump_stdio( if t == "started" and on_started: on_started(cast(int, _need(frame, "pid"))) elif t == "stdout" and on_stdout: - on_stdout(_need(frame, "data")) + _deliver(on_stdout, _need(frame, "data")) elif t == "stderr" and on_stderr: - on_stderr(_need(frame, "data")) + _deliver(on_stderr, _need(frame, "data")) elif t == "exit": return cast(int, _need(frame, "code")) return None +def _deliver(callback: Callable[[bytes], object], data: bytes) -> None: + """Runs a caller's output callback; an SDK error it raises never sends the running call to another node.""" + try: + callback(data) + except SandboxError as exc: + exc._owner_gone = None + raise + + def _expect(conn: Conn, frame_type: str) -> dict[str, Any]: frame = conn.recv() if frame["type"] != frame_type: diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 6ab7df5a..393b3866 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -10,6 +10,7 @@ from conftest import accept_upgrade, sandbox_at, wait_until from cocoonsandbox import APIError, Client, SandboxError, SilkdError +from cocoonsandbox.errors import _mark_gone INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") @@ -210,7 +211,7 @@ def test_a_handle_follows_its_sandbox_off_a_retired_node(spawn_node, dead_addr): agent.stop() -def test_a_run_whose_callback_fails_is_not_replayed_elsewhere(spawn_node): +def test_a_callback_error_tagged_for_this_sandbox_is_not_replayed(spawn_node): owner, other = FakeAgent(), FakeAgent() try: directory = spawn_node( @@ -219,9 +220,9 @@ def test_a_run_whose_callback_fails_is_not_replayed_elsewhere(spawn_node): sb = Client(directory).attach(owner.addr, "sb_1", "tok") def refuse(_: bytes) -> None: - raise APIError("another sandbox", 404, "gone") + raise _mark_gone(APIError("agent upgrade", 404, "gone"), (sb.id, sb.owner)) - with pytest.raises(APIError, match="another sandbox"): + with pytest.raises(APIError, match="agent upgrade"): sb.run(["echo", "42"], on_stdout=refuse) assert other.upgrades == 0 finally: