sdk: follow a sandbox that moved to another node

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`,

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

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

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.

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) }

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")
 }

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

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

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