Skip to content

pool: drop volume holds for archived claims; plan compat removals #399

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
CMGS merged 1 commit into from
Oct 29, 2026
Merged

pool: drop volume holds for archived claims; plan compat removals #399

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions sandboxd/pool/claim.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down Expand Up @@ -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]
Expand Down
5 changes: 0 additions & 5 deletions sandboxd/pool/failover.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand Down
61 changes: 0 additions & 61 deletions sandboxd/pool/failover_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
1 change: 0 additions & 1 deletion sandboxd/pool/reconcile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
15 changes: 0 additions & 15 deletions sandboxd/pool/testdata/legacy-volume-replica.json
View file
Open in desktop

This file was deleted.

154 changes: 0 additions & 154 deletions sandboxd/pool/volume_catalog_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package pool

import (
"context"
"encoding/json/v2"
"errors"
"fmt"
Expand All @@ -11,7 +10,6 @@ import (
"strings"
"sync"
"testing"
"testing/synctest"

"github.com/cocoonstack/sandbox/sandboxd/config"
"github.com/cocoonstack/sandbox/sandboxd/types"
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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
}
Loading