From 1c00d926a108b57e128039893aae0b9de3012091 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:19:23 +0800 Subject: [PATCH 1/7] cut: answer a repeated adoption from the held claim instead of settling a second take handleOwner reaches AdoptArchived for a held claim only when its AgentSocket miss predates the flight that adopted it. A check under m.mu at the top of the flight answers that case before any registry write, so the duplicate branch and adoptHeld cannot run: no other path adds an id a live row names while the server serves. --- sandboxd/pool/failover.go | 29 ++++------------------- sandboxd/pool/failover_test.go | 43 ++++++++++------------------------ 2 files changed, 18 insertions(+), 54 deletions(-) diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index ba6f95d12..aee2e90fc 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -31,9 +31,13 @@ func (m *Manager) AdoptArchived(ctx context.Context, id, token string) (bool, er func (m *Manager) adoptArchived(ctx context.Context, id, token string) (bool, error) { m.mu.Lock() draining := m.draining + _, held := m.authed(id, token) m.mu.Unlock() - if draining { + switch { + case draining: return false, nil + case held: + return true, nil } ctx = context.WithoutCancel(ctx) rec, ok, err := m.archives.Get(ctx, id) @@ -61,10 +65,6 @@ func (m *Manager) adoptArchived(ctx context.Context, id, token string) (bool, er if sb.Deadline.IsZero() { sb.Deadline = m.retentionDeadline(sb.Key) } - if cur, dup := m.claimed[id]; dup { - m.mu.Unlock() - return m.adoptHeld(ctx, cur, epoch), nil - } m.addClaimLocked(sb) js := m.store.set(sb) m.mu.Unlock() @@ -77,25 +77,6 @@ func (m *Manager) adoptArchived(ctx context.Context, id, token string) (bool, er return true, nil } -func (m *Manager) adoptHeld(ctx context.Context, cur *types.Sandbox, epoch int64) bool { - cur.Transition.Lock() - defer cur.Transition.Unlock() - m.mu.Lock() - held := m.claimed[cur.ID] == cur - m.mu.Unlock() - if held { - _, _ = m.commitIfLive(ctx, cur, func() { cur.ArchiveEpoch = epoch }) - return true - } - if rec, ok, err := m.archives.Get(ctx, cur.ID); err == nil && ok && rec.Owner == m.nodeID && !rec.Consumed { - if consumed, err := m.archives.Consume(ctx, cur.ID, rec.Epoch); err == nil { - rec.Epoch, rec.Consumed = consumed, true - m.purgeArchiveRow(ctx, rec) - } - } - return false -} - func (m *Manager) putArchive(ctx context.Context, sb *types.Sandbox, ck string) (int64, error) { return m.putRow(ctx, sb, ck, false, false) } diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index fe4c6a6aa..099d01388 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -1206,29 +1206,18 @@ func TestFailoverARepeatedAdoptionOfAReleasedClaimIsRefused(t *testing.T) { t.Fatalf("adopt: %v %v", adopted, err) } cur, _ := b.byID(sb.ID) - before, _, _ := c.reg.Get(t.Context(), sb.ID) - - cur.Transition.Lock() - adopted := make(chan bool, 1) - go func() { - ok, err := b.AdoptArchived(t.Context(), sb.ID, sb.Token) - if err != nil { - t.Errorf("repeated adopt: %v", err) - } - adopted <- ok - }() - waitFor(t, func() bool { rec, _, _ := c.reg.Get(t.Context(), sb.ID); return rec.Epoch != before.Epoch }) - time.Sleep(50 * time.Millisecond) - b.forgetClaim(t.Context(), cur) - cur.Transition.Unlock() + ck := cur.ArchiveCk + if err := b.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatalf("release: %v", err) + } - if <-adopted { - t.Error("a repeated adoption answered for a claim released under it") + if adopted, err := b.AdoptArchived(t.Context(), sb.ID, sb.Token); err != nil || adopted { + t.Errorf("repeated adopt of a released claim: %v %v, want refused", adopted, err) } if rec, ok, _ := c.reg.Get(t.Context(), sb.ID); ok { t.Errorf("row %+v outlived the released claim", rec) } - if ckExists(t, b, cur.ArchiveCk) { + if ckExists(t, b, ck) { t.Error("the retired row's checkpoint outlived it") } } @@ -1385,25 +1374,19 @@ func TestFailoverAWakeKeepsItsFenceAgainstThisNodesOwnAdoption(t *testing.T) { if adopted, err := b.AdoptArchived(t.Context(), sb.ID, sb.Token); err != nil || !adopted { t.Fatalf("adopt: %v %v", adopted, err) } - second := make(chan bool, 1) b.archives = &hookRegistry{Registry: c.reg, beforeConsume: func() { taken, _, _ := c.reg.Get(t.Context(), sb.ID) - go func() { - ok, err := b.AdoptArchived(t.Context(), sb.ID, sb.Token) - if err != nil { - t.Errorf("second adopt: %v", err) - } - second <- ok - }() - waitFor(t, func() bool { rec, _, _ := c.reg.Get(t.Context(), sb.ID); return rec.Epoch != taken.Epoch }) + if adopted, err := b.AdoptArchived(t.Context(), sb.ID, sb.Token); err != nil || !adopted { + t.Errorf("racing adopt: %v %v, want this node to answer for the claim", adopted, err) + } + if rec, _, _ := c.reg.Get(t.Context(), sb.ID); rec.Epoch != taken.Epoch { + t.Errorf("the racing adoption moved the row from epoch %d to %d under the wake", taken.Epoch, rec.Epoch) + } }} if _, _, _, err := b.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { t.Fatalf("wake raced by this node's own adoption: %v, want it to commit", err) } - if !<-second { - t.Error("the racing adoption answered that this node does not hold the claim") - } held, ok := b.byID(sb.ID) if !ok || held.ArchiveCk != "" { t.Fatalf("claim held %v, want it woken on b", ok) From e57466eee8646f471f4902cd6cec533b2f9784f3 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:20:11 +0800 Subject: [PATCH 2/7] cut: let the next renewal retry a contested confirm instead of a loop inside it A refused take no longer re-reads the row and retries within the pass. Any error leaves the node fenced and the next renewal, 5 s later, reads the rows again. An own claim is still never dropped on a refusal, and a contested pass costs replicated claims about 5 s more of 503. --- docs/cluster.md | 12 +++++----- sandboxd/pool/lease.go | 44 ++++++++++++------------------------- sandboxd/pool/lease_test.go | 16 ++++++++++---- 3 files changed, 32 insertions(+), 40 deletions(-) diff --git a/docs/cluster.md b/docs/cluster.md index 0a1b587f9..01e02fd1c 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -359,12 +359,12 @@ checkpoint store and keeps each claim's registry row on the latest snapshot: the node, and the node destroys nothing. Claims of pools that do not replicate stay served. A node starts fenced and lifts the fence after its first renewal that is followed by a confirmation. The confirmation runs beside the renew loop, which keeps - renewing: for each replicated claim the node takes the row back for itself in - one compare-and-set, which the row lock serializes against adopters, and - retries once on a stale epoch. A claim whose row another node took, or took - and finished, is forgotten at once and its VM removed in the background; a - claim whose row is still this node's but keeps refusing the write leaves the - node fenced and is retried at the next renewal. The lift replaces the fenced + renewing: for each replicated claim the node reads the row and takes it back + for itself in one compare-and-set, which the row lock serializes against + adopters. A claim whose row reads as another node's, or as gone after this + node held it, is forgotten at once and its VM removed in the background; a + refused write, whoever moved the row, leaves the node fenced until the next + renewal, about 5 seconds later, reads the row again. The lift replaces the fenced state this renewal saw, only if it is still current, and trips again at once if more than 12 seconds passed since the start of that renewal; a slower confirmation therefore leaves the node fenced for the next renewal to repeat. The margin makes the fence close 3 seconds before diff --git a/sandboxd/pool/lease.go b/sandboxd/pool/lease.go index 380d2b624..02bbab355 100644 --- a/sandboxd/pool/lease.go +++ b/sandboxd/pool/lease.go @@ -2,7 +2,6 @@ package pool import ( "context" - "errors" "fmt" "io" "maps" @@ -142,36 +141,21 @@ func (m *Manager) lift(ctx context.Context, fenced *fenceState, start time.Time) // confirmOwnership takes sb's row for this node in one compare-and-set; false means another node holds the claim. func (m *Manager) confirmOwnership(ctx context.Context, sb *types.Sandbox, rec archives.Record) (bool, error) { - for attempt := range 3 { - switch { - case rec.ID == "": - gone, err := m.replicaGone(ctx, sb) - return !gone, err - case rec.Owner != m.nodeID: - return false, nil - case rec.Consumed: - return true, nil - } - if attempt == 2 { - break - } - epoch, err := m.archives.Take(ctx, sb.ID, rec.Epoch, m.nodeID) - if err == nil { - _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = max(sb.ArchiveEpoch, epoch) }) - return true, nil - } - if !errors.Is(err, archives.ErrLost) { - return false, err - } - var found bool - if rec, found, err = m.archives.Get(ctx, sb.ID); err != nil { - return false, err - } - if !found { - rec = archives.Record{} - } + switch { + case rec.ID == "": + gone, err := m.replicaGone(ctx, sb) + return !gone, err + case rec.Owner != m.nodeID: + return false, nil + case rec.Consumed: + return true, nil + } + epoch, err := m.archives.Take(ctx, sb.ID, rec.Epoch, m.nodeID) + if err != nil { + return false, fmt.Errorf("confirm %s: %w", sb.ID, err) } - return false, fmt.Errorf("confirm %s: its row kept moving under this node", sb.ID) + _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = max(sb.ArchiveEpoch, epoch) }) + return true, nil } // fenceWatch gates sb at resolution and returns the channel that closes when its streams must end; nil for an unreplicated claim. diff --git a/sandboxd/pool/lease_test.go b/sandboxd/pool/lease_test.go index 022d3b2ef..247203039 100644 --- a/sandboxd/pool/lease_test.go +++ b/sandboxd/pool/lease_test.go @@ -284,7 +284,7 @@ func TestLeaseAdoptionDuringUnfenceDropsTheClaimWithoutServingIt(t *testing.T) { } } cut.cut.Store(false) - time.Sleep(leaseRenew) + time.Sleep(2 * leaseRenew) synctest.Wait() if !fencedAtTake { @@ -374,7 +374,7 @@ func TestLeaseConfirmThatKeepsLosingOnItsOwnRowStaysFencedAndKeepsTheClaim(t *te } } -func TestLeaseConfirmRetriesOnceOnAStaleEpoch(t *testing.T) { +func TestLeaseStaleEpochRefusesTheConfirmAndTheNextPassLifts(t *testing.T) { c := newReplicaCluster(t) a := c.node(t, "a", t.TempDir(), keepAnHour) sb := mustClaim(t, a, testKey) @@ -382,8 +382,16 @@ func TestLeaseConfirmRetriesOnceOnAStaleEpoch(t *testing.T) { stale, _, _ := c.reg.Get(t.Context(), sb.ID) replicateNow(t, a, sb) - if keep, err := a.confirmOwnership(t.Context(), sb, stale); err != nil || !keep { - t.Fatalf("confirm on a row read before a newer round: %v %v, want kept", keep, err) + if keep, err := a.confirmOwnership(t.Context(), sb, stale); !errors.Is(err, archives.ErrLost) || keep { + t.Fatalf("confirm on a row read before a newer round: %v %v, want ErrLost", keep, err) + } + if _, held := a.byID(sb.ID); !held { + t.Fatal("a refused confirm dropped a claim whose row is still this node's") + } + a.tripFence() + a.unfence(t.Context(), time.Now()) + if a.isFenced() { + t.Error("the next pass read a fresh row and did not lift the fence") } current, _, _ := c.reg.Get(t.Context(), sb.ID) if a.epochOf(sb) != current.Epoch { From c472aea9671e624e75b55aa36b058bfac2a98a71 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:21:36 +0800 Subject: [PATCH 3/7] cut: lock the tap the provisioned record names, with no lookup fallback Arm runs only from finalizeBatch on a sandbox startVM just built, and every claim path goes through startVM, which copies the tap from cocoon's clone or run record. Restart uses readoptEgressTap and Relock. An empty tap now fails the claim closed, which is the Arm contract, and outbound.Engine no longer needs Inspect. --- sandboxd/outbound/outbound.go | 35 ++++++------------------- sandboxd/outbound/outbound_test.go | 42 +----------------------------- 2 files changed, 9 insertions(+), 68 deletions(-) diff --git a/sandboxd/outbound/outbound.go b/sandboxd/outbound/outbound.go index 112ef45b9..d7ae597f3 100644 --- a/sandboxd/outbound/outbound.go +++ b/sandboxd/outbound/outbound.go @@ -21,9 +21,8 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) -// Engine is the VM engine surface the host needs: a VM's tap, and a guest's lane verdict and egress root. +// Engine is the VM engine surface the host needs: a guest's lane verdict and egress root. type Engine interface { - Inspect(ctx context.Context, name string) (types.VMRecord, bool, error) MarkLane(ctx context.Context, vsockSocket string, lane types.NetRoute) error InstallCACert(ctx context.Context, vsockSocket string, certPEM []byte) error } @@ -92,7 +91,7 @@ func New(o Options) *Host { // Arm locks an egress-lane NIC, then serves the proxy: fail-closed, never a free NIC. func (h *Host) Arm(ctx context.Context, sb *types.Sandbox) error { - if err := h.lockNIC(ctx, sb); err != nil { + if err := h.lockNIC(sb); err != nil { return err } return h.ArmProxy(ctx, sb) @@ -338,38 +337,20 @@ func (h *Host) markLane(ctx context.Context, key types.PoolKey, sock string) err return nil } -func (h *Host) lockNIC(ctx context.Context, sb *types.Sandbox) error { +func (h *Host) lockNIC(sb *types.Sandbox) error { if !h.LocksNIC(sb.Key) { return nil } - tap := sb.TAP - if tap == "" { - var err error - if tap, err = h.tapOf(ctx, sb.VMName); err != nil { - return err - } + if sb.TAP == "" { + return fmt.Errorf("no tap for %s", sb.VMName) } - if err := h.o.Lock(tap); err != nil { - return fmt.Errorf("nft lock %s: %w", tap, err) + if err := h.o.Lock(sb.TAP); err != nil { + return fmt.Errorf("nft lock %s: %w", sb.TAP, err) } - h.KeepLock(sb.ID, tap) + h.KeepLock(sb.ID, sb.TAP) return nil } -func (h *Host) tapOf(ctx context.Context, vmName string) (string, error) { - vm, ok, err := h.o.Engine.Inspect(ctx, vmName) - if err != nil { - return "", fmt.Errorf("list %s: %w", vmName, err) - } - if !ok { - return "", fmt.Errorf("no NIC config for %s", vmName) - } - if tap := vm.TapDevice(); tap != "" { - return tap, nil - } - return "", fmt.Errorf("no tap for %s", vmName) -} - // LoadCA reads cfg's egress root and this node's intermediate, nil when no pool intercepts. func LoadCA(cfg *config.Config) (*egress.CA, error) { if !slices.ContainsFunc(cfg.Pools, func(s config.PoolSpec) bool { return s.Egress.Intercepts() }) { diff --git a/sandboxd/outbound/outbound_test.go b/sandboxd/outbound/outbound_test.go index 33e76c6e6..1d9ec4b86 100644 --- a/sandboxd/outbound/outbound_test.go +++ b/sandboxd/outbound/outbound_test.go @@ -7,7 +7,6 @@ import ( "net/http" "net/http/httptest" "slices" - "sync" "sync/atomic" "testing" @@ -121,24 +120,6 @@ func TestArmEgressLocksEgressLaneWithoutPolicy(t *testing.T) { } } -func TestLockFallsBackToALookupForPreTapClaims(t *testing.T) { - eng := &fakeEngine{taps: map[string]string{"sbx-old": "tap-fake1"}} - var taps outboundtest.TapLog - h := newHost(t, eng, poolConfig(egKey, egPolicy), func(o *Options) { - o.LockNIC = true - o.Lock, o.Unlock = taps.Lock, taps.Unlock - }) - - sb := &types.Sandbox{ID: "sb_old", Key: egKey, VMName: "sbx-old"} - lockErr := h.Arm(t.Context(), sb) - if calls := eng.lookups(); calls != 1 { - t.Errorf("fallback looked the VM up %d times, want 1", calls) - } - if got := taps.Locked(); lockErr != nil || !slices.Equal(got, []string{"tap-fake1"}) { - t.Errorf("fallback locked %v (err %v), want tap-fake1", got, lockErr) - } -} - func TestEgressLaneLocksWithNoPolicyAnywhere(t *testing.T) { var taps outboundtest.TapLog h := newHost(t, &fakeEngine{}, poolConfig(egKey, nil), func(o *Options) { @@ -276,22 +257,7 @@ func testSecrets(t *testing.T) *egress.SecretStore { return s } -type fakeEngine struct { - mu sync.Mutex - taps map[string]string - inspects int -} - -func (f *fakeEngine) Inspect(_ context.Context, name string) (types.VMRecord, bool, error) { - f.mu.Lock() - defer f.mu.Unlock() - f.inspects++ - tap, ok := f.taps[name] - if !ok { - return types.VMRecord{}, false, nil - } - return types.VMRecord{Config: types.VMConfig{Name: name}, NetworkConfigs: []types.VMNetConfig{{TAP: tap}}}, true, nil -} +type fakeEngine struct{} func (f *fakeEngine) InstallCACert(context.Context, string, []byte) error { return nil @@ -300,9 +266,3 @@ func (f *fakeEngine) InstallCACert(context.Context, string, []byte) error { func (f *fakeEngine) MarkLane(context.Context, string, types.NetRoute) error { return nil } - -func (f *fakeEngine) lookups() int { - f.mu.Lock() - defer f.mu.Unlock() - return f.inspects -} From 532435d5a49b9acfaf1b12fe47eb32335b38740d Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:22:19 +0800 Subject: [PATCH 4/7] cut: stop tracking which volume-holding claims the replication pass logged The line only repeated a documented rule: a claim that holds volumes is not replicated. GET /v1/sandboxes/{id}/volumes shows what a claim holds, so the state stays visible. replicaState collapses to the time of the last try. --- docs/cluster.md | 4 ++-- sandboxd/pool/pool.go | 2 +- sandboxd/pool/replica.go | 29 ++++++++--------------------- sandboxd/pool/replica_test.go | 5 +---- 4 files changed, 12 insertions(+), 28 deletions(-) diff --git a/docs/cluster.md b/docs/cluster.md index 01e02fd1c..ddbb5836d 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -292,8 +292,8 @@ checkpoint store and keeps each claim's registry row on the latest snapshot: stale epoch and is refreshed, and an absent or foreign row drops the claim and its VM. A claim gets its first replica one interval after it is claimed (or after a restart). A hibernated, archived or failed claim, - one whose tenant was removed, and one with volumes (capture refuses them; - logged once per claim) is skipped; a replicated claim that gains volumes + one whose tenant was removed, and one with volumes (capture refuses them) is + skipped; a replicated claim that gains volumes drops its row and replica. The claim's policy pool, not its own template key, picks the interval, so a claim cloned from a template promoted out of a replicating pool follows that pool. Rounds share the maintenance budget with diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 50d4d018b..783a7bb65 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -371,7 +371,7 @@ type Manager struct { unfencing atomic.Bool replicaPools map[types.PoolKey]struct{} replicating atomic.Bool - replicas map[string]replicaState + replicas map[string]time.Time view atomic.Pointer[configView] // cfg is the config the view was built from; reloadMu orders reloads, the only writers of cfg and view. diff --git a/sandboxd/pool/replica.go b/sandboxd/pool/replica.go index b27c22be6..793fd40d5 100644 --- a/sandboxd/pool/replica.go +++ b/sandboxd/pool/replica.go @@ -13,22 +13,16 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) -type replicaState struct { - tried time.Time - skipLogged bool -} - func (m *Manager) replicateOnce(ctx context.Context) { if m.archives == nil || m.isFenced() || !m.replicating.CompareAndSwap(false, true) { return } now, v := time.Now(), m.view.Load() var due, retire []*types.Sandbox - var withVolumes []string - next := make(map[string]replicaState, len(m.replicas)) + next := make(map[string]time.Time, len(m.replicas)) m.mu.Lock() for id, sb := range m.claimed { - every, st := v.poolReplicate[sb.PolicyKey()], m.replicas[id] + every, tried := v.poolReplicate[sb.PolicyKey()], m.replicas[id] switch { case every <= 0: if sb.HasReplica() { @@ -39,29 +33,22 @@ func (m *Manager) replicateOnce(ctx context.Context) { if sb.HasReplica() { retire = append(retire, sb) } - if !st.skipLogged { - withVolumes = append(withVolumes, id) - } - next[id] = replicaState{skipLogged: true} - case st.tried.IsZero(): - next[id] = replicaState{tried: now} - case now.Sub(st.tried) < every: - next[id] = st + case tried.IsZero(): + next[id] = now + case now.Sub(tried) < every: + next[id] = tried default: - next[id] = replicaState{tried: now} + next[id] = now due = append(due, sb) } } m.mu.Unlock() m.replicas = next - logger := log.WithFunc("pool.replicateOnce") - for _, id := range withVolumes { - logger.Infof(ctx, "sandbox %s holds volumes and is not replicated", id) - } if len(due)+len(retire) == 0 { m.replicating.Store(false) return } + logger := log.WithFunc("pool.replicateOnce") go func() { defer m.replicating.Store(false) m.runBounded(ctx, len(due)+len(retire), func(ctx context.Context, i int) { diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index 4d0cb57e2..deed5a830 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -330,7 +330,7 @@ func TestReplicaFollowsThePolicyPoolOfAPromotedTemplate(t *testing.T) { a.replicateOnce(t.Context()) waitFor(t, func() bool { return !a.replicating.Load() }) a.mu.Lock() - a.replicas[sb.ID] = replicaState{tried: time.Now().Add(-time.Hour)} + a.replicas[sb.ID] = time.Now().Add(-time.Hour) a.mu.Unlock() a.replicateOnce(t.Context()) waitFor(t, func() bool { return !a.replicating.Load() }) @@ -579,9 +579,6 @@ func TestReplicaPassSnapshotsOnlyDueRunningClaims(t *testing.T) { if a.replicaOf(paused) != "" || a.replicaOf(withVolume) != "" { t.Error("a hibernated claim or one with volumes was replicated") } - if !a.replicas[withVolume.ID].skipLogged { - t.Error("a claim with volumes was not skipped and logged") - } if _, ok, _ := c.reg.Get(t.Context(), running.ID); !ok { t.Error("the due claim has no row") } From 53e6f7f03678710b9f09fca85adea08b33b1a750 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:24:24 +0800 Subject: [PATCH 5/7] cut: close a warm VM's doors at its claim on a node that guards nothing ArmProxy's early return now closes what refill bound, so Prebind's re-check and the reload sweep have nothing left to cover. After a reload that drops every egress policy, an unclaimed warm VM keeps its listeners until it is claimed or removed. No tenant code runs in it before the claim, and removeVM, the only removal path, closes them too. An unguarded claim gains one uncontended lock and one map lookup, about 20 ns. --- sandboxd/outbound/outbound.go | 18 +++--------------- sandboxd/pool/reload.go | 2 +- sandboxd/pool/reload_test.go | 21 +++++++++++++++++---- 3 files changed, 21 insertions(+), 20 deletions(-) diff --git a/sandboxd/outbound/outbound.go b/sandboxd/outbound/outbound.go index d7ae597f3..f662d8582 100644 --- a/sandboxd/outbound/outbound.go +++ b/sandboxd/outbound/outbound.go @@ -101,6 +101,7 @@ func (h *Host) Arm(ctx context.Context, sb *types.Sandbox) error { func (h *Host) ArmProxy(ctx context.Context, sb *types.Sandbox) error { v := h.o.View() if !v.guarded || sb.VsockSocket == "" { + h.ClosePrebound(sb.VMName) return nil } policy, ok := v.Resolve(sb, h.o.Pooled) @@ -169,11 +170,6 @@ func (h *Host) Prebind(ctx context.Context, sb *types.Sandbox) { return } h.mu.Lock() - if !h.o.View().guarded { - h.mu.Unlock() - d.close() - return - } h.prebound[sb.VMName] = d h.mu.Unlock() } @@ -208,19 +204,11 @@ func (h *Host) Disarm(id string, removed bool) { } } -// Reloaded follows a view swap: doors pre-bound for a node that now guards nothing close, and live proxies drop their pooled connections. -func (h *Host) Reloaded(v *View) { +// Reloaded follows a view swap: live proxies drop their pooled connections. +func (h *Host) Reloaded() { h.mu.Lock() live := slices.Collect(maps.Values(h.doors)) - var stale []*doors - if !v.guarded { - stale = slices.Collect(maps.Values(h.prebound)) - clear(h.prebound) - } h.mu.Unlock() - for _, d := range stale { - d.close() - } for _, d := range live { d.proxy.ResetPools() } diff --git a/sandboxd/pool/reload.go b/sandboxd/pool/reload.go index ab1d67d83..7a707c079 100644 --- a/sandboxd/pool/reload.go +++ b/sandboxd/pool/reload.go @@ -61,7 +61,7 @@ func (m *Manager) ReloadConfig(ctx context.Context, next *config.Config) (Reload } } m.mu.Unlock() - m.out.Reloaded(view.out) + m.out.Reloaded() m.destroyAll(context.WithoutCancel(ctx), trim).Wait() m.cfg = next m.recordAudit(ctx, "", auditFrame{Op: "config_reload", Changed: changed}) diff --git a/sandboxd/pool/reload_test.go b/sandboxd/pool/reload_test.go index 50e1ad45b..efa6cb645 100644 --- a/sandboxd/pool/reload_test.go +++ b/sandboxd/pool/reload_test.go @@ -334,7 +334,7 @@ func TestAnInterruptedReloadStillDestroysTheRetiredWarmVMs(t *testing.T) { } } -func TestAReloadThatUnguardsTheNodeClosesThePreboundDoors(t *testing.T) { +func TestAReloadThatUnguardsTheNodeClosesAWarmVMsDoorsAtItsClaim(t *testing.T) { eng := newFakeEngine() eng.sockRoot = outboundtest.SockRoot(t) m := egressManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1, Egress: &egress.Policy{Allow: []egress.Rule{{Host: "example.com"}}}}) @@ -345,12 +345,25 @@ func TestAReloadThatUnguardsTheNodeClosesThePreboundDoors(t *testing.T) { if _, err := m.ReloadConfig(t.Context(), &next); err != nil { t.Fatalf("reload: %v", err) } + conn, err := net.Dial("unix", pin) + if err != nil { + t.Fatalf("the unclaimed warm VM's door: %v, want it bound until the claim", err) + } + _ = conn.Close() + + sb := mustClaim(t, m, testKey) + if sb.VMName != warm.VMName { + t.Fatalf("claim took %s, want the warm VM %s", sb.VMName, warm.VMName) + } if conn, err := net.Dial("unix", pin); err == nil { _ = conn.Close() - t.Error("a prebound door is left open on a node with no egress policy") + t.Error("a prebound door is left open on a claim of a node with no egress policy") + } + if _, err := net.Dial("unix", engine.EgressSocketPath(sb.VsockSocket)); err == nil { + t.Error("a door refill bound still accepts on a claim of a node with no egress policy") } - if _, err := net.Dial("unix", engine.EgressSocketPath(warm.VsockSocket)); err == nil { - t.Error("a door refill bound still accepts after the reload removed every policy") + if route := m.NetRoute(sb); route != types.NetRouteNone { + t.Errorf("claim route %q, want none", route) } } From 54aad811081ff0fbe729426a18ac980df4c07410 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:43:07 +0800 Subject: [PATCH 6/7] pool: keep the retry inside confirmOwnership This reverts 413724c5. Without the retry, one release landing on a claim not yet confirmed aborts the whole unfence pass, and every replicated claim on the node stays at 503 for another renewal. Simulated with 1.3 ms per Take: with 1000 replicated claims and 5 releases/s the fence lasted a mean 1m57s against 1.3 s with the retry, and with 1/s a mean 4.0 s and a max 19.6 s against 1.3 s. 16 lines do not pay for that. --- docs/cluster.md | 12 +++++----- sandboxd/pool/lease.go | 44 +++++++++++++++++++++++++------------ sandboxd/pool/lease_test.go | 16 ++++---------- 3 files changed, 40 insertions(+), 32 deletions(-) diff --git a/docs/cluster.md b/docs/cluster.md index ddbb5836d..0eccef656 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -359,12 +359,12 @@ checkpoint store and keeps each claim's registry row on the latest snapshot: the node, and the node destroys nothing. Claims of pools that do not replicate stay served. A node starts fenced and lifts the fence after its first renewal that is followed by a confirmation. The confirmation runs beside the renew loop, which keeps - renewing: for each replicated claim the node reads the row and takes it back - for itself in one compare-and-set, which the row lock serializes against - adopters. A claim whose row reads as another node's, or as gone after this - node held it, is forgotten at once and its VM removed in the background; a - refused write, whoever moved the row, leaves the node fenced until the next - renewal, about 5 seconds later, reads the row again. The lift replaces the fenced + renewing: for each replicated claim the node takes the row back for itself in + one compare-and-set, which the row lock serializes against adopters, and + retries once on a stale epoch. A claim whose row another node took, or took + and finished, is forgotten at once and its VM removed in the background; a + claim whose row is still this node's but keeps refusing the write leaves the + node fenced and is retried at the next renewal. The lift replaces the fenced state this renewal saw, only if it is still current, and trips again at once if more than 12 seconds passed since the start of that renewal; a slower confirmation therefore leaves the node fenced for the next renewal to repeat. The margin makes the fence close 3 seconds before diff --git a/sandboxd/pool/lease.go b/sandboxd/pool/lease.go index 02bbab355..380d2b624 100644 --- a/sandboxd/pool/lease.go +++ b/sandboxd/pool/lease.go @@ -2,6 +2,7 @@ package pool import ( "context" + "errors" "fmt" "io" "maps" @@ -141,21 +142,36 @@ func (m *Manager) lift(ctx context.Context, fenced *fenceState, start time.Time) // confirmOwnership takes sb's row for this node in one compare-and-set; false means another node holds the claim. func (m *Manager) confirmOwnership(ctx context.Context, sb *types.Sandbox, rec archives.Record) (bool, error) { - switch { - case rec.ID == "": - gone, err := m.replicaGone(ctx, sb) - return !gone, err - case rec.Owner != m.nodeID: - return false, nil - case rec.Consumed: - return true, nil - } - epoch, err := m.archives.Take(ctx, sb.ID, rec.Epoch, m.nodeID) - if err != nil { - return false, fmt.Errorf("confirm %s: %w", sb.ID, err) + for attempt := range 3 { + switch { + case rec.ID == "": + gone, err := m.replicaGone(ctx, sb) + return !gone, err + case rec.Owner != m.nodeID: + return false, nil + case rec.Consumed: + return true, nil + } + if attempt == 2 { + break + } + epoch, err := m.archives.Take(ctx, sb.ID, rec.Epoch, m.nodeID) + if err == nil { + _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = max(sb.ArchiveEpoch, epoch) }) + return true, nil + } + if !errors.Is(err, archives.ErrLost) { + return false, err + } + var found bool + if rec, found, err = m.archives.Get(ctx, sb.ID); err != nil { + return false, err + } + if !found { + rec = archives.Record{} + } } - _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = max(sb.ArchiveEpoch, epoch) }) - return true, nil + return false, fmt.Errorf("confirm %s: its row kept moving under this node", sb.ID) } // fenceWatch gates sb at resolution and returns the channel that closes when its streams must end; nil for an unreplicated claim. diff --git a/sandboxd/pool/lease_test.go b/sandboxd/pool/lease_test.go index 247203039..022d3b2ef 100644 --- a/sandboxd/pool/lease_test.go +++ b/sandboxd/pool/lease_test.go @@ -284,7 +284,7 @@ func TestLeaseAdoptionDuringUnfenceDropsTheClaimWithoutServingIt(t *testing.T) { } } cut.cut.Store(false) - time.Sleep(2 * leaseRenew) + time.Sleep(leaseRenew) synctest.Wait() if !fencedAtTake { @@ -374,7 +374,7 @@ func TestLeaseConfirmThatKeepsLosingOnItsOwnRowStaysFencedAndKeepsTheClaim(t *te } } -func TestLeaseStaleEpochRefusesTheConfirmAndTheNextPassLifts(t *testing.T) { +func TestLeaseConfirmRetriesOnceOnAStaleEpoch(t *testing.T) { c := newReplicaCluster(t) a := c.node(t, "a", t.TempDir(), keepAnHour) sb := mustClaim(t, a, testKey) @@ -382,16 +382,8 @@ func TestLeaseStaleEpochRefusesTheConfirmAndTheNextPassLifts(t *testing.T) { stale, _, _ := c.reg.Get(t.Context(), sb.ID) replicateNow(t, a, sb) - if keep, err := a.confirmOwnership(t.Context(), sb, stale); !errors.Is(err, archives.ErrLost) || keep { - t.Fatalf("confirm on a row read before a newer round: %v %v, want ErrLost", keep, err) - } - if _, held := a.byID(sb.ID); !held { - t.Fatal("a refused confirm dropped a claim whose row is still this node's") - } - a.tripFence() - a.unfence(t.Context(), time.Now()) - if a.isFenced() { - t.Error("the next pass read a fresh row and did not lift the fence") + if keep, err := a.confirmOwnership(t.Context(), sb, stale); err != nil || !keep { + t.Fatalf("confirm on a row read before a newer round: %v %v, want kept", keep, err) } current, _, _ := c.reg.Get(t.Context(), sb.ID) if a.epochOf(sb) != current.Epoch { From 36bfa202403a365b2179ca104db06d8816a5bb93 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 3 Nov 2026 01:48:24 +0800 Subject: [PATCH 7/7] outbound: keep the tap lookup when a clone's record has none This reverts b3f0964e. runRecord swallows a parse failure of cocoon's output and returns an empty record, so without the fallback every egress claim on such a node answers 500 and burns its warm VM, where before it took one inspect and a warning: 5 of 5 claims succeeded with the lookup, 0 of 5 without. No cocoon version read today triggers it, but a test guaranteed the behaviour and 19 lines off a branch that never runs buy nothing. --- sandboxd/outbound/outbound.go | 35 +++++++++++++++++++------ sandboxd/outbound/outbound_test.go | 42 +++++++++++++++++++++++++++++- 2 files changed, 68 insertions(+), 9 deletions(-) diff --git a/sandboxd/outbound/outbound.go b/sandboxd/outbound/outbound.go index f662d8582..01854238f 100644 --- a/sandboxd/outbound/outbound.go +++ b/sandboxd/outbound/outbound.go @@ -21,8 +21,9 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) -// Engine is the VM engine surface the host needs: a guest's lane verdict and egress root. +// Engine is the VM engine surface the host needs: a VM's tap, and a guest's lane verdict and egress root. type Engine interface { + Inspect(ctx context.Context, name string) (types.VMRecord, bool, error) MarkLane(ctx context.Context, vsockSocket string, lane types.NetRoute) error InstallCACert(ctx context.Context, vsockSocket string, certPEM []byte) error } @@ -91,7 +92,7 @@ func New(o Options) *Host { // Arm locks an egress-lane NIC, then serves the proxy: fail-closed, never a free NIC. func (h *Host) Arm(ctx context.Context, sb *types.Sandbox) error { - if err := h.lockNIC(sb); err != nil { + if err := h.lockNIC(ctx, sb); err != nil { return err } return h.ArmProxy(ctx, sb) @@ -325,20 +326,38 @@ func (h *Host) markLane(ctx context.Context, key types.PoolKey, sock string) err return nil } -func (h *Host) lockNIC(sb *types.Sandbox) error { +func (h *Host) lockNIC(ctx context.Context, sb *types.Sandbox) error { if !h.LocksNIC(sb.Key) { return nil } - if sb.TAP == "" { - return fmt.Errorf("no tap for %s", sb.VMName) + tap := sb.TAP + if tap == "" { + var err error + if tap, err = h.tapOf(ctx, sb.VMName); err != nil { + return err + } } - if err := h.o.Lock(sb.TAP); err != nil { - return fmt.Errorf("nft lock %s: %w", sb.TAP, err) + if err := h.o.Lock(tap); err != nil { + return fmt.Errorf("nft lock %s: %w", tap, err) } - h.KeepLock(sb.ID, sb.TAP) + h.KeepLock(sb.ID, tap) return nil } +func (h *Host) tapOf(ctx context.Context, vmName string) (string, error) { + vm, ok, err := h.o.Engine.Inspect(ctx, vmName) + if err != nil { + return "", fmt.Errorf("list %s: %w", vmName, err) + } + if !ok { + return "", fmt.Errorf("no NIC config for %s", vmName) + } + if tap := vm.TapDevice(); tap != "" { + return tap, nil + } + return "", fmt.Errorf("no tap for %s", vmName) +} + // LoadCA reads cfg's egress root and this node's intermediate, nil when no pool intercepts. func LoadCA(cfg *config.Config) (*egress.CA, error) { if !slices.ContainsFunc(cfg.Pools, func(s config.PoolSpec) bool { return s.Egress.Intercepts() }) { diff --git a/sandboxd/outbound/outbound_test.go b/sandboxd/outbound/outbound_test.go index 1d9ec4b86..33e76c6e6 100644 --- a/sandboxd/outbound/outbound_test.go +++ b/sandboxd/outbound/outbound_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "slices" + "sync" "sync/atomic" "testing" @@ -120,6 +121,24 @@ func TestArmEgressLocksEgressLaneWithoutPolicy(t *testing.T) { } } +func TestLockFallsBackToALookupForPreTapClaims(t *testing.T) { + eng := &fakeEngine{taps: map[string]string{"sbx-old": "tap-fake1"}} + var taps outboundtest.TapLog + h := newHost(t, eng, poolConfig(egKey, egPolicy), func(o *Options) { + o.LockNIC = true + o.Lock, o.Unlock = taps.Lock, taps.Unlock + }) + + sb := &types.Sandbox{ID: "sb_old", Key: egKey, VMName: "sbx-old"} + lockErr := h.Arm(t.Context(), sb) + if calls := eng.lookups(); calls != 1 { + t.Errorf("fallback looked the VM up %d times, want 1", calls) + } + if got := taps.Locked(); lockErr != nil || !slices.Equal(got, []string{"tap-fake1"}) { + t.Errorf("fallback locked %v (err %v), want tap-fake1", got, lockErr) + } +} + func TestEgressLaneLocksWithNoPolicyAnywhere(t *testing.T) { var taps outboundtest.TapLog h := newHost(t, &fakeEngine{}, poolConfig(egKey, nil), func(o *Options) { @@ -257,7 +276,22 @@ func testSecrets(t *testing.T) *egress.SecretStore { return s } -type fakeEngine struct{} +type fakeEngine struct { + mu sync.Mutex + taps map[string]string + inspects int +} + +func (f *fakeEngine) Inspect(_ context.Context, name string) (types.VMRecord, bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.inspects++ + tap, ok := f.taps[name] + if !ok { + return types.VMRecord{}, false, nil + } + return types.VMRecord{Config: types.VMConfig{Name: name}, NetworkConfigs: []types.VMNetConfig{{TAP: tap}}}, true, nil +} func (f *fakeEngine) InstallCACert(context.Context, string, []byte) error { return nil @@ -266,3 +300,9 @@ func (f *fakeEngine) InstallCACert(context.Context, string, []byte) error { func (f *fakeEngine) MarkLane(context.Context, string, types.NetRoute) error { return nil } + +func (f *fakeEngine) lookups() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.inspects +}