From 668112eca7664c2ec2f4bb1a991dd9c111392711 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 29 Oct 2026 17:11:07 +0800 Subject: [PATCH] pool: drop volume holds for archived claims Archived claims counted their volumes to cover replica rows a live read-only attach could publish while an older replica export ran (v0.1.17-v0.1.19). #383 refuses that publication now, so no new row carries volumes; the adoption, restart, transfer and reap bookkeeping and their legacy-row tests go. --- sandboxd/pool/claim.go | 4 - sandboxd/pool/failover.go | 5 - sandboxd/pool/failover_test.go | 61 ------- sandboxd/pool/reconcile.go | 1 - .../pool/testdata/legacy-volume-replica.json | 15 -- sandboxd/pool/volume_catalog_test.go | 154 ------------------ 6 files changed, 240 deletions(-) delete mode 100644 sandboxd/pool/testdata/legacy-volume-replica.json diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 67334f2f..c13e8e75 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -607,18 +607,15 @@ func (m *Manager) reapOnce(ctx context.Context) { m.mu.Lock() keep := expired[:0] var purgeCks, purged []string - var purgeHolds []types.Volume for _, v := range expired { if v.action == reapPurge { if m.claimed[v.id] != v.sb || v.sb.ArchiveCk != v.ck { continue } m.pendingCks[v.ck] = struct{}{} - v.holds = m.cancelVolumeRun(v.sb) m.dropClaimLocked(v.sb) purgeCks = append(purgeCks, v.ck) purged = append(purged, v.id) - purgeHolds = append(purgeHolds, v.holds...) } keep = append(keep, v) } @@ -653,7 +650,6 @@ func (m *Manager) reapOnce(ctx context.Context) { logger.Error(ctx, saveErr, "persist reap; rolled back") return } - m.unreserveVolumes(purgeHolds) // engine teardown can take minutes, so fan out bounded and never stall the ticker loop m.runBounded(ctx, len(expired), func(ctx context.Context, i int) { v := expired[i] diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index bd5f6331..dfcb60ca 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -64,7 +64,6 @@ func (m *Manager) adoptArchived(ctx context.Context, id, token string) (bool, er m.mu.Unlock() return m.adoptHeld(ctx, cur, epoch), nil } - m.adoptVolumes(heldVolumes(sb)) m.addClaimLocked(sb) js := m.store.set(sb) m.mu.Unlock() @@ -420,10 +419,6 @@ func (m *Manager) forgetClaim(ctx context.Context, sb *types.Sandbox) ([]types.V return nil, false } holds := m.cancelVolumeRun(sb) - if sb.ArchiveCk != "" { - m.releaseVolumes(holds) - holds = nil - } m.dropClaimLocked(sb) js := m.store.del(sb.ID) m.mu.Unlock() diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index 507571eb..505dc55c 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -111,67 +111,6 @@ func TestFailoverOwnerWakeWithTheRegistryDownFailsClosed(t *testing.T) { } } -func TestFailoverLegacyVolumeTransferPreservesOtherReader(t *testing.T) { - c := newFailoverCluster(t) - b := c.node(t, "b", t.TempDir(), keepAnHour) - d := c.node(t, "d", t.TempDir(), keepAnHour) - backing := writeVolumeImage(t, "data", "volume bytes") - for _, m := range []*Manager{b, d} { - if err := m.RegisterVolume(t.Context(), "data", types.VolumeRegistration{Path: backing, Writable: true}); err != nil { - t.Fatal(err) - } - } - raw, readErr := os.ReadFile("testdata/legacy-volume-replica.json") - if readErr != nil { - t.Fatal(readErr) - } - var fixture struct { - Token string - Record archives.Record - } - if err := json.Unmarshal(raw, &fixture); err != nil { - t.Fatal(err) - } - if _, err := c.reg.Put(t.Context(), fixture.Record); err != nil { - t.Fatal(err) - } - c.expire(fixture.Record.Owner) - for range 2 { - if adopted, err := b.AdoptArchived(t.Context(), fixture.Record.ID, fixture.Token); err != nil || !adopted { - t.Fatalf("adopt recorded legacy row: adopted=%v err=%v", adopted, err) - } - } - writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} - if _, err := b.ClaimProvision(t.Context(), plainKey, writer); !errors.Is(err, ErrVolumeBusy) { - t.Fatalf("adopted reader failed to exclude writer: %v", err) - } - other, err := b.ClaimProvision(t.Context(), plainKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}) - if err != nil { - t.Fatal(err) - } - c.expire("b") - if adopted, err := d.AdoptArchived(t.Context(), fixture.Record.ID, fixture.Token); err != nil || !adopted { - t.Fatalf("transfer recorded row: adopted=%v err=%v", adopted, err) - } - for range 2 { - if err := b.Release(t.Context(), fixture.Record.ID, Cred{Token: fixture.Token}); !errors.Is(err, ErrUnknownSandbox) { - t.Fatalf("former owner release: %v", err) - } - if _, err := b.ClaimProvision(t.Context(), plainKey, writer); !errors.Is(err, ErrVolumeBusy) { - t.Fatalf("former owner released independent reader: %v", err) - } - } - if err := b.Release(t.Context(), other.ID, Cred{Token: other.Token}); err != nil { - t.Fatal(err) - } - if _, err := b.ClaimProvision(t.Context(), plainKey, writer); err != nil { - t.Fatalf("former owner leaked reader hold: %v", err) - } - if _, err := d.ClaimProvision(t.Context(), plainKey, writer); !errors.Is(err, ErrVolumeBusy) { - t.Fatalf("new owner failed to exclude writer: %v", err) - } -} - func TestFailoverAdoptionMovesTheClaimAndTheOldOwnerLetsGo(t *testing.T) { c := newFailoverCluster(t) a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index a776546d..a116b580 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -47,7 +47,6 @@ func (m *Manager) Reconcile(ctx context.Context) error { case sb.ArchiveCk != "": // archived: no local VM by design, so adopt the stub and let the first exec wake it m.addClaimLocked(sb) - m.adoptVolumes(heldVolumes(sb)) continue case ok && rec.State == vmStateRunning: sb.VsockSocket = rec.VsockSocket diff --git a/sandboxd/pool/testdata/legacy-volume-replica.json b/sandboxd/pool/testdata/legacy-volume-replica.json deleted file mode 100644 index b246bf44..00000000 --- a/sandboxd/pool/testdata/legacy-volume-replica.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "Source": "5cd5c4caa0b85e90489e62e25b8c8e7277040a78", - "Token": "6892970471eef08bd0c28f34aa10cb09", - "Record": { - "ID": "sb_9200d2ad82fc1f9c", - "Tenant": "", - "TokenSHA256": "eaf458d283c8c6fcf8f647f20336395939a074412bef97a19f9ca604caa977eb", - "Owner": "a", - "Epoch": 1, - "Consumed": false, - "Checkpoint": "ck_96a0c036f8fafaca", - "Deadline": "0001-01-01T00:00:00Z", - "Claim": {"id":"sb_9200d2ad82fc1f9c","vm_name":"","key":{"template":"rt:24.04","net":"none","size":"small"},"claimed_at":"2026-10-29T00:50:38.542878594Z","lease_seconds":300,"policy_layer":"pooled","volumes":[{"name":"data","mount":"/volumes/data"}],"archive_ck":"ck_96a0c036f8fafaca"} - } -} diff --git a/sandboxd/pool/volume_catalog_test.go b/sandboxd/pool/volume_catalog_test.go index 9a9b79d0..b9f989ee 100644 --- a/sandboxd/pool/volume_catalog_test.go +++ b/sandboxd/pool/volume_catalog_test.go @@ -1,7 +1,6 @@ package pool import ( - "context" "encoding/json/v2" "errors" "fmt" @@ -11,7 +10,6 @@ import ( "strings" "sync" "testing" - "testing/synctest" "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/types" @@ -85,134 +83,6 @@ func TestRuntimeCatalogPersistsNormalizedSemantics(t *testing.T) { } } -func TestRuntimeCatalogRecoversLegacyArchivedVolumeReference(t *testing.T) { - for _, tc := range []struct { - name string - finish func(*testing.T, *Manager, *types.Sandbox) error - }{ - {"release", func(t *testing.T, m *Manager, sb *types.Sandbox) error { - return m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}) - }}, - {"ownership transfer", func(t *testing.T, m *Manager, sb *types.Sandbox) error { - m.dropAdopted(t.Context(), sb) - return nil - }}, - {"reap", func(t *testing.T, m *Manager, sb *types.Sandbox) error { - expire(m, sb) - m.reapOnce(t.Context()) - return nil - }}, - {"reap with canceled cleanup", func(t *testing.T, m *Manager, sb *types.Sandbox) error { - for range cap(m.maintSem) { - m.maintSem <- struct{}{} - } - ctx, cancel := context.WithCancel(t.Context()) - cancel() - expire(m, sb) - m.reapOnce(ctx) - synctest.Wait() - for range cap(m.maintSem) { - <-m.maintSem - } - return nil - }}, - } { - t.Run(tc.name, func(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - m, recovered := recoverLegacyArchived(t) - if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeRegistrationConflict) { - t.Fatalf("unregister recovered reference: %v", err) - } - writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} - if _, err := m.ClaimProvision(t.Context(), testKey, writer); !errors.Is(err, ErrVolumeBusy) { - t.Fatalf("writer beside recovered reference: %v", err) - } - reader, claimErr := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}) - if claimErr != nil { - t.Fatal(claimErr) - } - for range 2 { - if err := tc.finish(t, m, recovered); err != nil && !errors.Is(err, ErrUnknownSandbox) { - t.Fatal(err) - } - synctest.Wait() - if _, ok := m.byID(recovered.ID); ok { - t.Fatal("archived claim survived retirement") - } - if _, err := m.ClaimProvision(t.Context(), testKey, writer); !errors.Is(err, ErrVolumeBusy) { - t.Fatalf("writer beside remaining reader: %v", err) - } - if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeRegistrationConflict) { - t.Fatalf("unregister beside remaining reader: %v", err) - } - } - if err := m.Release(t.Context(), reader.ID, Cred{Token: reader.Token}); err != nil { - t.Fatal(err) - } - writable, err := m.ClaimProvision(t.Context(), testKey, writer) - if err != nil { - t.Fatalf("writer after both readers retire: %v", err) - } - if err := m.Release(t.Context(), writable.ID, Cred{Token: writable.Token}); err != nil { - t.Fatal(err) - } - if err := m.UnregisterVolume(t.Context(), "data"); err != nil { - t.Fatalf("unregister after claim removal: %v", err) - } - }) - }) - } -} - -func TestRuntimeCatalogRetainsArchivedVolumeReferenceAfterFailedReap(t *testing.T) { - for _, tc := range []struct { - name string - breakStore func(*testing.T, *Manager) func() - }{ - {"delete marker", func(t *testing.T, m *Manager) func() { - blocker := filepath.Join(m.dataDir, archiveDeleteDir) - if err := os.WriteFile(blocker, nil, 0o600); err != nil { - t.Fatal(err) - } - return func() { - if err := os.Remove(blocker); err != nil { - t.Fatal(err) - } - } - }}, - {"claim store", func(t *testing.T, m *Manager) func() { - breakStore(t, m) - return func() { healStore(t, m) } - }}, - } { - t.Run(tc.name, func(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - m, recovered := recoverLegacyArchived(t) - restore := sync.OnceFunc(tc.breakStore(t, m)) - defer restore() - expire(m, recovered) - m.reapOnce(t.Context()) - if _, ok := m.byID(recovered.ID); !ok { - t.Fatal("failed reap removed the archived claim") - } - if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeRegistrationConflict) { - t.Fatalf("unregister after failed reap: %v", err) - } - writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} - if _, err := m.ClaimProvision(t.Context(), testKey, writer); !errors.Is(err, ErrVolumeBusy) { - t.Fatalf("writer after failed reap: %v", err) - } - restore() - m.reapOnce(t.Context()) - synctest.Wait() - if err := m.UnregisterVolume(t.Context(), "data"); err != nil { - t.Fatalf("unregister after reap retry: %v", err) - } - }) - }) - } -} - func TestRuntimeCatalogChangesAreAudited(t *testing.T) { cfg := &config.Config{DataDir: t.TempDir(), AuditLog: true} m, initErr := NewManager(t.Context(), cfg, newFakeEngine(), testSecrets(t), testBackends(t, cfg)) @@ -558,27 +428,3 @@ func claimAndRelease(t *testing.T, m *Manager, volumes ...types.Volume) error { } return m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}) } - -func recoverLegacyArchived(t *testing.T) (*Manager, *types.Sandbox) { - t.Helper() - m := newVolumeManager(t, newFakeEngine(), nil) - if err := m.RegisterVolume(t.Context(), "data", types.VolumeRegistration{Path: writeVolumeImage(t, "data", "bytes"), Writable: true}); err != nil { - t.Fatal(err) - } - sb := &types.Sandbox{ - ID: "sb_legacy", Key: testKey, Token: "legacy-token", ArchiveCk: "ck_00000000000000aa", - Volumes: []types.Volume{{Name: "data", Mount: "/volumes/data"}}, - } - if err := m.store.save(map[string]*types.Sandbox{sb.ID: sb}); err != nil { - t.Fatal(err) - } - m = newVolumeManagerAt(t, newFakeEngine(), m.dataDir, nil) - if err := m.Reconcile(t.Context()); err != nil { - t.Fatal(err) - } - recovered, ok := m.byID(sb.ID) - if !ok { - t.Fatal("archived claim did not survive restart") - } - return m, recovered -}