Fix the three refill-path stalls that cap a large warm-pool fill

sandboxd/engine/capacity_test.go

@@ -0,0 +1,28 @@
+package engine
+
+import (
+	"errors"
+	"testing"
+)
+
+// TestOnlyNodeWideFailuresCarryACapacitySignature guards the classifier.
+// Parking on an ordinary clone failure would stall a healthy node over one bad VM.
+func TestOnlyNodeWideFailuresCarryACapacitySignature(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"bridge port exhaustion", errors.New(`failed to connect "veth1" to bridge cni0: exchange full`), true},
+		{"disk exhaustion", errors.New("write overlay: no space left on device"), true},
+		{"one VM failed to boot", errors.New("cocoon vm clone: exit status 1: Error: guest did not respond"), false},
+		{"missing golden", errors.New("cocoon vm clone: exit status 1: Error: snapshot not found"), false},
+		{"no error", nil, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := CapacitySignature(tc.err) != ""; got != tc.want {
+				t.Errorf("CapacitySignature(%v) carries a signature: %v, want %v", tc.err, got, tc.want)
+			}
+		})
+	}
+}

sandboxd/engine/engine.go

@@ -47,6 +47,15 @@ const (
 	portForwardMax = 4096
 )
 
+// capacitySignatures mean "this node cannot attach another VM", not "this VM
+// failed"; matched as text because cocoon is a subprocess and the distinction
+// survives only in its stderr. "exchange full" is EXFULL from a bridge at
+// BR_MAX_PORTS — permanent until VMs are removed.
+var capacitySignatures = []string{
+	"exchange full",
+	"no space left on device",
+}
+
 // Engine runs cocoon commands on the local node.
 type Engine struct {
 	bin         string
@@ -463,6 +472,20 @@ func readLine(conn net.Conn, max int) (string, error) {
 	return "", fmt.Errorf("reply exceeds %d bytes", max)
 }
 
+// CapacitySignature returns the node-capacity signature err carries, or "".
+func CapacitySignature(err error) string {
+	if err == nil {
+		return ""
+	}
+	msg := strings.ToLower(err.Error())
+	for _, sig := range capacitySignatures {
+		if strings.Contains(msg, sig) {
+			return sig
+		}
+	}
+	return ""
+}
+
 func tail(s string) string {
 	s = strings.TrimSpace(s)
 	if len(s) <= outputTail {

sandboxd/pool/archive.go

@@ -141,9 +141,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error {
 	m.disarmEgress(sb.ID, true)
 	sb.Transition.Unlock()
 	// Committed: the store ck is authoritative now; reclaim the local footprint.
-	if rmErr := m.eng.Remove(ctx, vmName); rmErr != nil {
-		log.WithFunc("pool.archive").Warnf(ctx, "remove archived VM %s: %v", vmName, rmErr)
-	}
+	m.destroy(ctx, vmName)
 	m.dropSnap(ctx, snap)
 	m.counters.archives.Add(1)
 	m.recordUsage(ctx, usageEvent{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant})

sandboxd/pool/claim.go

@@ -129,8 +129,8 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand
 		m.purgeArchiveCk(ctx, id, ck, sb.Tenant) // archived: no local VM
 	}
 	var err error
-	if vmName != "" {
-		err = m.eng.Remove(ctx, vmName)
+	if vmName != "" && !m.removeOrRetry(ctx, vmName, id, "") {
+		err = fmt.Errorf("vm %s survived removal", vmName)
 	}
 	m.disarmEgress(id, err == nil)
 	m.dropSnap(ctx, snap)
@@ -291,7 +291,7 @@ func (m *Manager) rollbackClaim(ctx context.Context, sbs []*types.Sandbox) {
 	m.mu.Unlock()
 	m.recommit(ctx, rb)
 	for _, sb := range sbs {
-		m.disarmEgress(sb.ID, m.removeVM(ctx, sb.VMName))
+		m.disarmEgress(sb.ID, m.removeOrRetry(ctx, sb.VMName, sb.ID, ""))
 	}
 }
 
@@ -370,7 +370,7 @@ func (m *Manager) reapOnce(ctx context.Context) {
 				logger.Errorf(ctx, err, "archive expired %s", v.id)
 			}
 		default:
-			m.disarmEgress(v.id, m.removeVM(ctx, v.vmName))
+			m.disarmEgress(v.id, m.removeOrRetry(ctx, v.vmName, v.id, ""))
 			m.dropSnap(ctx, v.snap)
 			m.counters.reaps.Add(1)
 			m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id, VMName: v.vmName})

sandboxd/pool/egress.go

@@ -120,15 +120,14 @@ func (m *Manager) lockEgressNIC(ctx context.Context, sb *types.Sandbox) error {
 }
 
 func (m *Manager) tapOf(ctx context.Context, vmName string) (string, error) {
-	vms, err := m.eng.List(ctx, vmName)
+	vm, ok, err := m.findVM(ctx, vmName)
 	if err != nil {
 		return "", fmt.Errorf("list %s: %w", vmName, err)
 	}
-	i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == vmName })
-	if i < 0 {
+	if !ok {
 		return "", fmt.Errorf("no NIC config for %s", vmName)
 	}
-	if tap := vms[i].TapDevice(); tap != "" {
+	if tap := vm.TapDevice(); tap != "" {
 		return tap, nil
 	}
 	return "", fmt.Errorf("no tap for %s", vmName)

sandboxd/pool/egress_test.go

@@ -254,6 +254,9 @@ func TestBatchArmFailureRecordsNoUsage(t *testing.T) {
 func TestRollbackKeepsLockWhenRemoveFails(t *testing.T) {
 	eng := newFakeEngine()
 	eng.removeErrFor = "sbx-r1"
+	// Both VMs are running; sbx-r1's remove fails and the engine keeps listing it.
+	eng.vms["sbx-r1"] = "/run/sbx-r1.sock"
+	eng.vms["sbx-r2"] = "/run/sbx-r2.sock"
 	m := egressManager(t, eng, config.PoolSpec{PoolKey: egKey, Egress: egPolicy})
 	// Both members armed before a later batch member failed; sbx-r1's remove fails.
 	sbs := []*types.Sandbox{
@@ -287,6 +290,8 @@ func TestRollbackKeepsLockWhenRemoveFails(t *testing.T) {
 func TestQuarantineFailedRemoveStaysUnswept(t *testing.T) {
 	eng := newFakeEngine()
 	eng.removeErrFor = "sbx-q1"
+	// Running: removeVM reports "still here" only while the engine lists it.
+	eng.vms["sbx-q1"] = "/run/sbx-q1.sock"
 	m := egressManager(t, eng, config.PoolSpec{PoolKey: egKey, Egress: egPolicy})
 	sb := &types.Sandbox{ID: "sb_q1", VMName: "sbx-q1", Key: egKey, TAP: "tap-q1"}
 	m.mu.Lock()

sandboxd/pool/pool.go

@@ -35,11 +35,19 @@ import (
 )
 
 const (
-	refillInterval     = 2 * time.Second
-	reapInterval       = 5 * time.Second
-	buildRetryDelay    = 30 * time.Second
-	claimProbeTimeout  = 15 * time.Second
-	coldProbeTimeout   = 90 * time.Second
+	refillInterval    = 2 * time.Second
+	reapInterval      = 5 * time.Second
+	buildRetryDelay   = 30 * time.Second
+	claimProbeTimeout = 15 * time.Second
+	coldProbeTimeout  = 90 * time.Second
+	// A refill has no caller waiting and holds a target-accounting slot for
+	// its whole probe; a clone answers in ~1s even saturated, so a silent one
+	// is replaced, not waited on. Claims keep the generous deadline.
+	warmProbeTimeout = 5 * time.Second
+	// One list; a wrong answer only costs an extra sweep.
+	removeVerifyTimeout = 15 * time.Second
+	// A full node clears only when VMs go away; retrying sooner buys nothing.
+	capacityBackoff    = buildRetryDelay
 	vsockPollInterval  = 100 * time.Millisecond
 	defaultTTL         = 5 * time.Minute
 	maxTTL             = 24 * time.Hour
@@ -134,6 +142,15 @@ type Gauges struct {
 	Hibernated int
 	Archived   int
 	Draining   bool
+	// AtCapacity marks refill parked because the node refused another VM, so a
+	// placer reads a short warm count as "full", not "still filling".
+	AtCapacity       bool
+	AtCapacityReason string
+}
+
+type pendingRemoval struct {
+	sandboxID string
+	tap       string
 }
 
 type pool struct {
@@ -319,9 +336,15 @@ type Manager struct {
 	dial  egress.DialFunc
 	sweep func(map[string]bool) error
 
-	mu      sync.Mutex
-	pools   map[types.PoolKey]*pool
-	claimed map[string]*types.Sandbox
+	mu              sync.Mutex
+	pools           map[types.PoolKey]*pool
+	claimed         map[string]*types.Sandbox
+	pendingRemovals map[string]pendingRemoval
+
+	// Node-wide, unlike the per-pool streak backoff: the resource that ran
+	// out — bridge ports, disk — is shared by every pool.
+	atCapacityUntil  time.Time
+	atCapacityReason string
 
 	refillSem  chan struct{}
 	probeSem   chan struct{}
@@ -350,6 +373,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg
 		poolStore:       newPoolStore(cfg.DataDir),
 		pools:           make(map[types.PoolKey]*pool, len(cfg.Pools)),
 		claimed:         map[string]*types.Sandbox{},
+		pendingRemovals: map[string]pendingRemoval{},
 		tenantLive:      map[string]int{},
 		archiving:       map[string]struct{}{},
 		pendingCks:      map[string]struct{}{},
@@ -485,6 +509,7 @@ func (m *Manager) Run(ctx context.Context) {
 		case <-m.refillKick:
 			m.refillOnce(ctx)
 		case <-reap.C:
+			m.retryRemovals(ctx)
 			m.reapOnce(ctx)
 			m.idleOnce(ctx)
 			m.archiveOnce(ctx)
@@ -538,6 +563,9 @@ func (m *Manager) Info() ([]PoolInfo, Gauges) {
 	}
 	slices.SortFunc(pools, func(a, b PoolInfo) int { return strings.Compare(a.Key.Hash(), b.Key.Hash()) })
 	g := Gauges{Claimed: len(m.claimed), Draining: m.draining}
+	if now.Before(m.atCapacityUntil) {
+		g.AtCapacity, g.AtCapacityReason = true, m.atCapacityReason
+	}
 	for _, sb := range m.claimed {
 		if sb.HibernateSnap != "" {
 			g.Hibernated++

sandboxd/pool/reconcile.go

@@ -131,7 +131,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.removeVM(ctx, stale[i]) {
+		if m.removeOrRetry(ctx, stale[i], "", live[stale[i]].TapDevice()) {
 			gone[i] = true
 			logger.Infof(ctx, "removed stale VM %s", stale[i])
 		}
@@ -190,12 +190,9 @@ func (m *Manager) resyncEgress(ctx context.Context, live map[string]types.VMReco
 	}
 }
 
-// quarantineClaim removes a claim whose egress NIC could not be locked and drops
-// it from service; it returns whether the VM is confirmed gone (a failed remove
-// leaves it for the next restart's stale sweep). A never-applied lock plus a
-// failed remove leaves the VM unguarded until a later remove succeeds.
+// A failed remove stays out of service and queued until teardown succeeds.
 func (m *Manager) quarantineClaim(ctx context.Context, sb *types.Sandbox) bool {
-	gone := m.removeVM(ctx, sb.VMName)
+	gone := m.removeOrRetry(ctx, sb.VMName, sb.ID, "")
 	m.mu.Lock()
 	delete(m.claimed, sb.ID)
 	m.tenantDelta(sb.Tenant, -1)