diff --git a/docs/cluster.md b/docs/cluster.md index 0a1b587f9..0eccef656 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/outbound/outbound.go b/sandboxd/outbound/outbound.go index 112ef45b9..01854238f 100644 --- a/sandboxd/outbound/outbound.go +++ b/sandboxd/outbound/outbound.go @@ -102,6 +102,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) @@ -170,11 +171,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() } @@ -209,19 +205,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/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) 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/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) } } 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") }