detached b94f7d6

perf(daemon): move snapshot GC off startup and cut restore allocations

- Run snapshot GC in a background worker after restoration, then every
  30 minutes, instead of before the daemon socket is published.
- Durable mutations preempt a running GC pass through the recovery
  coordinator; a preempted pass retries after one minute. Shutdown
  cancels maintenance before publishing final checkpoints.
- GC rewinds a forward-orphan HEAD to the committed checkpoint instead
  of deleting it, so live sessions keep publishing.
- Sweep objects through one pinned root without per-shard fsync.
- Skip the checkpoint reload and HEAD rewrite in ReconcileCheckpoint
  when HEAD already matches.
- Read snapshot files into fstat-sized buffers and stop copying object
  payloads twice during restore.
commit
b94f7d6c232656d96fa6f95a004bf1a45f0f059b
author
brice · · 2026-10-02 05:46 UTC
parent
74436e9
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
20 files · +479 -82
Split
UnifiedSplit
M internal/adapters/snapshot/atomic.go +27 -1
@@ -342,7 +342,7 @@ func readBoundedFile(f *os.File) ([]byte, error) {
342342 _ = f.Close()
343343 return nil, fmt.Errorf("snapshot file too large")
344344 }
345- data, readErr := io.ReadAll(io.LimitReader(f, int64(maxRepositoryRead)+1))
345+ data, readErr := readSizedFile(f, int(stat.Size))
346346 closeErr := f.Close()
347347 if readErr != nil {
348348 return nil, readErr
@@ -355,3 +355,29 @@ func readBoundedFile(f *os.File) ([]byte, error) {
355355 }
356356 return data, nil
357357 }
358+
359+// readSizedFile reads f into a buffer sized from fstat, avoiding the repeated
360+// grow-and-copy of io.ReadAll for multi-megabyte snapshot objects. One spare
361+// byte detects a file that grew after fstat; the caller still enforces the
362+// read bound on the result.
363+func readSizedFile(f *os.File, size int) ([]byte, error) {
364+ data := make([]byte, 0, size+1)
365+ for {
366+ if len(data) > maxRepositoryRead {
367+ return data, nil
368+ }
369+ if len(data) == cap(data) {
370+ data = append(data, 0)[:len(data)]
371+ }
372+ // Never read past the bound plus the one detection byte, even when
373+ // append grew the buffer further.
374+ n, err := f.Read(data[len(data):min(cap(data), maxRepositoryRead+1)])
375+ data = data[:len(data)+n]
376+ if errors.Is(err, io.EOF) {
377+ return data, nil
378+ }
379+ if err != nil {
380+ return nil, err
381+ }
382+ }
383+}
@@ -342,7 +342,7 @@ func readBoundedFile(f *os.File) ([]byte, error) {
342 _ = f.Close()342 _ = f.Close()
343 return nil, fmt.Errorf("snapshot file too large")343 return nil, fmt.Errorf("snapshot file too large")
344 }344 }
345- data, readErr := io.ReadAll(io.LimitReader(f, int64(maxRepositoryRead)+1))345+ data, readErr := readSizedFile(f, int(stat.Size))
346 closeErr := f.Close()346 closeErr := f.Close()
347 if readErr != nil {347 if readErr != nil {
348 return nil, readErr348 return nil, readErr
@@ -355,3 +355,29 @@ func readBoundedFile(f *os.File) ([]byte, error) {
355 }355 }
356 return data, nil356 return data, nil
357 }357 }
358+
359+// readSizedFile reads f into a buffer sized from fstat, avoiding the repeated
360+// grow-and-copy of io.ReadAll for multi-megabyte snapshot objects. One spare
361+// byte detects a file that grew after fstat; the caller still enforces the
362+// read bound on the result.
363+func readSizedFile(f *os.File, size int) ([]byte, error) {
364+ data := make([]byte, 0, size+1)
365+ for {
366+ if len(data) > maxRepositoryRead {
367+ return data, nil
368+ }
369+ if len(data) == cap(data) {
370+ data = append(data, 0)[:len(data)]
371+ }
372+ // Never read past the bound plus the one detection byte, even when
373+ // append grew the buffer further.
374+ n, err := f.Read(data[len(data):min(cap(data), maxRepositoryRead+1)])
375+ data = data[:len(data)+n]
376+ if errors.Is(err, io.EOF) {
377+ return data, nil
378+ }
379+ if err != nil {
380+ return nil, err
381+ }
382+ }
383+}
M internal/adapters/snapshot/direct_load_test.go +40 -0
@@ -186,6 +186,46 @@ func TestReconcileCheckpoint(t *testing.T) {
186186 }
187187 }
188188
189+func TestReconcileCheckpointRepairsOnlyDivergentHead(t *testing.T) {
190+ tests := []struct {
191+ name string
192+ prepare func(t *testing.T, repo *Repository, id domain.IncarnationID)
193+ wantWrite bool
194+ }{
195+ {name: "matching HEAD is left untouched", prepare: func(*testing.T, *Repository, domain.IncarnationID) {}},
196+ {name: "missing HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
197+ require.NoError(t, os.Remove(repo.headPath(id)))
198+ }},
199+ {name: "corrupt HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
200+ require.NoError(t, os.WriteFile(repo.headPath(id), []byte("garbage"), 0o600))
201+ }},
202+ {name: "stale HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
203+ require.NoError(t, os.WriteFile(repo.headPath(id), marshalHead(1, [32]byte{9}), 0o600))
204+ }},
205+ }
206+ for _, tt := range tests {
207+ t.Run(tt.name, func(t *testing.T) {
208+ repo := NewRepository(privateDir(t))
209+ id := domain.IncarnationID{1}
210+ pub := incarnationPublication(t, id, "work", 1, nil)
211+ require.NoError(t, repo.Publish(t.Context(), pub))
212+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
213+ tt.prepare(t, repo, id)
214+ before, _ := os.Stat(repo.headPath(id))
215+
216+ require.NoError(t, repo.ReconcileCheckpoint(t.Context(), id, ref))
217+
218+ generation, digest, err := repo.readHead(id)
219+ require.NoError(t, err)
220+ require.Equal(t, ref.Generation, generation)
221+ require.Equal(t, ports.SnapshotDigest(ref.ManifestDigest), digest)
222+ after, err := os.Stat(repo.headPath(id))
223+ require.NoError(t, err)
224+ require.Equal(t, tt.wantWrite, before == nil || !os.SameFile(before, after), "HEAD rewrite")
225+ })
226+ }
227+}
228+
189229 func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
190230 for _, tt := range []struct {
191231 name string
@@ -186,6 +186,46 @@ func TestReconcileCheckpoint(t *testing.T) {
186 }186 }
187 }187 }
188 188
189+func TestReconcileCheckpointRepairsOnlyDivergentHead(t *testing.T) {
190+ tests := []struct {
191+ name string
192+ prepare func(t *testing.T, repo *Repository, id domain.IncarnationID)
193+ wantWrite bool
194+ }{
195+ {name: "matching HEAD is left untouched", prepare: func(*testing.T, *Repository, domain.IncarnationID) {}},
196+ {name: "missing HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
197+ require.NoError(t, os.Remove(repo.headPath(id)))
198+ }},
199+ {name: "corrupt HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
200+ require.NoError(t, os.WriteFile(repo.headPath(id), []byte("garbage"), 0o600))
201+ }},
202+ {name: "stale HEAD is repaired", wantWrite: true, prepare: func(t *testing.T, repo *Repository, id domain.IncarnationID) {
203+ require.NoError(t, os.WriteFile(repo.headPath(id), marshalHead(1, [32]byte{9}), 0o600))
204+ }},
205+ }
206+ for _, tt := range tests {
207+ t.Run(tt.name, func(t *testing.T) {
208+ repo := NewRepository(privateDir(t))
209+ id := domain.IncarnationID{1}
210+ pub := incarnationPublication(t, id, "work", 1, nil)
211+ require.NoError(t, repo.Publish(t.Context(), pub))
212+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
213+ tt.prepare(t, repo, id)
214+ before, _ := os.Stat(repo.headPath(id))
215+
216+ require.NoError(t, repo.ReconcileCheckpoint(t.Context(), id, ref))
217+
218+ generation, digest, err := repo.readHead(id)
219+ require.NoError(t, err)
220+ require.Equal(t, ref.Generation, generation)
221+ require.Equal(t, ports.SnapshotDigest(ref.ManifestDigest), digest)
222+ after, err := os.Stat(repo.headPath(id))
223+ require.NoError(t, err)
224+ require.Equal(t, tt.wantWrite, before == nil || !os.SameFile(before, after), "HEAD rewrite")
225+ })
226+ }
227+}
228+
189 func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {229 func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
190 for _, tt := range []struct {230 for _, tt := range []struct {
191 name string231 name string
M internal/adapters/snapshot/load.go +6 -0
@@ -78,6 +78,12 @@ func (r *Repository) ReconcileCheckpoint(ctx context.Context, id domain.Incarnat
7878 }
7979 lock := r.lockSession(key)
8080 defer r.unlockSession(lock)
81+ // HEAD normally already names the committed checkpoint. Skip the full
82+ // reload and the durable rewrite then; only a divergent or unreadable HEAD
83+ // needs repair, and repair still validates the checkpoint first.
84+ if generation, digest, err := r.readHead(id); err == nil && generation == ref.Generation && digest == ref.ManifestDigest {
85+ return nil
86+ }
8187 _, manifest, err := r.loadCheckpointLocked(ctx, id, "", ref)
8288 if err != nil {
8389 return fmt.Errorf("validate checkpoint before HEAD repair: %w", err)
@@ -78,6 +78,12 @@ func (r *Repository) ReconcileCheckpoint(ctx context.Context, id domain.Incarnat
78 }78 }
79 lock := r.lockSession(key)79 lock := r.lockSession(key)
80 defer r.unlockSession(lock)80 defer r.unlockSession(lock)
81+ // HEAD normally already names the committed checkpoint. Skip the full
82+ // reload and the durable rewrite then; only a divergent or unreadable HEAD
83+ // needs repair, and repair still validates the checkpoint first.
84+ if generation, digest, err := r.readHead(id); err == nil && generation == ref.Generation && digest == ref.ManifestDigest {
85+ return nil
86+ }
81 _, manifest, err := r.loadCheckpointLocked(ctx, id, "", ref)87 _, manifest, err := r.loadCheckpointLocked(ctx, id, "", ref)
82 if err != nil {88 if err != nil {
83 return fmt.Errorf("validate checkpoint before HEAD repair: %w", err)89 return fmt.Errorf("validate checkpoint before HEAD repair: %w", err)
M internal/adapters/snapshot/maintenance_gc.go +49 -15
@@ -7,6 +7,7 @@ import (
77 "fmt"
88 "os"
99 "path/filepath"
10+ "syscall"
1011
1112 "github.com/bnema/vev/internal/domain"
1213 "github.com/bnema/vev/internal/ports"
@@ -128,7 +129,7 @@ func (r *Repository) pruneGenerations(ctx context.Context, id domain.Incarnation
128129 }
129130 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)
130131 }
131- if err := r.removeHeadForDiscardedGenerations(id, kept); err != nil {
132+ if err := r.removeHeadForDiscardedGenerations(id, kept, committed); err != nil {
132133 return err
133134 }
134135 return r.sweepUnreferencedObjects(ctx, id, references)
@@ -137,8 +138,11 @@ func (r *Repository) pruneGenerations(ctx context.Context, id domain.Incarnation
137138 // removeHeadForDiscardedGenerations removes a repository HEAD only when its
138139 // generation was removed by the retention policy. In particular, an
139140 // incarnation with no committed checkpoint retains no generations, so its
140-// crash-published HEAD cannot make a retry conflict with an orphan.
141-func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}) error {
141+// crash-published HEAD cannot make a retry conflict with an orphan. When a
142+// committed checkpoint exists, HEAD is instead rewound to it: GC also runs
143+// while the daemon serves, and a live session's next publication must still
144+// find its committed parent.
145+func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}, committed domain.CheckpointRef) error {
142146 removeHead := len(kept) == 0
143147 if !removeHead {
144148 generation, _, err := r.readHead(id)
@@ -154,6 +158,13 @@ func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID,
154158 if !removeHead {
155159 return nil
156160 }
161+ if committed.Generation > 0 {
162+ // referencesForGenerations already validated the committed manifest digest.
163+ if err := r.writeMutable(r.headPath(id), marshalHead(committed.Generation, committed.ManifestDigest)); err != nil {
164+ return fmt.Errorf("rewind snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
165+ }
166+ return nil
167+ }
157168 if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
158169 return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
159170 }
@@ -229,6 +240,24 @@ func (r *Repository) readGarbageCollectionDirectory(path string) (entries []os.D
229240 if err != nil {
230241 return nil, err
231242 }
243+ return readAndCloseDirectory(directory)
244+}
245+
246+// readRootDirectory lists rel through an already pinned repository root, so a
247+// sweep over thousands of objects does not reopen and revalidate the root for
248+// every shard.
249+func readRootDirectory(root *os.Root, rel string) ([]os.DirEntry, error) {
250+ if err := rejectFinalSymlink(root, rel); err != nil {
251+ return nil, err
252+ }
253+ directory, err := root.OpenFile(rel, syscall.O_RDONLY|syscall.O_DIRECTORY, 0)
254+ if err != nil {
255+ return nil, err
256+ }
257+ return readAndCloseDirectory(directory)
258+}
259+
260+func readAndCloseDirectory(directory *os.File) (entries []os.DirEntry, err error) {
232261 defer func() {
233262 if closeErr := directory.Close(); closeErr != nil {
234263 err = errors.Join(err, closeErr)
@@ -237,9 +266,17 @@ func (r *Repository) readGarbageCollectionDirectory(path string) (entries []os.D
237266 return directory.ReadDir(-1)
238267 }
239268
240-func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.IncarnationID, references map[ports.SnapshotDigest]struct{}) error {
241- objectsRoot := filepath.Join(r.sessionPath(id), repositoryObjectsDir)
242- shards, err := r.readGarbageCollectionDirectory(objectsRoot)
269+func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.IncarnationID, references map[ports.SnapshotDigest]struct{}) (err error) {
270+ objectsRoot, ok := r.repositoryRelative(filepath.Join(r.sessionPath(id), repositoryObjectsDir))
271+ if !ok {
272+ return fmt.Errorf("snapshot path outside repository")
273+ }
274+ root, err := r.openRoot()
275+ if err != nil {
276+ return fmt.Errorf("open snapshot root for %s: %w", id.String(), safeFilesystemError(err))
277+ }
278+ defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
279+ shards, err := readRootDirectory(root, objectsRoot)
243280 if errors.Is(err, os.ErrNotExist) {
244281 return nil
245282 }
@@ -255,11 +292,14 @@ func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.Inc
255292 continue
256293 }
257294 dir := filepath.Join(objectsRoot, shard.Name())
258- objects, err := r.readGarbageCollectionDirectory(dir)
295+ objects, err := readRootDirectory(root, dir)
259296 if err != nil {
260297 return fmt.Errorf("read snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
261298 }
262- shardRemoved := false
299+ // Object unlinks are deliberately not fsynced per shard. Objects are
300+ // immutable and content-addressed: an unlink lost to a crash only brings
301+ // back an unreferenced object, which the next pass collects again, and a
302+ // later publication revalidates any existing object before reuse.
263303 for _, object := range objects {
264304 digest, canonical := parseObjectDigest(object.Name())
265305 if !canonical || object.IsDir() {
@@ -268,16 +308,10 @@ func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.Inc
268308 if _, retained := references[digest]; retained {
269309 continue
270310 }
271- if err := r.remove(filepath.Join(dir, object.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
311+ if err := root.Remove(filepath.Join(dir, object.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
272312 return fmt.Errorf("remove snapshot object for %s: %w", id.String(), safeFilesystemError(err))
273313 }
274314 removed++
275- shardRemoved = true
276- }
277- if shardRemoved {
278- if err := r.syncDirectory(dir); err != nil {
279- return fmt.Errorf("sync snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
280- }
281315 }
282316 }
283317 if removed > 0 {
@@ -7,6 +7,7 @@ import (
7 "fmt"7 "fmt"
8 "os"8 "os"
9 "path/filepath"9 "path/filepath"
10+ "syscall"
10 11
11 "github.com/bnema/vev/internal/domain"12 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"13 "github.com/bnema/vev/internal/ports"
@@ -128,7 +129,7 @@ func (r *Repository) pruneGenerations(ctx context.Context, id domain.Incarnation
128 }129 }
129 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)130 r.log.Info("snapshot_garbage_collected", "incarnation", id.String(), "action", "remove-generations", "removed", removed)
130 }131 }
131- if err := r.removeHeadForDiscardedGenerations(id, kept); err != nil {132+ if err := r.removeHeadForDiscardedGenerations(id, kept, committed); err != nil {
132 return err133 return err
133 }134 }
134 return r.sweepUnreferencedObjects(ctx, id, references)135 return r.sweepUnreferencedObjects(ctx, id, references)
@@ -137,8 +138,11 @@ func (r *Repository) pruneGenerations(ctx context.Context, id domain.Incarnation
137 // removeHeadForDiscardedGenerations removes a repository HEAD only when its138 // removeHeadForDiscardedGenerations removes a repository HEAD only when its
138 // generation was removed by the retention policy. In particular, an139 // generation was removed by the retention policy. In particular, an
139 // incarnation with no committed checkpoint retains no generations, so its140 // incarnation with no committed checkpoint retains no generations, so its
140-// crash-published HEAD cannot make a retry conflict with an orphan.141+// crash-published HEAD cannot make a retry conflict with an orphan. When a
141-func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}) error {142+// committed checkpoint exists, HEAD is instead rewound to it: GC also runs
143+// while the daemon serves, and a live session's next publication must still
144+// find its committed parent.
145+func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID, kept map[uint64]struct{}, committed domain.CheckpointRef) error {
142 removeHead := len(kept) == 0146 removeHead := len(kept) == 0
143 if !removeHead {147 if !removeHead {
144 generation, _, err := r.readHead(id)148 generation, _, err := r.readHead(id)
@@ -154,6 +158,13 @@ func (r *Repository) removeHeadForDiscardedGenerations(id domain.IncarnationID,
154 if !removeHead {158 if !removeHead {
155 return nil159 return nil
156 }160 }
161+ if committed.Generation > 0 {
162+ // referencesForGenerations already validated the committed manifest digest.
163+ if err := r.writeMutable(r.headPath(id), marshalHead(committed.Generation, committed.ManifestDigest)); err != nil {
164+ return fmt.Errorf("rewind snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
165+ }
166+ return nil
167+ }
157 if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {168 if err := r.remove(r.headPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
158 return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))169 return fmt.Errorf("remove snapshot HEAD for %s: %w", id.String(), safeFilesystemError(err))
159 }170 }
@@ -229,6 +240,24 @@ func (r *Repository) readGarbageCollectionDirectory(path string) (entries []os.D
229 if err != nil {240 if err != nil {
230 return nil, err241 return nil, err
231 }242 }
243+ return readAndCloseDirectory(directory)
244+}
245+
246+// readRootDirectory lists rel through an already pinned repository root, so a
247+// sweep over thousands of objects does not reopen and revalidate the root for
248+// every shard.
249+func readRootDirectory(root *os.Root, rel string) ([]os.DirEntry, error) {
250+ if err := rejectFinalSymlink(root, rel); err != nil {
251+ return nil, err
252+ }
253+ directory, err := root.OpenFile(rel, syscall.O_RDONLY|syscall.O_DIRECTORY, 0)
254+ if err != nil {
255+ return nil, err
256+ }
257+ return readAndCloseDirectory(directory)
258+}
259+
260+func readAndCloseDirectory(directory *os.File) (entries []os.DirEntry, err error) {
232 defer func() {261 defer func() {
233 if closeErr := directory.Close(); closeErr != nil {262 if closeErr := directory.Close(); closeErr != nil {
234 err = errors.Join(err, closeErr)263 err = errors.Join(err, closeErr)
@@ -237,9 +266,17 @@ func (r *Repository) readGarbageCollectionDirectory(path string) (entries []os.D
237 return directory.ReadDir(-1)266 return directory.ReadDir(-1)
238 }267 }
239 268
240-func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.IncarnationID, references map[ports.SnapshotDigest]struct{}) error {269+func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.IncarnationID, references map[ports.SnapshotDigest]struct{}) (err error) {
241- objectsRoot := filepath.Join(r.sessionPath(id), repositoryObjectsDir)270+ objectsRoot, ok := r.repositoryRelative(filepath.Join(r.sessionPath(id), repositoryObjectsDir))
242- shards, err := r.readGarbageCollectionDirectory(objectsRoot)271+ if !ok {
272+ return fmt.Errorf("snapshot path outside repository")
273+ }
274+ root, err := r.openRoot()
275+ if err != nil {
276+ return fmt.Errorf("open snapshot root for %s: %w", id.String(), safeFilesystemError(err))
277+ }
278+ defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
279+ shards, err := readRootDirectory(root, objectsRoot)
243 if errors.Is(err, os.ErrNotExist) {280 if errors.Is(err, os.ErrNotExist) {
244 return nil281 return nil
245 }282 }
@@ -255,11 +292,14 @@ func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.Inc
255 continue292 continue
256 }293 }
257 dir := filepath.Join(objectsRoot, shard.Name())294 dir := filepath.Join(objectsRoot, shard.Name())
258- objects, err := r.readGarbageCollectionDirectory(dir)295+ objects, err := readRootDirectory(root, dir)
259 if err != nil {296 if err != nil {
260 return fmt.Errorf("read snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))297 return fmt.Errorf("read snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
261 }298 }
262- shardRemoved := false299+ // Object unlinks are deliberately not fsynced per shard. Objects are
300+ // immutable and content-addressed: an unlink lost to a crash only brings
301+ // back an unreferenced object, which the next pass collects again, and a
302+ // later publication revalidates any existing object before reuse.
263 for _, object := range objects {303 for _, object := range objects {
264 digest, canonical := parseObjectDigest(object.Name())304 digest, canonical := parseObjectDigest(object.Name())
265 if !canonical || object.IsDir() {305 if !canonical || object.IsDir() {
@@ -268,16 +308,10 @@ func (r *Repository) sweepUnreferencedObjects(ctx context.Context, id domain.Inc
268 if _, retained := references[digest]; retained {308 if _, retained := references[digest]; retained {
269 continue309 continue
270 }310 }
271- if err := r.remove(filepath.Join(dir, object.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {311+ if err := root.Remove(filepath.Join(dir, object.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
272 return fmt.Errorf("remove snapshot object for %s: %w", id.String(), safeFilesystemError(err))312 return fmt.Errorf("remove snapshot object for %s: %w", id.String(), safeFilesystemError(err))
273 }313 }
274 removed++314 removed++
275- shardRemoved = true
276- }
277- if shardRemoved {
278- if err := r.syncDirectory(dir); err != nil {
279- return fmt.Errorf("sync snapshot object shard for %s: %w", id.String(), safeFilesystemError(err))
280- }
281 }315 }
282 }316 }
283 if removed > 0 {317 if removed > 0 {
M internal/adapters/snapshot/maintenance_gc_test.go +18 -0
@@ -37,6 +37,24 @@ func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
3737 require.NoError(t, err)
3838 }
3939
40+func TestCollectGarbageRewindsForwardOrphanHeadForLivePublication(t *testing.T) {
41+ t.Parallel()
42+ repository := NewRepository(privateDir(t))
43+ // Generation 4 reached the repository but its catalogue commit failed.
44+ publications := publishGarbageCollectionGenerations(t, repository, "work", 4)
45+ id := publications[0].IncarnationID
46+ committed := domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256(publications[2].Manifest)}
47+
48+ require.NoError(t, repository.CollectGarbage(t.Context(), map[domain.IncarnationID]domain.CheckpointRef{id: committed}))
49+
50+ generation, digest, err := repository.readHead(id)
51+ require.NoError(t, err)
52+ require.Equal(t, committed.Generation, generation)
53+ require.Equal(t, ports.SnapshotDigest(committed.ManifestDigest), digest)
54+ // The live session retries generation 4 against its committed parent.
55+ require.NoError(t, repository.Publish(t.Context(), publications[3]))
56+}
57+
4058 func TestCollectGarbageRejectsMalformedAuthorityWithoutMutation(t *testing.T) {
4159 for _, tt := range []struct {
4260 name string
@@ -37,6 +37,24 @@ func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
37 require.NoError(t, err)37 require.NoError(t, err)
38 }38 }
39 39
40+func TestCollectGarbageRewindsForwardOrphanHeadForLivePublication(t *testing.T) {
41+ t.Parallel()
42+ repository := NewRepository(privateDir(t))
43+ // Generation 4 reached the repository but its catalogue commit failed.
44+ publications := publishGarbageCollectionGenerations(t, repository, "work", 4)
45+ id := publications[0].IncarnationID
46+ committed := domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256(publications[2].Manifest)}
47+
48+ require.NoError(t, repository.CollectGarbage(t.Context(), map[domain.IncarnationID]domain.CheckpointRef{id: committed}))
49+
50+ generation, digest, err := repository.readHead(id)
51+ require.NoError(t, err)
52+ require.Equal(t, committed.Generation, generation)
53+ require.Equal(t, ports.SnapshotDigest(committed.ManifestDigest), digest)
54+ // The live session retries generation 4 against its committed parent.
55+ require.NoError(t, repository.Publish(t.Context(), publications[3]))
56+}
57+
40 func TestCollectGarbageRejectsMalformedAuthorityWithoutMutation(t *testing.T) {58 func TestCollectGarbageRejectsMalformedAuthorityWithoutMutation(t *testing.T) {
41 for _, tt := range []struct {59 for _, tt := range []struct {
42 name string60 name string
M internal/adapters/snapshot/repository_hardening_test.go +34 -0
@@ -62,6 +62,40 @@ func TestReadBoundedRejectsUnsafeFiles(t *testing.T) {
6262 }
6363 }
6464
65+func TestReadSizedFile(t *testing.T) {
66+ tests := []struct {
67+ name string
68+ content []byte
69+ hintSize int
70+ }{
71+ {name: "empty", content: nil, hintSize: 0},
72+ {name: "exact size hint", content: []byte("snapshot object"), hintSize: 15},
73+ {name: "file grew after stat", content: []byte("snapshot object grew"), hintSize: 4},
74+ {name: "file shrank after stat", content: []byte("tiny"), hintSize: 64},
75+ }
76+ for _, tt := range tests {
77+ t.Run(tt.name, func(t *testing.T) {
78+ path := filepath.Join(t.TempDir(), "object")
79+ if err := os.WriteFile(path, tt.content, 0o600); err != nil {
80+ t.Fatal(err)
81+ }
82+ f, err := os.Open(path)
83+ if err != nil {
84+ t.Fatal(err)
85+ }
86+ defer func() { _ = f.Close() }()
87+
88+ got, err := readSizedFile(f, tt.hintSize)
89+ if err != nil {
90+ t.Fatal(err)
91+ }
92+ if string(got) != string(tt.content) {
93+ t.Fatalf("readSizedFile = %q, want %q", got, tt.content)
94+ }
95+ })
96+ }
97+}
98+
6599 func TestValidatePublicationRejectsUnloadableAggregate(t *testing.T) {
66100 pub := repositoryPublication(t, "named", 1, []byte("state"))
67101 manifest, err := codec.UnmarshalManifest(pub.Manifest)
@@ -62,6 +62,40 @@ func TestReadBoundedRejectsUnsafeFiles(t *testing.T) {
62 }62 }
63 }63 }
64 64
65+func TestReadSizedFile(t *testing.T) {
66+ tests := []struct {
67+ name string
68+ content []byte
69+ hintSize int
70+ }{
71+ {name: "empty", content: nil, hintSize: 0},
72+ {name: "exact size hint", content: []byte("snapshot object"), hintSize: 15},
73+ {name: "file grew after stat", content: []byte("snapshot object grew"), hintSize: 4},
74+ {name: "file shrank after stat", content: []byte("tiny"), hintSize: 64},
75+ }
76+ for _, tt := range tests {
77+ t.Run(tt.name, func(t *testing.T) {
78+ path := filepath.Join(t.TempDir(), "object")
79+ if err := os.WriteFile(path, tt.content, 0o600); err != nil {
80+ t.Fatal(err)
81+ }
82+ f, err := os.Open(path)
83+ if err != nil {
84+ t.Fatal(err)
85+ }
86+ defer func() { _ = f.Close() }()
87+
88+ got, err := readSizedFile(f, tt.hintSize)
89+ if err != nil {
90+ t.Fatal(err)
91+ }
92+ if string(got) != string(tt.content) {
93+ t.Fatalf("readSizedFile = %q, want %q", got, tt.content)
94+ }
95+ })
96+ }
97+}
98+
65 func TestValidatePublicationRejectsUnloadableAggregate(t *testing.T) {99 func TestValidatePublicationRejectsUnloadableAggregate(t *testing.T) {
66 pub := repositoryPublication(t, "named", 1, []byte("state"))100 pub := repositoryPublication(t, "named", 1, []byte("state"))
67 manifest, err := codec.UnmarshalManifest(pub.Manifest)101 manifest, err := codec.UnmarshalManifest(pub.Manifest)
M internal/app/daemon_integration_test.go +1 -4
@@ -1044,9 +1044,6 @@ func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
10441044 daemon.WithDurableMaintenance(catalogue, maintenanceRepository),
10451045 daemon.WithCatalogue(catalogue, nil),
10461046 )
1047- if err := d.CollectStartupGarbage(ctx); err != nil {
1048- return err
1049- }
10501047 listener, err := ipc.Listen(runtimeDir)
10511048 if err != nil {
10521049 return err
@@ -1055,7 +1052,7 @@ func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
10551052 })
10561053 }()
10571054
1058- awaitLifecycleStage(t, maintenanceRepository.entered, "pre-publication maintenance repository call")
1055+ awaitLifecycleStage(t, maintenanceRepository.entered, "background maintenance repository call")
10591056 cancel()
10601057 select {
10611058 case <-callbackReturned:
@@ -1044,9 +1044,6 @@ func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
1044 daemon.WithDurableMaintenance(catalogue, maintenanceRepository),1044 daemon.WithDurableMaintenance(catalogue, maintenanceRepository),
1045 daemon.WithCatalogue(catalogue, nil),1045 daemon.WithCatalogue(catalogue, nil),
1046 )1046 )
1047- if err := d.CollectStartupGarbage(ctx); err != nil {
1048- return err
1049- }
1050 listener, err := ipc.Listen(runtimeDir)1047 listener, err := ipc.Listen(runtimeDir)
1051 if err != nil {1048 if err != nil {
1052 return err1049 return err
@@ -1055,7 +1052,7 @@ func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
1055 })1052 })
1056 }()1053 }()
1057 1054
1058- awaitLifecycleStage(t, maintenanceRepository.entered, "pre-publication maintenance repository call")1055+ awaitLifecycleStage(t, maintenanceRepository.entered, "background maintenance repository call")
1059 cancel()1056 cancel()
1060 select {1057 select {
1061 case <-callbackReturned:1058 case <-callbackReturned:
M internal/app/run.go +2 -13
@@ -706,13 +706,9 @@ func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord
706706
707707 func constructDaemonBeforeSocketPublication(
708708 construct func() *daemon.Daemon,
709- prepare func(*daemon.Daemon) error,
710709 listen func() (wire.Listener, error),
711710 ) (*daemon.Daemon, wire.Listener, error) {
712711 d := construct()
713- if err := prepare(d); err != nil {
714- return d, nil, err
715- }
716712 ln, err := listen()
717713 return d, ln, err
718714 }
@@ -832,16 +828,9 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
832828 }()
833829
834830 d, ln, err := constructDaemonBeforeSocketPublication(
831+ // Snapshot GC runs in the background once restoration finishes, so it
832+ // never delays socket publication.
835833 func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
836- func(d *daemon.Daemon) error {
837- if err := d.CollectStartupGarbage(ctx); err != nil {
838- // GC is best-effort, but it is fully finished before socket
839- // publication. A failed pass leaves durable state untouched and
840- // restoration retains its per-session failure isolation.
841- log.Warn("snapshot_garbage_collection_failed", "err", err)
842- }
843- return nil
844- },
845834 func() (wire.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
846835 )
847836 if err != nil {
@@ -706,13 +706,9 @@ func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord
706 706
707 func constructDaemonBeforeSocketPublication(707 func constructDaemonBeforeSocketPublication(
708 construct func() *daemon.Daemon,708 construct func() *daemon.Daemon,
709- prepare func(*daemon.Daemon) error,
710 listen func() (wire.Listener, error),709 listen func() (wire.Listener, error),
711 ) (*daemon.Daemon, wire.Listener, error) {710 ) (*daemon.Daemon, wire.Listener, error) {
712 d := construct()711 d := construct()
713- if err := prepare(d); err != nil {
714- return d, nil, err
715- }
716 ln, err := listen()712 ln, err := listen()
717 return d, ln, err713 return d, ln, err
718 }714 }
@@ -832,16 +828,9 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
832 }()828 }()
833 829
834 d, ln, err := constructDaemonBeforeSocketPublication(830 d, ln, err := constructDaemonBeforeSocketPublication(
831+ // Snapshot GC runs in the background once restoration finishes, so it
832+ // never delays socket publication.
835 func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },833 func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
836- func(d *daemon.Daemon) error {
837- if err := d.CollectStartupGarbage(ctx); err != nil {
838- // GC is best-effort, but it is fully finished before socket
839- // publication. A failed pass leaves durable state untouched and
840- // restoration retains its per-session failure isolation.
841- log.Warn("snapshot_garbage_collection_failed", "err", err)
842- }
843- return nil
844- },
845 func() (wire.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },834 func() (wire.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
846 )835 )
847 if err != nil {836 if err != nil {
M internal/app/run_test.go +2 -7
@@ -176,19 +176,14 @@ func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
176176 events = append(events, "catalogue-registry")
177177 return nil
178178 },
179- func(*daemon.Daemon) error {
180- require.Equal(t, []string{"catalogue-registry"}, events)
181- events = append(events, "startup-garbage-collection")
182- return nil
183- },
184179 func() (wire.Listener, error) {
185- require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection"}, events)
180+ require.Equal(t, []string{"catalogue-registry"}, events)
186181 events = append(events, "socket-publication")
187182 return nil, nil
188183 },
189184 )
190185 require.NoError(t, err)
191- require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection", "socket-publication"}, events)
186+ require.Equal(t, []string{"catalogue-registry", "socket-publication"}, events)
192187 }
193188
194189 func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {
@@ -176,19 +176,14 @@ func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
176 events = append(events, "catalogue-registry")176 events = append(events, "catalogue-registry")
177 return nil177 return nil
178 },178 },
179- func(*daemon.Daemon) error {
180- require.Equal(t, []string{"catalogue-registry"}, events)
181- events = append(events, "startup-garbage-collection")
182- return nil
183- },
184 func() (wire.Listener, error) {179 func() (wire.Listener, error) {
185- require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection"}, events)180+ require.Equal(t, []string{"catalogue-registry"}, events)
186 events = append(events, "socket-publication")181 events = append(events, "socket-publication")
187 return nil, nil182 return nil, nil
188 },183 },
189 )184 )
190 require.NoError(t, err)185 require.NoError(t, err)
191- require.Equal(t, []string{"catalogue-registry", "startup-garbage-collection", "socket-publication"}, events)186+ require.Equal(t, []string{"catalogue-registry", "socket-publication"}, events)
192 }187 }
193 188
194 func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {189 func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {
M internal/usecase/daemon/daemon.go +9 -3
@@ -299,6 +299,7 @@ type Daemon struct {
299299 // snapshotRepository is the sole checkpoint storage contract.
300300 snapshotRepository ports.SnapshotRepository
301301 recovery *recoveryusecase.Coordinator
302+ durableMaintenance bool
302303 maintenanceWorkerCancel context.CancelFunc
303304 maintenanceWorkerDone chan struct{}
304305 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
@@ -555,14 +556,15 @@ func WithNoticeStore(store ports.NoticeStore) Option {
555556 return func(d *Daemon) { d.noticeStore = store }
556557 }
557558
558-// WithDurableMaintenance retains the application wiring for the one
559-// pre-publication GC pass. Standalone users without an explicitly supplied
560-// coordinator receive the same canonical coordinator path.
559+// WithDurableMaintenance enables background snapshot GC. Standalone users
560+// without an explicitly supplied coordinator receive the same canonical
561+// coordinator path.
561562 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
562563 return func(d *Daemon) {
563564 if d.recovery == nil {
564565 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, nil)
565566 }
567+ d.durableMaintenance = true
566568 }
567569 }
568570
@@ -830,6 +832,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.ServerListener) error {
830832 d.snapshotRepositorySaver(d.serveCtx)
831833 })
832834 d.startSnapshotRestoration()
835+ d.startDurableMaintenance()
833836 } else {
834837 d.closeRestoreDone()
835838 }
@@ -926,6 +929,9 @@ func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
926929 // shutdownAll and Serve's exit path. It is separate from purgeAllSessions,
927930 // which leaves every daemon-wide shutdown fact untouched.
928931 func (d *Daemon) terminateAllForShutdown(reason uint8, deadline *snapshotShutdownDeadline) (checkpointIncomplete bool) {
932+ // Stop background GC first so it releases the recovery fence before final
933+ // checkpoints are published under the shutdown deadline.
934+ d.cancelDurableMaintenance()
929935 d.closeMoveLifecycles()
930936 d.mu.Lock()
931937 d.closing = true
@@ -299,6 +299,7 @@ type Daemon struct {
299 // snapshotRepository is the sole checkpoint storage contract.299 // snapshotRepository is the sole checkpoint storage contract.
300 snapshotRepository ports.SnapshotRepository300 snapshotRepository ports.SnapshotRepository
301 recovery *recoveryusecase.Coordinator301 recovery *recoveryusecase.Coordinator
302+ durableMaintenance bool
302 maintenanceWorkerCancel context.CancelFunc303 maintenanceWorkerCancel context.CancelFunc
303 maintenanceWorkerDone chan struct{}304 maintenanceWorkerDone chan struct{}
304 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup305 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
@@ -555,14 +556,15 @@ func WithNoticeStore(store ports.NoticeStore) Option {
555 return func(d *Daemon) { d.noticeStore = store }556 return func(d *Daemon) { d.noticeStore = store }
556 }557 }
557 558
558-// WithDurableMaintenance retains the application wiring for the one559+// WithDurableMaintenance enables background snapshot GC. Standalone users
559-// pre-publication GC pass. Standalone users without an explicitly supplied560+// without an explicitly supplied coordinator receive the same canonical
560-// coordinator receive the same canonical coordinator path.561+// coordinator path.
561 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {562 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
562 return func(d *Daemon) {563 return func(d *Daemon) {
563 if d.recovery == nil {564 if d.recovery == nil {
564 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, nil)565 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, nil)
565 }566 }
567+ d.durableMaintenance = true
566 }568 }
567 }569 }
568 570
@@ -830,6 +832,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.ServerListener) error {
830 d.snapshotRepositorySaver(d.serveCtx)832 d.snapshotRepositorySaver(d.serveCtx)
831 })833 })
832 d.startSnapshotRestoration()834 d.startSnapshotRestoration()
835+ d.startDurableMaintenance()
833 } else {836 } else {
834 d.closeRestoreDone()837 d.closeRestoreDone()
835 }838 }
@@ -926,6 +929,9 @@ func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
926 // shutdownAll and Serve's exit path. It is separate from purgeAllSessions,929 // shutdownAll and Serve's exit path. It is separate from purgeAllSessions,
927 // which leaves every daemon-wide shutdown fact untouched.930 // which leaves every daemon-wide shutdown fact untouched.
928 func (d *Daemon) terminateAllForShutdown(reason uint8, deadline *snapshotShutdownDeadline) (checkpointIncomplete bool) {931 func (d *Daemon) terminateAllForShutdown(reason uint8, deadline *snapshotShutdownDeadline) (checkpointIncomplete bool) {
932+ // Stop background GC first so it releases the recovery fence before final
933+ // checkpoints are published under the shutdown deadline.
934+ d.cancelDurableMaintenance()
929 d.closeMoveLifecycles()935 d.closeMoveLifecycles()
930 d.mu.Lock()936 d.mu.Lock()
931 d.closing = true937 d.closing = true
M internal/usecase/daemon/maintenance.go +80 -4
@@ -2,18 +2,94 @@ package daemon
22
33 import (
44 "context"
5+ "errors"
56 "fmt"
7+ "time"
8+
9+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
610 )
711
8-// CollectStartupGarbage runs the one coordinator-owned GC pass before the
9-// daemon socket is published.
10-func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {
12+// snapshotGarbageCollectionInterval spaces the periodic snapshot GC passes. A
13+// long-running daemon otherwise accumulates superseded generations and objects
14+// until its next restart, which then pays for the whole backlog.
15+const snapshotGarbageCollectionInterval = 30 * time.Minute
16+
17+// snapshotGarbageCollectionRetryInterval follows a pass that yielded to a
18+// durable mutation. Work already done is kept, so a retry resumes cheaply.
19+const snapshotGarbageCollectionRetryInterval = time.Minute
20+
21+// startDurableMaintenance runs snapshot GC off the startup critical path: once
22+// catalogue restoration has finished, then periodically. The recovery
23+// coordinator fences every pass against catalogue mutations and checkpoint
24+// publications, so the keep set is never stale relative to a concurrent writer,
25+// and any mutation preempts a running pass so it never waits for maintenance.
26+func (d *Daemon) startDurableMaintenance() {
27+ if d == nil || d.recovery == nil || !d.durableMaintenance {
28+ return
29+ }
30+ d.snapshotWorkerMu.Lock()
31+ if d.maintenanceWorkerCancel != nil {
32+ d.snapshotWorkerMu.Unlock()
33+ return
34+ }
35+ ctx, cancel := context.WithCancel(d.serveCtx)
36+ d.maintenanceWorkerCancel = cancel
37+ done := make(chan struct{})
38+ d.maintenanceWorkerDone = done
39+ d.snapshotWorkerMu.Unlock()
40+
41+ go func() {
42+ defer close(done)
43+ d.runDurableMaintenance(ctx)
44+ }()
45+}
46+
47+// cancelDurableMaintenance stops the GC worker without waiting for it;
48+// WaitDurableWriters owns the join.
49+func (d *Daemon) cancelDurableMaintenance() {
50+ d.snapshotWorkerMu.Lock()
51+ defer d.snapshotWorkerMu.Unlock()
52+ if d.maintenanceWorkerCancel != nil {
53+ d.maintenanceWorkerCancel()
54+ }
55+}
56+
57+func (d *Daemon) runDurableMaintenance(ctx context.Context) {
58+ // Restoration reads committed checkpoints; collecting afterwards keeps the
59+ // pass from competing with it for disk and for the coordinator fence.
60+ select {
61+ case <-ctx.Done():
62+ return
63+ case <-d.restoreDone:
64+ }
65+ for {
66+ interval := snapshotGarbageCollectionInterval
67+ switch err := d.collectSnapshotGarbage(ctx); {
68+ case err == nil || ctx.Err() != nil:
69+ case errors.Is(err, recoveryusecase.ErrGarbageCollectionPreempted):
70+ d.log.Debug("snapshot_garbage_collection_preempted")
71+ interval = snapshotGarbageCollectionRetryInterval
72+ default:
73+ d.log.Warn("snapshot_garbage_collection_failed", "err", err)
74+ }
75+ timer := d.clock.NewTimer(interval)
76+ select {
77+ case <-ctx.Done():
78+ timer.Stop()
79+ return
80+ case <-timer.C():
81+ }
82+ }
83+}
84+
85+// collectSnapshotGarbage runs one coordinator-owned GC pass.
86+func (d *Daemon) collectSnapshotGarbage(ctx context.Context) error {
1187 if d == nil || d.recovery == nil {
1288 return nil
1389 }
1490 incarnations, err := d.recovery.CollectGarbage(ctx)
1591 if err != nil {
16- return fmt.Errorf("collect startup snapshots: %w", err)
92+ return fmt.Errorf("collect snapshots: %w", err)
1793 }
1894 d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
1995 return nil
@@ -2,18 +2,94 @@ package daemon
2 2
3 import (3 import (
4 "context"4 "context"
5+ "errors"
5 "fmt"6 "fmt"
7+ "time"
8+
9+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
6 )10 )
7 11
8-// CollectStartupGarbage runs the one coordinator-owned GC pass before the12+// snapshotGarbageCollectionInterval spaces the periodic snapshot GC passes. A
9-// daemon socket is published.13+// long-running daemon otherwise accumulates superseded generations and objects
10-func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {14+// until its next restart, which then pays for the whole backlog.
15+const snapshotGarbageCollectionInterval = 30 * time.Minute
16+
17+// snapshotGarbageCollectionRetryInterval follows a pass that yielded to a
18+// durable mutation. Work already done is kept, so a retry resumes cheaply.
19+const snapshotGarbageCollectionRetryInterval = time.Minute
20+
21+// startDurableMaintenance runs snapshot GC off the startup critical path: once
22+// catalogue restoration has finished, then periodically. The recovery
23+// coordinator fences every pass against catalogue mutations and checkpoint
24+// publications, so the keep set is never stale relative to a concurrent writer,
25+// and any mutation preempts a running pass so it never waits for maintenance.
26+func (d *Daemon) startDurableMaintenance() {
27+ if d == nil || d.recovery == nil || !d.durableMaintenance {
28+ return
29+ }
30+ d.snapshotWorkerMu.Lock()
31+ if d.maintenanceWorkerCancel != nil {
32+ d.snapshotWorkerMu.Unlock()
33+ return
34+ }
35+ ctx, cancel := context.WithCancel(d.serveCtx)
36+ d.maintenanceWorkerCancel = cancel
37+ done := make(chan struct{})
38+ d.maintenanceWorkerDone = done
39+ d.snapshotWorkerMu.Unlock()
40+
41+ go func() {
42+ defer close(done)
43+ d.runDurableMaintenance(ctx)
44+ }()
45+}
46+
47+// cancelDurableMaintenance stops the GC worker without waiting for it;
48+// WaitDurableWriters owns the join.
49+func (d *Daemon) cancelDurableMaintenance() {
50+ d.snapshotWorkerMu.Lock()
51+ defer d.snapshotWorkerMu.Unlock()
52+ if d.maintenanceWorkerCancel != nil {
53+ d.maintenanceWorkerCancel()
54+ }
55+}
56+
57+func (d *Daemon) runDurableMaintenance(ctx context.Context) {
58+ // Restoration reads committed checkpoints; collecting afterwards keeps the
59+ // pass from competing with it for disk and for the coordinator fence.
60+ select {
61+ case <-ctx.Done():
62+ return
63+ case <-d.restoreDone:
64+ }
65+ for {
66+ interval := snapshotGarbageCollectionInterval
67+ switch err := d.collectSnapshotGarbage(ctx); {
68+ case err == nil || ctx.Err() != nil:
69+ case errors.Is(err, recoveryusecase.ErrGarbageCollectionPreempted):
70+ d.log.Debug("snapshot_garbage_collection_preempted")
71+ interval = snapshotGarbageCollectionRetryInterval
72+ default:
73+ d.log.Warn("snapshot_garbage_collection_failed", "err", err)
74+ }
75+ timer := d.clock.NewTimer(interval)
76+ select {
77+ case <-ctx.Done():
78+ timer.Stop()
79+ return
80+ case <-timer.C():
81+ }
82+ }
83+}
84+
85+// collectSnapshotGarbage runs one coordinator-owned GC pass.
86+func (d *Daemon) collectSnapshotGarbage(ctx context.Context) error {
11 if d == nil || d.recovery == nil {87 if d == nil || d.recovery == nil {
12 return nil88 return nil
13 }89 }
14 incarnations, err := d.recovery.CollectGarbage(ctx)90 incarnations, err := d.recovery.CollectGarbage(ctx)
15 if err != nil {91 if err != nil {
16- return fmt.Errorf("collect startup snapshots: %w", err)92+ return fmt.Errorf("collect snapshots: %w", err)
17 }93 }
18 d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)94 d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
19 return nil95 return nil
M internal/usecase/daemon/maintenance_test.go +81 -16
@@ -4,38 +4,103 @@ import (
44 "context"
55 "errors"
66 "testing"
7+ "time"
78
89 "github.com/bnema/vev/internal/domain"
910 portsmocks "github.com/bnema/vev/internal/ports/mocks"
11+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
1012 "github.com/stretchr/testify/mock"
1113 "github.com/stretchr/testify/require"
1214 )
1315
14-func TestStartupGarbageCollectionUsesValidatedCatalogue(t *testing.T) {
16+func TestDurableMaintenanceCollectsSnapshotGarbage(t *testing.T) {
1517 committed := domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}
1618 records := []domain.CatalogueRecord{
1719 {Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed},
1820 {Name: "fresh", IncarnationID: domain.IncarnationID{2}},
1921 }
20- catalogue := portsmocks.NewMockCatalogue(t)
21- catalogue.EXPECT().Records().Return(records, nil).Once()
22- repository := portsmocks.NewMockSnapshotRepository(t)
23- repository.EXPECT().CollectGarbage(mock.Anything, mock.MatchedBy(func(keep map[domain.IncarnationID]domain.CheckpointRef) bool {
22+ validKeep := mock.MatchedBy(func(keep map[domain.IncarnationID]domain.CheckpointRef) bool {
2423 return len(keep) == 2 && keep[records[0].IncarnationID] == committed && keep[records[1].IncarnationID] == (domain.CheckpointRef{})
25- })).Return(nil).Once()
26- d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
27- WithDurableMaintenance(catalogue, repository)(d)
24+ })
25+
26+ tests := []struct {
27+ name string
28+ // passes is the number of GC passes expected: the first after
29+ // restoration, each later one after a periodic timer tick.
30+ passes int
31+ recordsErr error
32+ collectErr error
33+ wantCollect bool
34+ wantInterval time.Duration
35+ }{
36+ {name: "first pass after restoration", passes: 1, wantCollect: true, wantInterval: snapshotGarbageCollectionInterval},
37+ {name: "periodic pass after timer tick", passes: 2, wantCollect: true, wantInterval: snapshotGarbageCollectionInterval},
38+ {name: "unreadable catalogue never collects", passes: 1, recordsErr: errors.New("catalogue read failed"), wantInterval: snapshotGarbageCollectionInterval},
39+ {name: "preempted pass retries sooner", passes: 1, wantCollect: true, collectErr: recoveryusecase.ErrGarbageCollectionPreempted, wantInterval: snapshotGarbageCollectionRetryInterval},
40+ }
41+ for _, tt := range tests {
42+ t.Run(tt.name, func(t *testing.T) {
43+ catalogue := portsmocks.NewMockCatalogue(t)
44+ repository := portsmocks.NewMockSnapshotRepository(t)
45+ collected := make(chan struct{}, tt.passes)
46+ if tt.recordsErr != nil {
47+ catalogue.EXPECT().Records().Return(nil, tt.recordsErr).Times(tt.passes)
48+ } else {
49+ catalogue.EXPECT().Records().Return(records, nil).Times(tt.passes)
50+ }
51+ if tt.wantCollect {
52+ repository.EXPECT().CollectGarbage(mock.Anything, validKeep).
53+ Run(func(context.Context, map[domain.IncarnationID]domain.CheckpointRef) { collected <- struct{}{} }).
54+ Return(tt.collectErr).Times(tt.passes)
55+ }
56+
57+ clk := &signalClock{timers: make(chan *signalTimer, 1)}
58+ d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clk)
59+ WithDurableMaintenance(catalogue, repository)(d)
60+ d.startDurableMaintenance()
61+
62+ select {
63+ case <-collected:
64+ t.Fatal("GC must wait for restoration to finish")
65+ case <-time.After(20 * time.Millisecond):
66+ }
67+ d.closeRestoreDone()
2868
29- require.NoError(t, d.CollectStartupGarbage(context.Background()))
69+ for pass := range tt.passes {
70+ timer := receiveWithin(t, clk.timers, "GC interval timer")
71+ require.Equal(t, tt.wantInterval, timer.duration)
72+ if tt.wantCollect {
73+ receiveWithin(t, collected, "GC pass")
74+ }
75+ if pass < tt.passes-1 {
76+ timer.ch <- time.Time{}
77+ }
78+ }
79+
80+ d.serveCancel()
81+ d.WaitDurableWriters()
82+ })
83+ }
84+}
85+
86+func receiveWithin[T any](t *testing.T, ch <-chan T, what string) T {
87+ t.Helper()
88+ select {
89+ case v := <-ch:
90+ return v
91+ case <-time.After(5 * time.Second):
92+ t.Fatalf("timed out waiting for %s", what)
93+ panic("unreachable")
94+ }
3095 }
3196
32-func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {
33- catalogue := portsmocks.NewMockCatalogue(t)
34- catalogue.EXPECT().Records().Return(nil, errors.New("catalogue read failed")).Once()
35- repository := portsmocks.NewMockSnapshotRepository(t)
97+func TestDurableMaintenanceRequiresExplicitOption(t *testing.T) {
3698 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
37- WithDurableMaintenance(catalogue, repository)(d)
99+ coordinator := recoveryusecase.NewCoordinator(portsmocks.NewMockCatalogue(t), portsmocks.NewMockSnapshotRepository(t), nil)
100+ WithRecoveryCoordinator(coordinator)(d)
101+ d.startDurableMaintenance()
38102
39- require.Error(t, d.CollectStartupGarbage(context.Background()))
40- repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)
103+ d.snapshotWorkerMu.Lock()
104+ defer d.snapshotWorkerMu.Unlock()
105+ require.Nil(t, d.maintenanceWorkerDone)
41106 }
@@ -4,38 +4,103 @@ import (
4 "context"4 "context"
5 "errors"5 "errors"
6 "testing"6 "testing"
7+ "time"
7 8
8 "github.com/bnema/vev/internal/domain"9 "github.com/bnema/vev/internal/domain"
9 portsmocks "github.com/bnema/vev/internal/ports/mocks"10 portsmocks "github.com/bnema/vev/internal/ports/mocks"
11+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
10 "github.com/stretchr/testify/mock"12 "github.com/stretchr/testify/mock"
11 "github.com/stretchr/testify/require"13 "github.com/stretchr/testify/require"
12 )14 )
13 15
14-func TestStartupGarbageCollectionUsesValidatedCatalogue(t *testing.T) {16+func TestDurableMaintenanceCollectsSnapshotGarbage(t *testing.T) {
15 committed := domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}17 committed := domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}
16 records := []domain.CatalogueRecord{18 records := []domain.CatalogueRecord{
17 {Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed},19 {Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed},
18 {Name: "fresh", IncarnationID: domain.IncarnationID{2}},20 {Name: "fresh", IncarnationID: domain.IncarnationID{2}},
19 }21 }
20- catalogue := portsmocks.NewMockCatalogue(t)22+ validKeep := mock.MatchedBy(func(keep map[domain.IncarnationID]domain.CheckpointRef) bool {
21- catalogue.EXPECT().Records().Return(records, nil).Once()
22- repository := portsmocks.NewMockSnapshotRepository(t)
23- repository.EXPECT().CollectGarbage(mock.Anything, mock.MatchedBy(func(keep map[domain.IncarnationID]domain.CheckpointRef) bool {
24 return len(keep) == 2 && keep[records[0].IncarnationID] == committed && keep[records[1].IncarnationID] == (domain.CheckpointRef{})23 return len(keep) == 2 && keep[records[0].IncarnationID] == committed && keep[records[1].IncarnationID] == (domain.CheckpointRef{})
25- })).Return(nil).Once()24+ })
26- d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})25+
27- WithDurableMaintenance(catalogue, repository)(d)26+ tests := []struct {
27+ name string
28+ // passes is the number of GC passes expected: the first after
29+ // restoration, each later one after a periodic timer tick.
30+ passes int
31+ recordsErr error
32+ collectErr error
33+ wantCollect bool
34+ wantInterval time.Duration
35+ }{
36+ {name: "first pass after restoration", passes: 1, wantCollect: true, wantInterval: snapshotGarbageCollectionInterval},
37+ {name: "periodic pass after timer tick", passes: 2, wantCollect: true, wantInterval: snapshotGarbageCollectionInterval},
38+ {name: "unreadable catalogue never collects", passes: 1, recordsErr: errors.New("catalogue read failed"), wantInterval: snapshotGarbageCollectionInterval},
39+ {name: "preempted pass retries sooner", passes: 1, wantCollect: true, collectErr: recoveryusecase.ErrGarbageCollectionPreempted, wantInterval: snapshotGarbageCollectionRetryInterval},
40+ }
41+ for _, tt := range tests {
42+ t.Run(tt.name, func(t *testing.T) {
43+ catalogue := portsmocks.NewMockCatalogue(t)
44+ repository := portsmocks.NewMockSnapshotRepository(t)
45+ collected := make(chan struct{}, tt.passes)
46+ if tt.recordsErr != nil {
47+ catalogue.EXPECT().Records().Return(nil, tt.recordsErr).Times(tt.passes)
48+ } else {
49+ catalogue.EXPECT().Records().Return(records, nil).Times(tt.passes)
50+ }
51+ if tt.wantCollect {
52+ repository.EXPECT().CollectGarbage(mock.Anything, validKeep).
53+ Run(func(context.Context, map[domain.IncarnationID]domain.CheckpointRef) { collected <- struct{}{} }).
54+ Return(tt.collectErr).Times(tt.passes)
55+ }
56+
57+ clk := &signalClock{timers: make(chan *signalTimer, 1)}
58+ d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clk)
59+ WithDurableMaintenance(catalogue, repository)(d)
60+ d.startDurableMaintenance()
61+
62+ select {
63+ case <-collected:
64+ t.Fatal("GC must wait for restoration to finish")
65+ case <-time.After(20 * time.Millisecond):
66+ }
67+ d.closeRestoreDone()
28 68
29- require.NoError(t, d.CollectStartupGarbage(context.Background()))69+ for pass := range tt.passes {
70+ timer := receiveWithin(t, clk.timers, "GC interval timer")
71+ require.Equal(t, tt.wantInterval, timer.duration)
72+ if tt.wantCollect {
73+ receiveWithin(t, collected, "GC pass")
74+ }
75+ if pass < tt.passes-1 {
76+ timer.ch <- time.Time{}
77+ }
78+ }
79+
80+ d.serveCancel()
81+ d.WaitDurableWriters()
82+ })
83+ }
84+}
85+
86+func receiveWithin[T any](t *testing.T, ch <-chan T, what string) T {
87+ t.Helper()
88+ select {
89+ case v := <-ch:
90+ return v
91+ case <-time.After(5 * time.Second):
92+ t.Fatalf("timed out waiting for %s", what)
93+ panic("unreachable")
94+ }
30 }95 }
31 96
32-func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {97+func TestDurableMaintenanceRequiresExplicitOption(t *testing.T) {
33- catalogue := portsmocks.NewMockCatalogue(t)
34- catalogue.EXPECT().Records().Return(nil, errors.New("catalogue read failed")).Once()
35- repository := portsmocks.NewMockSnapshotRepository(t)
36 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})98 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
37- WithDurableMaintenance(catalogue, repository)(d)99+ coordinator := recoveryusecase.NewCoordinator(portsmocks.NewMockCatalogue(t), portsmocks.NewMockSnapshotRepository(t), nil)
100+ WithRecoveryCoordinator(coordinator)(d)
101+ d.startDurableMaintenance()
38 102
39- require.Error(t, d.CollectStartupGarbage(context.Background()))103+ d.snapshotWorkerMu.Lock()
40- repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)104+ defer d.snapshotWorkerMu.Unlock()
105+ require.Nil(t, d.maintenanceWorkerDone)
41 }106 }
M internal/usecase/daemon/session.go +1 -1
@@ -1272,7 +1272,7 @@ func (d *Daemon) beginSnapshotPurge(_ string, _ domain.IncarnationID) error {
12721272 }
12731273
12741274 // finishSnapshotPurge removes catalogue metadata first, then deletes the
1275-// incarnation directory. Startup garbage collection removes the directory if
1275+// incarnation directory. Background garbage collection removes the directory if
12761276 // the second step is interrupted.
12771277 func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error {
12781278 if !d.persistEnabled {
@@ -1272,7 +1272,7 @@ func (d *Daemon) beginSnapshotPurge(_ string, _ domain.IncarnationID) error {
1272 }1272 }
1273 1273
1274 // finishSnapshotPurge removes catalogue metadata first, then deletes the1274 // finishSnapshotPurge removes catalogue metadata first, then deletes the
1275-// incarnation directory. Startup garbage collection removes the directory if1275+// incarnation directory. Background garbage collection removes the directory if
1276 // the second step is interrupted.1276 // the second step is interrupted.
1277 func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error {1277 func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error {
1278 if !d.persistEnabled {1278 if !d.persistEnabled {
M internal/usecase/daemon/snapshot_restore_incremental.go +5 -2
@@ -428,9 +428,12 @@ func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectR
428428 if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {
429429 return nil, fmt.Errorf("snapshot: missing object")
430430 }
431- gotKind, payload, err := snapcodec.UnmarshalObject(data)
431+ // The payload aliases the loaded object. A generation's objects are private
432+ // to this restore and are only decoded, never mutated, so copying here would
433+ // double the transient footprint of every restored history chunk.
434+ gotKind, payload, err := snapcodec.PreflightObject(data)
432435 if err != nil || gotKind != kind {
433436 return nil, fmt.Errorf("snapshot: invalid object")
434437 }
435- return append([]byte(nil), payload...), nil
438+ return payload, nil
436439 }
@@ -428,9 +428,12 @@ func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectR
428 if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {428 if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {
429 return nil, fmt.Errorf("snapshot: missing object")429 return nil, fmt.Errorf("snapshot: missing object")
430 }430 }
431- gotKind, payload, err := snapcodec.UnmarshalObject(data)431+ // The payload aliases the loaded object. A generation's objects are private
432+ // to this restore and are only decoded, never mutated, so copying here would
433+ // double the transient footprint of every restored history chunk.
434+ gotKind, payload, err := snapcodec.PreflightObject(data)
432 if err != nil || gotKind != kind {435 if err != nil || gotKind != kind {
433 return nil, fmt.Errorf("snapshot: invalid object")436 return nil, fmt.Errorf("snapshot: invalid object")
434 }437 }
435- return append([]byte(nil), payload...), nil438+ return payload, nil
436 }439 }
M internal/usecase/recovery/broken.go +1 -1
@@ -21,7 +21,7 @@ func (c *Coordinator) MarkBroken(ctx context.Context, name string, expected doma
2121 if reason == "" {
2222 return domain.CatalogueRecord{}, false, errors.New("recovery: empty broken-session reason")
2323 }
24- c.mutationMu.Lock()
24+ c.lockMutation()
2525 defer c.mutationMu.Unlock()
2626 unlock := c.locks.Lock([]string{name})
2727 defer unlock()
@@ -21,7 +21,7 @@ func (c *Coordinator) MarkBroken(ctx context.Context, name string, expected doma
21 if reason == "" {21 if reason == "" {
22 return domain.CatalogueRecord{}, false, errors.New("recovery: empty broken-session reason")22 return domain.CatalogueRecord{}, false, errors.New("recovery: empty broken-session reason")
23 }23 }
24- c.mutationMu.Lock()24+ c.lockMutation()
25 defer c.mutationMu.Unlock()25 defer c.mutationMu.Unlock()
26 unlock := c.locks.Lock([]string{name})26 unlock := c.locks.Lock([]string{name})
27 defer unlock()27 defer unlock()
M internal/usecase/recovery/checkpoint.go +1 -1
@@ -25,7 +25,7 @@ func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, public
2525 if err := ctx.Err(); err != nil {
2626 return domain.CatalogueRecord{}, err
2727 }
28- c.mutationMu.Lock()
28+ c.lockMutation()
2929 defer c.mutationMu.Unlock()
3030 unlock := c.locks.Lock([]string{name})
3131 defer unlock()
@@ -25,7 +25,7 @@ func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, public
25 if err := ctx.Err(); err != nil {25 if err := ctx.Err(); err != nil {
26 return domain.CatalogueRecord{}, err26 return domain.CatalogueRecord{}, err
27 }27 }
28- c.mutationMu.Lock()28+ c.lockMutation()
29 defer c.mutationMu.Unlock()29 defer c.mutationMu.Unlock()
30 unlock := c.locks.Lock([]string{name})30 unlock := c.locks.Lock([]string{name})
31 defer unlock()31 defer unlock()
M internal/usecase/recovery/coordinator.go +60 -11
@@ -85,12 +85,36 @@ type Coordinator struct {
8585 catalogue ports.Catalogue
8686 repository ports.SnapshotRepository
8787 locks *KeyLocks
88- // mutationMu serializes the startup GC snapshot and collection with every
89- // catalogue or checkpoint mutation. Startup normally provides this fence by
90- // running before socket publication; retaining it here also makes the
91- // coordinator safe when used directly by a composition root.
88+ // mutationMu serializes the GC keep-set snapshot and collection with every
89+ // catalogue or checkpoint mutation. It is the only fence: GC runs while the
90+ // daemon serves. Take it through lockMutation, never directly.
9291 mutationMu sync.Mutex
93- random io.Reader
92+ // gcMu guards gcCancel and mutationWaiters. A mutation preempts a running
93+ // GC pass so durable writes never wait for maintenance.
94+ gcMu sync.Mutex
95+ gcCancel context.CancelFunc
96+ mutationWaiters int
97+ random io.Reader
98+}
99+
100+// ErrGarbageCollectionPreempted reports a GC pass that yielded to a durable
101+// mutation. Work completed before the yield is kept; the caller retries later.
102+var ErrGarbageCollectionPreempted = errors.New("recovery: garbage collection preempted by a durable mutation")
103+
104+// lockMutation acquires mutationMu after cancelling any running GC pass. The
105+// waiter count closes the window where GC has not yet published its cancel
106+// function: a pass that starts while a mutation is waiting does not run.
107+func (c *Coordinator) lockMutation() {
108+ c.gcMu.Lock()
109+ c.mutationWaiters++
110+ if c.gcCancel != nil {
111+ c.gcCancel()
112+ }
113+ c.gcMu.Unlock()
114+ c.mutationMu.Lock()
115+ c.gcMu.Lock()
116+ c.mutationWaiters--
117+ c.gcMu.Unlock()
94118 }
95119
96120 func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, random io.Reader) *Coordinator {
@@ -106,7 +130,7 @@ func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord)
106130 if err := ctx.Err(); err != nil {
107131 return domain.CatalogueRecord{}, err
108132 }
109- c.mutationMu.Lock()
133+ c.lockMutation()
110134 defer c.mutationMu.Unlock()
111135 unlock := c.locks.Lock([]string{record.Name})
112136 defer unlock()
@@ -142,7 +166,7 @@ func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (doma
142166 if err := ctx.Err(); err != nil {
143167 return domain.CatalogueRecord{}, err
144168 }
145- c.mutationMu.Lock()
169+ c.lockMutation()
146170 defer c.mutationMu.Unlock()
147171 unlock := c.locks.Lock([]string{oldName, newName})
148172 defer unlock()
@@ -179,7 +203,7 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
179203 if err := ctx.Err(); err != nil {
180204 return err
181205 }
182- c.mutationMu.Lock()
206+ c.lockMutation()
183207 defer c.mutationMu.Unlock()
184208 unlock := c.locks.Lock([]string{name})
185209 defer unlock()
@@ -206,7 +230,7 @@ func (c *Coordinator) DeleteExact(ctx context.Context, name string, incarnation
206230 if err := ctx.Err(); err != nil {
207231 return err
208232 }
209- c.mutationMu.Lock()
233+ c.lockMutation()
210234 defer c.mutationMu.Unlock()
211235 unlock := c.locks.Lock([]string{name})
212236 defer unlock()
@@ -234,8 +258,33 @@ func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
234258 if err := ctx.Err(); err != nil {
235259 return 0, err
236260 }
261+ c.gcMu.Lock()
262+ if c.mutationWaiters > 0 {
263+ c.gcMu.Unlock()
264+ return 0, ErrGarbageCollectionPreempted
265+ }
266+ gcCtx, cancel := context.WithCancel(ctx)
267+ defer cancel()
268+ c.gcCancel = cancel
269+ c.gcMu.Unlock()
270+ defer func() {
271+ c.gcMu.Lock()
272+ c.gcCancel = nil
273+ c.gcMu.Unlock()
274+ }()
275+
276+ // GC takes mutationMu directly: it must not preempt itself.
237277 c.mutationMu.Lock()
238278 defer c.mutationMu.Unlock()
279+ preempted := func(err error) error {
280+ if gcCtx.Err() != nil && ctx.Err() == nil {
281+ return ErrGarbageCollectionPreempted
282+ }
283+ return err
284+ }
285+ if err := gcCtx.Err(); err != nil {
286+ return 0, preempted(err)
287+ }
239288
240289 records, err := c.catalogue.Records()
241290 if err != nil {
@@ -249,8 +298,8 @@ func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
249298 }
250299 keep[record.IncarnationID] = domain.CheckpointRef{}
251300 }
252- if err := c.repository.CollectGarbage(ctx, keep); err != nil {
253- return len(keep), err
301+ if err := c.repository.CollectGarbage(gcCtx, keep); err != nil {
302+ return len(keep), preempted(err)
254303 }
255304 return len(keep), nil
256305 }
@@ -85,12 +85,36 @@ type Coordinator struct {
85 catalogue ports.Catalogue85 catalogue ports.Catalogue
86 repository ports.SnapshotRepository86 repository ports.SnapshotRepository
87 locks *KeyLocks87 locks *KeyLocks
88- // mutationMu serializes the startup GC snapshot and collection with every88+ // mutationMu serializes the GC keep-set snapshot and collection with every
89- // catalogue or checkpoint mutation. Startup normally provides this fence by89+ // catalogue or checkpoint mutation. It is the only fence: GC runs while the
90- // running before socket publication; retaining it here also makes the90+ // daemon serves. Take it through lockMutation, never directly.
91- // coordinator safe when used directly by a composition root.
92 mutationMu sync.Mutex91 mutationMu sync.Mutex
93- random io.Reader92+ // gcMu guards gcCancel and mutationWaiters. A mutation preempts a running
93+ // GC pass so durable writes never wait for maintenance.
94+ gcMu sync.Mutex
95+ gcCancel context.CancelFunc
96+ mutationWaiters int
97+ random io.Reader
98+}
99+
100+// ErrGarbageCollectionPreempted reports a GC pass that yielded to a durable
101+// mutation. Work completed before the yield is kept; the caller retries later.
102+var ErrGarbageCollectionPreempted = errors.New("recovery: garbage collection preempted by a durable mutation")
103+
104+// lockMutation acquires mutationMu after cancelling any running GC pass. The
105+// waiter count closes the window where GC has not yet published its cancel
106+// function: a pass that starts while a mutation is waiting does not run.
107+func (c *Coordinator) lockMutation() {
108+ c.gcMu.Lock()
109+ c.mutationWaiters++
110+ if c.gcCancel != nil {
111+ c.gcCancel()
112+ }
113+ c.gcMu.Unlock()
114+ c.mutationMu.Lock()
115+ c.gcMu.Lock()
116+ c.mutationWaiters--
117+ c.gcMu.Unlock()
94 }118 }
95 119
96 func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, random io.Reader) *Coordinator {120 func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, random io.Reader) *Coordinator {
@@ -106,7 +130,7 @@ func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord)
106 if err := ctx.Err(); err != nil {130 if err := ctx.Err(); err != nil {
107 return domain.CatalogueRecord{}, err131 return domain.CatalogueRecord{}, err
108 }132 }
109- c.mutationMu.Lock()133+ c.lockMutation()
110 defer c.mutationMu.Unlock()134 defer c.mutationMu.Unlock()
111 unlock := c.locks.Lock([]string{record.Name})135 unlock := c.locks.Lock([]string{record.Name})
112 defer unlock()136 defer unlock()
@@ -142,7 +166,7 @@ func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (doma
142 if err := ctx.Err(); err != nil {166 if err := ctx.Err(); err != nil {
143 return domain.CatalogueRecord{}, err167 return domain.CatalogueRecord{}, err
144 }168 }
145- c.mutationMu.Lock()169+ c.lockMutation()
146 defer c.mutationMu.Unlock()170 defer c.mutationMu.Unlock()
147 unlock := c.locks.Lock([]string{oldName, newName})171 unlock := c.locks.Lock([]string{oldName, newName})
148 defer unlock()172 defer unlock()
@@ -179,7 +203,7 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
179 if err := ctx.Err(); err != nil {203 if err := ctx.Err(); err != nil {
180 return err204 return err
181 }205 }
182- c.mutationMu.Lock()206+ c.lockMutation()
183 defer c.mutationMu.Unlock()207 defer c.mutationMu.Unlock()
184 unlock := c.locks.Lock([]string{name})208 unlock := c.locks.Lock([]string{name})
185 defer unlock()209 defer unlock()
@@ -206,7 +230,7 @@ func (c *Coordinator) DeleteExact(ctx context.Context, name string, incarnation
206 if err := ctx.Err(); err != nil {230 if err := ctx.Err(); err != nil {
207 return err231 return err
208 }232 }
209- c.mutationMu.Lock()233+ c.lockMutation()
210 defer c.mutationMu.Unlock()234 defer c.mutationMu.Unlock()
211 unlock := c.locks.Lock([]string{name})235 unlock := c.locks.Lock([]string{name})
212 defer unlock()236 defer unlock()
@@ -234,8 +258,33 @@ func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
234 if err := ctx.Err(); err != nil {258 if err := ctx.Err(); err != nil {
235 return 0, err259 return 0, err
236 }260 }
261+ c.gcMu.Lock()
262+ if c.mutationWaiters > 0 {
263+ c.gcMu.Unlock()
264+ return 0, ErrGarbageCollectionPreempted
265+ }
266+ gcCtx, cancel := context.WithCancel(ctx)
267+ defer cancel()
268+ c.gcCancel = cancel
269+ c.gcMu.Unlock()
270+ defer func() {
271+ c.gcMu.Lock()
272+ c.gcCancel = nil
273+ c.gcMu.Unlock()
274+ }()
275+
276+ // GC takes mutationMu directly: it must not preempt itself.
237 c.mutationMu.Lock()277 c.mutationMu.Lock()
238 defer c.mutationMu.Unlock()278 defer c.mutationMu.Unlock()
279+ preempted := func(err error) error {
280+ if gcCtx.Err() != nil && ctx.Err() == nil {
281+ return ErrGarbageCollectionPreempted
282+ }
283+ return err
284+ }
285+ if err := gcCtx.Err(); err != nil {
286+ return 0, preempted(err)
287+ }
239 288
240 records, err := c.catalogue.Records()289 records, err := c.catalogue.Records()
241 if err != nil {290 if err != nil {
@@ -249,8 +298,8 @@ func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
249 }298 }
250 keep[record.IncarnationID] = domain.CheckpointRef{}299 keep[record.IncarnationID] = domain.CheckpointRef{}
251 }300 }
252- if err := c.repository.CollectGarbage(ctx, keep); err != nil {301+ if err := c.repository.CollectGarbage(gcCtx, keep); err != nil {
253- return len(keep), err302+ return len(keep), preempted(err)
254 }303 }
255 return len(keep), nil304 return len(keep), nil
256 }305 }
M internal/usecase/recovery/discard.go +2 -2
@@ -14,7 +14,7 @@ var (
1414 // Discard throws away a broken session's persisted state and replaces it with a
1515 // fresh incarnation. It is retry-idempotent, not crash-resumable: a crash
1616 // leaves either the old record (rerun discard) or a new record plus an orphan
17-// directory that startup garbage collection removes.
17+// directory that background garbage collection removes.
1818 func (c *Coordinator) Discard(ctx context.Context, name string) error {
1919 if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil {
2020 return errors.New("recovery: incomplete discard dependencies")
@@ -22,7 +22,7 @@ func (c *Coordinator) Discard(ctx context.Context, name string) error {
2222 if err := ctx.Err(); err != nil {
2323 return err
2424 }
25- c.mutationMu.Lock()
25+ c.lockMutation()
2626 defer c.mutationMu.Unlock()
2727 unlock := c.locks.Lock([]string{name})
2828 defer unlock()
@@ -14,7 +14,7 @@ var (
14 // Discard throws away a broken session's persisted state and replaces it with a14 // Discard throws away a broken session's persisted state and replaces it with a
15 // fresh incarnation. It is retry-idempotent, not crash-resumable: a crash15 // fresh incarnation. It is retry-idempotent, not crash-resumable: a crash
16 // leaves either the old record (rerun discard) or a new record plus an orphan16 // leaves either the old record (rerun discard) or a new record plus an orphan
17-// directory that startup garbage collection removes.17+// directory that background garbage collection removes.
18 func (c *Coordinator) Discard(ctx context.Context, name string) error {18 func (c *Coordinator) Discard(ctx context.Context, name string) error {
19 if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil {19 if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil {
20 return errors.New("recovery: incomplete discard dependencies")20 return errors.New("recovery: incomplete discard dependencies")
@@ -22,7 +22,7 @@ func (c *Coordinator) Discard(ctx context.Context, name string) error {
22 if err := ctx.Err(); err != nil {22 if err := ctx.Err(); err != nil {
23 return err23 return err
24 }24 }
25- c.mutationMu.Lock()25+ c.lockMutation()
26 defer c.mutationMu.Unlock()26 defer c.mutationMu.Unlock()
27 unlock := c.locks.Lock([]string{name})27 unlock := c.locks.Lock([]string{name})
28 defer unlock()28 defer unlock()
M internal/usecase/recovery/reset.go +1 -1
@@ -17,7 +17,7 @@ func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expect
1717 if err := ctx.Err(); err != nil {
1818 return domain.CatalogueRecord{}, false, err
1919 }
20- c.mutationMu.Lock()
20+ c.lockMutation()
2121 defer c.mutationMu.Unlock()
2222 unlock := c.locks.Lock([]string{name})
2323 defer unlock()
@@ -17,7 +17,7 @@ func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expect
17 if err := ctx.Err(); err != nil {17 if err := ctx.Err(); err != nil {
18 return domain.CatalogueRecord{}, false, err18 return domain.CatalogueRecord{}, false, err
19 }19 }
20- c.mutationMu.Lock()20+ c.lockMutation()
21 defer c.mutationMu.Unlock()21 defer c.mutationMu.Unlock()
22 unlock := c.locks.Lock([]string{name})22 unlock := c.locks.Lock([]string{name})
23 defer unlock()23 defer unlock()
M internal/usecase/recovery/transactions_test.go +59 -0
@@ -6,6 +6,7 @@ import (
66 "errors"
77 "sync"
88 "testing"
9+ "time"
910
1011 "github.com/bnema/vev/internal/domain"
1112 "github.com/bnema/vev/internal/ports"
@@ -462,6 +463,64 @@ func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(
462463 })
463464 }
464465
466+// cooperativeGarbageCollectionRepository blocks a GC pass until its context is
467+// cancelled, like the filesystem repository between entries.
468+type cooperativeGarbageCollectionRepository struct {
469+ garbageCollectionFenceRepository
470+}
471+
472+func (r *cooperativeGarbageCollectionRepository) CollectGarbage(ctx context.Context, _ map[domain.IncarnationID]domain.CheckpointRef) error {
473+ close(r.collectionEntered)
474+ <-ctx.Done()
475+ return ctx.Err()
476+}
477+
478+func TestDurableMutationPreemptsRunningGarbageCollection(t *testing.T) {
479+ t.Parallel()
480+ committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
481+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
482+ tests := []struct {
483+ name string
484+ mutate func(*Coordinator) error
485+ }{
486+ {name: "create", mutate: func(c *Coordinator) error {
487+ _, err := c.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
488+ return err
489+ }},
490+ {name: "checkpoint publication", mutate: func(c *Coordinator) error {
491+ _, err := c.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
492+ Name: "work", IncarnationID: record.IncarnationID, Generation: 2,
493+ ParentCheckpoint: &committed, Manifest: []byte("generation two"),
494+ })
495+ return err
496+ }},
497+ }
498+ for _, tt := range tests {
499+ t.Run(tt.name, func(t *testing.T) {
500+ t.Parallel()
501+ repository := &cooperativeGarbageCollectionRepository{garbageCollectionFenceRepository{collectionEntered: make(chan struct{})}}
502+ coordinator := NewCoordinator(newTransactionCatalogue(record), repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
503+
504+ collected := make(chan error, 1)
505+ go func() {
506+ _, err := coordinator.CollectGarbage(context.Background())
507+ collected <- err
508+ }()
509+ <-repository.collectionEntered
510+
511+ mutated := make(chan error, 1)
512+ go func() { mutated <- tt.mutate(coordinator) }()
513+ select {
514+ case err := <-mutated:
515+ require.NoError(t, err)
516+ case <-time.After(5 * time.Second):
517+ t.Fatal("mutation waited for the GC pass instead of preempting it")
518+ }
519+ require.ErrorIs(t, <-collected, ErrGarbageCollectionPreempted)
520+ })
521+ }
522+}
523+
465524 func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
466525 t.Parallel()
467526 old := degradedTransactionRecord()
@@ -6,6 +6,7 @@ import (
6 "errors"6 "errors"
7 "sync"7 "sync"
8 "testing"8 "testing"
9+ "time"
9 10
10 "github.com/bnema/vev/internal/domain"11 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"12 "github.com/bnema/vev/internal/ports"
@@ -462,6 +463,64 @@ func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(
462 })463 })
463 }464 }
464 465
466+// cooperativeGarbageCollectionRepository blocks a GC pass until its context is
467+// cancelled, like the filesystem repository between entries.
468+type cooperativeGarbageCollectionRepository struct {
469+ garbageCollectionFenceRepository
470+}
471+
472+func (r *cooperativeGarbageCollectionRepository) CollectGarbage(ctx context.Context, _ map[domain.IncarnationID]domain.CheckpointRef) error {
473+ close(r.collectionEntered)
474+ <-ctx.Done()
475+ return ctx.Err()
476+}
477+
478+func TestDurableMutationPreemptsRunningGarbageCollection(t *testing.T) {
479+ t.Parallel()
480+ committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
481+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
482+ tests := []struct {
483+ name string
484+ mutate func(*Coordinator) error
485+ }{
486+ {name: "create", mutate: func(c *Coordinator) error {
487+ _, err := c.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
488+ return err
489+ }},
490+ {name: "checkpoint publication", mutate: func(c *Coordinator) error {
491+ _, err := c.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
492+ Name: "work", IncarnationID: record.IncarnationID, Generation: 2,
493+ ParentCheckpoint: &committed, Manifest: []byte("generation two"),
494+ })
495+ return err
496+ }},
497+ }
498+ for _, tt := range tests {
499+ t.Run(tt.name, func(t *testing.T) {
500+ t.Parallel()
501+ repository := &cooperativeGarbageCollectionRepository{garbageCollectionFenceRepository{collectionEntered: make(chan struct{})}}
502+ coordinator := NewCoordinator(newTransactionCatalogue(record), repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
503+
504+ collected := make(chan error, 1)
505+ go func() {
506+ _, err := coordinator.CollectGarbage(context.Background())
507+ collected <- err
508+ }()
509+ <-repository.collectionEntered
510+
511+ mutated := make(chan error, 1)
512+ go func() { mutated <- tt.mutate(coordinator) }()
513+ select {
514+ case err := <-mutated:
515+ require.NoError(t, err)
516+ case <-time.After(5 * time.Second):
517+ t.Fatal("mutation waited for the GC pass instead of preempting it")
518+ }
519+ require.ErrorIs(t, <-collected, ErrGarbageCollectionPreempted)
520+ })
521+ }
522+}
523+
465 func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {524 func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
466 t.Parallel()525 t.Parallel()
467 old := degradedTransactionRecord()526 old := degradedTransactionRecord()

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close