From 99d1707b142dce3f5cef1823b3af387143c2379e Mon Sep 17 00:00:00 2001 From: doge Date: Wed, 29 Jul 2026 00:29:54 +0800 Subject: [PATCH 1/3] Spread egress-lane VMs over several CNI networks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Linux bridge holds at most BR_MAX_PORTS ports — 1024, a compile-time kernel constant with no sysctl — and answers EXFULL for the next one. One bridge per node therefore caps egress density near a thousand VMs however much CPU and memory the node has, and a fleet run reached exactly that: nineteen of twenty nodes wedged between 1004 and 1014 ports. Measured on one node afterwards, the cost is not only the ceiling. Filling a single bridge to 1000 took 86s and the per-second rate collapsed from 105 to single digits as it filled, because br_add_if walks the bridge's port list while holding the global rtnl lock. The same node filled 1000 none-lane VMs, which take no bridge port at all, in 5s at a flat 200/s. Accept a list of conflists and hash the VM name to pick one, so N bridges give N×1024 and each stays in the range where that walk is short. Hashing rather than a counter keeps the choice free of process state: a VM resolves to the same conflist whoever asks and whenever, which matters because its record persists the network it was built on and restore has to agree. The attachment key is now `networks`, a list; a one-entry list is byte-for-byte the old argv. The scalar `network` key is retired — strict config decoding fails the load loudly instead of silently dropping the egress lane. --- docs/cluster.md | 2 +- docs/deploy.md | 2 +- docs/egress.md | 2 +- e2e/fakeengine_test.go | 2 +- sandboxd/config/config.go | 39 ++++++++++++++++----- sandboxd/config/config_test.go | 42 +++++++++++++++++----- sandboxd/engine/cloneargs_test.go | 58 +++++++++++++++++++++++++++++-- sandboxd/engine/engine.go | 32 +++++++++++------ sandboxd/engine/engine_test.go | 16 ++++----- sandboxd/engine/installca_test.go | 6 ++-- sandboxd/main.go | 2 +- 11 files changed, 159 insertions(+), 44 deletions(-) diff --git a/docs/cluster.md b/docs/cluster.md index dee252f..eab2dfa 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -179,7 +179,7 @@ fully from it: | state | source of truth | survives restart | |---|---|---| -| operator config (`tenants`, `secrets`, egress policies, `bridge`/`network`, `mesh`, `preview_secret`, `egress_ca`) | `config.json` (human/deploy-tool owned) | re-read at boot | +| operator config (`tenants`, `secrets`, egress policies, `bridge`/`networks`, `mesh`, `preview_secret`, `egress_ca`) | `config.json` (human/deploy-tool owned) | re-read at boot | | API-applied pool targets (`PUT /v1/pools`) | `/pools.json` (machine owned) | yes | | claims | the claims journal + `Reconcile` | yes | | placement hints (warm counts, template sets) | gossip | rebuilt | diff --git a/docs/deploy.md b/docs/deploy.md index f284e1d..7e3b0b7 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -68,7 +68,7 @@ sandboxd reads one JSON file (`-config`, default | `restore_mode` | unset | clone and wake-restore memory mode: `copy`, `ondemand`, or `mmap`; use `mmap` for dense pools | | `no_direct_io` | false | use buffered writable disks for Cloud Hypervisor cold boots and clones; recommended for dense ephemeral pools to avoid direct-I/O CoW journal contention | | `advertise_addr` | = `listen` | the host:port clients reach this node at; returned as a claim's owner address and gossiped to peers. Must be routable when `listen` is a wildcard | -| `bridge` / `network` | unset | egress-lane attachment: a host bridge device, or a CNI conflist name. Mutually exclusive; with neither set the node serves only the no-network lane. [Guarded egress](egress.md) needs the bridge form and rejects a CNI network at load | +| `bridge` / `networks` | unset | egress-lane attachment: a host bridge device, or a list of CNI conflist names. Mutually exclusive; with neither set the node serves only the no-network lane. Each conflist attaches its own Linux bridge and a bridge holds at most 1024 ports (kernel `BR_MAX_PORTS`), so N conflists raise the node's egress ceiling to N×1024 — VMs spread over them by a stable hash of the VM name, so size the list with headroom (the spread is statistical, not exact). [Guarded egress](egress.md) needs the bridge form and rejects a CNI network at load | | `egress_ca` | unset | [HTTPS-interception](egress.md#https-interception) PKI: `root_cert` (the cluster root baked into intercepted guests; may bundle old+new roots during rotation) plus this node's `intermediate_cert`/`intermediate_key` from `sandboxd ca issue-intermediate`. Required when any pool rule sets `intercept` | | `api_token` | unset | the operator (root) credential: when set, guards the node-level endpoints (Bearer) with full access, including release-by-id cleanup. Per-sandbox tokens guard ordinary sandbox-scoped calls | | `tenants` | unset | multi-tenant tokens next to `api_token`: `[{"name": "acme", "token": "…", "max_claims": 50}]`. A tenant token reaches the resource-creating verbs (claim, fork, promote, checkpoint, preview) and everything it creates is stamped with the tenant name; operator surfaces (`GET /v1/sandboxes` and the per-id reads under it, `GET /v1/info`, `PUT /v1/pools`, `POST/DELETE /v1/drain`, `/metrics`) answer it 403. `max_claims` (0 = unlimited) caps that tenant's live claims next to the node-wide cap. Requires `api_token` set (operator surfaces need it). Names and tokens must be unique, tokens distinct from `api_token`. On a cluster all nodes must carry the same tenants set (the SDK replays a tenant token across a redirect; a peer missing that tenant answers 401), and per-node caps mean a tenant's effective cluster limit is `max_claims` × nodes. Empty = exactly the single-token behavior | diff --git a/docs/egress.md b/docs/egress.md index 76abcdf..9aca988 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -84,7 +84,7 @@ domain policy first; the allow-list widens the IP gate only. bridge they do not share with an untrusted listener. - **Bridge lane only (egress lane).** A CNI network's tap lives in the VM netns, out of reach of the root-netns lock, so a guarded egress *lane* needs a bridge - and is rejected on a CNI `network`. None-lane policies ride the proxy and work + and is rejected on CNI `networks`. None-lane policies ride the proxy and work on either. A bridge egress lane locks every NIC default-deny, even with no policy configured. - **No custom NAT64/DNS64 prefix routed to the host.** The SSRF guard folds the diff --git a/e2e/fakeengine_test.go b/e2e/fakeengine_test.go index 989377f..1103ab7 100644 --- a/e2e/fakeengine_test.go +++ b/e2e/fakeengine_test.go @@ -31,7 +31,7 @@ type fakeEngine struct { func newFakeEngine(dir string) *fakeEngine { return &fakeEngine{ - real: engine.New("cocoon", "", "", false, ""), + real: engine.New("cocoon", "", nil, false, ""), dir: dir, listeners: map[string]io.Closer{}, socks: map[string]string{}, diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 7253169..522f63a 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -170,11 +170,15 @@ type Config struct { // to Listen, which is correct when Listen is a routable host:port. AdvertiseAddr string `json:"advertise_addr,omitempty"` - // Bridge and Network pick the egress-lane attachment (TAP-on-bridge vs - // CNI conflist); mutually exclusive. With neither set the node serves + // Bridge and Networks pick the egress-lane attachment (TAP-on-bridge vs + // CNI conflists); mutually exclusive. With neither set the node serves // only the no-network lane. - Bridge string `json:"bridge,omitempty"` - Network string `json:"network,omitempty"` + Bridge string `json:"bridge,omitempty"` + + // Networks shards egress-lane VMs over several CNI conflists, one Linux + // bridge each: a bridge holds at most 1024 ports (BR_MAX_PORTS, no + // sysctl), so N conflists raise the node's egress ceiling to N×1024. + Networks []string `json:"networks,omitempty"` // RestoreMode rides clones and wake restores as cocoon's --restore-mode. Opt-in: mmap // needs every node's cloud-hypervisor to carry CoW restore — an older CH @@ -272,7 +276,7 @@ type Config struct { // HasEgress reports whether the node can attach egress-lane VMs. func (c *Config) HasEgress() bool { - return c.Bridge != "" || c.Network != "" + return c.Bridge != "" || len(c.Networks) > 0 } // ClusterDigest fingerprints the must-match config so a divergent node is @@ -334,12 +338,12 @@ func autoRefillConcurrency(cpus int) int { } func (c *Config) validate() error { - if c.Bridge != "" && c.Network != "" { - return fmt.Errorf("bridge and network are mutually exclusive") + if err := c.validateNetworks(); err != nil { + return err } // A CNI network's tap lives in the VM netns, unreachable from the root-netns // nft lock; guarded egress needs a bridge. - if c.Network != "" && c.guardsEgressLane() { + if len(c.Networks) > 0 && c.guardsEgressLane() { return fmt.Errorf("guarded egress needs a bridge lane, not a CNI network: the tap lives in the VM netns and cannot be locked") } if c.MaxForkCount < 1 { @@ -457,6 +461,25 @@ func (c *Config) validateSecrets() (map[string]struct{}, error) { return names, nil } +// validateNetworks checks the egress-lane attachment; a repeated conflist +// would report N shards while filling one bridge. +func (c *Config) validateNetworks() error { + if c.Bridge != "" && len(c.Networks) > 0 { + return fmt.Errorf("bridge and networks are mutually exclusive") + } + seen := make(map[string]struct{}, len(c.Networks)) + for _, n := range c.Networks { + if n == "" { + return fmt.Errorf("networks must not contain an empty conflist name") + } + if _, ok := seen[n]; ok { + return fmt.Errorf("duplicate conflist %q", n) + } + seen[n] = struct{}{} + } + return nil +} + // validateEgressAllow rejects a malformed prefix at load: a typo would // otherwise surface as a refused dial long after startup. func (c *Config) validateEgressAllow() error { diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go index 1b8fb27..e757e83 100644 --- a/sandboxd/config/config_test.go +++ b/sandboxd/config/config_test.go @@ -85,7 +85,7 @@ func TestLoadRejectsInvalid(t *testing.T) { name, body, want string }{ {"bad json", `{`, "config"}, - {"bridge and network", `{"bridge":"br0","network":"cni","pools":[]}`, "mutually exclusive"}, + {"bridge and networks", `{"bridge":"br0","networks":["cni"],"pools":[]}`, "mutually exclusive"}, {"bad fork count", `{"max_fork_count":-1,"pools":[]}`, "max_fork_count"}, {"negative refill concurrency", `{"refill_concurrency":-1,"pools":[]}`, "refill_concurrency"}, {"bad restore mode", `{"restore_mode":"Mmap","pools":[]}`, "restore_mode"}, @@ -114,9 +114,9 @@ func TestLoadRejectsInvalid(t *testing.T) { {"pool egress empty host", `{"pools":[{"template":"rt:24.04","net":"none","size":"small","egress":{"allow":[{"host":""}]}}]}`, "must not be empty"}, {"pool egress unknown secret", `{"pools":[{"template":"rt:24.04","net":"none","size":"small","egress":{"allow":[{"host":"api.github.com","secret":"gh"}]}}]}`, "unknown secret"}, {"tenant egress unknown secret", `{"api_token":"root","pools":[],"tenants":[{"name":"acme","token":"t1","egress":{"allow":[{"host":"x","secret":"gh"}]}}]}`, "unknown secret"}, - {"guarded egress on cni pool", `{"network":"cni","pools":[{"template":"rt:24.04","net":"egress","size":"small","egress":{"allow":[{"host":"x"}]}}]}`, "needs a bridge lane"}, - {"guarded egress on cni tenant", `{"api_token":"root","network":"cni","pools":[{"template":"rt:24.04","net":"egress","size":"small"}],"tenants":[{"name":"acme","token":"t1","egress":{"allow":[{"host":"x"}]}}]}`, "needs a bridge lane"}, - {"cni tenant egress no egress pool", `{"api_token":"root","network":"cni","pools":[{"template":"rt:24.04","net":"none","size":"small"}],"tenants":[{"name":"acme","token":"t1","egress":{"allow":[{"host":"x"}]}}]}`, "needs a bridge lane"}, + {"guarded egress on cni pool", `{"networks":["cni"],"pools":[{"template":"rt:24.04","net":"egress","size":"small","egress":{"allow":[{"host":"x"}]}}]}`, "needs a bridge lane"}, + {"guarded egress on cni tenant", `{"api_token":"root","networks":["cni"],"pools":[{"template":"rt:24.04","net":"egress","size":"small"}],"tenants":[{"name":"acme","token":"t1","egress":{"allow":[{"host":"x"}]}}]}`, "needs a bridge lane"}, + {"cni tenant egress no egress pool", `{"api_token":"root","networks":["cni"],"pools":[{"template":"rt:24.04","net":"none","size":"small"}],"tenants":[{"name":"acme","token":"t1","egress":{"allow":[{"host":"x"}]}}]}`, "needs a bridge lane"}, {"mesh bind missing port", `{"pools":[],"mesh":{"bind":"node1"}}`, "mesh bind"}, {"mesh bind wildcard host", `{"pools":[],"mesh":{"bind":":7946"}}`, "explicit host"}, {"mesh cluster key not base64", `{"pools":[],"mesh":{"bind":"node1:7946","cluster_key":"not!base64"}}`, "not valid base64"}, @@ -169,7 +169,7 @@ func TestLoadAcceptsEgressPolicy(t *testing.T) { func TestLoadAcceptsUnguardedCNINetwork(t *testing.T) { // Only guarded egress needs a bridge; an unguarded CNI network lane is fine. - path := writeConfig(t, `{"network":"cni","pools":[{"template":"rt:24.04","net":"egress","size":"small"}]}`) + path := writeConfig(t, `{"networks":["cni"],"pools":[{"template":"rt:24.04","net":"egress","size":"small"}]}`) if _, err := Load(path); err != nil { t.Fatalf("Load: %v", err) } @@ -179,8 +179,8 @@ func TestLoadAcceptsNoneLanePolicyOnCNI(t *testing.T) { // A none-lane policy rides the vsock proxy and locks no tap, so it is valid on // a CNI network — the bridge requirement is only for a guarded egress lane. for _, body := range []string{ - `{"network":"cni","pools":[{"template":"rt:24.04","net":"none","size":"small","egress":{"allow":[{"host":"x"}]}}]}`, - `{"network":"cni","pools":[{"template":"rt:24.04","net":"egress","size":"small"},{"template":"rt:24.04","net":"none","size":"medium","egress":{"allow":[{"host":"x"}]}}]}`, + `{"networks":["cni"],"pools":[{"template":"rt:24.04","net":"none","size":"small","egress":{"allow":[{"host":"x"}]}}]}`, + `{"networks":["cni"],"pools":[{"template":"rt:24.04","net":"egress","size":"small"},{"template":"rt:24.04","net":"none","size":"medium","egress":{"allow":[{"host":"x"}]}}]}`, } { if _, err := Load(writeConfig(t, body)); err != nil { t.Errorf("Load rejected a none-lane policy on CNI: %v", err) @@ -218,7 +218,7 @@ func TestHasEgress(t *testing.T) { if (&Config{}).HasEgress() { t.Error("no attachment must mean no egress") } - if !(&Config{Bridge: "br0"}).HasEgress() || !(&Config{Network: "cni"}).HasEgress() { + if !(&Config{Bridge: "br0"}).HasEgress() || !(&Config{Networks: []string{"cni"}}).HasEgress() { t.Error("bridge or network must enable egress") } } @@ -238,6 +238,32 @@ func TestLoadKeepsExplicitValues(t *testing.T) { } } +func TestLoadRejectsUnusableNetworkLists(t *testing.T) { + for name, body := range map[string]string{ + "bridge and networks together": `{"bridge":"br0","networks":["a"],"pools":[]}`, + "retired scalar network key": `{"network":"a","pools":[]}`, + "empty conflist name": `{"networks":["a",""],"pools":[]}`, + "repeated conflist": `{"networks":["a","b","a"],"pools":[]}`, + } { + t.Run(name, func(t *testing.T) { + if _, err := Load(writeConfig(t, body)); err == nil { + t.Error("Load accepted an unusable network list; want rejection") + } + }) + } +} + +func TestLoadAcceptsAShardedNetworkList(t *testing.T) { + path := writeConfig(t, `{"networks":["cocoon-sbx0","cocoon-sbx1","cocoon-sbx2","cocoon-sbx3"],"pools":[]}`) + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(cfg.Networks) != 4 || !cfg.HasEgress() { + t.Errorf("Networks = %v, HasEgress = %v", cfg.Networks, cfg.HasEgress()) + } +} + func writeConfig(t *testing.T, body string) string { t.Helper() path := filepath.Join(t.TempDir(), "config.json") diff --git a/sandboxd/engine/cloneargs_test.go b/sandboxd/engine/cloneargs_test.go index 0adc762..f828a2d 100644 --- a/sandboxd/engine/cloneargs_test.go +++ b/sandboxd/engine/cloneargs_test.go @@ -23,7 +23,7 @@ func TestCloneArgsRestoreMode(t *testing.T) { } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - e := New("cocoon", "br0", "", false, tc.mode) + e := New("cocoon", "br0", nil, false, tc.mode) for _, args := range [][]string{ e.cloneArgs("/goldens/g1", "sbx-1", tc.key), e.cloneSnapArgs("ck_1", "sbx-1", tc.key), @@ -45,7 +45,7 @@ func TestLifecycleArgsApplyDirectIOPolicy(t *testing.T) { key := types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeMedium} for _, noDirectIO := range []bool{false, true} { t.Run(strconv.FormatBool(noDirectIO), func(t *testing.T) { - e := New("cocoon", "", "", noDirectIO, "") + e := New("cocoon", "", nil, noDirectIO, "") want := "--no-direct-io=" + strconv.FormatBool(noDirectIO) cold := e.runColdArgs("sbx-1", key) for _, args := range [][]string{ @@ -63,3 +63,57 @@ func TestLifecycleArgsApplyDirectIOPolicy(t *testing.T) { }) } } + +// TestEgressVMsSpreadOverEveryConfiguredNetwork pins the point of the list: +// a shard the hash never picks is bridge capacity that does not exist. +func TestEgressVMsSpreadOverEveryConfiguredNetwork(t *testing.T) { + nets := []string{"cocoon-sbx0", "cocoon-sbx1", "cocoon-sbx2", "cocoon-sbx3"} + e := New("cocoon", "", nets, false, "") + key := types.PoolKey{Template: "rt:24.04", Net: types.NetEgress, Size: types.SizeMedium} + + counts := map[string]int{} + for i := range 4000 { + args := e.cloneArgs("/goldens/g1", "sbx-pool1-"+strconv.Itoa(i), key) + j := slices.Index(args, "--network") + if j < 0 { + t.Fatalf("args %v carry no --network", args) + } + counts[args[j+1]]++ + } + for _, n := range nets { + // The bound catches a starved shard, not non-uniformity. + if counts[n] < 4000/len(nets)/2 { + t.Errorf("shard %s got %d of 4000, want a fair share: %v", n, counts[n], counts) + } + } +} + +// TestNetworkChoiceIsStableForAName guards restore and teardown: the record +// persists the network a VM was built on, so the choice must not move. +func TestNetworkChoiceIsStableForAName(t *testing.T) { + e := New("cocoon", "", []string{"a", "b", "c"}, false, "") + first := e.networkFor("sbx-pool1-42") + for range 100 { + if got := e.networkFor("sbx-pool1-42"); got != first { + t.Fatalf("networkFor returned %q then %q for one name", first, got) + } + } +} + +// TestNetArgsHonorsTheLaneAndTheAttachment keeps the three attachment shapes +// distinct: no NIC on the none lane, a device on the bridge lane, and a +// single-entry list byte-for-byte the old scalar. +func TestNetArgsHonorsTheLaneAndTheAttachment(t *testing.T) { + none := types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeMedium} + egress := types.PoolKey{Template: "rt:24.04", Net: types.NetEgress, Size: types.SizeMedium} + + if args := New("cocoon", "", []string{"cni"}, false, "").netArgs("sbx-1", none, false); len(args) != 0 { + t.Errorf("none lane took an attachment: %v", args) + } + if args := New("cocoon", "br0", nil, false, "").netArgs("sbx-1", egress, false); !slices.Equal(args, []string{"--bridge", "br0"}) { + t.Errorf("bridge lane args = %v", args) + } + if args := New("cocoon", "", []string{"cocoon-dhcp"}, false, "").netArgs("sbx-1", egress, false); !slices.Equal(args, []string{"--network", "cocoon-dhcp"}) { + t.Errorf("single-network args = %v", args) + } +} diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index a3b1da5..d86fe1d 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -13,6 +13,7 @@ import ( "context" "encoding/json" "fmt" + "hash/fnv" "io" "net" "os/exec" @@ -60,14 +61,15 @@ var capacitySignatures = []string{ type Engine struct { bin string bridge string - network string + networks []string noDirectIO bool restoreMode types.RestoreMode } -// New returns a cocoon engine with node-wide network and disk policy. -func New(bin, bridge, network string, noDirectIO bool, restoreMode types.RestoreMode) *Engine { - return &Engine{bin: bin, bridge: bridge, network: network, noDirectIO: noDirectIO, restoreMode: restoreMode} +// New returns a cocoon engine with node-wide network and disk policy. networks +// are CNI conflists to spread egress-lane VMs over; see networkFor. +func New(bin, bridge string, networks []string, noDirectIO bool, restoreMode types.RestoreMode) *Engine { + return &Engine{bin: bin, bridge: bridge, networks: networks, noDirectIO: noDirectIO, restoreMode: restoreMode} } // Version reports cocoon's version string — a "vX.Y.Z" release or a @@ -304,13 +306,13 @@ func (e *Engine) cloneArgs(fromDir, name string, key types.PoolKey) []string { // by digest. args := []string{"vm", "clone", "--from-dir", fromDir, argName, name, "--pull", argOutput, formatJSON, e.directIOArg()} args = append(args, e.restoreArgs()...) - return append(args, e.netArgs(key, false)...) + return append(args, e.netArgs(name, key, false)...) } func (e *Engine) cloneSnapArgs(snap, name string, key types.PoolKey) []string { args := []string{"vm", "clone", snap, argName, name, "--pull", argOutput, formatJSON, e.directIOArg()} args = append(args, e.restoreArgs()...) - return append(args, e.netArgs(key, false)...) + return append(args, e.netArgs(name, key, false)...) } func (e *Engine) restoreCmdArgs(vmName, snapRef string) []string { @@ -337,23 +339,33 @@ func (e *Engine) runColdArgs(name string, key types.PoolKey) []string { // hypervisor from the golden's pinned snapshot, so only RunCold flags it. args = append(args, "--fc") } - args = append(args, e.netArgs(key, true)...) + args = append(args, e.netArgs(name, key, true)...) return append(args, key.Template) } -func (e *Engine) netArgs(key types.PoolKey, cold bool) []string { +func (e *Engine) netArgs(name string, key types.PoolKey, cold bool) []string { if key.Net == types.NetNone { if cold { return []string{"--nics", "0"} } return nil } - if e.network != "" { - return []string{"--network", e.network} + if len(e.networks) > 0 { + return []string{"--network", e.networkFor(name)} } return []string{"--bridge", e.bridge} } +// networkFor picks a conflist by hashing the VM name, not by counter: the +// record persists the network a VM was built on, so the choice must be +// reproducible without process state. +func (e *Engine) networkFor(name string) string { + h := fnv.New32a() + _, _ = h.Write([]byte(name)) + // >>1 keeps the index conversion positive even where int is 32-bit. + return e.networks[int(h.Sum32()>>1)%len(e.networks)] +} + func (e *Engine) run(ctx context.Context, args ...string) ([]byte, error) { ctx, cancel := context.WithTimeout(ctx, cmdTimeout) defer cancel() diff --git a/sandboxd/engine/engine_test.go b/sandboxd/engine/engine_test.go index d9429cd..5981167 100644 --- a/sandboxd/engine/engine_test.go +++ b/sandboxd/engine/engine_test.go @@ -24,7 +24,7 @@ func TestDialSilkdConsumesOnlyHandshake(t *testing.T) { // over-reads past the newline, the first Read below loses it. listenMuxer(t, path, "OK 2048\nX") - conn, err := New("cocoon", "", "", false, "").DialSilkd(t.Context(), path) + conn, err := New("cocoon", "", nil, false, "").DialSilkd(t.Context(), path) if err != nil { t.Fatalf("DialSilkd: %v", err) } @@ -42,7 +42,7 @@ func TestDialSilkdRejectedHandshake(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "ERR no guest listener\n") - _, err := New("cocoon", "", "", false, "").DialSilkd(t.Context(), path) + _, err := New("cocoon", "", nil, false, "").DialSilkd(t.Context(), path) if err == nil || !strings.Contains(err.Error(), "no guest listener") { t.Errorf("got %v, want handshake rejection", err) } @@ -52,7 +52,7 @@ func TestProbeSucceeds(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "OK 2048\n", infoFrame) - if err := New("cocoon", "", "", false, "").Probe(t.Context(), path, 2*time.Second); err != nil { + if err := New("cocoon", "", nil, false, "").Probe(t.Context(), path, 2*time.Second); err != nil { t.Errorf("Probe: %v", err) } } @@ -61,7 +61,7 @@ func TestInfoRoundTripRejectsErrorFrame(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "OK 2048\n", errFrame) - err := New("cocoon", "", "", false, "").infoRoundTrip(t.Context(), path) + err := New("cocoon", "", nil, false, "").infoRoundTrip(t.Context(), path) if err == nil || !strings.Contains(err.Error(), `info reply type "error"`) { t.Errorf("got %v, want error-frame rejection", err) } @@ -71,7 +71,7 @@ func TestProbeRetriesPastFailures(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "OK 2048\n", errFrame, errFrame, infoFrame) - if err := New("cocoon", "", "", false, "").Probe(t.Context(), path, 2*time.Second); err != nil { + if err := New("cocoon", "", nil, false, "").Probe(t.Context(), path, 2*time.Second); err != nil { t.Errorf("Probe: %v", err) } } @@ -80,7 +80,7 @@ func TestProbeRetriesUntilListenerAppears(t *testing.T) { path := sockPath(t) done := make(chan error, 1) go func() { - done <- New("cocoon", "", "", false, "").Probe(t.Context(), path, 2*time.Second) + done <- New("cocoon", "", nil, false, "").Probe(t.Context(), path, 2*time.Second) }() time.Sleep(60 * time.Millisecond) @@ -118,7 +118,7 @@ func TestDialGuestPortCtxCancel(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond) defer cancel() - if _, err := New("cocoon", "", "", false, "").DialGuestPort(ctx, path, 8080); !errors.Is(err, context.DeadlineExceeded) { + if _, err := New("cocoon", "", nil, false, "").DialGuestPort(ctx, path, 8080); !errors.Is(err, context.DeadlineExceeded) { t.Errorf("got %v, want context.DeadlineExceeded", err) } } @@ -126,7 +126,7 @@ func TestDialGuestPortCtxCancel(t *testing.T) { func TestProbeTimeout(t *testing.T) { path := sockPath(t) - err := New("cocoon", "", "", false, "").Probe(t.Context(), path, 150*time.Millisecond) + err := New("cocoon", "", nil, false, "").Probe(t.Context(), path, 150*time.Millisecond) if err == nil || !strings.Contains(err.Error(), "silkd probe") { t.Errorf("got %v, want probe timeout", err) } diff --git a/sandboxd/engine/installca_test.go b/sandboxd/engine/installca_test.go index 6bc2320..5f7c125 100644 --- a/sandboxd/engine/installca_test.go +++ b/sandboxd/engine/installca_test.go @@ -14,7 +14,7 @@ import ( func TestInstallCACertWritesCertAndUpdates(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) - if err := New("cocoon", "", "", false, "").InstallCACert(t.Context(), path, []byte("CERT-PEM")); err != nil { + if err := New("cocoon", "", nil, false, "").InstallCACert(t.Context(), path, []byte("CERT-PEM")); err != nil { t.Fatalf("InstallCACert: %v", err) } fake.mu.Lock() @@ -44,7 +44,7 @@ func TestInstallCACertNonzeroExitFails(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) fake.execCode = 3 - err := New("cocoon", "", "", false, "").InstallCACert(t.Context(), path, []byte("x")) + err := New("cocoon", "", nil, false, "").InstallCACert(t.Context(), path, []byte("x")) if err == nil || !strings.Contains(err.Error(), "exit code 3") { t.Errorf("got %v, want exit code 3 failure", err) } @@ -54,7 +54,7 @@ func TestInstallCACertWriteErrorFrameFails(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) fake.writeErr = "disk full" - err := New("cocoon", "", "", false, "").InstallCACert(t.Context(), path, []byte("x")) + err := New("cocoon", "", nil, false, "").InstallCACert(t.Context(), path, []byte("x")) if err == nil || !strings.Contains(err.Error(), "disk full") { t.Errorf("got %v, want fs_write error-frame failure", err) } diff --git a/sandboxd/main.go b/sandboxd/main.go index 42c92d9..4356e27 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -70,7 +70,7 @@ func main() { if err != nil { logger.Fatalf(ctx, err, "load config") } - eng := engine.New(cfg.CocoonBin, cfg.Bridge, cfg.Network, cfg.NoDirectIO, cfg.RestoreMode) + eng := engine.New(cfg.CocoonBin, cfg.Bridge, cfg.Networks, cfg.NoDirectIO, cfg.RestoreMode) if v, warn := eng.VersionWarning(ctx); warn != "" { logger.Warn(ctx, warn) } else { From 50ebf6d2edbdaedf57bf58de333d31e8277bece0 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 30 Jul 2026 13:22:39 +0800 Subject: [PATCH 2/3] review: partition unexported methods below the exported set asl findings in pool (checkpoint/claim/pool/telemetry/pool_test) and store/s3; layout only, no behavior change. --- sandboxd/pool/checkpoint.go | 368 ++++++++++++++++++------------------ sandboxd/pool/claim.go | 80 ++++---- sandboxd/pool/pool.go | 64 +++---- sandboxd/pool/pool_test.go | 122 ++++++------ sandboxd/pool/telemetry.go | 10 +- sandboxd/store/s3/s3.go | 88 ++++----- 6 files changed, 366 insertions(+), 366 deletions(-) diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index c9a6a42..7cb1908 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -57,42 +57,6 @@ func (m *Manager) Checkpoint(ctx context.Context, id string, cred Cred, name, te return ckpt, nil } -// publishCheckpoint stages the sandbox's exported state, writes the meta, and -// publishes it to the store, returning the record and the source snapshot the -// export captured. Shared by Checkpoint and archive; a hibernated source -// exports its wake image directly (no VM start — refill.sourceSnap). -func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID, name, tenant string, archive bool) (types.Checkpoint, string, error) { - ckpt := types.Checkpoint{ - ID: ckID, - Name: name, - SandboxID: sb.ID, - Key: sb.Key, - Tenant: tenant, - CreatedAt: time.Now(), - Archive: archive, - } - staging, err := m.ckpts.Stage(ckpt.ID) - if err != nil { - return types.Checkpoint{}, "", fmt.Errorf("stage checkpoint: %w", err) - } - defer func() { _ = os.RemoveAll(staging) }() - srcSnap, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) - if err != nil { - return types.Checkpoint{}, "", fmt.Errorf("checkpoint %s: %w", sb.ID, err) - } - meta, err := json.Marshal(ckpt) - if err != nil { - return types.Checkpoint{}, "", fmt.Errorf("encode checkpoint meta: %w", err) - } - if err := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil { - return types.Checkpoint{}, "", fmt.Errorf("write checkpoint meta: %w", err) - } - if err := m.ckpts.Publish(ctx, staging, ckpt.ID); err != nil { - return types.Checkpoint{}, "", fmt.Errorf("commit checkpoint: %w", err) - } - return ckpt, srcSnap, nil -} - // ClaimCheckpoint provisions a fresh claim cloned from a checkpoint — a // branch. The checkpoint's recorded key applies (snapshots pin size and // lane); the checkpoint itself is read-only and reusable. The claim is @@ -132,6 +96,137 @@ func (m *Manager) ClaimCheckpointHeal(ctx context.Context, ckptID string, ttl ti return m.claimLoaded(ctx, ckpt, ttl, tenant) } +// Checkpoints lists the store's checkpoints, newest first — on a shared +// backend (a FUSE mount, a bucket), that is the cluster's set, not one +// node's. A non-empty tenant filters to that tenant's records; empty (root) +// lists everything. Checkpoints backing a live archived claim are hidden: +// they are lifecycle-internal wake images, not user checkpoints. +func (m *Manager) Checkpoints(ctx context.Context, tenant string) ([]types.Checkpoint, error) { + metas, err := m.ckpts.Metas(ctx) + if err != nil { + return nil, fmt.Errorf("list checkpoints: %w", err) + } + pinned := m.pinnedArchiveCks() + ckpts := make([]types.Checkpoint, 0, len(metas)) + for _, raw := range metas { + var ckpt types.Checkpoint + if err := json.Unmarshal(raw, &ckpt); err != nil || !tenantOwns(tenant, ckpt.Tenant) { + continue + } + if _, archived := pinned[ckpt.ID]; archived || ckpt.Archive { + continue + } + ckpts = append(ckpts, ckpt) + } + slices.SortFunc(ckpts, func(a, b types.Checkpoint) int { return b.CreatedAt.Compare(a.CreatedAt) }) + return ckpts, nil +} + +// DeleteCheckpoint removes a checkpoint's snapshot and record, then +// broadcasts to peers when fleet-scoped so a healed copy does not outlive it. +// A tenant may delete only its own records — anything else answers +// ErrUnknownCheckpoint, never a hint the id exists; root deletes anything. +// Existence is checked under the record lock (heal broke the "local miss +// means truly gone" assumption), plus vetoIfHealPending for a heal whose +// transfer runs unlocked. Every exit evicts the lock entry (recDoneEvict) — +// a checkpoint id is effectively one-shot, so the map must not grow per call. +func (m *Manager) DeleteCheckpoint(ctx context.Context, ckptID, tenant string, scope DeleteScope) error { + // Reject a bad id before recLock: a rejected id must not leave a + // lock-map entry. + if !store.CheckpointIDRe.MatchString(ckptID) { + return ErrUnknownCheckpoint + } + l := m.recLock(ckptID) + l.Lock() + defer func() { l.Unlock(); m.recDoneEvict(ckptID) }() + ckpt, err := m.loadCheckpoint(ctx, ckptID) + if err != nil { + m.vetoIfHealPending(ckptID) + return err + } + if !tenantOwns(tenant, ckpt.Tenant) { + return ErrUnknownCheckpoint + } + // ckpt.Archive guards wake images across every node sharing the store; + // the pin set guards this node's not-yet-committed ones. + if _, pinned := m.pinnedArchiveCks()[ckptID]; pinned || ckpt.Archive { + return ErrUnknownCheckpoint // backs an archived sandbox, not a deletable checkpoint + } + if err := m.ckpts.Delete(ctx, ckptID); err != nil { + return fmt.Errorf("delete checkpoint: %w", err) + } + // A shared backend has no per-node replicas to chase. + if scope == DeleteFleet && m.peerDelete != nil && !m.ckptsShared { + m.peerDelete(context.WithoutCancel(ctx), ckptID) + } + return nil +} + +// HasCheckpoint answers the ownership probe: a branchable local record, never +// a fetch. +func (m *Manager) HasCheckpoint(ctx context.Context, ckptID string) bool { + ckpt, err := m.loadCheckpoint(ctx, ckptID) + return err == nil && !ckpt.Archive +} + +// FetchCheckpoint materializes a checkpoint's export for a peer transfer, +// returning the local directory, its meta, and the release to call when done. +func (m *Manager) FetchCheckpoint(ctx context.Context, ckptID string) (string, []byte, func(), error) { + if _, err := m.loadCheckpoint(ctx, ckptID); err != nil { + return "", nil, nil, err + } + l := m.recLock(ckptID) + l.RLock() + dir, meta, release, err := m.ckpts.Fetch(ctx, ckptID) + if err != nil { + l.RUnlock() + m.recDone(ckptID) + if errors.Is(err, store.ErrNotFound) { + return "", nil, nil, ErrUnknownCheckpoint + } + return "", nil, nil, fmt.Errorf("fetch checkpoint: %w", err) + } + // The read lock spans the transfer: a delete must not pull the export out + // from under a stream already writing it to a peer. + return dir, meta, func() { release(); l.RUnlock(); m.recDone(ckptID) }, nil +} + +// publishCheckpoint stages the sandbox's exported state, writes the meta, and +// publishes it to the store, returning the record and the source snapshot the +// export captured. Shared by Checkpoint and archive; a hibernated source +// exports its wake image directly (no VM start — refill.sourceSnap). +func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID, name, tenant string, archive bool) (types.Checkpoint, string, error) { + ckpt := types.Checkpoint{ + ID: ckID, + Name: name, + SandboxID: sb.ID, + Key: sb.Key, + Tenant: tenant, + CreatedAt: time.Now(), + Archive: archive, + } + staging, err := m.ckpts.Stage(ckpt.ID) + if err != nil { + return types.Checkpoint{}, "", fmt.Errorf("stage checkpoint: %w", err) + } + defer func() { _ = os.RemoveAll(staging) }() + srcSnap, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) + if err != nil { + return types.Checkpoint{}, "", fmt.Errorf("checkpoint %s: %w", sb.ID, err) + } + meta, err := json.Marshal(ckpt) + if err != nil { + return types.Checkpoint{}, "", fmt.Errorf("encode checkpoint meta: %w", err) + } + if err := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil { + return types.Checkpoint{}, "", fmt.Errorf("write checkpoint meta: %w", err) + } + if err := m.ckpts.Publish(ctx, staging, ckpt.ID); err != nil { + return types.Checkpoint{}, "", fmt.Errorf("commit checkpoint: %w", err) + } + return ckpt, srcSnap, nil +} + // claimLoaded is the shared body once ckpt's meta is resolved: it re-fetches // under the record lock, so a delete racing the pre-check cannot slip through. func (m *Manager) claimLoaded(ctx context.Context, ckpt types.Checkpoint, ttl time.Duration, tenant string) (*types.Sandbox, error) { @@ -268,85 +363,6 @@ func (m *Manager) vetoIfHealPending(ckptID string) { } } -// validateHealedCheckpoint checks a staged pull's shape before publishing it: -// an unreadable or misattributed record would suppress every later heal. -func validateHealedCheckpoint(staging, wantID string) error { - raw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // staging dir is this manager's own - if err != nil { - return fmt.Errorf("read healed meta: %w", err) - } - ckpt, err := parseCheckpoint(raw) - if err != nil { - return fmt.Errorf("parse healed meta: %w", err) - } - if ckpt.ID != wantID { - return fmt.Errorf("healed record id %q does not match requested %q", ckpt.ID, wantID) - } - if ckpt.Archive { - return fmt.Errorf("healed record %s is a wake image, not a checkpoint", wantID) - } - if keyErr := ckpt.Key.Validate(); keyErr != nil { - return fmt.Errorf("healed record %s has an invalid key: %w", wantID, keyErr) - } - if !ckpt.Key.Capturable() { - // A well-formed egress key is not branchable: publishing it would - // poison the id and suppress a good owner; reject, try the next owner. - return fmt.Errorf("healed record %s has a non-branchable (egress) key", wantID) - } - export, err := os.ReadDir(filepath.Join(staging, store.ExportDir)) - if err != nil { - return fmt.Errorf("healed record %s missing export dir: %w", wantID, err) - } - // A present export with no regular file clones to nothing (an empty dir, or - // only empty subdirs); the byte content is cocoon's own format and not - // sandboxd's to validate further. - if !hasRegularFile(export) { - return fmt.Errorf("healed record %s has no export content", wantID) - } - entries, err := os.ReadDir(staging) - if err != nil { - return fmt.Errorf("read staging: %w", err) - } - for _, e := range entries { - if e.Name() != store.MetaFile && e.Name() != store.ExportDir { - return fmt.Errorf("healed record %s has unexpected entry %q", wantID, e.Name()) - } - } - return nil -} - -// hasRegularFile reports whether entries holds at least one regular file, so an -// export of only empty subdirectories is treated as empty. -func hasRegularFile(entries []os.DirEntry) bool { - return slices.ContainsFunc(entries, func(e os.DirEntry) bool { return e.Type().IsRegular() }) -} - -// Checkpoints lists the store's checkpoints, newest first — on a shared -// backend (a FUSE mount, a bucket), that is the cluster's set, not one -// node's. A non-empty tenant filters to that tenant's records; empty (root) -// lists everything. Checkpoints backing a live archived claim are hidden: -// they are lifecycle-internal wake images, not user checkpoints. -func (m *Manager) Checkpoints(ctx context.Context, tenant string) ([]types.Checkpoint, error) { - metas, err := m.ckpts.Metas(ctx) - if err != nil { - return nil, fmt.Errorf("list checkpoints: %w", err) - } - pinned := m.pinnedArchiveCks() - ckpts := make([]types.Checkpoint, 0, len(metas)) - for _, raw := range metas { - var ckpt types.Checkpoint - if err := json.Unmarshal(raw, &ckpt); err != nil || !tenantOwns(tenant, ckpt.Tenant) { - continue - } - if _, archived := pinned[ckpt.ID]; archived || ckpt.Archive { - continue - } - ckpts = append(ckpts, ckpt) - } - slices.SortFunc(ckpts, func(a, b types.Checkpoint) int { return b.CreatedAt.Compare(a.CreatedAt) }) - return ckpts, nil -} - // pinnedArchiveCks is the set of checkpoint ids backing a live archived claim // or an archive publish in flight (pendingCks): wake images the listing hides // and delete/TTL must spare (deleting one would strand its sandbox). @@ -365,46 +381,6 @@ func (m *Manager) pinnedArchiveCks() map[string]struct{} { return pinned } -// DeleteCheckpoint removes a checkpoint's snapshot and record, then -// broadcasts to peers when fleet-scoped so a healed copy does not outlive it. -// A tenant may delete only its own records — anything else answers -// ErrUnknownCheckpoint, never a hint the id exists; root deletes anything. -// Existence is checked under the record lock (heal broke the "local miss -// means truly gone" assumption), plus vetoIfHealPending for a heal whose -// transfer runs unlocked. Every exit evicts the lock entry (recDoneEvict) — -// a checkpoint id is effectively one-shot, so the map must not grow per call. -func (m *Manager) DeleteCheckpoint(ctx context.Context, ckptID, tenant string, scope DeleteScope) error { - // Reject a bad id before recLock: a rejected id must not leave a - // lock-map entry. - if !store.CheckpointIDRe.MatchString(ckptID) { - return ErrUnknownCheckpoint - } - l := m.recLock(ckptID) - l.Lock() - defer func() { l.Unlock(); m.recDoneEvict(ckptID) }() - ckpt, err := m.loadCheckpoint(ctx, ckptID) - if err != nil { - m.vetoIfHealPending(ckptID) - return err - } - if !tenantOwns(tenant, ckpt.Tenant) { - return ErrUnknownCheckpoint - } - // ckpt.Archive guards wake images across every node sharing the store; - // the pin set guards this node's not-yet-committed ones. - if _, pinned := m.pinnedArchiveCks()[ckptID]; pinned || ckpt.Archive { - return ErrUnknownCheckpoint // backs an archived sandbox, not a deletable checkpoint - } - if err := m.ckpts.Delete(ctx, ckptID); err != nil { - return fmt.Errorf("delete checkpoint: %w", err) - } - // A shared backend has no per-node replicas to chase. - if scope == DeleteFleet && m.peerDelete != nil && !m.ckptsShared { - m.peerDelete(context.WithoutCancel(ctx), ckptID) - } - return nil -} - // sweepExpiredCheckpoints ages out checkpoints older than the configured // TTL; explicit deletes never wait for it. It runs detached from the Run // loop, so the guard keeps a slow backend from stacking sweeps. @@ -464,39 +440,63 @@ func (m *Manager) loadCheckpoint(ctx context.Context, ckptID string) (types.Chec return parseCheckpoint(raw) } -func parseCheckpoint(raw []byte) (types.Checkpoint, error) { - var ckpt types.Checkpoint - if err := json.Unmarshal(raw, &ckpt); err != nil { - return types.Checkpoint{}, ErrUnknownCheckpoint +// validateHealedCheckpoint checks a staged pull's shape before publishing it: +// an unreadable or misattributed record would suppress every later heal. +func validateHealedCheckpoint(staging, wantID string) error { + raw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // staging dir is this manager's own + if err != nil { + return fmt.Errorf("read healed meta: %w", err) } - return ckpt, nil + ckpt, err := parseCheckpoint(raw) + if err != nil { + return fmt.Errorf("parse healed meta: %w", err) + } + if ckpt.ID != wantID { + return fmt.Errorf("healed record id %q does not match requested %q", ckpt.ID, wantID) + } + if ckpt.Archive { + return fmt.Errorf("healed record %s is a wake image, not a checkpoint", wantID) + } + if keyErr := ckpt.Key.Validate(); keyErr != nil { + return fmt.Errorf("healed record %s has an invalid key: %w", wantID, keyErr) + } + if !ckpt.Key.Capturable() { + // A well-formed egress key is not branchable: publishing it would + // poison the id and suppress a good owner; reject, try the next owner. + return fmt.Errorf("healed record %s has a non-branchable (egress) key", wantID) + } + export, err := os.ReadDir(filepath.Join(staging, store.ExportDir)) + if err != nil { + return fmt.Errorf("healed record %s missing export dir: %w", wantID, err) + } + // A present export with no regular file clones to nothing (an empty dir, or + // only empty subdirs); the byte content is cocoon's own format and not + // sandboxd's to validate further. + if !hasRegularFile(export) { + return fmt.Errorf("healed record %s has no export content", wantID) + } + entries, err := os.ReadDir(staging) + if err != nil { + return fmt.Errorf("read staging: %w", err) + } + for _, e := range entries { + if e.Name() != store.MetaFile && e.Name() != store.ExportDir { + return fmt.Errorf("healed record %s has unexpected entry %q", wantID, e.Name()) + } + } + return nil } -// HasCheckpoint answers the ownership probe: a branchable local record, never -// a fetch. -func (m *Manager) HasCheckpoint(ctx context.Context, ckptID string) bool { - ckpt, err := m.loadCheckpoint(ctx, ckptID) - return err == nil && !ckpt.Archive +// hasRegularFile reports whether entries holds at least one regular file, so an +// export of only empty subdirectories is treated as empty. +func hasRegularFile(entries []os.DirEntry) bool { + return slices.ContainsFunc(entries, func(e os.DirEntry) bool { return e.Type().IsRegular() }) } -// FetchCheckpoint materializes a checkpoint's export for a peer transfer, -// returning the local directory, its meta, and the release to call when done. -func (m *Manager) FetchCheckpoint(ctx context.Context, ckptID string) (string, []byte, func(), error) { - if _, err := m.loadCheckpoint(ctx, ckptID); err != nil { - return "", nil, nil, err - } - l := m.recLock(ckptID) - l.RLock() - dir, meta, release, err := m.ckpts.Fetch(ctx, ckptID) - if err != nil { - l.RUnlock() - m.recDone(ckptID) - if errors.Is(err, store.ErrNotFound) { - return "", nil, nil, ErrUnknownCheckpoint - } - return "", nil, nil, fmt.Errorf("fetch checkpoint: %w", err) +func parseCheckpoint(raw []byte) (types.Checkpoint, error) { + var ckpt types.Checkpoint + if err := json.Unmarshal(raw, &ckpt); err != nil { + return types.Checkpoint{}, ErrUnknownCheckpoint } - // The read lock spans the transfer: a delete must not pull the export out - // from under a stream already writing it to a peer. - return dir, meta, func() { release(); l.RUnlock(); m.recDone(ckptID) }, nil + return ckpt, nil } diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index b628eb5..e57ac6e 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -99,46 +99,6 @@ func (m *Manager) Release(ctx context.Context, id string, cred Cred) error { return m.releaseResolved(ctx, id, sb) } -// releaseResolved drops the resolved claim and tears down its VM. resolve -// unlocks before this runs, so it re-checks under m.mu that sb is still the -// live claim: a second release racing in must not tear down twice. -func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sandbox) error { - m.mu.Lock() - if m.claimed[id] != sb { - m.mu.Unlock() - return ErrUnknownSandbox - } - delete(m.claimed, id) - m.tenantDelta(sb.Tenant, -1) - snap, ck, vmName := sb.HibernateSnap, sb.ArchiveCk, sb.VMName - js := m.store.snapshot(m.claimed) - m.mu.Unlock() - if saveErr := m.store.commit(js); saveErr != nil { - m.mu.Lock() - m.claimed[id] = sb // roll back so memory matches the still-durable claim; ck stays pinned - m.tenantDelta(sb.Tenant, 1) - rb := m.store.snapshot(m.claimed) - m.mu.Unlock() - m.recommit(ctx, rb) - log.WithFunc("pool.releaseResolved").Errorf(ctx, saveErr, "persist release of %s", id) - return fmt.Errorf("release %s: %w", id, saveErr) - } - // Cleanup must survive the caller hanging up; the claim is already dropped. - ctx = context.WithoutCancel(ctx) - if ck != "" { - m.purgeArchiveCk(ctx, id, ck, sb.Tenant) // archived: no local VM - } - var err error - if vmName != "" && !m.removeOrRetry(ctx, vmName, id, "") { - err = fmt.Errorf("vm %s survived removal", vmName) - } - m.disarmEgress(id, err == nil) - m.dropSnap(ctx, snap) - m.counters.releases.Add(1) - m.recordUsage(ctx, usageEvent{Event: "release", ID: id, VMName: vmName}) - return err -} - // ClaimDeadline authorizes a sandbox by token and returns its lease // deadline — the preview mint clamps a URL's life to it. func (m *Manager) ClaimDeadline(id, token string) (time.Time, error) { @@ -189,6 +149,46 @@ func (m *Manager) AgentSocket(id, token string) (string, error) { return sb.VsockSocket, nil } +// releaseResolved drops the resolved claim and tears down its VM. resolve +// unlocks before this runs, so it re-checks under m.mu that sb is still the +// live claim: a second release racing in must not tear down twice. +func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sandbox) error { + m.mu.Lock() + if m.claimed[id] != sb { + m.mu.Unlock() + return ErrUnknownSandbox + } + delete(m.claimed, id) + m.tenantDelta(sb.Tenant, -1) + snap, ck, vmName := sb.HibernateSnap, sb.ArchiveCk, sb.VMName + js := m.store.snapshot(m.claimed) + m.mu.Unlock() + if saveErr := m.store.commit(js); saveErr != nil { + m.mu.Lock() + m.claimed[id] = sb // roll back so memory matches the still-durable claim; ck stays pinned + m.tenantDelta(sb.Tenant, 1) + rb := m.store.snapshot(m.claimed) + m.mu.Unlock() + m.recommit(ctx, rb) + log.WithFunc("pool.releaseResolved").Errorf(ctx, saveErr, "persist release of %s", id) + return fmt.Errorf("release %s: %w", id, saveErr) + } + // Cleanup must survive the caller hanging up; the claim is already dropped. + ctx = context.WithoutCancel(ctx) + if ck != "" { + m.purgeArchiveCk(ctx, id, ck, sb.Tenant) // archived: no local VM + } + var err error + if vmName != "" && !m.removeOrRetry(ctx, vmName, id, "") { + err = fmt.Errorf("vm %s survived removal", vmName) + } + m.disarmEgress(id, err == nil) + m.dropSnap(ctx, snap) + m.counters.releases.Add(1) + m.recordUsage(ctx, usageEvent{Event: "release", ID: id, VMName: vmName}) + return err +} + // overQuota is a cheap advisory precheck; finalizeBatch does the authoritative // admission check, so this just spares a doomed request the provision cost. func (m *Manager) overQuota(extra int, tenant string) error { diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 9f76ab7..3d5d592 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -525,18 +525,6 @@ func (m *Manager) FlushClaims() error { return m.store.commit(m.claimsSnapshot()) } -func (m *Manager) claimsSnapshot() claimSnapshot { - m.mu.Lock() - defer m.mu.Unlock() - return m.store.snapshot(m.claimed) -} - -func (m *Manager) untrack(set map[string]struct{}, key string) { - m.mu.Lock() - delete(set, key) - m.mu.Unlock() -} - // SetTemplateNotifier wires the immediate-republish hook; call it before // the server starts serving. func (m *Manager) SetTemplateNotifier(fn func()) { @@ -610,26 +598,6 @@ func (m *Manager) WarmCounts() map[string]int { return counts } -// activePool looks up key treating a removed pool as absent; callers hold m.mu. -func (m *Manager) activePool(key types.PoolKey) (*pool, bool) { - p, ok := m.pools[key] - return p, ok && !p.removed -} - -func (m *Manager) validate(key types.PoolKey) error { - if err := key.Validate(); err != nil { - return fmt.Errorf("%w: %v", ErrBadKey, err) - } - if key.Net == types.NetEgress && !m.egress { - return ErrNoEgress - } - return nil -} - -func (m *Manager) goldensDir() string { - return filepath.Join(m.dataDir, "goldens") -} - func loadEgressCA(cfg *config.EgressCAConfig) (*egress.CA, error) { root, err := os.ReadFile(cfg.RootCert) //nolint:gosec // operator-configured ca path if err != nil { @@ -671,6 +639,38 @@ func (m *Manager) WithPeerDelete(fn PeerDeleteFunc) { m.peerDelete = fn } +func (m *Manager) claimsSnapshot() claimSnapshot { + m.mu.Lock() + defer m.mu.Unlock() + return m.store.snapshot(m.claimed) +} + +func (m *Manager) untrack(set map[string]struct{}, key string) { + m.mu.Lock() + delete(set, key) + m.mu.Unlock() +} + +// activePool looks up key treating a removed pool as absent; callers hold m.mu. +func (m *Manager) activePool(key types.PoolKey) (*pool, bool) { + p, ok := m.pools[key] + return p, ok && !p.removed +} + +func (m *Manager) validate(key types.PoolKey) error { + if err := key.Validate(); err != nil { + return fmt.Errorf("%w: %v", ErrBadKey, err) + } + if key.Net == types.NetEgress && !m.egress { + return ErrNoEgress + } + return nil +} + +func (m *Manager) goldensDir() string { + return filepath.Join(m.dataDir, "goldens") +} + func dirExists(path string) bool { fi, err := os.Stat(path) return err == nil && fi.IsDir() diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 83ee48b..59b0c80 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -798,28 +798,6 @@ func (f *fakeEngine) CloneSnap(_ context.Context, snap, name string, _ types.Poo return f.clone(snap, name) } -func (f *fakeEngine) clone(from, name string) (types.VMRecord, error) { - f.mu.Lock() - f.clones = append(f.clones, name) - f.cloneFroms = append(f.cloneFroms, from) - call := len(f.clones) - stall, err, failNth := f.cloneStall, f.cloneErr, f.cloneFailNth - f.mu.Unlock() - if stall != nil { - <-stall - } - if err != nil { - return types.VMRecord{}, err - } - if failNth > 0 && call == failNth { - return types.VMRecord{}, errors.New("clone failed") - } - f.mu.Lock() - defer f.mu.Unlock() - f.vms[name] = "/vsock/" + name - return f.record(name), nil -} - func (f *fakeEngine) RunCold(_ context.Context, name string, _ types.PoolKey) (types.VMRecord, error) { f.mu.Lock() defer f.mu.Unlock() @@ -831,33 +809,6 @@ func (f *fakeEngine) RunCold(_ context.Context, name string, _ types.PoolKey) (t return f.record(name), nil } -// record builds a lifecycle result for name; callers hold f.mu. -func (f *fakeEngine) record(name string) types.VMRecord { - rec := types.VMRecord{VsockSocket: f.lateVsock(f.vms[name]), Config: types.VMConfig{Name: name}} - if f.tap != "" { - rec.NetworkConfigs = []types.VMNetConfig{{TAP: f.tap}} - } - return rec -} - -// lateVsock models cocoon's report-once-running lag: while vsockLateN ticks -// remain it reports no socket (consuming one), forcing the caller's poll. -func (f *fakeEngine) lateVsock(sock string) string { - if f.vsockLateN > 0 { - f.vsockLateN-- - return "" - } - return sock -} - -// removed reports whether name was destroyed, under the fake's own lock — -// bounded batches append concurrently with test polls. -func (f *fakeEngine) removed(name string) bool { - f.mu.Lock() - defer f.mu.Unlock() - return slices.Contains(f.removes, name) -} - func (f *fakeEngine) Remove(ctx context.Context, name string) error { // Models exec.CommandContext: a canceled ctx never runs cocoon at all. if err := ctx.Err(); err != nil { @@ -909,18 +860,6 @@ func (f *fakeEngine) SnapshotList(_ context.Context) ([]string, error) { return snaps, nil } -func (f *fakeEngine) listCalls() int { - f.mu.Lock() - defer f.mu.Unlock() - return f.listCount -} - -func (f *fakeEngine) snapListCalls() int { - f.mu.Lock() - defer f.mu.Unlock() - return f.snapListCount -} - func (f *fakeEngine) SnapshotRemove(_ context.Context, snapName string) error { f.mu.Lock() stall := f.snapRemoveStall @@ -1014,6 +953,67 @@ func (f *fakeEngine) InstallCACert(_ context.Context, vsockSocket string, _ []by return f.installCAErr } +func (f *fakeEngine) clone(from, name string) (types.VMRecord, error) { + f.mu.Lock() + f.clones = append(f.clones, name) + f.cloneFroms = append(f.cloneFroms, from) + call := len(f.clones) + stall, err, failNth := f.cloneStall, f.cloneErr, f.cloneFailNth + f.mu.Unlock() + if stall != nil { + <-stall + } + if err != nil { + return types.VMRecord{}, err + } + if failNth > 0 && call == failNth { + return types.VMRecord{}, errors.New("clone failed") + } + f.mu.Lock() + defer f.mu.Unlock() + f.vms[name] = "/vsock/" + name + return f.record(name), nil +} + +// record builds a lifecycle result for name; callers hold f.mu. +func (f *fakeEngine) record(name string) types.VMRecord { + rec := types.VMRecord{VsockSocket: f.lateVsock(f.vms[name]), Config: types.VMConfig{Name: name}} + if f.tap != "" { + rec.NetworkConfigs = []types.VMNetConfig{{TAP: f.tap}} + } + return rec +} + +// lateVsock models cocoon's report-once-running lag: while vsockLateN ticks +// remain it reports no socket (consuming one), forcing the caller's poll. +func (f *fakeEngine) lateVsock(sock string) string { + if f.vsockLateN > 0 { + f.vsockLateN-- + return "" + } + return sock +} + +// removed reports whether name was destroyed, under the fake's own lock — +// bounded batches append concurrently with test polls. +func (f *fakeEngine) removed(name string) bool { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Contains(f.removes, name) +} + +func (f *fakeEngine) listCalls() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.listCount +} + +func (f *fakeEngine) snapListCalls() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.snapListCount +} + func (f *fakeEngine) cloneCount() int { f.mu.Lock() defer f.mu.Unlock() diff --git a/sandboxd/pool/telemetry.go b/sandboxd/pool/telemetry.go index 35b2459..ffddd16 100644 --- a/sandboxd/pool/telemetry.go +++ b/sandboxd/pool/telemetry.go @@ -113,6 +113,11 @@ func (m *Manager) Audit(ctx context.Context, id string, line []byte) { m.recordAudit(ctx, id, frame) } +// AuditEnabled reports whether the relay should tap request frames at all. +func (m *Manager) AuditEnabled() bool { + return m.audit != nil +} + func (m *Manager) recordAudit(ctx context.Context, id string, frame auditFrame) { if m.audit == nil { return @@ -127,11 +132,6 @@ func (m *Manager) recordAudit(ctx context.Context, id string, frame auditFrame) } } -// AuditEnabled reports whether the relay should tap request frames at all. -func (m *Manager) AuditEnabled() bool { - return m.audit != nil -} - // recordEgress logs one guarded-egress decision to the audit tap (secret name // only, never its value) and meters it as an always-on usage event. func (m *Manager) recordEgress(ctx context.Context, id, tenant string, ev egress.Event) { diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index e202583..1971395 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -173,50 +173,6 @@ func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, func(), e return export, meta, func() {}, nil } -// populate downloads one cache generation and installs it atomically. -func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string) error { - if _, err := os.Stat(gen); err == nil { - return nil // another flight installed it between stat and Do - } - local, err := os.MkdirTemp(s.staging, id+"-fetch-*") - if err != nil { - return err - } - defer func() { _ = os.RemoveAll(local) }() - exportPrefix := s.key(id, exportGen(meta)) + "/" - keys, err := s.list(ctx, exportPrefix) - if err != nil { - return err - } - if len(keys) == 0 { - // Records published before per-generation prefixes. - exportPrefix = s.key(id, store.ExportDir) + "/" - if keys, err = s.list(ctx, exportPrefix); err != nil { - return err - } - } - if len(keys) == 0 { - return fmt.Errorf("record %s has no export", id) - } - g, gctx := errgroup.WithContext(ctx) - g.SetLimit(4) - for _, key := range keys { - g.Go(func() error { - return s.download(gctx, key, filepath.Join(local, store.ExportDir, strings.TrimPrefix(key, exportPrefix))) - }) - } - if err := g.Wait(); err != nil { - return err - } - if err := os.WriteFile(filepath.Join(local, store.MetaFile), meta, 0o600); err != nil { - return err - } - if err := os.MkdirAll(filepath.Dir(gen), 0o750); err != nil { - return err - } - return os.Rename(local, gen) -} - func (s *Store) ReadMeta(ctx context.Context, id string) ([]byte, error) { out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{ Bucket: &s.bucket, Key: aws.String(s.key(id, store.MetaFile)), @@ -302,6 +258,50 @@ func (s *Store) SweepStaging() error { return nil } +// populate downloads one cache generation and installs it atomically. +func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string) error { + if _, err := os.Stat(gen); err == nil { + return nil // another flight installed it between stat and Do + } + local, err := os.MkdirTemp(s.staging, id+"-fetch-*") + if err != nil { + return err + } + defer func() { _ = os.RemoveAll(local) }() + exportPrefix := s.key(id, exportGen(meta)) + "/" + keys, err := s.list(ctx, exportPrefix) + if err != nil { + return err + } + if len(keys) == 0 { + // Records published before per-generation prefixes. + exportPrefix = s.key(id, store.ExportDir) + "/" + if keys, err = s.list(ctx, exportPrefix); err != nil { + return err + } + } + if len(keys) == 0 { + return fmt.Errorf("record %s has no export", id) + } + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(4) + for _, key := range keys { + g.Go(func() error { + return s.download(gctx, key, filepath.Join(local, store.ExportDir, strings.TrimPrefix(key, exportPrefix))) + }) + } + if err := g.Wait(); err != nil { + return err + } + if err := os.WriteFile(filepath.Join(local, store.MetaFile), meta, 0o600); err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(gen), 0o750); err != nil { + return err + } + return os.Rename(local, gen) +} + func (s *Store) key(id, rest string) string { if rest == "" { return s.prefix + id From 89ad6d884ec9ddf93e72ca401ae7af65ba03961a Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 30 Jul 2026 13:29:01 +0800 Subject: [PATCH 3/3] review: apply the LOC-audit cut-list - e2e: shared harness.Claim replaces the per-tool connect/claim prologue (7 tools; the tools with per-step timing or per-node error context keep their own) - egress: one relay() carries the forward path shared by the plain proxy and the intercept handler - mcp: parseAndBox generic replaces the per-handler parse-then-resolve prologue (10 handlers) - types: TTLField embed dedups the wire TTL conversion (4 request bodies) - config, store/peer: cmp.Or for two hand-rolled defaults --- e2e/cmd/egresssmoke/main.go | 7 +- e2e/cmd/interceptsmoke/main.go | 7 +- e2e/cmd/lifecycle/main.go | 8 +- e2e/cmd/pullbench/main.go | 7 +- e2e/cmd/pushbench/main.go | 7 +- e2e/cmd/rpcbench/main.go | 7 +- e2e/cmd/smoke/main.go | 9 +- e2e/internal/harness/harness.go | 22 ++++ mcp/tools.go | 168 ++++++++++++++----------------- sandboxd/config/config.go | 4 +- sandboxd/egress/intercept.go | 39 ++----- sandboxd/egress/proxy.go | 11 +- sandboxd/server/preview.go | 2 +- sandboxd/store/peer/transport.go | 5 +- sandboxd/types/api.go | 52 +++++----- 15 files changed, 161 insertions(+), 194 deletions(-) create mode 100644 e2e/internal/harness/harness.go diff --git a/e2e/cmd/egresssmoke/main.go b/e2e/cmd/egresssmoke/main.go index e62af4b..0004688 100644 --- a/e2e/cmd/egresssmoke/main.go +++ b/e2e/cmd/egresssmoke/main.go @@ -23,6 +23,7 @@ import ( "strings" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -58,14 +59,10 @@ func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string defer func() { _ = origin.Close() }() target := fmt.Sprintf("http://%s:%d/", reach, port) - c, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + _, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetShape(netShape)), sandbox.WithSize(sandbox.Small)) if err != nil { return err } - sb, err := c.New(ctx, template, sandbox.WithNetwork(sandbox.NetShape(netShape)), sandbox.WithSize(sandbox.Small)) - if err != nil { - return fmt.Errorf("claim: %w", err) - } defer func() { _ = sb.Close() }() fmt.Printf(" claimed %s-lane sandbox %s\n", netShape, sb.ID) diff --git a/e2e/cmd/interceptsmoke/main.go b/e2e/cmd/interceptsmoke/main.go index abe7f45..5f59f93 100644 --- a/e2e/cmd/interceptsmoke/main.go +++ b/e2e/cmd/interceptsmoke/main.go @@ -13,6 +13,7 @@ import ( "strings" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -36,14 +37,10 @@ func run(addr, token, template, echo, secret, issuer string) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() - c, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + _, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetNone), sandbox.WithSize(sandbox.Small)) if err != nil { return err } - sb, err := c.New(ctx, template, sandbox.WithNetwork(sandbox.NetNone), sandbox.WithSize(sandbox.Small)) - if err != nil { - return fmt.Errorf("claim: %w", err) - } defer func() { _ = sb.Close() }() fmt.Printf(" claimed none-lane sandbox %s\n", sb.ID) diff --git a/e2e/cmd/lifecycle/main.go b/e2e/cmd/lifecycle/main.go index 9e45cbc..df1ba5a 100644 --- a/e2e/cmd/lifecycle/main.go +++ b/e2e/cmd/lifecycle/main.go @@ -12,6 +12,7 @@ import ( "os" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -34,15 +35,10 @@ func main() { func run(addr, token, template, netShape string, wait time.Duration) error { ctx := context.Background() - client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + client, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetShape(netShape))) if err != nil { return err } - - sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetShape(netShape))) - if err != nil { - return fmt.Errorf("claim: %w", err) - } id := sb.ID marker := fmt.Appendf(nil, "archive-marker-%d", time.Now().UnixNano()) if err = sb.WriteFile(ctx, markerPath, marker, nil); err != nil { diff --git a/e2e/cmd/pullbench/main.go b/e2e/cmd/pullbench/main.go index b83c5e6..e71fbdc 100644 --- a/e2e/cmd/pullbench/main.go +++ b/e2e/cmd/pullbench/main.go @@ -11,6 +11,7 @@ import ( "os" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -30,14 +31,10 @@ func main() { func run(addr, token, template string, size, n int) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() - client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + _, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetNone)) if err != nil { return err } - sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetNone)) - if err != nil { - return fmt.Errorf("claim: %w", err) - } defer func() { _ = sb.Close() }() if _, err := sb.Exec(ctx, "sh", "-c", diff --git a/e2e/cmd/pushbench/main.go b/e2e/cmd/pushbench/main.go index 78d8f73..17543bf 100644 --- a/e2e/cmd/pushbench/main.go +++ b/e2e/cmd/pushbench/main.go @@ -9,6 +9,7 @@ import ( "os" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -28,14 +29,10 @@ func main() { func run(addr, token, template string, size, n int) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() - client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + _, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetNone)) if err != nil { return err } - sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetNone)) - if err != nil { - return fmt.Errorf("claim: %w", err) - } defer func() { _ = sb.Close() }() if _, err := sb.Exec(ctx, "mkdir", "-p", "/work/bench"); err != nil { diff --git a/e2e/cmd/rpcbench/main.go b/e2e/cmd/rpcbench/main.go index 51e5705..f60f52d 100644 --- a/e2e/cmd/rpcbench/main.go +++ b/e2e/cmd/rpcbench/main.go @@ -18,6 +18,7 @@ import ( "slices" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" "github.com/cocoonstack/sandbox/protocol/wire" sandbox "github.com/cocoonstack/sandbox/sdk/go" "github.com/cocoonstack/sandbox/sdk/go/silkd" @@ -38,14 +39,10 @@ func main() { func run(addr, token, template string, n int) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() - client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + _, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetNone)) if err != nil { return err } - sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetNone)) - if err != nil { - return fmt.Errorf("claim: %w", err) - } defer func() { _ = sb.Close() }() dial := func() (net.Conn, error) { return dialAgent(ctx, sb.Owner(), sb.ID, sb.Token()) } diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index f391932..e389ea4 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -21,6 +21,7 @@ import ( "testing/iotest" "time" + "github.com/cocoonstack/sandbox/e2e/internal/harness" "github.com/cocoonstack/sandbox/protocol/wire" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -48,14 +49,10 @@ func run(addr, token, template, lspTemplate string, egress bool) error { ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) - if err != nil { - return err - } // Explicitly the no-network lane: the git step asserts its typed error. - sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetNone)) + client, sb, err := harness.Claim(ctx, addr, token, template, sandbox.WithNetwork(sandbox.NetNone)) if err != nil { - return fmt.Errorf("claim: %w", err) + return err } defer func() { _ = sb.Close() }() // Info is node-local: on a cluster the claim may have been redirected, so diff --git a/e2e/internal/harness/harness.go b/e2e/internal/harness/harness.go new file mode 100644 index 0000000..264c09b --- /dev/null +++ b/e2e/internal/harness/harness.go @@ -0,0 +1,22 @@ +// Package harness carries the claim prologue shared by the e2e tools. +package harness + +import ( + "context" + "fmt" + + sandbox "github.com/cocoonstack/sandbox/sdk/go" +) + +// Claim connects to a node and claims one sandbox; the caller owns Close. +func Claim(ctx context.Context, addr, token, template string, opts ...sandbox.Option) (*sandbox.Client, *sandbox.Sandbox, error) { + client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) + if err != nil { + return nil, nil, err + } + sb, err := client.New(ctx, template, opts...) + if err != nil { + return nil, nil, fmt.Errorf("claim: %w", err) + } + return client, sb, nil +} diff --git a/mcp/tools.go b/mcp/tools.go index e4f4c97..7e4d1d4 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -111,16 +111,35 @@ func toolCreateSandbox(ctx context.Context, s *server, raw json.RawMessage) (str return jsonText(map[string]any{"sandbox_id": sb.ID, "deadline": sb.Deadline}), nil } -func toolExec(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Command string `json:"command"` - Cwd string `json:"cwd"` - } +// sandboxArg is the shared sandbox_id field of the tool argument structs; +// parseAndBox reads it through id(). +type sandboxArg struct { + SandboxID string `json:"sandbox_id"` +} + +func (a sandboxArg) id() string { return a.SandboxID } + +type sandboxIDer interface{ id() string } + +// parseAndBox decodes a tool's arguments and resolves their sandbox_id to a +// live handle — the prologue every sandbox-scoped tool shares. +func parseAndBox[T sandboxIDer](s *server, raw json.RawMessage) (T, *sandbox.Sandbox, error) { + var args T if err := parse(raw, &args); err != nil { - return "", err + return args, nil, err } - sb, err := s.box(args.SandboxID) + sb, err := s.box(args.id()) + return args, sb, err +} + +type cmdArgs struct { + sandboxArg + Command string `json:"command"` + Cwd string `json:"cwd"` +} + +func toolExec(ctx context.Context, s *server, raw json.RawMessage) (string, error) { + args, sb, err := parseAndBox[cmdArgs](s, raw) if err != nil { return "", err } @@ -136,15 +155,7 @@ func toolExec(ctx context.Context, s *server, raw json.RawMessage) (string, erro } func toolSpawn(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Command string `json:"command"` - Cwd string `json:"cwd"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[cmdArgs](s, raw) if err != nil { return "", err } @@ -167,16 +178,14 @@ func toolPs(ctx context.Context, s *server, raw json.RawMessage) (string, error) return jsonText(procs), nil } +type killArgs struct { + sandboxArg + PID uint32 `json:"pid"` + Signal int32 `json:"signal"` +} + func toolKill(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - PID uint32 `json:"pid"` - Signal int32 `json:"signal"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[killArgs](s, raw) if err != nil { return "", err } @@ -186,15 +195,13 @@ func toolKill(ctx context.Context, s *server, raw json.RawMessage) (string, erro return "killed", nil } +type logsArgs struct { + sandboxArg + PID uint32 `json:"pid"` +} + func toolLogs(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - PID uint32 `json:"pid"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[logsArgs](s, raw) if err != nil { return "", err } @@ -210,16 +217,14 @@ func toolLogs(ctx context.Context, s *server, raw json.RawMessage) (string, erro return jsonText(out), nil } +type writeFileArgs struct { + sandboxArg + Path string `json:"path"` + Content string `json:"content"` +} + func toolWriteFile(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Path string `json:"path"` - Content string `json:"content"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[writeFileArgs](s, raw) if err != nil { return "", err } @@ -229,15 +234,13 @@ func toolWriteFile(ctx context.Context, s *server, raw json.RawMessage) (string, return fmt.Sprintf("wrote %d bytes to %s", len(args.Content), args.Path), nil } +type pathArgs struct { + sandboxArg + Path string `json:"path"` +} + func toolReadFile(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Path string `json:"path"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[pathArgs](s, raw) if err != nil { return "", err } @@ -249,14 +252,7 @@ func toolReadFile(ctx context.Context, s *server, raw json.RawMessage) (string, } func toolListDir(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Path string `json:"path"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[pathArgs](s, raw) if err != nil { return "", err } @@ -267,15 +263,13 @@ func toolListDir(ctx context.Context, s *server, raw json.RawMessage) (string, e return jsonText(entries), nil } +type forkArgs struct { + sandboxArg + Count int `json:"count"` +} + func toolFork(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Count int `json:"count"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[forkArgs](s, raw) if err != nil { return "", err } @@ -291,15 +285,13 @@ func toolFork(ctx context.Context, s *server, raw json.RawMessage) (string, erro return jsonText(map[string]any{"sandbox_ids": ids}), nil } +type checkpointArgs struct { + sandboxArg + Name string `json:"name"` +} + func toolCheckpoint(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - Name string `json:"name"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[checkpointArgs](s, raw) if err != nil { return "", err } @@ -359,15 +351,13 @@ func toolHibernate(ctx context.Context, s *server, raw json.RawMessage) (string, return "hibernated (any later call wakes it transparently)", nil } +type promoteArgs struct { + sandboxArg + TemplateName string `json:"template_name"` +} + func toolPromote(ctx context.Context, s *server, raw json.RawMessage) (string, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - TemplateName string `json:"template_name"` - } - if err := parse(raw, &args); err != nil { - return "", err - } - sb, err := s.box(args.SandboxID) + args, sb, err := parseAndBox[promoteArgs](s, raw) if err != nil { return "", err } @@ -398,16 +388,10 @@ func toolNodeInfo(ctx context.Context, s *server, _ json.RawMessage) (string, er return jsonText(info), nil } -// boxArg resolves a sandbox_id argument to a live handle; handlers with -// more arguments parse their own struct from the same raw bytes. +// boxArg resolves a bare sandbox_id argument to a live handle. func (s *server) boxArg(raw json.RawMessage) (*sandbox.Sandbox, error) { - var args struct { - SandboxID string `json:"sandbox_id"` - } - if err := parse(raw, &args); err != nil { - return nil, err - } - return s.box(args.SandboxID) + _, sb, err := parseAndBox[sandboxArg](s, raw) + return sb, err } // checkpointArg resolves a checkpoint_id argument: a handle minted in this diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 522f63a..c51f3ee 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -320,9 +320,7 @@ func (c *Config) applyDefaults() { c.DataDir = cmp.Or(c.DataDir, defaultDataDir) c.CocoonBin = cmp.Or(c.CocoonBin, defaultCocoonBin) c.AdvertiseAddr = cmp.Or(c.AdvertiseAddr, c.Listen) - if c.PreviewListen != "" && c.PreviewAdvertise == "" { - c.PreviewAdvertise = c.PreviewListen - } + c.PreviewAdvertise = cmp.Or(c.PreviewAdvertise, c.PreviewListen) c.MaxForkCount = cmp.Or(c.MaxForkCount, defaultMaxForkCount) if c.RefillConcurrency == 0 { c.RefillConcurrency = autoRefillConcurrency(runtime.NumCPU()) diff --git a/sandboxd/egress/intercept.go b/sandboxd/egress/intercept.go index 6134b81..93f6818 100644 --- a/sandboxd/egress/intercept.go +++ b/sandboxd/egress/intercept.go @@ -3,7 +3,6 @@ package egress import ( "crypto/tls" "io" - "maps" "net" "net/http" "strings" @@ -94,34 +93,16 @@ type interceptHandler struct { func (h *interceptHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { p := h.proxy rule, decision := p.policy.EvalInner(h.host, r.Method) - if decision == DecisionDeny { - p.record(Event{Method: r.Method, Host: h.host, Decision: decision}) - denied(w, h.host) - return - } - out := r.Clone(r.Context()) - out.URL.Scheme = "https" - out.URL.Host = h.authority - // Keep the guest's Host when it names the CONNECT host (SigV4-style signing - // breaks on the ":443" rewrite); a foreign Host snaps to the authority. - if !strings.EqualFold(hostOnly(r.Host), h.host) { - out.Host = h.authority - } - out.RequestURI = "" - stripHop(out.Header) - injected := p.inject(rule, out.Header) - p.record(Event{Method: r.Method, Host: h.host, Decision: decision, Injected: injected}) - - resp, err := p.mitmTr.RoundTrip(out) - if err != nil { - http.Error(w, "egress: upstream unreachable", http.StatusBadGateway) - return - } - defer func() { _ = resp.Body.Close() }() - maps.Copy(w.Header(), resp.Header) - stripHop(w.Header()) - w.WriteHeader(resp.StatusCode) - _, _ = io.Copy(w, resp.Body) + p.relay(w, r, h.host, rule, decision, p.mitmTr, func(out *http.Request) { + out.URL.Scheme = "https" + out.URL.Host = h.authority + // Keep the guest's Host when it names the CONNECT host (SigV4-style + // signing breaks on the ":443" rewrite); a foreign Host snaps to the + // authority. + if !strings.EqualFold(hostOnly(r.Host), h.host) { + out.Host = h.authority + } + }) } // singleConnListener feeds one already-accepted conn to an http.Server and then diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index 7f3c211..2b671e8 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -186,6 +186,12 @@ func (p *Proxy) serveForward(w http.ResponseWriter, r *http.Request) { } host := hostOnly(r.URL.Host) rule, decision := p.policy.Eval(host, r.Method) + p.relay(w, r, host, rule, decision, p.tr, nil) +} + +// relay forwards one policy-evaluated request upstream and mirrors the +// response; prepare, when set, rewrites the clone before it leaves. +func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, host string, rule Rule, decision Decision, tr *http.Transport, prepare func(*http.Request)) { if decision == DecisionDeny { p.record(Event{Method: r.Method, Host: host, Decision: decision}) denied(w, host) @@ -193,12 +199,15 @@ func (p *Proxy) serveForward(w http.ResponseWriter, r *http.Request) { } out := r.Clone(r.Context()) + if prepare != nil { + prepare(out) + } out.RequestURI = "" stripHop(out.Header) injected := p.inject(rule, out.Header) p.record(Event{Method: r.Method, Host: host, Decision: decision, Injected: injected}) - resp, err := p.tr.RoundTrip(out) + resp, err := tr.RoundTrip(out) if err != nil { http.Error(w, "egress: upstream unreachable", http.StatusBadGateway) return diff --git a/sandboxd/server/preview.go b/sandboxd/server/preview.go index 883c2d4..1c34d7d 100644 --- a/sandboxd/server/preview.go +++ b/sandboxd/server/preview.go @@ -189,7 +189,7 @@ func (s *Server) handlePreview(w http.ResponseWriter, r *http.Request) { writePoolErr(w, err) return } - ttl := time.Duration(req.TTLSeconds) * time.Second + ttl := req.TTL() switch { case !deadline.IsZero(): if lease := time.Until(deadline); ttl <= 0 || ttl > lease { diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index b3e06b5..e79a3a4 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -5,6 +5,7 @@ package peer import ( "archive/tar" + "cmp" "context" "errors" "fmt" @@ -216,9 +217,7 @@ func Untar(r io.Reader, dst string) error { } func writeFile(target string, r io.Reader, mode os.FileMode) error { - if mode == 0 { - mode = 0o600 - } + mode = cmp.Or(mode, os.FileMode(0o600)) f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, mode) //nolint:gosec // target resolved by safeJoin if err != nil { return fmt.Errorf("untar create %s: %w", target, err) diff --git a/sandboxd/types/api.go b/sandboxd/types/api.go index e0c2e1d..de4ad7c 100644 --- a/sandboxd/types/api.go +++ b/sandboxd/types/api.go @@ -4,16 +4,27 @@ import ( "time" ) +// TTLField is the requested-lease field shared by the claiming request +// bodies; zero means the server default. +type TTLField struct { + TTLSeconds int `json:"ttl_seconds,omitempty"` +} + +// TTL converts the wire seconds to a duration. +func (f TTLField) TTL() time.Duration { + return time.Duration(f.TTLSeconds) * time.Second +} + // ClaimRequest is the wire body of POST /v1/claim. NoRedirect is set by the // SDK on a claim it is retrying at a redirect target, so that node warm-or- // provisions locally instead of bouncing the claim back on a stale view. type ClaimRequest struct { - Template string `json:"template"` - Net NetShape `json:"net,omitempty"` - Size Size `json:"size,omitempty"` - Engine Engine `json:"engine,omitempty"` - TTLSeconds int `json:"ttl_seconds,omitempty"` - NoRedirect bool `json:"no_redirect,omitempty"` + Template string `json:"template"` + Net NetShape `json:"net,omitempty"` + Size Size `json:"size,omitempty"` + Engine Engine `json:"engine,omitempty"` + TTLField + NoRedirect bool `json:"no_redirect,omitempty"` // ClaimRef is an opaque caller reference (the aggregated apiserver passes // the k8s "/") recorded on the claim so the read path can // map a listed sandbox back to the name it was claimed under. Optional. @@ -25,11 +36,6 @@ func (r ClaimRequest) Key() PoolKey { return PoolKey{Template: r.Template, Net: r.Net, Size: r.Size, Engine: r.Engine}.Defaulted() } -// TTL converts the wire seconds to a duration; zero means server default. -func (r ClaimRequest) TTL() time.Duration { - return time.Duration(r.TTLSeconds) * time.Second -} - // ClaimResponse is the wire reply of POST /v1/claim. A successful claim // carries ID/Token/Deadline/OwnerAddr; a mesh miss carries Redirect (peer // addresses to retry, MOVED-style), and the two are mutually exclusive. @@ -53,14 +59,9 @@ type ClaimResponse struct { // lease is a per-sandbox resource bound, so children never inherit the // parent's remainder. type ForkRequest struct { - Token string `json:"token"` - Count int `json:"count"` - TTLSeconds int `json:"ttl_seconds,omitempty"` -} - -// TTL converts the wire seconds to a duration; zero means server default. -func (r ForkRequest) TTL() time.Duration { - return time.Duration(r.TTLSeconds) * time.Second + Token string `json:"token"` + Count int `json:"count"` + TTLField } // ForkResponse carries one claim per child. @@ -83,17 +84,12 @@ type CheckpointResponse struct { // CheckpointClaimRequest is the wire body of POST /v1/checkpoints/{id}/claim. // TTLSeconds semantics match a claim's; zero means the server default. type CheckpointClaimRequest struct { - TTLSeconds int `json:"ttl_seconds,omitempty"` + TTLField // NoRedirect is set by a client retrying at a redirect target, so the retry // resolves locally instead of bouncing between two nodes. NoRedirect bool `json:"no_redirect,omitempty"` } -// TTL converts the requested lease. -func (r CheckpointClaimRequest) TTL() time.Duration { - return time.Duration(r.TTLSeconds) * time.Second -} - // CheckpointListResponse is the wire reply of GET /v1/checkpoints. type CheckpointListResponse struct { Checkpoints []Checkpoint `json:"checkpoints"` @@ -116,9 +112,9 @@ type PromoteResponse struct { // PreviewRequest is the wire body of POST /v1/sandboxes/{id}/preview; auth // mirrors ForkRequest (control token in the header, ownership in Token). type PreviewRequest struct { - Token string `json:"token"` - Port uint16 `json:"port"` - TTLSeconds int `json:"ttl_seconds,omitempty"` + Token string `json:"token"` + Port uint16 `json:"port"` + TTLField } // PreviewResponse carries the minted URL.