diff --git a/docs/deploy.md b/docs/deploy.md index cff6ddd..61b24e8 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -187,6 +187,45 @@ The last setting also removes the VMM's network-namespace quarantine. On a measurement. Pause `cocoon daemon` or lengthen its reconcile interval during a mass fill. +### Host tuning for dense egress-lane nodes + +The no-network lane is CPU-bound and needs none of this. The egress lane +serializes on the kernel's global rtnl lock — tap create, bridge attach, and +the VMM's own tap open all take it — so at hundreds of VMs host configuration +dominates fill throughput: + +- **Keep the kernel console quiet** (`loglevel=4` on the kernel cmdline, or + drop `console=ttyS0`; dmesg and the journal keep everything either way). + Bridge port transitions printk synchronously to every registered console + *while holding rtnl*: measured on a serial+fbcon host, one bridge attach is + 44 ms noisy vs 3 ms quiet, and a 1000-VM fill roughly triples. Fix the + console before tuning anything else — netlink-level optimizations are + unmeasurable, or negative, on a noisy console. +- **Keep udev off the sandbox taps.** Every tap add/remove fires a udev + event; rule evaluation plus hooks like `ifupdown-hotplug` (a fork+exec per + interface) contend on the same rtnl lock — stopping the udev exec queue + during a fill measured ~4x throughput, and a backed-up event queue can + collapse a fill outright. Shadow the net-setup rules for these taps, or + pause the exec queue around planned mass fills. +- **Set `refill_concurrency` explicitly (~64) on egress-heavy nodes.** The + auto default scales with cores (a 384-core node gets 256), which suits the + no-network lane but overshoots the bridge lane: rtnl does not plateau under + pressure, it collapses — on a quiet host RC 256 halves the fill rate RC 64 + achieves. + +### Reserving CPU for the control plane + +A saturated node can leave clone/wake execution and sandboxd itself nothing +to run on — under full-node load CH clone p95 has measured in the tens of +seconds. cocoon newer than v0.5.8 runs every VMM in its own cgroup v2 CPU +scope (Guaranteed-at-N: a VM's CPU count is also its hard host-CPU cap) and +adds `cgroup_cpus`, a one-time cpuset fence keeping the whole VM population — +including the virtio and io-wq worker threads vCPU affinity cannot reach — +off reserved host cores. On dense nodes set the fence in cocoon's config +(e.g. `cgroup_cpus: "0-379"` on a 384-core host) so the reserved cores stay +free for sandboxd, its cocoon invocations, and the OS; cocoon's CPU-isolation +docs cover validation semantics and the per-VM knobs. + ### Auth model Three token kinds. The root `api_token` has full access — operators and @@ -210,7 +249,11 @@ sandboxd -config /etc/sandboxd/config.json ``` On start the node reconciles: persisted claims whose VMs still run are -re-adopted, everything else `sbx-`-prefixed is removed. Then the refill loop +re-adopted, everything else `sbx-`-prefixed is removed. A record still in +cocoon's `creating` state is reclaimed through `vm reconcile-stale-create` +(cocoon ≥ v0.5.8), which refuses while a clone is in flight instead of +deleting the VM out from under it; older cocoons fall back to forced +removal. Then the refill loop builds one golden snapshot per pool (a one-time cold boot + snapshot export, tens of seconds) and keeps each pool topped up with claim-ready clones. `GET /v1/info` shows `"golden": true` and `warm` at target when the node is diff --git a/e2e/fakeengine_test.go b/e2e/fakeengine_test.go index d33124e..4823596 100644 --- a/e2e/fakeengine_test.go +++ b/e2e/fakeengine_test.go @@ -61,6 +61,10 @@ func (f *fakeEngine) Remove(_ context.Context, name string) error { return nil } +func (f *fakeEngine) ReconcileStaleCreate(context.Context, string) (engine.StaleCreateOutcome, error) { + return engine.StaleCreateNotCreating, nil +} + func (f *fakeEngine) SnapshotSave(_ context.Context, _, _ string) error { return nil } func (f *fakeEngine) SnapshotExport(_ context.Context, _, toDir string) error { diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index ffdaf17..8dda469 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -713,8 +713,7 @@ func fastBulk(tag string, slow func([]byte) (Response, error), mk func([]byte) R // AppendBulkRequest renders a data-carrying request frame — // {"v":1,"op":,"data":""} plus newline — into buf, reused across -// calls: the zero-alloc twin of EncodeRequest for the bulk send paths -// (base64's alphabet needs no JSON escaping). +// calls on the bulk send paths (base64's alphabet needs no JSON escaping). func AppendBulkRequest(buf []byte, op string, data []byte) []byte { buf = append(buf[:0], requestHead...) buf = append(buf, op...) diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index c07d072..fb645f8 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -32,6 +32,12 @@ const ( // RequiredCocoon carries the snapshot/store performance baseline. RequiredCocoon = "v0.5.2" + // Stale-create verdicts, mirroring cocoon's reconcile-stale-create verb. + StaleCreateCollected StaleCreateOutcome = "collected" + StaleCreateBusy StaleCreateOutcome = "busy" + StaleCreateNotCreating StaleCreateOutcome = "not-creating" + StaleCreateNotFound StaleCreateOutcome = "not-found" + argName = "--name" argOutput = "--output" argNetwork = "--network" @@ -59,6 +65,9 @@ var capacitySignatures = []string{ "no space left on device", } +// StaleCreateOutcome reports what reconcile-stale-create did with a record. +type StaleCreateOutcome string + // Engine runs cocoon commands on the local node. type Engine struct { bin string @@ -141,6 +150,22 @@ func (e *Engine) Remove(ctx context.Context, name string) error { return err } +// ReconcileStaleCreate reclaims a creating-state record via cocoon's +// free-ops-lock predicate — safe against an in-flight clone where rm --force is not. +func (e *Engine) ReconcileStaleCreate(ctx context.Context, name string) (StaleCreateOutcome, error) { + out, err := e.run(ctx, "vm", "reconcile-stale-create", name, argOutput, formatJSON) + if err != nil { + return "", err + } + var res struct { + Outcome StaleCreateOutcome `json:"outcome"` + } + if err := json.Unmarshal(out, &res); err != nil { + return "", fmt.Errorf("parse reconcile-stale-create: %w", err) + } + return res.Outcome, nil +} + // SnapshotSave snapshots a running VM under snapName. func (e *Engine) SnapshotSave(ctx context.Context, vmName, snapName string) error { _, err := e.run(ctx, "snapshot", "save", argName, snapName, vmName) diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index c34d919..090cc55 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -67,12 +67,7 @@ func (m *Manager) archiveOnce(ctx context.Context) { logger := log.WithFunc("pool.archiveOnce") m.runBounded(ctx, len(victims), func(ctx context.Context, i int) { sb := victims[i] - switch err := m.archive(ctx, sb); { - case err == nil: - logger.Infof(ctx, "archived %s", sb.ID) - case !benignSweepErr(err): - logger.Errorf(ctx, err, "archive %s", sb.ID) - } + logSweepResult(ctx, logger, m.archive(ctx, sb), "archived "+sb.ID, "archive "+sb.ID) }).Wait() }() } diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index e57ac6e..a83c497 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -363,12 +363,7 @@ func (m *Manager) reapOnce(ctx context.Context) { m.purgeArchiveCk(ctx, v.id, v.ck, v.tenant) logger.Infof(ctx, "purged archived sandbox %s", v.id) case reapArchive: - switch err := m.archive(ctx, v.sb); { - case err == nil: - logger.Infof(ctx, "archived expired sandbox %s", v.id) - case !benignSweepErr(err): - logger.Errorf(ctx, err, "archive expired %s", v.id) - } + logSweepResult(ctx, logger, m.archive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) default: m.disarmEgress(v.id, m.removeOrRetry(ctx, v.vmName, v.id, "")) m.dropSnap(ctx, v.snap) diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index a9228e1..b9231c7 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -220,12 +220,7 @@ func (m *Manager) idleOnce(ctx context.Context) { logger := log.WithFunc("pool.idleOnce") m.runBounded(ctx, len(victims), func(ctx context.Context, i int) { v := victims[i] - switch err := m.idleHibernate(ctx, v.id, v.token, now); { - case err == nil: - logger.Infof(ctx, "idle-hibernated %s", v.id) - case !benignSweepErr(err): - logger.Errorf(ctx, err, "idle-hibernate %s", v.id) - } + logSweepResult(ctx, logger, m.idleHibernate(ctx, v.id, v.token, now), "idle-hibernated "+v.id, "idle-hibernate "+v.id) }).Wait() }() } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index f7f213d..00981e3 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -26,6 +26,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/egress" + "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/netfilter" "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/store/dir" @@ -73,6 +74,7 @@ const ( hibernatePrefix = "sbx-hib-" forkPrefix = "sbx-fork-" vmStateRunning = "running" + vmStateCreating = "creating" caSidecarSuffix = ".cafp" ) @@ -100,6 +102,7 @@ type Engine interface { CloneSnap(ctx context.Context, snap, name string, key types.PoolKey) (types.VMRecord, error) RunCold(ctx context.Context, name string, key types.PoolKey) (types.VMRecord, error) Remove(ctx context.Context, name string) error + ReconcileStaleCreate(ctx context.Context, name string) (engine.StaleCreateOutcome, error) SnapshotSave(ctx context.Context, vmName, snapName string) error SnapshotExport(ctx context.Context, snapName, toDir string) error SnapshotRemove(ctx context.Context, snapName string) error @@ -149,8 +152,9 @@ type Gauges struct { } type pendingRemoval struct { - sandboxID string - tap string + sandboxID string + tap string + staleCreate bool } type pool struct { @@ -685,6 +689,16 @@ func tenantOwns(tenant, owner string) bool { // benignSweepErr reports whether err is the expected outcome of a housekeeping // sweep (victim released, woke mid-sweep, or a lane that never hibernates). +// logSweepResult reports one background-sweep outcome; benign races stay silent. +func logSweepResult(ctx context.Context, logger *log.Fields, err error, okMsg, failMsg string) { + switch { + case err == nil: + logger.Info(ctx, okMsg) + case !benignSweepErr(err): + logger.Error(ctx, err, failMsg) + } +} + func benignSweepErr(err error) bool { return errors.Is(err, ErrUnknownSandbox) || errors.Is(err, errWokeMeanwhile) || errors.Is(err, ErrNoEgressHibernate) diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 59b0c80..fd2d855 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -16,6 +16,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/egress" + "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" ) @@ -765,8 +766,12 @@ type fakeEngine struct { hibernates, restores, snapRemoves []string snapSaves, exports, snapshots []string caInstalls []string // vsock sockets InstallCACert was called on + staleReconciles []string // VM names ReconcileStaleCreate was called on installCAErr error stopped map[string]bool + creating map[string]bool // VMs List reports in the creating state + staleOutcome engine.StaleCreateOutcome + staleErr error pids map[string]int // VM name → PID, for Stats' resident-set lookup vsockLateN int // List calls that report no socket yet @@ -787,7 +792,7 @@ type fakeEngine struct { } func newFakeEngine() *fakeEngine { - return &fakeEngine{vms: map[string]string{}, stopped: map[string]bool{}, pids: map[string]int{}} + return &fakeEngine{vms: map[string]string{}, stopped: map[string]bool{}, creating: map[string]bool{}, pids: map[string]int{}} } func (f *fakeEngine) Clone(_ context.Context, fromDir, name string, _ types.PoolKey) (types.VMRecord, error) { @@ -830,6 +835,24 @@ func (f *fakeEngine) Remove(ctx context.Context, name string) error { return nil } +func (f *fakeEngine) ReconcileStaleCreate(ctx context.Context, name string) (engine.StaleCreateOutcome, error) { + // Models exec.CommandContext: a canceled ctx never runs cocoon at all. + if err := ctx.Err(); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + f.staleReconciles = append(f.staleReconciles, name) + if f.staleErr != nil { + return "", f.staleErr + } + if f.staleOutcome == engine.StaleCreateCollected || f.staleOutcome == engine.StaleCreateNotFound { + delete(f.vms, name) + delete(f.creating, name) + } + return f.staleOutcome, nil +} + func (f *fakeEngine) SnapshotSave(_ context.Context, _, snapName string) error { f.mu.Lock() defer f.mu.Unlock() @@ -918,7 +941,10 @@ func (f *fakeEngine) List(_ context.Context, filters ...string) ([]types.VMRecor } sock = f.lateVsock(sock) state := vmStateRunning - if f.stopped[name] { + switch { + case f.creating[name]: + state = vmStateCreating + case f.stopped[name]: state = "stopped" } rec := types.VMRecord{State: state, PID: f.pids[name], VsockSocket: sock, Config: types.VMConfig{Name: name}} diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index 4edc817..d069e67 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -12,6 +12,7 @@ import ( "github.com/projecteru2/core/log" + "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/netfilter" "github.com/cocoonstack/sandbox/sandboxd/types" ) @@ -131,7 +132,7 @@ func (m *Manager) sweepStaleVMs(ctx context.Context, live map[string]types.VMRec } gone := make([]bool, len(stale)) // distinct indices: no lock under the Wait barrier m.runBounded(ctx, len(stale), func(ctx context.Context, i int) { - if m.removeOrRetry(ctx, stale[i], "", live[stale[i]].TapDevice()) { + if m.removeStaleVM(ctx, stale[i], live[stale[i]]) { gone[i] = true logger.Infof(ctx, "removed stale VM %s", stale[i]) } @@ -145,6 +146,31 @@ func (m *Manager) sweepStaleVMs(ctx context.Context, live map[string]types.VMRec return removed } +// removeStaleVM reclaims one unowned VM, reporting whether it is gone; a +// creating-state record goes through reconcile-stale-create first, since +// rm --force would queue on the ops lock and then delete the VM an +// in-flight clone just produced. +func (m *Manager) removeStaleVM(ctx context.Context, name string, rec types.VMRecord) bool { + logger := log.WithFunc("pool.removeStaleVM") + if rec.State == vmStateCreating { + // Cancellation-immune like removeVM: a canceled ctx must not skip + // the busy check and fall through to the forced remove it guards. + switch outcome, err := m.eng.ReconcileStaleCreate(context.WithoutCancel(ctx), name); { + case err != nil: + // Verb missing (cocoon < v0.5.8) or failed: keep the old sweep. + logger.Warnf(ctx, "reconcile stale create %s: %v; removing", name, err) + case outcome == engine.StaleCreateCollected, outcome == engine.StaleCreateNotFound: + return true + case outcome == engine.StaleCreateBusy: + logger.Infof(ctx, "stale create %s has an in-flight owner; queued for retry", name) + m.queueStaleCreate(name, rec.TapDevice()) + return false + } + // not-creating: the record moved on under the lock; remove normally. + } + return m.removeOrRetry(ctx, name, "", rec.TapDevice()) +} + // resyncEgress re-locks adopted egress claims after a restart, quarantines any // it cannot lock, and sweeps tables orphaned by VMs confirmed gone (in removed). func (m *Manager) resyncEgress(ctx context.Context, live map[string]types.VMRecord, removed map[string]bool) { diff --git a/sandboxd/pool/reconcile_test.go b/sandboxd/pool/reconcile_test.go new file mode 100644 index 0000000..96ce2b1 --- /dev/null +++ b/sandboxd/pool/reconcile_test.go @@ -0,0 +1,164 @@ +package pool + +import ( + "context" + "errors" + "slices" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/engine" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestReconcileStaleCreateSweep(t *testing.T) { + tests := []staleCreateCase{ + {"collected frees the record", engine.StaleCreateCollected, nil, false, false, false}, + {"not-found treats the record as gone", engine.StaleCreateNotFound, nil, false, false, false}, + {"busy leaves the in-flight clone", engine.StaleCreateBusy, nil, true, false, true}, + {"not-creating removes normally", engine.StaleCreateNotCreating, nil, false, true, false}, + {"verb failure falls back to remove", "", errors.New("unknown command"), false, true, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + eng := newFakeEngine() + eng.vms["sbx-stale-1"] = "/vsock/sbx-stale-1" + eng.creating["sbx-stale-1"] = true + eng.staleOutcome, eng.staleErr = tt.outcome, tt.err + m := newTestManager(t, eng) + + if err := m.Reconcile(t.Context()); err != nil { + t.Fatalf("Reconcile: %v", err) + } + if !slices.Contains(eng.staleReconciles, "sbx-stale-1") { + t.Fatal("reconcile-stale-create not attempted for the creating VM") + } + eng.mu.Lock() + _, present := eng.vms["sbx-stale-1"] + eng.mu.Unlock() + if present != tt.wantPresent { + t.Errorf("present=%v, want %v", present, tt.wantPresent) + } + if eng.removed("sbx-stale-1") != tt.wantRemove { + t.Errorf("removed=%v, want %v", eng.removed("sbx-stale-1"), tt.wantRemove) + } + m.mu.Lock() + _, pending := m.pendingRemovals["sbx-stale-1"] + m.mu.Unlock() + if pending != tt.wantPending { + t.Errorf("pending=%v, want %v", pending, tt.wantPending) + } + }) + } +} + +func TestBusyStaleCreateRetryConverges(t *testing.T) { + tests := []staleCreateCase{ + {"collected", engine.StaleCreateCollected, nil, false, false, false}, + {"not-found", engine.StaleCreateNotFound, nil, false, false, false}, + {"not-creating", engine.StaleCreateNotCreating, nil, false, true, false}, + {"still busy", engine.StaleCreateBusy, nil, true, false, true}, + {"verb failure", "", errors.New("temporary failure"), true, false, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + eng := newFakeEngine() + eng.vms["sbx-stale-1"] = "/vsock/sbx-stale-1" + eng.creating["sbx-stale-1"] = true + eng.staleOutcome = engine.StaleCreateBusy + m := newTestManager(t, eng) + + if err := m.Reconcile(t.Context()); err != nil { + t.Fatalf("Reconcile: %v", err) + } + eng.mu.Lock() + eng.staleOutcome, eng.staleErr = tt.outcome, tt.err + eng.mu.Unlock() + m.retryRemovals(t.Context()).Wait() + + eng.mu.Lock() + _, present := eng.vms["sbx-stale-1"] + eng.mu.Unlock() + m.mu.Lock() + _, pending := m.pendingRemovals["sbx-stale-1"] + m.mu.Unlock() + removed := eng.removed("sbx-stale-1") + if present != tt.wantPresent || removed != tt.wantRemove || pending != tt.wantPending { + t.Errorf("present=%v removed=%v pending=%v, want %v %v %v", + present, removed, pending, + tt.wantPresent, tt.wantRemove, tt.wantPending) + } + }) + } +} + +func TestReconcileBusyStaleCreateKeepsTap(t *testing.T) { + eng := newFakeEngine() + eng.tap = "tap-busy" + eng.vms["sbx-stale-1"] = "/vsock/sbx-stale-1" + eng.creating["sbx-stale-1"] = true + eng.staleOutcome = engine.StaleCreateBusy + m := newTestManager(t, eng) + + var gotKeep map[string]bool + m.sweep = func(keep map[string]bool) error { gotKeep = keep; return nil } + if err := m.Reconcile(t.Context()); err != nil { + t.Fatalf("Reconcile: %v", err) + } + if !gotKeep["tap-busy"] { + t.Error("busy VM's tap not kept; the sweep would tear down an in-flight clone's tables") + } + m.mu.Lock() + pending, ok := m.pendingRemovals["sbx-stale-1"] + m.mu.Unlock() + if !ok || !pending.staleCreate || pending.tap != "tap-busy" { + t.Errorf("pending=%+v present=%v, want stale-create retry with tap-busy", pending, ok) + } +} + +func TestRemoveStaleVMCanceledCtxStillChecksBusy(t *testing.T) { + eng := newFakeEngine() + eng.vms["sbx-stale-1"] = "/vsock/sbx-stale-1" + eng.creating["sbx-stale-1"] = true + eng.staleOutcome = engine.StaleCreateBusy + m := newTestManager(t, eng) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + rec := types.VMRecord{State: vmStateCreating, Config: types.VMConfig{Name: "sbx-stale-1"}} + if m.removeStaleVM(ctx, "sbx-stale-1", rec) { + t.Error("busy VM reported gone under a canceled ctx") + } + if len(eng.staleReconciles) != 1 { + t.Errorf("verb ran %d times, want 1 (cancellation-immune)", len(eng.staleReconciles)) + } + if eng.removed("sbx-stale-1") { + t.Error("canceled ctx fell through to the forced remove the verb guards") + } +} + +func TestReconcileStaleRunningVMSkipsVerb(t *testing.T) { + eng := newFakeEngine() + eng.vms["sbx-orphan-1"] = "/vsock/sbx-orphan-1" + m := newTestManager(t, eng) + + if err := m.Reconcile(t.Context()); err != nil { + t.Fatalf("Reconcile: %v", err) + } + if len(eng.staleReconciles) != 0 { + t.Errorf("reconcile-stale-create ran on a running VM: %v", eng.staleReconciles) + } + if !eng.removed("sbx-orphan-1") { + t.Error("unowned running VM not removed") + } +} + +// staleCreateCase drives both the startup sweep and the reap-tick retry +// through the same outcome/end-state row shape. +type staleCreateCase struct { + name string + outcome engine.StaleCreateOutcome + err error + wantPresent bool + wantRemove bool + wantPending bool +} diff --git a/sandboxd/pool/remove.go b/sandboxd/pool/remove.go index 122ebff..1fae125 100644 --- a/sandboxd/pool/remove.go +++ b/sandboxd/pool/remove.go @@ -9,6 +9,7 @@ import ( "github.com/projecteru2/core/log" + "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/netfilter" ) @@ -58,6 +59,12 @@ func (m *Manager) queueRemoval(name, sandboxID, tap string) { m.mu.Unlock() } +func (m *Manager) queueStaleCreate(name, tap string) { + m.mu.Lock() + m.pendingRemovals[name] = pendingRemoval{tap: tap, staleCreate: true} + m.mu.Unlock() +} + // retryRemovals dispatches by emptying the queue, so absence also means "in // flight" and the next tick cannot double-dispatch; a failed retry re-queues // itself on completion. Callers that need completion Wait. @@ -77,10 +84,28 @@ func (m *Manager) retryRemovals(ctx context.Context) *sync.WaitGroup { } func (m *Manager) retryRemoval(ctx context.Context, name string, pending pendingRemoval) { + if pending.staleCreate { + switch outcome, err := m.eng.ReconcileStaleCreate(ctx, name); { + case err != nil: + log.WithFunc("pool.retryRemoval").Warnf(ctx, "reconcile stale create %s: %v; retrying", name, err) + m.queueStaleCreate(name, pending.tap) + return + case outcome == engine.StaleCreateBusy: + m.queueStaleCreate(name, pending.tap) + return + case outcome == engine.StaleCreateCollected, outcome == engine.StaleCreateNotFound: + m.finishRemoval(pending) + return + } + } if !m.removeVM(ctx, name) { m.queueRemoval(name, pending.sandboxID, pending.tap) return } + m.finishRemoval(pending) +} + +func (m *Manager) finishRemoval(pending pendingRemoval) { if pending.sandboxID != "" { m.disarmEgress(pending.sandboxID, true) } else if pending.tap != "" { diff --git a/sandboxd/store/peer/peer.go b/sandboxd/store/peer/peer.go index 0a2688e..a439618 100644 --- a/sandboxd/store/peer/peer.go +++ b/sandboxd/store/peer/peer.go @@ -57,7 +57,6 @@ func (h *Healer) Pull(ctx context.Context, id, staging string, validate Validate return h.pullFrom(ctx, id, staging, addrs, budget, validate) } -// pullFrom tries each owner in turn, giving each an even slice of budget. func (h *Healer) pullFrom(ctx context.Context, id, staging string, addrs []string, budget time.Duration, validate Validate) error { perOwner := budget / time.Duration(len(addrs)) var errs []error diff --git a/sandboxd/store/peer/peer_test.go b/sandboxd/store/peer/peer_test.go index 50f1554..db03dfa 100644 --- a/sandboxd/store/peer/peer_test.go +++ b/sandboxd/store/peer/peer_test.go @@ -98,13 +98,9 @@ func TestHealErrorIsReported(t *testing.T) { } } -// TestPullDoesNotDedupConcurrentCalls: Pull used to share one singleflight -// result across concurrent callers, which was only safe because every -// caller happened to pass the SAME staging dir. With independent dirs (the -// only safe assumption once dedup is not Pull's job), sharing a result would -// leave whichever caller did not trigger the shared call with an untouched, -// empty directory published as if it were a real record. Two concurrent -// Pulls for one id but different dirs must each run their own full transfer. +// TestPullDoesNotDedupConcurrentCalls: concurrent Pulls for one id but +// different staging dirs must each run their own transfer — sharing a result +// would leave one caller's dir empty but published as real. func TestPullDoesNotDedupConcurrentCalls(t *testing.T) { puller := &fakePuller{records: map[string]map[string]string{"peer-a:7777": record()}} h := NewHealer(func(string) []string { return []string{"peer-a:7777"} }, puller) diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index e79a3a4..524f9e1 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -52,7 +52,6 @@ type HTTPPuller struct { Token string } -// Pull implements Puller. func (p *HTTPPuller) Pull(ctx context.Context, addr, id, dst string) error { ctx, cancel := context.WithTimeout(ctx, pullTimeout) defer cancel() diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index b8dbdf4..3e3d28a 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -3,7 +3,6 @@ package sandbox import ( "bytes" "context" - "fmt" "net/http" "time" ) @@ -35,28 +34,15 @@ func (ck *Checkpoint) New(ctx context.Context, opts ...Option) (*Sandbox, error) if err := claim.rejectPinnedAxes(); err != nil { return nil, err } - body, err := encodeBody("checkpoint claim", checkpointClaimRequest{TTLSeconds: claim.TTLSeconds}) - if err != nil { - return nil, err - } - cr, err := ck.claimAt(ctx, ck.addr, body) - if err != nil { - return nil, err - } - if len(cr.Redirect) == 0 { - return ck.c.handleFrom(ck.addr, cr), nil - } - body, err = encodeBody("checkpoint claim", checkpointClaimRequest{TTLSeconds: claim.TTLSeconds, NoRedirect: true}) - if err != nil { - return nil, err - } - addr, target, err := redirectFallback(ck.addr, cr.Redirect, func(a string) (claimResponse, error) { + addr, cr, err := claimFollow(ck.addr, "claim checkpoint", func(noRedirect bool) ([]byte, error) { + return encodeBody("checkpoint claim", checkpointClaimRequest{TTLSeconds: claim.TTLSeconds, NoRedirect: noRedirect}) + }, func(a string, body []byte) (claimResponse, error) { return ck.claimAt(ctx, a, body) }) if err != nil { - return nil, fmt.Errorf("claim checkpoint: %w", err) + return nil, err } - return ck.c.handleFrom(addr, target), nil + return ck.c.handleFrom(addr, cr), nil } // Delete removes the checkpoint from its node and asks every peer that node diff --git a/sdk/go/checkpoint_test.go b/sdk/go/checkpoint_test.go index 46a5495..4f3f0d2 100644 --- a/sdk/go/checkpoint_test.go +++ b/sdk/go/checkpoint_test.go @@ -124,10 +124,8 @@ func TestCheckpointNewRedirectFallbackHeals(t *testing.T) { } } -// TestCheckpointNewRedirectNeverYieldsEmptyID is a regression test: New used -// to hand a bare redirect reply straight to handleFrom, producing a Sandbox -// with no id/token. It must now follow the redirect and fail loudly when no -// candidate answers, never return a Sandbox with an empty ID. +// TestCheckpointNewRedirectNeverYieldsEmptyID: New must fail loudly, not +// return a Sandbox with an empty ID, when no redirect candidate answers. func TestCheckpointNewRedirectNeverYieldsEmptyID(t *testing.T) { entry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _ = json.NewEncoder(w).Encode(claimResponse{Redirect: []string{"127.0.0.1:1"}}) diff --git a/sdk/go/client.go b/sdk/go/client.go index 5da411f..d6c339e 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -46,30 +46,16 @@ func (c *Client) New(ctx context.Context, template string, opts ...Option) (*San for _, opt := range opts { opt(&claim) } - body, err := encodeBody("claim", claim) - if err != nil { - return nil, err - } - - cr, err := c.claimAt(ctx, c.addr, body) - if err != nil { - return nil, err - } - if len(cr.Redirect) == 0 { - return c.handleFrom(c.addr, cr), nil - } - claim.NoRedirect = true - body, err = encodeBody("claim", claim) - if err != nil { - return nil, err - } - addr, target, err := redirectFallback(c.addr, cr.Redirect, func(a string) (claimResponse, error) { + addr, cr, err := claimFollow(c.addr, "claim", func(noRedirect bool) ([]byte, error) { + claim.NoRedirect = noRedirect + return encodeBody("claim", claim) + }, func(a string, body []byte) (claimResponse, error) { return c.claimAt(ctx, a, body) }) if err != nil { - return nil, fmt.Errorf("claim: %w", err) + return nil, err } - return c.handleFrom(addr, target), nil + return c.handleFrom(addr, cr), nil } // Lookup relocates a sandbox handle whose owner address was lost, given its @@ -222,6 +208,14 @@ func doJSON[T any](ctx context.Context, c *Client, method, addr, path string, bo return out, nil } +func doJSONPtr[T any](ctx context.Context, c *Client, method, addr, path string, body io.Reader, bearer, verb string) (*T, error) { + out, err := doJSON[T](ctx, c, method, addr, path, body, bearer, verb) + if err != nil { + return nil, err + } + return &out, nil +} + // doNoContent is doJSON's reply-less twin: 204 or an apiError. func doNoContent(ctx context.Context, c *Client, method, addr, path string, body io.Reader, bearer, verb string) error { resp, err := c.roundTrip(ctx, method, addr, path, body, bearer) @@ -295,6 +289,33 @@ func retryTransient(err error) bool { // an auth rejection, a conflict) skips the fallback: origin would fail the // same way. A second-level redirect (a compliant server never sends one // once no_redirect is set) fails the candidate rather than being followed. +// claimFollow runs the claim protocol from origin: claim there, and on a +// redirect re-encode with no_redirect and follow via redirectFallback. Only +// the fallback error carries the verb — first-contact errors return raw. +func claimFollow(origin, verb string, encode func(noRedirect bool) ([]byte, error), claimAt func(addr string, body []byte) (claimResponse, error)) (string, claimResponse, error) { + body, err := encode(false) + if err != nil { + return "", claimResponse{}, err + } + cr, err := claimAt(origin, body) + if err != nil { + return "", claimResponse{}, err + } + if len(cr.Redirect) == 0 { + return origin, cr, nil + } + if body, err = encode(true); err != nil { + return "", claimResponse{}, err + } + addr, target, err := redirectFallback(origin, cr.Redirect, func(a string) (claimResponse, error) { + return claimAt(a, body) + }) + if err != nil { + return "", claimResponse{}, fmt.Errorf("%s: %w", verb, err) + } + return addr, target, nil +} + func redirectFallback(origin string, candidates []string, claimAt func(addr string) (claimResponse, error)) (string, claimResponse, error) { claimNoRedirect := func(target string) (claimResponse, error) { cr, err := claimAt(target) diff --git a/sdk/go/drain.go b/sdk/go/drain.go index 6a5550f..a5246f3 100644 --- a/sdk/go/drain.go +++ b/sdk/go/drain.go @@ -8,18 +8,10 @@ import ( // Drain cordons the entry node (root token): new claims/forks/branches are // refused, live claims run to their leases; poll Info until Claimed is zero. func (c *Client) Drain(ctx context.Context) (*NodeInfo, error) { - info, err := doJSON[NodeInfo](ctx, c, http.MethodPost, c.addr, "/v1/drain", nil, c.apiToken, "drain") - if err != nil { - return nil, err - } - return &info, nil + return doJSONPtr[NodeInfo](ctx, c, http.MethodPost, c.addr, "/v1/drain", nil, c.apiToken, "drain") } // Uncordon lifts a drain on the entry node (root token). func (c *Client) Uncordon(ctx context.Context) (*NodeInfo, error) { - info, err := doJSON[NodeInfo](ctx, c, http.MethodDelete, c.addr, "/v1/drain", nil, c.apiToken, "uncordon") - if err != nil { - return nil, err - } - return &info, nil + return doJSONPtr[NodeInfo](ctx, c, http.MethodDelete, c.addr, "/v1/drain", nil, c.apiToken, "uncordon") } diff --git a/sdk/go/files_test.go b/sdk/go/files_test.go index 5acf2cb..df56365 100644 --- a/sdk/go/files_test.go +++ b/sdk/go/files_test.go @@ -93,8 +93,6 @@ func TestSessionLifecycle(t *testing.T) { } } -// fakeSandbox wires a Sandbox whose data plane is served by a temp-dir-backed -// silkd Fake behind a hijacking agent endpoint. func TestReadFileMultiChunk(t *testing.T) { sb := fakeSandbox(t) ctx := t.Context() @@ -111,6 +109,8 @@ func TestReadFileMultiChunk(t *testing.T) { } } +// fakeSandbox wires a Sandbox whose data plane is served by a temp-dir-backed +// silkd Fake behind a hijacking agent endpoint. func fakeSandbox(t *testing.T) *Sandbox { t.Helper() fake := silkdtest.NewFake(t.TempDir()) diff --git a/sdk/go/info.go b/sdk/go/info.go index e294ffb..9e672d7 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -38,11 +38,7 @@ type PoolStatus struct { // Info reports the entry node's pools, claim counts, and mesh peers. func (c *Client) Info(ctx context.Context) (*NodeInfo, error) { - info, err := doJSON[NodeInfo](ctx, c, http.MethodGet, c.addr, "/v1/info", nil, c.apiToken, "info") - if err != nil { - return nil, err - } - return &info, nil + return doJSONPtr[NodeInfo](ctx, c, http.MethodGet, c.addr, "/v1/info", nil, c.apiToken, "info") } // peers fetches the cluster's node addresses, best-effort (nil on failure). diff --git a/sdk/go/pools.go b/sdk/go/pools.go index 9395ca5..4f61767 100644 --- a/sdk/go/pools.go +++ b/sdk/go/pools.go @@ -73,9 +73,5 @@ func (c *Client) setPoolsAt(ctx context.Context, addr string, pools []PoolSpec) if err != nil { return nil, err } - info, err := doJSON[NodeInfo](ctx, c, http.MethodPut, addr, "/v1/pools", bytes.NewReader(body), c.apiToken, "pools") - if err != nil { - return nil, err - } - return &info, nil + return doJSONPtr[NodeInfo](ctx, c, http.MethodPut, addr, "/v1/pools", bytes.NewReader(body), c.apiToken, "pools") } diff --git a/sdk/go/pty_test.go b/sdk/go/pty_test.go index 5aee3b3..389b6fe 100644 --- a/sdk/go/pty_test.go +++ b/sdk/go/pty_test.go @@ -31,7 +31,6 @@ func TestPtyEchoAndExit(t *testing.T) { t.Errorf("read %q, want ping", line) } - // Resize is a separate RPC. if err := pty.Resize(t.Context(), 120, 40); err != nil { t.Fatalf("resize: %v", err) } diff --git a/sdk/go/silkd/silkdtest/fake.go b/sdk/go/silkd/silkdtest/fake.go index 4991762..aaa99cf 100644 --- a/sdk/go/silkd/silkdtest/fake.go +++ b/sdk/go/silkd/silkdtest/fake.go @@ -19,13 +19,13 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) +// readChunk mirrors silkd's BULK_CHUNK so downloads exercise real framing. +const readChunk = 256 * 1024 + // Fake is a stateful silkd fake backing the fs verbs with a real directory // and tracking sessions, so an SDK write-then-read round-trips through it. // exec/info reuse the stateless handlers. It exists for host-side unit tests; // the authoritative fs/session behavior is silkd's own Rust test suite. -// readChunk mirrors silkd's BULK_CHUNK so downloads exercise real framing. -const readChunk = 256 * 1024 - type Fake struct { Root string diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index 2d8d31b..234dddf 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -1,6 +1,7 @@ //! Small OS helpers: the base environment for spawned commands, best-effort -//! user de-escalation, and the one signal syscall — all the crate's unsafe -//! lives here. +//! user de-escalation, and the one signal syscall — the crate's unsafe work +//! lives here (pty.rs holds the one other unsafe block, its pre_exec +//! registration). use std::fmt::Write as _; use std::io::Read;