diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index af26aa17..9624529c 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -122,10 +122,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { intent := m.store.set(sb) m.mu.Unlock() if err = m.store.commit(intent); err != nil { - m.mu.Lock() - sb.ArchivePending = "" - rb := m.store.set(sb) - m.mu.Unlock() + rb, _ := m.setIfLive(sb, func() { sb.ArchivePending = "" }) sb.Transition.Unlock() m.store.recommit(ctx, rb) m.deleteOrphanArchiveCk(ctx, ck.ID) @@ -280,7 +277,7 @@ func (m *Manager) commitWake(ctx context.Context, sb *types.Sandbox, vmName, soc } func (m *Manager) deleteOrphanArchiveCk(ctx context.Context, ckID string) { - if err := m.deleteArchiveCk(ctx, ckID); err != nil { + if _, err := m.deleteArchiveCk(ctx, ckID); err != nil { log.WithFunc("pool.deleteOrphanArchiveCk").Warnf(ctx, "delete orphaned archive ck %s: %v", ckID, err) } } @@ -351,17 +348,28 @@ func (m *Manager) clearArchiveCk(ckID string) error { return err } -func (m *Manager) deleteArchiveCk(ctx context.Context, ckID string) error { +func (m *Manager) deleteArchiveCk(ctx context.Context, ckID string) (bool, error) { if err := m.markArchiveCk(ckID); err != nil { - return fmt.Errorf("track: %w", err) + return false, fmt.Errorf("track: %w", err) } - if err := m.deleteCkLocked(ctx, ckID); err != nil { - return err + return m.finishArchiveCkDelete(ctx, ckID) +} + +func (m *Manager) finishArchiveCkDelete(ctx context.Context, ckID string) (bool, error) { + unlock := m.lockRec(ckID) + if m.store.referencesCheckpoint(ckID) { + unlock() + return false, nil + } + err := m.ckpts.Delete(ctx, ckID) + unlock() + if err != nil { + return false, err } if err := m.clearArchiveCk(ckID); err != nil { - return fmt.Errorf("clear: %w", err) + return false, fmt.Errorf("clear: %w", err) } - return nil + return true, nil } func (m *Manager) retryArchiveDeletes(ctx context.Context) { diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index c9ffeee9..f3012e57 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -522,7 +522,7 @@ func TestArchiveWakeRequiresDeleteMarker(t *testing.T) { if !archived || !ckExists(t, m, ck) { t.Fatal("failed wake did not retain the archived claim and checkpoint") } - claims, err := m.store.load() + claims, err := newClaimStore(m.dataDir, false).load() if err != nil || claims[sb.ID] == nil || claims[sb.ID].ArchiveCk != ck { t.Fatalf("failed wake changed the durable archive claim: %+v, %v", claims[sb.ID], err) } @@ -680,11 +680,14 @@ func TestArchiveDeleteRetryRechecksWakeRollback(t *testing.T) { m.retryArchiveDeletes(t.Context()) close(retried) }() - waitFor(t, func() bool { - m.recLocksMu.Lock() - defer m.recLocksMu.Unlock() - return m.recRefs[ck] >= 2 - }) + select { + case <-retried: + case <-time.After(3 * time.Second): + t.Fatal("retry did not defer the checkpoint still referenced by the journal") + } + if !ckExists(t, m, ck) { + t.Fatal("retry deleted the checkpoint before wake persistence completed") + } breakStore(t, m) m.store.writeMu.Unlock() locked = false diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index e02b70b2..e6a32474 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -318,18 +318,8 @@ func (m *Manager) vetoIfHealPending(ckptID string) { func (m *Manager) pinnedArchiveCks() map[string]struct{} { m.mu.Lock() defer m.mu.Unlock() - pinned := maps.Clone(m.pendingCks) - for _, sb := range m.claimed { - if sb.ArchiveCk != "" { - pinned[sb.ArchiveCk] = struct{}{} - } - if sb.ReplicaCk != "" { - pinned[sb.ReplicaCk] = struct{}{} - } - if sb.ReplicaPending != "" { - pinned[sb.ReplicaPending] = struct{}{} - } - } + pinned := m.store.checkpointReferences() + maps.Copy(pinned, m.pendingCks) return pinned } diff --git a/sandboxd/pool/checkpoint_bench_test.go b/sandboxd/pool/checkpoint_bench_test.go new file mode 100644 index 00000000..e3ba1b6c --- /dev/null +++ b/sandboxd/pool/checkpoint_bench_test.go @@ -0,0 +1,67 @@ +package pool + +import ( + "fmt" + "path/filepath" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func BenchmarkClaimCommitWithCheckpoints(b *testing.B) { + for _, checkpoints := range []bool{false, true} { + b.Run(fmt.Sprintf("claims=5000/checkpoints=%t", checkpoints), func(b *testing.B) { + claims := checkpointBenchClaims(checkpoints) + s := newClaimStore(b.TempDir(), false) + if err := s.save(claims); err != nil { + b.Fatal(err) + } + one := claims["sb_0000000000000000"] + b.ReportAllocs() + for b.Loop() { + one.Deadline = one.Deadline.Add(time.Second) + if err := s.commit(s.set(one)); err != nil { + b.Fatal(err) + } + } + loaded, err := newClaimStore(filepath.Dir(s.path), false).load() + if err != nil || len(loaded) != 5000 || !loaded[one.ID].Deadline.Equal(one.Deadline) || loaded[one.ID].ReplicaCk != one.ReplicaCk { + b.Fatalf("commit output: rows=%d err=%v", len(loaded), err) + } + }) + } +} + +func BenchmarkPinnedArchiveCheckpoints(b *testing.B) { + for _, checkpoints := range []bool{false, true} { + b.Run(fmt.Sprintf("claims=5000/checkpoints=%t", checkpoints), func(b *testing.B) { + m := newTestManager(b, newFakeEngine(), config.PoolSpec{PoolKey: testKey}) + m.claimed = checkpointBenchClaims(checkpoints) + if err := m.store.save(m.claimed); err != nil { + b.Fatal(err) + } + want := 0 + if checkpoints { + want = 5000 + } + b.ReportAllocs() + for b.Loop() { + if got := len(m.pinnedArchiveCks()); got != want { + b.Fatalf("pinned=%d want=%d", got, want) + } + } + }) + } +} + +func checkpointBenchClaims(checkpoints bool) map[string]*types.Sandbox { + claims := benchClaims(5000) + if checkpoints { + for id, sb := range claims { + sb.ReplicaCk = "ck_" + id[3:] + } + } + return claims +} diff --git a/sandboxd/pool/checkpoint_cleanup_test.go b/sandboxd/pool/checkpoint_cleanup_test.go new file mode 100644 index 00000000..907ef02b --- /dev/null +++ b/sandboxd/pool/checkpoint_cleanup_test.go @@ -0,0 +1,223 @@ +package pool + +import ( + "errors" + "fmt" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestCheckpointCleanupWaitsForTheClaimWrite(t *testing.T) { + replicated := func(t *testing.T, m *Manager, _ *failoverCluster, sb *types.Sandbox) string { + replicateNow(t, m, sb) + return sb.ReplicaCk + } + keepsTombstone := func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) { + rec, ok, err := c.reg.Get(t.Context(), sb.ID) + if err != nil || !ok || !rec.Consumed { + t.Fatalf("retirement lost its tombstone: %+v, %t, %v", rec, ok, err) + } + m.reconcileOwnRow(t.Context(), rec, time.Now()) + if _, ok, err = c.reg.Get(t.Context(), sb.ID); err != nil || !ok { + t.Fatalf("sweep erased an unacknowledged retirement: %t, %v", ok, err) + } + } + for _, tc := range []struct { + name string + setup func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) string + act func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) + }{ + { + name: "replica retirement", + setup: replicated, + act: func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) { + sb.Transition.Lock() + err := m.dropReplica(t.Context(), sb) + sb.Transition.Unlock() + if err == nil || sb.HasReplica() { + t.Fatalf("dropReplica: replica=%t, %v; want a memory clear and the write error", sb.HasReplica(), err) + } + keepsTombstone(t, m, c, sb) + }, + }, + { + name: "retirement of a released claim", + setup: replicated, + act: func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) { + m.retireReplica(t.Context(), sb) + m.mu.Lock() + delete(m.claimed, sb.ID) + m.store.del(sb.ID) + m.mu.Unlock() + keepsTombstone(t, m, c, sb) + }, + }, + { + name: "unpublished replica intent", + setup: func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) string { + m.archives = &hookRegistry{Registry: c.reg, failPut: true} + if err := m.replicate(t.Context(), sb); !errors.Is(err, ErrArchiveStoreDown) { + t.Fatal(err) + } + m.archives = c.reg + return sb.ReplicaPending + }, + act: func(t *testing.T, m *Manager, _ *failoverCluster, sb *types.Sandbox) { + m.retireReplica(t.Context(), sb) + if sb.ReplicaPending != "" { + t.Fatal("retirement kept the intent in memory") + } + }, + }, + { + name: "settled replica intent", + setup: func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) string { + replicateNow(t, m, sb) + previous := sb.ReplicaCk + m.archives = &hookRegistry{Registry: c.reg, losePutReply: true} + if err := m.replicate(t.Context(), sb); !errors.Is(err, ErrArchiveStoreDown) { + t.Fatal(err) + } + m.archives = c.reg + return previous + }, + act: func(t *testing.T, m *Manager, _ *failoverCluster, sb *types.Sandbox) { + pending := sb.ReplicaPending + sb.Transition.Lock() + err := m.settleReplicaIntent(t.Context(), sb) + sb.Transition.Unlock() + if err == nil || sb.ReplicaPending != "" || sb.ReplicaCk != pending { + t.Fatalf("settle: %+v, %v; want the intent settled in memory and the write error", sb, err) + } + }, + }, + } { + for _, restart := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/restart=%t", tc.name, restart), func(t *testing.T) { + c := newReplicaCluster(t) + m := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, m, testKey) + ck := tc.setup(t, m, c, sb) + restoreWrites := failClaimWrites(t, m.store) + tc.act(t, m, c, sb) + m.retryArchiveDeletes(t.Context()) + if !ckExists(t, m, ck) || !m.deleteMarked(ck) { + t.Fatal("cleanup dropped the checkpoint or its delete marker before the claim write landed") + } + restoreWrites() + if restart { + m = c.restart(t, m) + if err := m.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + var ok bool + if sb, ok = m.byID(sb.ID); !ok { + t.Fatal("restart dropped the claim") + } + m.retireReplica(t.Context(), sb) + } else if err := m.store.commit(m.store.mark()); err != nil { + t.Fatal(err) + } + other := mustClaim(t, m, testKey) + m.mu.Lock() + other.LeaseSeconds++ + m.store.set(other) + m.mu.Unlock() + m.retryArchiveDeletes(t.Context()) + if ckExists(t, m, ck) || m.deleteMarked(ck) { + t.Fatal("cleanup did not finish once the claim write landed") + } + for _, live := range checkpointRefs(sb) { + if live != "" && !ckExists(t, m, live) { + t.Errorf("cleanup deleted live checkpoint %s", live) + } + } + m.sweepOrphanArchives(t.Context()) + if rec, ok, _ := c.reg.Get(t.Context(), sb.ID); ok && rec.Consumed { + t.Error("sweep kept the tombstone after the claim write landed") + } + }) + } + } +} + +func TestFailoverDisabledExitKeepsReplicaCheckpoint(t *testing.T) { + states := []struct { + name string + setup func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) string + }{ + {"published replica", func(t *testing.T, m *Manager, _ *failoverCluster, sb *types.Sandbox) string { + replicateNow(t, m, sb) + return sb.ReplicaCk + }}, + {"pending replica", func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) string { + m.archives = &hookRegistry{Registry: c.reg, failPut: true} + if err := m.replicate(t.Context(), sb); !errors.Is(err, ErrArchiveStoreDown) { + t.Fatal(err) + } + return sb.ReplicaPending + }}, + {"archive intent", func(t *testing.T, m *Manager, c *failoverCluster, sb *types.Sandbox) string { + if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatal(err) + } + m.archives = &hookRegistry{Registry: c.reg, failPut: true} + if err := m.archive(t.Context(), sb); !errors.Is(err, ErrArchiveStoreDown) { + t.Fatal(err) + } + return sb.ArchivePending + }}, + } + exits := []struct { + name string + run func(t *testing.T, m *Manager, sb *types.Sandbox) + }{ + {"release", func(t *testing.T, m *Manager, sb *types.Sandbox) { + if err := m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatal(err) + } + }}, + {"expire", func(t *testing.T, m *Manager, sb *types.Sandbox) { + m.mu.Lock() + sb.Deadline = time.Now().Add(-time.Minute) + m.mu.Unlock() + m.reapOnce(t.Context()) + waitFor(t, func() bool { return m.Counters().Reaps == 1 }) + }}, + } + for _, state := range states { + for _, exit := range exits { + t.Run(state.name+"/"+exit.name, func(t *testing.T) { + c := newReplicaCluster(t) + old := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, old, testKey) + ck := state.setup(t, old, c, sb) + cfg := &config.Config{ + APIToken: "root", DataDir: old.dataDir, CheckpointDir: c.ckDir, + Pools: []config.PoolSpec{{PoolKey: testKey}}, + } + m, err := NewManager(t.Context(), cfg, old.eng, testSecrets(t), testBackends(t, cfg)) + if err != nil { + t.Fatal(err) + } + if err = m.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + restored, ok := m.byID(sb.ID) + if !ok || checkpointRefs(restored) != checkpointRefs(sb) { + t.Fatalf("startup without a registry changed the claim: %+v", restored) + } + if !ckExists(t, m, ck) || m.deleteMarked(ck) { + t.Fatalf("startup without a registry queued or deleted checkpoint %s", ck) + } + exit.run(t, m, restored) + if _, held := m.byID(sb.ID); held || !ckExists(t, m, ck) { + t.Fatalf("%s without a registry: held=%t checkpoint kept=%t", exit.name, held, ckExists(t, m, ck)) + } + }) + } + } +} diff --git a/sandboxd/pool/checkpoint_references_test.go b/sandboxd/pool/checkpoint_references_test.go new file mode 100644 index 00000000..3125c68a --- /dev/null +++ b/sandboxd/pool/checkpoint_references_test.go @@ -0,0 +1,210 @@ +package pool + +import ( + "errors" + "io" + "io/fs" + "maps" + "os" + "slices" + "sync" + "syscall" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +const ( + failWrite = iota + captureWrite + failAfterRename +) + +func TestClaimCheckpointReferencesOutliveUnacknowledgedWrites(t *testing.T) { + recreate := func(s *claimStore) { s.set(&types.Sandbox{ID: "owner", ReplicaCk: "recreated"}) } + for _, tc := range []struct { + name string + fault int + acked, next *types.Sandbox + startup func(claims map[string]*types.Sandbox) + mutate func(s *claimStore) + onDisk, afterFault, afterCommit []string + }{ + { + name: "captured write then clear", fault: captureWrite, + next: &types.Sandbox{ID: "owner", ReplicaPending: "captured"}, + mutate: func(s *claimStore) { s.set(&types.Sandbox{ID: "owner"}) }, + onDisk: []string{"captured"}, afterFault: []string{"captured"}, + }, + { + name: "captured write then delete and recreate", fault: captureWrite, + next: &types.Sandbox{ID: "owner", ReplicaCk: "captured"}, + mutate: func(s *claimStore) { s.del("owner"); recreate(s) }, + onDisk: []string{"captured"}, afterFault: []string{"captured", "recreated"}, afterCommit: []string{"recreated"}, + }, + { + name: "captured write then reset and recreate", fault: captureWrite, + next: &types.Sandbox{ID: "owner", ReplicaCk: "captured"}, + mutate: func(s *claimStore) { s.reset(map[string]*types.Sandbox{}); recreate(s) }, + onDisk: []string{"captured"}, afterFault: []string{"captured", "recreated"}, afterCommit: []string{"recreated"}, + }, + { + name: "failed startup rewrite clears the loaded reference", fault: failWrite, + acked: &types.Sandbox{ID: "owner", ReplicaPending: "loaded"}, + startup: func(claims map[string]*types.Sandbox) { claims["owner"].ReplicaPending = "" }, + onDisk: []string{"loaded"}, afterFault: []string{"loaded"}, + }, + { + name: "failed startup rewrite drops the loaded claim", fault: failWrite, + acked: &types.Sandbox{ID: "owner", ReplicaPending: "loaded"}, + startup: func(claims map[string]*types.Sandbox) { delete(claims, "owner") }, + onDisk: []string{"loaded"}, afterFault: []string{"loaded"}, + }, + { + name: "failed writes then a return to the original", fault: failWrite, + acked: &types.Sandbox{ID: "owner", ReplicaCk: "original"}, + next: &types.Sandbox{ID: "owner", ReplicaCk: "attempt-1"}, + mutate: func(s *claimStore) { + for _, ck := range []string{"attempt-2", "attempt-3", "attempt-4"} { + _ = s.commit(s.set(&types.Sandbox{ID: "owner", ReplicaCk: ck})) + } + s.set(&types.Sandbox{ID: "owner", ReplicaCk: "original"}) + }, + onDisk: []string{"original"}, + afterFault: []string{"attempt-1", "attempt-2", "attempt-3", "attempt-4", "original"}, + afterCommit: []string{"original"}, + }, + { + name: "write fails after the rename", fault: failAfterRename, + acked: &types.Sandbox{ID: "owner", ArchiveCk: "archive", ArchivePending: "archive-pending", ReplicaCk: "replica", ReplicaPending: "replica-pending", FromCheckpoint: "branch-origin"}, + next: &types.Sandbox{ID: "owner", ReplicaPending: "uncertain"}, + mutate: func(s *claimStore) { + s.set(&types.Sandbox{ID: "owner"}, &types.Sandbox{ID: "other", ReplicaPending: "other-replica"}) + }, + onDisk: []string{"uncertain"}, + afterFault: []string{"archive", "archive-pending", "other-replica", "replica", "replica-pending", "uncertain"}, + afterCommit: []string{"other-replica"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + if tc.fault == failAfterRename && os.Geteuid() == 0 { + t.Skip("requires non-root: root can read the write-only directory") + } + dir := t.TempDir() + s := newClaimStore(dir, tc.fault == failAfterRename) + if tc.acked != nil { + if err := s.commit(s.set(tc.acked)); err != nil { + t.Fatal(err) + } + } + var snap claimSnapshot + if tc.startup != nil { + s = newClaimStore(dir, false) + claims, err := s.load() + if err != nil { + t.Fatal(err) + } + tc.startup(claims) + snap = s.reset(claims) + } else { + snap = s.set(tc.next) + } + var finish func() + switch tc.fault { + case failWrite: + finish = failClaimWrites(t, s) + if err := s.commit(snap); err == nil { + t.Fatal("write succeeded under the fault") + } + case failAfterRename: + if err := os.Chmod(dir, 0o300); err != nil { + t.Fatal(err) + } + finish = func() { _ = os.Chmod(dir, 0o700) } + t.Cleanup(finish) + if err := s.commit(snap); !errors.Is(err, fs.ErrPermission) { + t.Fatalf("want a failed directory sync after the rename: %v", err) + } + case captureWrite: + if err := syscall.Mkfifo(s.path+".tmp", 0o600); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- s.commit(snap) }() + waitFor(t, func() bool { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.retained["owner"]) != 0 + }) + finish = func() { + r, err := os.Open(s.path + ".tmp") + if err != nil { + t.Fatal(err) + } + raw, err := io.ReadAll(r) + _ = r.Close() + if err = errors.Join(err, <-done); err != nil { + t.Fatal(err) + } + if err = os.Remove(s.path); err != nil { + t.Fatal(err) + } + if err = os.WriteFile(s.path, raw, 0o600); err != nil { + t.Fatal(err) + } + } + } + if tc.mutate != nil { + tc.mutate(s) + } + finish() + loaded, err := newClaimStore(dir, false).load() + if err != nil { + t.Fatal(err) + } + var onDisk []string + for _, sb := range loaded { + onDisk = appendRefs(onDisk, checkpointRefs(sb)) + } + slices.Sort(onDisk) + if pinned := slices.Sorted(maps.Keys(s.checkpointReferences())); !slices.Equal(onDisk, tc.onDisk) || !slices.Equal(pinned, tc.afterFault) { + t.Fatalf("after the fault: on disk %v, pinned %v; want %v, %v", onDisk, pinned, tc.onDisk, tc.afterFault) + } + if err = s.commit(s.mark()); err != nil { + t.Fatal(err) + } + if pinned := slices.Sorted(maps.Keys(s.checkpointReferences())); !slices.Equal(pinned, tc.afterCommit) { + t.Fatalf("after the next commit: pinned %v, want %v", pinned, tc.afterCommit) + } + }) + } +} + +func TestClaimSharedCheckpointReferencesRetirePerOwner(t *testing.T) { + s := newClaimStore(t.TempDir(), false) + if err := s.commit(s.set(&types.Sandbox{ID: "first", ReplicaCk: "shared", ArchivePending: "shared"}, &types.Sandbox{ID: "second", ArchiveCk: "shared"})); err != nil { + t.Fatal(err) + } + if err := s.commit(s.del("first")); err != nil || !s.referencesCheckpoint("shared") { + t.Fatalf("one owner's retirement released another owner's checkpoint: %v", err) + } + if err := s.commit(s.del("second")); err != nil || len(s.checkpointReferences()) != 0 { + t.Fatalf("last owner left the shared checkpoint pinned: %v", err) + } +} + +func failClaimWrites(t *testing.T, s *claimStore) func() { + t.Helper() + tmp := s.path + ".tmp" + if err := os.Mkdir(tmp, 0o700); err != nil { + t.Fatal(err) + } + s.recommitting.Store(true) + restore := sync.OnceFunc(func() { + if err := os.Remove(tmp); err != nil { + t.Error(err) + } + }) + t.Cleanup(restore) + return restore +} diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 683074ee..f9427792 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -696,24 +696,25 @@ func (m *Manager) pauseExpired(ctx context.Context, sb *types.Sandbox) error { func (m *Manager) purgeArchiveCk(ctx context.Context, id, ck, tenant string) { logger := log.WithFunc("pool.purgeArchiveCk") - if err := m.deleteCkLocked(ctx, ck); err != nil { + if _, err := m.finishArchiveCkDelete(ctx, ck); err != nil { logger.Warnf(ctx, "delete archive ck %s: %v", ck, err) - } else if err := m.clearArchiveCk(ck); err != nil { - logger.Warnf(ctx, "clear archive ck %s: %v", ck, err) } m.counters.archiveDeletes.Add(1) m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant}) } -func (m *Manager) commitIfLive(ctx context.Context, sb *types.Sandbox, mutate func()) (bool, error) { +func (m *Manager) setIfLive(sb *types.Sandbox, mutate func()) (claimSnapshot, bool) { m.mu.Lock() - live := m.claimed[sb.ID] == sb - var js claimSnapshot - if live { - mutate() - js = m.store.set(sb) + defer m.mu.Unlock() + if m.claimed[sb.ID] != sb { + return claimSnapshot{}, false } - m.mu.Unlock() + mutate() + return m.store.set(sb), true +} + +func (m *Manager) commitIfLive(ctx context.Context, sb *types.Sandbox, mutate func()) (bool, error) { + js, live := m.setIfLive(sb, mutate) if !live { return false, nil } diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index 5d7111df..2f07ecaf 100644 --- a/sandboxd/pool/claims.go +++ b/sandboxd/pool/claims.go @@ -81,16 +81,24 @@ type claimStore struct { writeMu sync.Mutex buf []byte - mu sync.Mutex - rows map[string]*claimRow - seq uint64 - written uint64 + mu sync.Mutex + rows map[string]*claimRow + seq uint64 + written uint64 + refs map[string][4]string + retained map[string][]string + changed map[string]struct{} + checkpoints map[string]int recommitting atomic.Bool } func newClaimStore(dataDir string, sync bool) *claimStore { - return &claimStore{path: filepath.Join(dataDir, "claims.json"), sync: sync, rows: map[string]*claimRow{}} + return &claimStore{ + path: filepath.Join(dataDir, "claims.json"), sync: sync, + rows: map[string]*claimRow{}, refs: map[string][4]string{}, retained: map[string][]string{}, + changed: map[string]struct{}{}, checkpoints: map[string]int{}, + } } func (s *claimStore) load() (map[string]*types.Sandbox, error) { @@ -105,10 +113,16 @@ func (s *claimStore) load() (map[string]*types.Sandbox, error) { if err := json.Unmarshal(raw, &claims); err != nil { return nil, fmt.Errorf("parse claims: %w", err) } + s.mu.Lock() + defer s.mu.Unlock() for _, sb := range claims { if sb.VolumesWant == nil { sb.VolumesWant = sb.Volumes } + if refs := checkpointRefs(sb); refs != [4]string{} { + s.changeRetainedRefs(sb.ID, appendRefs(nil, refs)) + s.changed[sb.ID] = struct{}{} + } } return claims, nil } @@ -118,7 +132,7 @@ func (s *claimStore) set(sbs ...*types.Sandbox) claimSnapshot { s.mu.Lock() defer s.mu.Unlock() for _, sb := range sbs { - s.rows[sb.ID] = &claimRow{dto: dtoOf(sb)} + s.put(sb.ID, sb) } s.seq++ return claimSnapshot{seq: s.seq} @@ -129,12 +143,34 @@ func (s *claimStore) del(ids ...string) claimSnapshot { s.mu.Lock() defer s.mu.Unlock() for _, id := range ids { - delete(s.rows, id) + s.put(id, nil) } s.seq++ return claimSnapshot{seq: s.seq} } +func (s *claimStore) put(id string, sb *types.Sandbox) { + var refs [4]string + if sb == nil { + delete(s.rows, id) + } else { + refs = checkpointRefs(sb) + s.rows[id] = &claimRow{dto: dtoOf(sb)} + } + before := s.refs[id] + if before == refs { + return + } + s.changeCheckpointCounts(before[:], -1) + s.changeCheckpointCounts(refs[:], 1) + if refs == [4]string{} { + delete(s.refs, id) + } else { + s.refs[id] = refs + } + s.changed[id] = struct{}{} +} + // mark sequences a persist of the projection as it already stands. func (s *claimStore) mark() claimSnapshot { s.mu.Lock() @@ -152,9 +188,13 @@ func (s *claimStore) pending(snap claimSnapshot) bool { func (s *claimStore) reset(claims map[string]*types.Sandbox) claimSnapshot { s.mu.Lock() defer s.mu.Unlock() - s.rows = make(map[string]*claimRow, len(claims)) + for id := range s.rows { + if _, ok := claims[id]; !ok { + s.put(id, nil) + } + } for id, sb := range claims { - s.rows[id] = &claimRow{dto: dtoOf(sb)} + s.put(id, sb) } s.seq++ return claimSnapshot{seq: s.seq} @@ -170,6 +210,11 @@ func (s *claimStore) commit(snap claimSnapshot) error { return nil } rows, seq := slices.AppendSeq(make([]*claimRow, 0, len(s.rows)), maps.Values(s.rows)), s.seq + captured := make(map[string][4]string, len(s.changed)) + for id := range s.changed { + captured[id] = s.refs[id] + s.changeRetainedRefs(id, appendRefs(s.retained[id], s.refs[id])) + } s.mu.Unlock() s.buf = append(s.buf[:0], '{') for i, row := range rows { @@ -190,6 +235,12 @@ func (s *claimStore) commit(snap claimSnapshot) error { return err } s.mu.Lock() + for id, refs := range captured { + s.changeRetainedRefs(id, appendRefs(nil, refs)) + if s.refs[id] == refs { + delete(s.changed, id) + } + } s.written = seq s.mu.Unlock() return nil @@ -251,6 +302,44 @@ func (s *claimStore) synced() bool { return s.written == s.seq } +func (s *claimStore) changeCheckpointCounts(refs []string, delta int) { + for _, ck := range refs { + if ck == "" { + continue + } + s.checkpoints[ck] += delta + if s.checkpoints[ck] == 0 { + delete(s.checkpoints, ck) + } + } +} + +func (s *claimStore) changeRetainedRefs(id string, after []string) { + s.changeCheckpointCounts(s.retained[id], -1) + s.changeCheckpointCounts(after, 1) + if len(after) == 0 { + delete(s.retained, id) + } else { + s.retained[id] = after + } +} + +func (s *claimStore) referencesCheckpoint(id string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.checkpoints[id] > 0 +} + +func (s *claimStore) checkpointReferences() map[string]struct{} { + s.mu.Lock() + defer s.mu.Unlock() + refs := make(map[string]struct{}, len(s.checkpoints)) + for ck := range s.checkpoints { + refs[ck] = struct{}{} + } + return refs +} + func encodeRow(dto *claimDTO) ([]byte, error) { raw, err := json.Marshal(map[string]*claimDTO{dto.ID: dto}) if err != nil { @@ -279,3 +368,16 @@ func volumesWantDTO(sb *types.Sandbox) []types.Volume { } return append([]types.Volume{}, sb.VolumesWant...) } + +func checkpointRefs(sb *types.Sandbox) [4]string { + return [4]string{sb.ArchiveCk, sb.ArchivePending, sb.ReplicaCk, sb.ReplicaPending} +} + +func appendRefs(list []string, refs [4]string) []string { + for _, ck := range refs { + if ck != "" && !slices.Contains(list, ck) { + list = append(list, ck) + } + } + return list +} diff --git a/sandboxd/pool/claims_test.go b/sandboxd/pool/claims_test.go index 0539a537..f7e19311 100644 --- a/sandboxd/pool/claims_test.go +++ b/sandboxd/pool/claims_test.go @@ -88,7 +88,7 @@ func TestStoreRoundTrip(t *testing.T) { if err := s.save(claims); err != nil { t.Fatalf("save (sync=%v): %v", fsync, err) } - got, err := s.load() + got, err := newClaimStore(filepath.Dir(s.path), false).load() if err != nil { t.Fatalf("load (sync=%v): %v", fsync, err) } @@ -109,7 +109,7 @@ func TestClaimStoreSnapshotDetachesVolumes(t *testing.T) { if err := s.commit(snap); err != nil { t.Fatalf("commit: %v", err) } - got, err := s.load() + got, err := newClaimStore(filepath.Dir(s.path), false).load() if err != nil { t.Fatalf("load: %v", err) } @@ -165,7 +165,7 @@ func TestClaimStoreCommitCoalesces(t *testing.T) { if err := s.commit(older); err != nil { t.Fatalf("commit older: %v", err) } - if got, err := s.load(); err != nil || len(got) != 2 { + if got, err := newClaimStore(filepath.Dir(s.path), false).load(); err != nil || len(got) != 2 { t.Errorf("stale snapshot overwrote the newer state: %d claims (%v), want 2", len(got), err) } } @@ -181,7 +181,7 @@ func TestClaimStoreIncrementalProjection(t *testing.T) { if err := s.commit(s.set(a)); err != nil { t.Fatalf("commit update: %v", err) } - got, err := s.load() + got, err := newClaimStore(filepath.Dir(s.path), false).load() if err != nil { t.Fatalf("load: %v", err) } @@ -191,13 +191,13 @@ func TestClaimStoreIncrementalProjection(t *testing.T) { if delErr := s.commit(s.del("sb_a")); delErr != nil { t.Fatalf("commit del: %v", delErr) } - if got, err = s.load(); err != nil || len(got) != 1 || got["sb_b"] == nil { + if got, err = newClaimStore(filepath.Dir(s.path), false).load(); err != nil || len(got) != 1 || got["sb_b"] == nil { t.Fatalf("after del: %+v, %v", got, err) } if resetErr := s.commit(s.reset(map[string]*types.Sandbox{"sb_c": {ID: "sb_c", Key: testKey}})); resetErr != nil { t.Fatalf("commit reset: %v", resetErr) } - if got, err = s.load(); err != nil || len(got) != 1 || got["sb_c"] == nil { + if got, err = newClaimStore(filepath.Dir(s.path), false).load(); err != nil || len(got) != 1 || got["sb_c"] == nil { t.Fatalf("after reset: %+v, %v", got, err) } } @@ -214,7 +214,7 @@ func TestClaimStoreMarkRepersists(t *testing.T) { if !s.synced() { t.Error("not synced after flushing the mark") } - if got, err := s.load(); err != nil || len(got) != 1 { + if got, err := newClaimStore(filepath.Dir(s.path), false).load(); err != nil || len(got) != 1 { t.Errorf("after mark flush: %+v, %v", got, err) } } @@ -227,7 +227,7 @@ func TestClaimStoreConcurrentCommits(t *testing.T) { wg.Go(func() { _ = s.commit(s.set(sb)) }) } wg.Wait() - if got, err := s.load(); err != nil || len(got) != 1 { + if got, err := newClaimStore(filepath.Dir(s.path), false).load(); err != nil || len(got) != 1 { t.Fatalf("after concurrent commits: %+v, %v", got, err) } } diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index dd0896f4..b02b20dd 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -477,10 +477,14 @@ func (m *Manager) dropStaleRow(ctx context.Context, sb *types.Sandbox, rec archi func (m *Manager) purgeArchiveRow(ctx context.Context, rec archives.Record) { logger := log.WithFunc("pool.purgeArchiveRow") - if err := m.deleteArchiveCk(ctx, rec.Checkpoint); err != nil { + deleted, err := m.deleteArchiveCk(ctx, rec.Checkpoint) + if err != nil { logger.Warnf(ctx, "purge archive %s: delete ck %s: %v", rec.ID, rec.Checkpoint, err) return } + if !deleted { + return + } switch err := m.archives.Delete(ctx, rec.ID, rec.Epoch); { case errors.Is(err, archives.ErrLost): return @@ -647,5 +651,5 @@ func (m *Manager) archiveRowed(sb *types.Sandbox) bool { } func (m *Manager) expiryDropsRow(sb *types.Sandbox) bool { - return m.archiveRowed(sb) || sb.HasReplica() && !m.archivesAtDeadline(sb) + return m.archiveRowed(sb) || m.archives != nil && sb.HasReplica() && !m.archivesAtDeadline(sb) } diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index 0adb7413..505dc55c 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -40,7 +40,7 @@ func TestFailoverArchiveWritesTheRow(t *testing.T) { if err = json.Unmarshal(rec.Claim, claim); err != nil || claim.ID != sb.ID || claim.Token != "" || claim.VMName != "" || claim.ArchiveCk != sb.ArchiveCk { t.Errorf("row claim %+v %v, want the archived record without token or VM", claim, err) } - stored, err := a.store.load() + stored, err := newClaimStore(a.dataDir, false).load() if err != nil || stored[sb.ID].ArchiveEpoch != 1 { t.Errorf("claims.json epoch %v %v, want 1", stored[sb.ID], err) } @@ -1492,7 +1492,7 @@ func TestFailoverArchivePersistsItsIntentBeforeThePut(t *testing.T) { } var intent string a.archives = &hookRegistry{Registry: c.reg, beforePut: func() { - stored, err := a.store.load() + stored, err := newClaimStore(a.dataDir, false).load() if err != nil { t.Errorf("load: %v", err) return @@ -1502,7 +1502,7 @@ func TestFailoverArchivePersistsItsIntentBeforeThePut(t *testing.T) { if err := a.archive(t.Context(), sb); err != nil { t.Fatalf("archive: %v", err) } - stored, err := a.store.load() + stored, err := newClaimStore(a.dataDir, false).load() if err != nil { t.Fatalf("load: %v", err) } diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index cac827ed..4e464fbf 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -93,11 +93,12 @@ func (m *Manager) hibernateLocked(ctx context.Context, sb *types.Sandbox) error // cocoon rejects sb.ID's underscore, so the snapshot name is VMName-based snap := hibernatePrefix + strings.TrimPrefix(sb.VMName, vmPrefix) + "-" + randHex(3) // journal the intent first so Reconcile can adopt a hibernate whose commit never landed - if err := m.store.commit(m.setPendingSnap(sb, snap)); err != nil { - m.mu.Lock() - sb.PendingSnap = "" - m.store.set(sb) - m.mu.Unlock() + intent, live := m.setIfLive(sb, func() { sb.PendingSnap = snap }) + if !live { + return ErrUnknownSandbox + } + if err := m.store.commit(intent); err != nil { + m.setIfLive(sb, func() { sb.PendingSnap = "" }) return fmt.Errorf("hibernate %s: persist intent: %w", sb.ID, err) } readerSafe := false @@ -323,13 +324,6 @@ func (m *Manager) dropStale(ctx context.Context, sb *types.Sandbox) { } } -func (m *Manager) setPendingSnap(sb *types.Sandbox, snap string) claimSnapshot { - m.mu.Lock() - defer m.mu.Unlock() - sb.PendingSnap = snap - return m.store.set(sb) -} - // settlePendingSnap settles a dangling hibernate intent and bills an adopted one, which never was. func (m *Manager) settlePendingSnap(ctx context.Context, sb *types.Sandbox) (adopted bool, err error) { adopted, err = m.resolvePendingSnap(ctx, sb) diff --git a/sandboxd/pool/hibernate_test.go b/sandboxd/pool/hibernate_test.go index 1d355272..ffc4dd9e 100644 --- a/sandboxd/pool/hibernate_test.go +++ b/sandboxd/pool/hibernate_test.go @@ -35,7 +35,7 @@ func TestHibernatePersistFailureSurfacesAndConverges(t *testing.T) { if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { t.Fatalf("Hibernate retry after heal: %v", err) } - got, err := (&claimStore{path: broken}).load() + got, err := newClaimStore(filepath.Dir(broken), false).load() if err != nil || got[sb.ID] == nil || got[sb.ID].HibernateSnap == "" || got[sb.ID].PendingSnap != "" { t.Fatalf("journal after healed retry: %+v, %v", got[sb.ID], err) } @@ -75,7 +75,7 @@ func TestWakePersistFailureSurfaces(t *testing.T) { } return eng.snapRemoved(snap) }) - got, loadErr := (&claimStore{path: broken}).load() + got, loadErr := newClaimStore(filepath.Dir(broken), false).load() if loadErr != nil || got[sb.ID] == nil || got[sb.ID].HibernateSnap != "" { t.Fatalf("journal after converge: %+v, %v", got[sb.ID], loadErr) } @@ -305,7 +305,7 @@ func TestResolveAdoptBillsWhenPersistFails(t *testing.T) { t.Fatalf("heal store dir: %v", err) } waitFor(t, func() bool { - got, err := (&claimStore{path: broken}).load() + got, err := newClaimStore(filepath.Dir(broken), false).load() return err == nil && got[sb.ID] != nil && got[sb.ID].HibernateSnap != "" }) }) @@ -365,7 +365,7 @@ func TestFlushClaimsClosesShutdownWindow(t *testing.T) { if err := m.FlushClaims(); err != nil { t.Fatalf("FlushClaims: %v", err) } - got, err := (&claimStore{path: broken}).load() + got, err := newClaimStore(filepath.Dir(broken), false).load() if err != nil || got[sb.ID] == nil || got[sb.ID].HibernateSnap != "" { t.Fatalf("journal after flush: %+v, %v", got[sb.ID], err) } diff --git a/sandboxd/pool/imds_test.go b/sandboxd/pool/imds_test.go index 011d5693..5cca29d2 100644 --- a/sandboxd/pool/imds_test.go +++ b/sandboxd/pool/imds_test.go @@ -54,7 +54,10 @@ func TestSetInstanceMetadataRefusesAPausedSandboxWithoutWaking(t *testing.T) { if err := m.SetInstanceMetadata(t.Context(), running.ID, doc); !errors.Is(err, ErrPaused) { t.Errorf("pending hibernate: %v, want ErrPaused", err) } + m.mu.Lock() running.PendingSnap, running.ArchiveCk = "", "ck" + m.store.set(running) + m.mu.Unlock() if err := m.SetInstanceMetadata(t.Context(), running.ID, doc); !errors.Is(err, ErrArchived) { t.Errorf("archived: %v, want ErrArchived", err) } diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 136cd118..56e4148f 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -970,6 +970,7 @@ func TestRenewRefusesAnArchivedClaim(t *testing.T) { sb := mustClaim(t, m, testKey) m.mu.Lock() sb.ArchiveCk, sb.Deadline = "ck_1", time.Time{} + m.store.set(sb) m.mu.Unlock() if _, err := m.Renew(t.Context(), sb.ID, Cred{Token: sb.Token}, 0, ""); !errors.Is(err, ErrArchived) { @@ -1087,7 +1088,10 @@ func TestOperatorPortDialNeverWakesAPausedSandbox(t *testing.T) { "pending hibernate": func() { running.PendingSnap = "snap" }, "archived": func() { running.PendingSnap, running.ArchiveCk = "", "ck" }, } { + m.mu.Lock() set() + m.store.set(running) + m.mu.Unlock() if _, err := m.DialPort(t.Context(), running.ID, operator, 49983); !errors.Is(err, ErrPaused) { t.Errorf("%s: %v, want ErrPaused", name, err) } @@ -1939,6 +1943,17 @@ func newTestManagerAt(t testing.TB, eng *fakeEngine, dataDir string, pools ...co if err != nil { t.Fatalf("setup manager: %v", err) } + t.Cleanup(func() { + m.mu.Lock() + defer m.mu.Unlock() + m.store.mu.Lock() + defer m.store.mu.Unlock() + for id, sb := range m.claimed { + if m.store.rows[id] == nil || m.store.refs[id] != checkpointRefs(sb) { + t.Errorf("claim %s references differ from its store row", id) + } + } + }) return m } diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index 3bfa3cf1..a116b580 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -296,7 +296,10 @@ func (m *Manager) reclaimOrphanArchiveCks(ctx context.Context, claims map[string continue } orig, mine := claims[ckpt.SandboxID] - if !mine || ckpt.Node != m.nodeID || orig.ArchiveCk == ckpt.ID || orig.ReplicaCk == ckpt.ID || orig.ReplicaPending == ckpt.ID || orig.ArchivePending == ckpt.ID { + if !mine || ckpt.Node != m.nodeID { + continue + } + if refs := checkpointRefs(orig); slices.Contains(refs[:], ckpt.ID) { continue } orphans = append(orphans, ckpt) @@ -313,9 +316,9 @@ func (m *Manager) reclaimOrphanArchiveCks(ctx context.Context, claims map[string if row, ok := rows[ckpt.SandboxID]; ok && (row.Owner != m.nodeID || row.Checkpoint == ckpt.ID) { continue } - if err := m.deleteArchiveCk(ctx, ckpt.ID); err != nil { + if deleted, err := m.deleteArchiveCk(ctx, ckpt.ID); err != nil { logger.Warnf(ctx, "reclaim %s: %v", ckpt.ID, err) - } else { + } else if deleted { logger.Infof(ctx, "reclaimed orphan archive ck %s", ckpt.ID) } } diff --git a/sandboxd/pool/reconcile_test.go b/sandboxd/pool/reconcile_test.go index 907c2ba4..b50d78dc 100644 --- a/sandboxd/pool/reconcile_test.go +++ b/sandboxd/pool/reconcile_test.go @@ -24,7 +24,8 @@ func TestReconcileJournalsAnAdoptedHibernate(t *testing.T) { if err := eng.Hibernate(t.Context(), sb.VMName, snap); err != nil { t.Fatalf("engine hibernate: %v", err) } - if err := m.store.commit(m.setPendingSnap(sb, snap)); err != nil { + intent, _ := m.setIfLive(sb, func() { sb.PendingSnap = snap }) + if err := m.store.commit(intent); err != nil { t.Fatalf("journal intent: %v", err) } diff --git a/sandboxd/pool/replica.go b/sandboxd/pool/replica.go index 4706ddd4..6524154d 100644 --- a/sandboxd/pool/replica.go +++ b/sandboxd/pool/replica.go @@ -124,15 +124,13 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { return errWokeMeanwhile } if err := m.store.commit(intent); err != nil { - m.clearReplicaIntent(ctx, sb, ckID) - m.deleteOrphanArchiveCk(ctx, ckID) + _ = m.clearReplicaIntent(ctx, sb, ckID) return fmt.Errorf("replicate %s: persist intent: %w", sb.ID, err) } epoch, err := m.replaceReplica(ctx, sb, ckID) switch { case errors.Is(err, archives.ErrLost): - m.clearReplicaIntent(ctx, sb, ckID) - m.deleteOrphanArchiveCk(ctx, ckID) + _ = m.clearReplicaIntent(ctx, sb, ckID) if settleErr := m.refusedReplace(ctx, sb); settleErr != nil && !errors.Is(settleErr, ErrUnknownSandbox) { return fmt.Errorf("replicate %s: settle refused write: %w", sb.ID, settleErr) } @@ -184,7 +182,7 @@ func (m *Manager) settleReplicaIntent(ctx context.Context, sb *types.Sandbox) er return ErrUnknownSandbox case ok && !rec.Consumed && rec.Checkpoint == pending: var prev string - live, _ := m.commitIfLive(ctx, sb, func() { + live, commitErr := m.commitIfLive(ctx, sb, func() { prev, sb.ReplicaCk, sb.ReplicaPending, sb.ArchiveEpoch = sb.ReplicaCk, pending, "", rec.Epoch }) if !live { @@ -193,11 +191,12 @@ func (m *Manager) settleReplicaIntent(ctx context.Context, sb *types.Sandbox) er if prev != "" { m.deleteOrphanArchiveCk(ctx, prev) } + if commitErr != nil { + return fmt.Errorf("persist settled replica: %w", commitErr) + } return nil } - m.clearReplicaIntent(ctx, sb, pending) - m.deleteOrphanArchiveCk(ctx, pending) - return nil + return m.clearReplicaIntent(ctx, sb, pending) } // replicaGone reports whether the absence of sb's row means another node adopted the claim and finished with it. @@ -218,12 +217,20 @@ func (m *Manager) replicaGone(ctx context.Context, sb *types.Sandbox) (bool, err return false, err } -func (m *Manager) clearReplicaIntent(ctx context.Context, sb *types.Sandbox, ck string) { - _, _ = m.commitIfLive(ctx, sb, func() { +func (m *Manager) clearReplicaIntent(ctx context.Context, sb *types.Sandbox, ck string) error { + live, err := m.commitIfLive(ctx, sb, func() { if sb.ReplicaPending == ck { sb.ReplicaPending = "" } }) + m.deleteOrphanArchiveCk(ctx, ck) + if !live { + return ErrUnknownSandbox + } + if err != nil { + return fmt.Errorf("persist replica intent clear: %w", err) + } + return nil } func (m *Manager) retireReplica(ctx context.Context, sb *types.Sandbox) { @@ -247,12 +254,16 @@ func (m *Manager) dropReplica(ctx context.Context, sb *types.Sandbox) error { return err } var ck string - live, _ := m.commitIfLive(ctx, sb, func() { + live, err := m.commitIfLive(ctx, sb, func() { ck, sb.ReplicaCk = sb.ReplicaCk, "" }) if !live { return ErrUnknownSandbox } + if err != nil { + m.deleteOrphanArchiveCk(ctx, ck) + return fmt.Errorf("persist replica retirement: %w", err) + } m.dropTombstone(ctx, sb.ID, consumed) m.deleteOrphanArchiveCk(ctx, ck) return nil diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index e36dee45..0a157b61 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -32,7 +32,7 @@ func TestReplicaRoundPublishesMovesTheRowAndDropsThePrevious(t *testing.T) { if first == "" || rec.Owner != "a" || rec.Checkpoint != first || rec.Consumed || rec.TokenSHA256 != tenants.TokenSHA256(sb.Token) || !rec.Deadline.IsZero() { t.Fatalf("row %+v with replica %q, want a live row of a on the replica with the token hash and no deadline", rec, first) } - if stored, loadErr := a.store.load(); loadErr != nil || stored[sb.ID].ReplicaCk != first || stored[sb.ID].ArchiveEpoch != rec.Epoch { + if stored, loadErr := newClaimStore(a.dataDir, false).load(); loadErr != nil || stored[sb.ID].ReplicaCk != first || stored[sb.ID].ArchiveEpoch != rec.Epoch { t.Errorf("claims.json replica %q epoch %d (%v), want %q at %d", stored[sb.ID].ReplicaCk, stored[sb.ID].ArchiveEpoch, loadErr, first, rec.Epoch) } meta, err := a.loadCheckpoint(t.Context(), first) @@ -694,11 +694,18 @@ func TestReplicaExpiryDropsTheRowThenTheClaim(t *testing.T) { vm := lockedVMName(sb) replicateNow(t, a, sb) ck := a.replicaOf(sb) + checkpointBeforeDelete := make(chan bool, 1) + a.archives = &hookRegistry{Registry: c.reg, beforeDelete: func() { + checkpointBeforeDelete <- ckExists(t, a, ck) + }} a.mu.Lock() sb.Deadline = time.Now().Add(-time.Second) a.mu.Unlock() a.reapOnce(t.Context()) + if !<-checkpointBeforeDelete { + t.Fatal("checkpoint disappeared before its registry tombstone was deleted") + } waitFor(t, func() bool { return a.replicaOf(sb) == "" && !ckExists(t, a, ck) }) if _, ok, _ := c.reg.Get(t.Context(), sb.ID); ok { t.Error("the row outlived the first reap") diff --git a/sandboxd/pool/stale_claim_test.go b/sandboxd/pool/stale_claim_test.go new file mode 100644 index 00000000..885069aa --- /dev/null +++ b/sandboxd/pool/stale_claim_test.go @@ -0,0 +1,118 @@ +package pool + +import ( + "errors" + "fmt" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestStaleClaimOperationsPreserveTheCurrentClaim(t *testing.T) { + hibernate := func(t *testing.T, m *Manager, sb *types.Sandbox) error { + return m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}) + } + rollback := func(op func(*testing.T, *Manager, *types.Sandbox) error) func(*testing.T, *Manager, *types.Sandbox, func()) { + return func(t *testing.T, m *Manager, sb *types.Sandbox, stale func()) { + m.store.writeMu.Lock() + done := make(chan error, 1) + go func() { done <- op(t, m, sb) }() + waitFor(t, func() bool { + m.mu.Lock() + defer m.mu.Unlock() + return sb.PendingSnap != "" || sb.ArchivePending != "" + }) + stale() + restore := failClaimWrites(t, m.store) + m.store.writeMu.Unlock() + if err := <-done; err == nil { + t.Fatal("intent write succeeded despite filesystem failure") + } + restore() + } + } + for _, tc := range []struct { + name string + newManager func(t *testing.T) *Manager + prepare func(t *testing.T, m *Manager, sb *types.Sandbox) + run func(t *testing.T, m *Manager, sb *types.Sandbox, stale func()) + }{ + { + name: "hibernate", + newManager: func(t *testing.T) *Manager { return newTestManager(t, newFakeEngine()) }, + run: func(t *testing.T, m *Manager, sb *types.Sandbox, stale func()) { + stale() + sb.Transition.Lock() + err := m.hibernateLocked(t.Context(), sb) + sb.Transition.Unlock() + if !errors.Is(err, ErrUnknownSandbox) { + t.Errorf("stale Hibernate: %v, want ErrUnknownSandbox", err) + } + if m.eng.(*fakeEngine).hibernateCount() != 0 { + t.Error("stale Hibernate reached the engine") + } + }, + }, + { + name: "hibernate intent rollback", + newManager: func(t *testing.T) *Manager { return newTestManager(t, newFakeEngine()) }, + run: rollback(hibernate), + }, + { + name: "archive intent rollback", + newManager: func(t *testing.T) *Manager { return newFailoverCluster(t).node(t, "a", t.TempDir(), keepAnHour) }, + prepare: func(t *testing.T, m *Manager, sb *types.Sandbox) { + if err := hibernate(t, m, sb); err != nil { + t.Fatal(err) + } + }, + run: rollback(func(t *testing.T, m *Manager, sb *types.Sandbox) error { return m.archive(t.Context(), sb) }), + }, + } { + for _, replace := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/replace=%t", tc.name, replace), func(t *testing.T) { + m := tc.newManager(t) + sb := mustClaim(t, m, testKey) + ck, err := m.Checkpoint(t.Context(), sb.ID, Cred{Token: sb.Token}, "", "") + if err != nil { + t.Fatal(err) + } + if tc.prepare != nil { + tc.prepare(t, m, sb) + } + tc.run(t, m, sb, func() { + m.mu.Lock() + defer m.mu.Unlock() + m.dropClaimLocked(sb) + m.store.del(sb.ID) + if replace { + replacement := &types.Sandbox{ID: sb.ID, ReplicaCk: ck.ID} + m.addClaimLocked(replacement) + m.store.set(replacement) + } + }) + if err = m.store.commit(m.store.mark()); err != nil { + t.Fatal(err) + } + loaded, err := newClaimStore(m.dataDir, false).load() + if err != nil { + t.Fatal(err) + } + if replace { + if loaded[sb.ID] == nil || loaded[sb.ID].ReplicaCk != ck.ID || loaded[sb.ID].PendingSnap != "" { + t.Errorf("stale operation overwrote replacement on disk: %+v", loaded[sb.ID]) + } + } else if loaded[sb.ID] != nil { + t.Errorf("stale operation resurrected released claim on disk: %+v", loaded[sb.ID]) + } + if err = m.markArchiveCk(ck.ID); err != nil { + t.Fatal(err) + } + m.retryArchiveDeletes(t.Context()) + if exists := ckExists(t, m, ck.ID); exists != replace { + t.Errorf("checkpoint survived cleanup = %v, want %v", exists, replace) + } + }) + } + } +} diff --git a/sandboxd/pool/volume_mutation_test.go b/sandboxd/pool/volume_mutation_test.go index ce30fd1a..4f94b749 100644 --- a/sandboxd/pool/volume_mutation_test.go +++ b/sandboxd/pool/volume_mutation_test.go @@ -287,7 +287,7 @@ func TestLiveVolumeDoesNotBlockLifecycle(t *testing.T) { if attachErr == nil { t.Fatal("released worker reported success") } - claims, err := m.store.load() + claims, err := newClaimStore(m.dataDir, false).load() if err != nil || claims[sb.ID] != nil { t.Fatalf("late worker resurrected claim: %v %v", claims, err) } @@ -547,7 +547,7 @@ func TestReleaseOwnsUncertainHoldsUntilVMRemoval(t *testing.T) { if volumeHoldersOf(m, "data").readers != 1 { t.Fatal("late writer freed teardown-owned hold") } - claims, err := m.store.load() + claims, err := newClaimStore(m.dataDir, false).load() if err != nil || claims[sb.ID] != nil { t.Fatalf("late writer resurrected claim: %v %v", claims, err) }