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..f71b6dd4 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -142,9 +142,13 @@ 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. 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 c596f3f0..f89020ed 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -212,9 +212,15 @@ 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. 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 of a sandbox whose tun is turned off. Passing diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index e49669a7..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.c.doJSON[checkpointResponse](ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/checkpoint", bytes.NewReader(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 9e07cde4..6e14f191 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -10,8 +10,11 @@ import ( "reflect" "strings" "sync/atomic" + "syscall" "testing" "time" + + "github.com/cocoonstack/sandbox/sdk/go/silkd/silkdtest" ) func TestConnectAddress(t *testing.T) { @@ -160,6 +163,121 @@ 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 TestAWriterErrorTaggedForThisSandboxIsNotReplayed(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) + 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 { + t.Errorf("the command was sent again to another node %d times", n) + } +} + +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 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 @@ -600,6 +718,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") @@ -619,3 +770,7 @@ func testClient(t testing.TB, ts *httptest.Server, opts ...ClientOption) *Client } return c } + +type writerFunc func([]byte) (int, error) + +func (f writerFunc) Write(p []byte) (int, error) { return f(p) } diff --git a/sdk/go/env.go b/sdk/go/env.go index 9e76dc57..899d201a 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..35f3cb0e 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..48bae30d 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" @@ -47,6 +48,23 @@ 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 + id, addr string +} + +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 { @@ -67,13 +85,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 +125,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 +138,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, 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 } @@ -165,20 +155,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, 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 } // 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 +177,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 +191,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 +203,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 +258,60 @@ func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err e return false, apiError("release", resp) } +// 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 _, 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 + } + 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 +} + +// 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) { + addr = s.Owner() + out, err = s.c.doJSON[T](ctx, method, addr, "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb) + 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 { + addr := s.Owner() + return markGone(s.c.doNoContent(ctx, method, addr, "/v1/sandboxes/"+s.ID+suffix, bodyReader(body), bearer, verb), s.ID, addr) + }) +} + // 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 +337,22 @@ 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 } +// 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, 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, id, addr} + } + return err +} + // pumpStdin chunks the reader into stdin frames. func pumpStdin(conn *silkd.Conn, r io.Reader) { buf := make([]byte, stdinChunk) diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 117a69e9..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 = 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, 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/utils.go b/sdk/go/utils.go index 7d414b98..eb7f053b 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) } } } @@ -145,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 @@ -173,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 @@ -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..bd9f9070 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,26 @@ 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) { + 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, addr string, body []byte, verb string) (SandboxVolumesResponse, error) { + var out SandboxVolumesResponse + 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/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..0dfebfb8 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) @@ -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 b37cc17e..e8d7a3ca 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 @@ -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 @@ -192,7 +201,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, 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(): @@ -210,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)) @@ -222,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) @@ -269,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: @@ -293,4 +303,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, tag) if code == 404 else err diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index ef50fdfd..0112610d 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: tuple[str, str] | None = None + 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, 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 cc34f614..fed19f15 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 @@ -20,6 +21,7 @@ SandboxTimeout, SilkdError, VolumeError, + _mark_gone, _require_field, ) from .frames import BULK_CHUNK, STREAM_CHUNK @@ -32,6 +34,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 SandboxError as exc: + if not self._rebind(exc): + raise + return method(self, *args, **kwargs) + + return cast(_F, wrapper) + class Sandbox: """One claimed microVM.""" @@ -124,6 +143,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 +243,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 +275,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 +298,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 +356,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") - 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.""" 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,17 +376,17 @@ 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") - 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._client._request(self.owner, "POST", f"/v1/sandboxes/{self.id}/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", ""), @@ -398,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._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,14 +454,16 @@ 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: 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, @@ -447,11 +472,12 @@ def _volumes_request( ) except APIError as exc: if exc.status == 0: + _owner_answer(exc, (self.id, addr)) 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), (self.id, addr)) from exc view = _volume_response(reply) if view is None: raise APIError(verb, 0, "reply without complete volume view") @@ -462,6 +488,35 @@ 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: + 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: + 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 = "" + ) -> 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, (self.id, addr)) + raise + return reply, addr + def _dial(self, deadline: float | None = None) -> Conn: return self._client._dial(self.owner, self.id, self.token, deadline) @@ -492,6 +547,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 +605,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 +613,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, @@ -743,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: @@ -778,3 +845,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, tag: tuple[str, str]) -> APIError: + if exc.status == 404 or isinstance(exc.__cause__, ConnectionRefusedError): + return _mark_gone(exc, tag) + return exc 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..393b3866 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -9,7 +9,8 @@ import pytest from conftest import accept_upgrade, sandbox_at, wait_until -from cocoonsandbox import SilkdError +from cocoonsandbox import APIError, Client, SandboxError, SilkdError +from cocoonsandbox.errors import _mark_gone INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") @@ -181,6 +182,74 @@ 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() + + +def test_a_callback_error_tagged_for_this_sandbox_is_not_replayed(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 _mark_gone(APIError("agent upgrade", 404, "gone"), (sb.id, sb.owner)) + + with pytest.raises(APIError, match="agent upgrade"): + sb.run(["echo", "42"], on_stdout=refuse) + assert other.upgrades == 0 + finally: + owner.stop() + 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 = []