detached b1b3fa9

fix(snapshot): bound repository maintenance

commit
b1b3fa9f0359cfd78f4eb6173d14e19a4055f74f
author
brice · · 2026-07-21 20:47 UTC
committer
brice · · 2026-07-21 20:48 UTC
parent
b5f28c1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
7 files · +426 -43
Unified
UnifiedSplit
M internal/adapters/snapshot/legacy.go +48 -16
@@ -4,6 +4,7 @@ import (
44 "context"
55 "errors"
66 "fmt"
7+ "io"
78 "os"
89 "path/filepath"
910 "sort"
@@ -19,32 +20,60 @@ func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, er
1920 if err := ctx.Err(); err != nil {
2021 return nil, err
2122 }
22- entries, err := os.ReadDir(r.dir)
23+ f, err := os.Open(r.dir)
2324 if errors.Is(err, os.ErrNotExist) {
2425 return nil, nil
2526 }
2627 if err != nil {
2728 return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err))
2829 }
29- sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
30- out := make([]ports.LegacySnapshot, 0, len(entries))
31- for _, entry := range entries {
32- if err := ctx.Err(); err != nil {
33- return nil, err
34- }
35- if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".snap") {
36- continue
30+ defer f.Close()
31+
32+ out := make([]ports.LegacySnapshot, 0, maxLegacySnapshotFiles)
33+ files := 0
34+ total := 0
35+ for {
36+ entries, readErr := f.ReadDir(maintenanceBatch)
37+ for _, entry := range entries {
38+ if err := ctx.Err(); err != nil {
39+ return nil, err
40+ }
41+ if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".snap") {
42+ continue
43+ }
44+ files++
45+ if files > maxLegacySnapshotFiles {
46+ return nil, fmt.Errorf("legacy snapshot import has too many files (maximum %d)", maxLegacySnapshotFiles)
47+ }
48+ path := filepath.Join(r.dir, entry.Name())
49+ // Check the prospective size before readBounded allocates a buffer.
50+ // readBounded repeats the security validation and the actual length is
51+ // checked below to close the metadata/read race.
52+ if info, statErr := os.Lstat(path); statErr == nil && info.Mode().IsRegular() && info.Size() > int64(maxLegacySnapshotBytes-total) {
53+ return nil, fmt.Errorf("legacy snapshot import exceeds aggregate byte limit (%d bytes)", maxLegacySnapshotBytes)
54+ }
55+ data, dataErr := readBounded(path)
56+ if dataErr != nil {
57+ continue
58+ }
59+ if len(data) > maxLegacySnapshotBytes-total {
60+ return nil, fmt.Errorf("legacy snapshot import exceeds aggregate byte limit (%d bytes)", maxLegacySnapshotBytes)
61+ }
62+ total += len(data)
63+ name := strings.TrimSuffix(entry.Name(), ".snap")
64+ if strings.HasPrefix(name, "@") || !safeNameRE.MatchString(name) {
65+ name = ""
66+ }
67+ out = append(out, ports.LegacySnapshot{Name: name, Data: data})
3768 }
38- data, err := readBounded(filepath.Join(r.dir, entry.Name()))
39- if err != nil {
40- continue
69+ if errors.Is(readErr, io.EOF) {
70+ break
4171 }
42- name := strings.TrimSuffix(entry.Name(), ".snap")
43- if strings.HasPrefix(name, "@") || !safeNameRE.MatchString(name) {
44- name = ""
72+ if readErr != nil {
73+ return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(readErr))
4574 }
46- out = append(out, ports.LegacySnapshot{Name: name, Data: data})
4775 }
76+ sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
4877 return out, nil
4978 }
5079
@@ -61,6 +90,9 @@ func (r *Repository) DeleteLegacy(ctx context.Context, name string) error {
6190 } else if err != nil {
6291 return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err))
6392 }
93+ if err := ctx.Err(); err != nil {
94+ return err
95+ }
6496 if err := r.syncDirectory(r.dir); err != nil {
6597 return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err))
6698 }
@@ -4,6 +4,7 @@ import (
4 "context"4 "context"
5 "errors"5 "errors"
6 "fmt"6 "fmt"
7+ "io"
7 "os"8 "os"
8 "path/filepath"9 "path/filepath"
9 "sort"10 "sort"
@@ -19,32 +20,60 @@ func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, er
19 if err := ctx.Err(); err != nil {20 if err := ctx.Err(); err != nil {
20 return nil, err21 return nil, err
21 }22 }
22- entries, err := os.ReadDir(r.dir)23+ f, err := os.Open(r.dir)
23 if errors.Is(err, os.ErrNotExist) {24 if errors.Is(err, os.ErrNotExist) {
24 return nil, nil25 return nil, nil
25 }26 }
26 if err != nil {27 if err != nil {
27 return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err))28 return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err))
28 }29 }
29- sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })30+ defer f.Close()
30- out := make([]ports.LegacySnapshot, 0, len(entries))31+
31- for _, entry := range entries {32+ out := make([]ports.LegacySnapshot, 0, maxLegacySnapshotFiles)
32- if err := ctx.Err(); err != nil {33+ files := 0
33- return nil, err34+ total := 0
34- }35+ for {
35- if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".snap") {36+ entries, readErr := f.ReadDir(maintenanceBatch)
36- continue37+ for _, entry := range entries {
38+ if err := ctx.Err(); err != nil {
39+ return nil, err
40+ }
41+ if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".snap") {
42+ continue
43+ }
44+ files++
45+ if files > maxLegacySnapshotFiles {
46+ return nil, fmt.Errorf("legacy snapshot import has too many files (maximum %d)", maxLegacySnapshotFiles)
47+ }
48+ path := filepath.Join(r.dir, entry.Name())
49+ // Check the prospective size before readBounded allocates a buffer.
50+ // readBounded repeats the security validation and the actual length is
51+ // checked below to close the metadata/read race.
52+ if info, statErr := os.Lstat(path); statErr == nil && info.Mode().IsRegular() && info.Size() > int64(maxLegacySnapshotBytes-total) {
53+ return nil, fmt.Errorf("legacy snapshot import exceeds aggregate byte limit (%d bytes)", maxLegacySnapshotBytes)
54+ }
55+ data, dataErr := readBounded(path)
56+ if dataErr != nil {
57+ continue
58+ }
59+ if len(data) > maxLegacySnapshotBytes-total {
60+ return nil, fmt.Errorf("legacy snapshot import exceeds aggregate byte limit (%d bytes)", maxLegacySnapshotBytes)
61+ }
62+ total += len(data)
63+ name := strings.TrimSuffix(entry.Name(), ".snap")
64+ if strings.HasPrefix(name, "@") || !safeNameRE.MatchString(name) {
65+ name = ""
66+ }
67+ out = append(out, ports.LegacySnapshot{Name: name, Data: data})
37 }68 }
38- data, err := readBounded(filepath.Join(r.dir, entry.Name()))69+ if errors.Is(readErr, io.EOF) {
39- if err != nil {70+ break
40- continue
41 }71 }
42- name := strings.TrimSuffix(entry.Name(), ".snap")72+ if readErr != nil {
43- if strings.HasPrefix(name, "@") || !safeNameRE.MatchString(name) {73+ return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(readErr))
44- name = ""
45 }74 }
46- out = append(out, ports.LegacySnapshot{Name: name, Data: data})
47 }75 }
76+ sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
48 return out, nil77 return out, nil
49 }78 }
50 79
@@ -61,6 +90,9 @@ func (r *Repository) DeleteLegacy(ctx context.Context, name string) error {
61 } else if err != nil {90 } else if err != nil {
62 return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err))91 return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err))
63 }92 }
93+ if err := ctx.Err(); err != nil {
94+ return err
95+ }
64 if err := r.syncDirectory(r.dir); err != nil {96 if err := r.syncDirectory(r.dir); err != nil {
65 return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err))97 return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err))
66 }98 }
M internal/adapters/snapshot/legacy_test.go +33 -0
@@ -3,8 +3,10 @@ package snapshot
33 import (
44 "context"
55 "errors"
6+ "fmt"
67 "os"
78 "path/filepath"
9+ "strings"
810 "testing"
911 )
1012
@@ -32,6 +34,37 @@ func TestRepositoryLegacyReadsOnlyBoundedRootSnapshots(t *testing.T) {
3234 }
3335 }
3436
37+func TestRepositoryLegacyRejectsTooManyAndTooLargeAggregateSnapshots(t *testing.T) {
38+ t.Run("count", func(t *testing.T) {
39+ repo := NewRepository(privateDir(t))
40+ if err := os.Mkdir(repo.dir, 0o700); err != nil {
41+ t.Fatal(err)
42+ }
43+ for i := 0; i < maxLegacySnapshotFiles+1; i++ {
44+ if err := os.WriteFile(filepath.Join(repo.dir, fmt.Sprintf("%03d.snap", i)), []byte("legacy"), 0o600); err != nil {
45+ t.Fatal(err)
46+ }
47+ }
48+ if _, err := repo.LoadLegacy(context.Background()); err == nil || !strings.Contains(err.Error(), "too many") {
49+ t.Fatalf("LoadLegacy error = %v, want actionable count limit", err)
50+ }
51+ })
52+ t.Run("aggregate bytes", func(t *testing.T) {
53+ repo := NewRepository(privateDir(t))
54+ if err := os.Mkdir(repo.dir, 0o700); err != nil {
55+ t.Fatal(err)
56+ }
57+ for i, size := range []int{maxLegacySnapshotBytes / 2, maxLegacySnapshotBytes/2 + 1} {
58+ if err := os.WriteFile(filepath.Join(repo.dir, fmt.Sprintf("%d.snap", i)), make([]byte, size), 0o600); err != nil {
59+ t.Fatal(err)
60+ }
61+ }
62+ if _, err := repo.LoadLegacy(context.Background()); err == nil || !strings.Contains(err.Error(), "aggregate") {
63+ t.Fatalf("LoadLegacy error = %v, want actionable aggregate limit", err)
64+ }
65+ })
66+}
67+
3568 func TestRepositoryDeleteLegacyMissingFileDoesNotRequireRoot(t *testing.T) {
3669 repo := NewRepository(privateDir(t))
3770 if err := repo.DeleteLegacy(context.Background(), "missing"); err != nil {
@@ -3,8 +3,10 @@ package snapshot
3 import (3 import (
4 "context"4 "context"
5 "errors"5 "errors"
6+ "fmt"
6 "os"7 "os"
7 "path/filepath"8 "path/filepath"
9+ "strings"
8 "testing"10 "testing"
9 )11 )
10 12
@@ -32,6 +34,37 @@ func TestRepositoryLegacyReadsOnlyBoundedRootSnapshots(t *testing.T) {
32 }34 }
33 }35 }
34 36
37+func TestRepositoryLegacyRejectsTooManyAndTooLargeAggregateSnapshots(t *testing.T) {
38+ t.Run("count", func(t *testing.T) {
39+ repo := NewRepository(privateDir(t))
40+ if err := os.Mkdir(repo.dir, 0o700); err != nil {
41+ t.Fatal(err)
42+ }
43+ for i := 0; i < maxLegacySnapshotFiles+1; i++ {
44+ if err := os.WriteFile(filepath.Join(repo.dir, fmt.Sprintf("%03d.snap", i)), []byte("legacy"), 0o600); err != nil {
45+ t.Fatal(err)
46+ }
47+ }
48+ if _, err := repo.LoadLegacy(context.Background()); err == nil || !strings.Contains(err.Error(), "too many") {
49+ t.Fatalf("LoadLegacy error = %v, want actionable count limit", err)
50+ }
51+ })
52+ t.Run("aggregate bytes", func(t *testing.T) {
53+ repo := NewRepository(privateDir(t))
54+ if err := os.Mkdir(repo.dir, 0o700); err != nil {
55+ t.Fatal(err)
56+ }
57+ for i, size := range []int{maxLegacySnapshotBytes / 2, maxLegacySnapshotBytes/2 + 1} {
58+ if err := os.WriteFile(filepath.Join(repo.dir, fmt.Sprintf("%d.snap", i)), make([]byte, size), 0o600); err != nil {
59+ t.Fatal(err)
60+ }
61+ }
62+ if _, err := repo.LoadLegacy(context.Background()); err == nil || !strings.Contains(err.Error(), "aggregate") {
63+ t.Fatalf("LoadLegacy error = %v, want actionable aggregate limit", err)
64+ }
65+ })
66+}
67+
35 func TestRepositoryDeleteLegacyMissingFileDoesNotRequireRoot(t *testing.T) {68 func TestRepositoryDeleteLegacyMissingFileDoesNotRequireRoot(t *testing.T) {
36 repo := NewRepository(privateDir(t))69 repo := NewRepository(privateDir(t))
37 if err := repo.DeleteLegacy(context.Background(), "missing"); err != nil {70 if err := repo.DeleteLegacy(context.Background(), "missing"); err != nil {
M internal/adapters/snapshot/load.go +3 -0
@@ -65,6 +65,9 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener
6565 lock := r.sessionLock(key)
6666 lock.Lock()
6767 defer lock.Unlock()
68+ if err := ctx.Err(); err != nil {
69+ return ports.SnapshotGeneration{}, err
70+ }
6871 return r.load(ctx, name, key)
6972 }
7073
@@ -65,6 +65,9 @@ func (r *Repository) Load(ctx context.Context, name string) (ports.SnapshotGener
65 lock := r.sessionLock(key)65 lock := r.sessionLock(key)
66 lock.Lock()66 lock.Lock()
67 defer lock.Unlock()67 defer lock.Unlock()
68+ if err := ctx.Err(); err != nil {
69+ return ports.SnapshotGeneration{}, err
70+ }
68 return r.load(ctx, name, key)71 return r.load(ctx, name, key)
69 }72 }
70 73
M internal/adapters/snapshot/maintenance.go +195 -27
@@ -4,6 +4,7 @@ import (
44 "context"
55 "errors"
66 "fmt"
7+ "io"
78 "os"
89 "path/filepath"
910 "strings"
@@ -13,6 +14,9 @@ import (
1314 codec "github.com/bnema/vev/internal/usecase/snapshot"
1415 )
1516
17+// maintenanceBatch bounds both a single directory read and the number of
18+// namespace entries removed by one maintenance pass. A pass never materializes
19+// an unbounded directory in memory.
1620 const maintenanceBatch = 64
1721
1822 // Delete makes a session unavailable by durably moving it out of the canonical
@@ -26,34 +30,86 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
2630 lock := r.sessionLock(key)
2731 lock.Lock()
2832 defer lock.Unlock()
33+ if err := ctx.Err(); err != nil {
34+ return err
35+ }
2936
3037 canonical := r.sessionPath(key)
38+ sessions := filepath.Dir(canonical)
39+ // A prior rename may have succeeded while its parent sync failed. Complete
40+ // that durability boundary before considering a canonical directory: this
41+ // also leaves a newly recreated session untouched.
42+ pending, err := pendingQuarantine(sessions, key)
43+ if err != nil {
44+ return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
45+ }
46+ if pending {
47+ if err := ctx.Err(); err != nil {
48+ return err
49+ }
50+ if err := r.syncDirectory(sessions); err != nil {
51+ return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
52+ }
53+ return nil
54+ }
3155 if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {
3256 return nil
3357 } else if err != nil {
3458 return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
3559 }
36- quarantine := filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
60+ quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
3761 for attempt := 0; ; attempt++ {
3862 if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
3963 break
4064 } else if err != nil {
4165 return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
4266 }
43- quarantine = filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
67+ quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
68+ }
69+ if err := ctx.Err(); err != nil {
70+ return err
4471 }
4572 if err := r.rename(canonical, quarantine); err != nil {
4673 return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
4774 }
48- if err := r.syncDirectory(filepath.Dir(canonical)); err != nil {
75+ if err := ctx.Err(); err != nil {
76+ return err
77+ }
78+ if err := r.syncDirectory(sessions); err != nil {
4979 return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
5080 }
5181 return nil
5282 }
5383
54-// Maintain reaps a fixed number of stale entries. It serializes each session
55-// with publication and loading so objects written by an in-flight publication
56-// cannot be collected.
84+func pendingQuarantine(dir, key string) (bool, error) {
85+ f, err := os.Open(dir)
86+ if errors.Is(err, os.ErrNotExist) {
87+ return false, nil
88+ }
89+ if err != nil {
90+ return false, err
91+ }
92+ defer f.Close()
93+ prefix := ".deleting-" + key + "-"
94+ for {
95+ entries, err := f.ReadDir(maintenanceBatch)
96+ for _, entry := range entries {
97+ if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
98+ return true, nil
99+ }
100+ }
101+ if errors.Is(err, io.EOF) {
102+ return false, nil
103+ }
104+ if err != nil {
105+ return false, err
106+ }
107+ }
108+}
109+
110+// Maintain reaps a bounded amount of stale state. Every scan uses ReadDir with
111+// a fixed batch, and a quarantine tree is peeled one entry at a time so an
112+// interrupted pass can continue after another call or restart.
57113 func (r *Repository) Maintain(ctx context.Context) error {
58114 if err := ctx.Err(); err != nil {
59115 return err
@@ -62,8 +118,11 @@ func (r *Repository) Maintain(ctx context.Context) error {
62118 if err := r.removeTemps(ctx, r.dir, &budget); err != nil {
63119 return err
64120 }
121+ if budget == 0 {
122+ return nil
123+ }
65124 sessions := filepath.Join(r.dir, repositorySessionsDir)
66- entries, err := os.ReadDir(sessions)
125+ entries, err := readDirBatch(sessions, maintenanceBatch)
67126 if errors.Is(err, os.ErrNotExist) {
68127 return nil
69128 }
@@ -79,13 +138,18 @@ func (r *Repository) Maintain(ctx context.Context) error {
79138 }
80139 path := filepath.Join(sessions, entry.Name())
81140 if isQuarantine(entry.Name()) {
82- if err := r.removeTree(path); err != nil && !errors.Is(err, os.ErrNotExist) {
141+ changed, err := r.removeTreeBatch(ctx, path, &budget)
142+ if err != nil && !errors.Is(err, os.ErrNotExist) {
83143 return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
84144 }
85- if err := r.syncDirectory(sessions); err != nil {
86- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
145+ if changed {
146+ if err := ctx.Err(); err != nil {
147+ return err
148+ }
149+ if err := r.syncDirectory(sessions); err != nil {
150+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
151+ }
87152 }
88- budget--
89153 continue
90154 }
91155 if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {
@@ -93,6 +157,10 @@ func (r *Repository) Maintain(ctx context.Context) error {
93157 }
94158 lock := r.sessionLock(entry.Name())
95159 lock.Lock()
160+ if err := ctx.Err(); err != nil {
161+ lock.Unlock()
162+ return err
163+ }
96164 err := r.maintainSession(ctx, entry.Name(), &budget)
97165 lock.Unlock()
98166 if err != nil {
@@ -104,12 +172,81 @@ func (r *Repository) Maintain(ctx context.Context) error {
104172
105173 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
106174
175+func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
176+ f, err := os.Open(dir)
177+ if err != nil {
178+ return nil, err
179+ }
180+ entries, readErr := f.ReadDir(n)
181+ closeErr := f.Close()
182+ if readErr != nil && !errors.Is(readErr, io.EOF) {
183+ return nil, readErr
184+ }
185+ if closeErr != nil {
186+ return nil, closeErr
187+ }
188+ return entries, nil
189+}
190+
191+func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
192+ changed := false
193+ for *budget > 0 {
194+ did, err := r.removeTreeStep(ctx, path, budget)
195+ if err != nil || !did {
196+ return changed, err
197+ }
198+ changed = true
199+ }
200+ return changed, nil
201+}
202+
203+func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
204+ if *budget == 0 {
205+ return false, nil
206+ }
207+ info, err := os.Lstat(path)
208+ if errors.Is(err, os.ErrNotExist) {
209+ return false, nil
210+ }
211+ if err != nil {
212+ return false, err
213+ }
214+ if !info.IsDir() {
215+ if err := ctx.Err(); err != nil {
216+ return false, err
217+ }
218+ if err := r.remove(path); err != nil {
219+ return false, err
220+ }
221+ *budget--
222+ return true, nil
223+ }
224+ entries, err := readDirBatch(path, 1)
225+ if err != nil {
226+ return false, err
227+ }
228+ if len(entries) == 0 {
229+ if err := ctx.Err(); err != nil {
230+ return false, err
231+ }
232+ if err := r.remove(path); err != nil {
233+ return false, err
234+ }
235+ *budget--
236+ return true, nil
237+ }
238+ return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
239+}
240+
107241 func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {
242+ if err := ctx.Err(); err != nil {
243+ return err
244+ }
108245 if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil {
109246 return err
110247 }
111248 objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
112- shards, err := os.ReadDir(objectRoot)
249+ shards, err := readDirBatch(objectRoot, maintenanceBatch)
113250 if err != nil && !errors.Is(err, os.ErrNotExist) {
114251 return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
115252 }
@@ -124,7 +261,7 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
124261 }
125262 }
126263
127- generations, err := r.generationNumbers(key)
264+ generations, err := r.generationNumbersBatch(key)
128265 if err != nil {
129266 return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
130267 }
@@ -155,7 +292,7 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
155292 }
156293
157294 retained := make(map[uint64]bool, 2)
158- for _, state := range states { // generationNumbers is newest first.
295+ for _, state := range states {
159296 if state.complete && complete < 2 {
160297 retained[state.generation] = true
161298 complete++
@@ -166,25 +303,59 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
166303 keep := retained[state.generation] || !state.complete
167304 if !keep && *budget > 0 {
168305 path := r.manifestPath(key, state.generation)
306+ if err := ctx.Err(); err != nil {
307+ return err
308+ }
169309 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
170310 return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
171311 }
312+ if err := ctx.Err(); err != nil {
313+ return err
314+ }
172315 if err := r.syncDirectory(filepath.Dir(path)); err != nil {
173316 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
174317 }
175318 *budget--
176319 continue
177320 }
178- for digest := range state.refs { // incomplete manifests protect in-flight objects.
321+ for digest := range state.refs {
179322 referenced[digest] = struct{}{}
180323 }
181324 }
182325 return r.removeUnreferencedObjects(ctx, key, referenced, budget)
183326 }
184327
328+func (r *Repository) generationNumbersBatch(key string) ([]uint64, error) {
329+ entries, err := readDirBatch(filepath.Join(r.sessionPath(key), repositoryGenerations), maintenanceBatch)
330+ if errors.Is(err, os.ErrNotExist) {
331+ return nil, nil
332+ }
333+ if err != nil {
334+ return nil, err
335+ }
336+ numbers := make([]uint64, 0, len(entries))
337+ for _, entry := range entries {
338+ if !entry.IsDir() {
339+ if generation, ok := parseGenerationFilename(entry.Name()); ok {
340+ numbers = append(numbers, generation)
341+ }
342+ }
343+ }
344+ // Generation file names are zero-padded, but keep the explicit comparison
345+ // in case the representation changes.
346+ for i := range numbers {
347+ for j := i + 1; j < len(numbers); j++ {
348+ if numbers[j] > numbers[i] {
349+ numbers[i], numbers[j] = numbers[j], numbers[i]
350+ }
351+ }
352+ }
353+ return numbers, nil
354+}
355+
185356 func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error {
186357 root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
187- shards, err := os.ReadDir(root)
358+ shards, err := readDirBatch(root, maintenanceBatch)
188359 if errors.Is(err, os.ErrNotExist) {
189360 return nil
190361 }
@@ -199,7 +370,7 @@ func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string,
199370 continue
200371 }
201372 dir := filepath.Join(root, shard.Name())
202- objects, err := os.ReadDir(dir)
373+ objects, err := readDirBatch(dir, maintenanceBatch)
203374 if err != nil {
204375 return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
205376 }
@@ -221,6 +392,9 @@ func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string,
221392 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
222393 return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
223394 }
395+ if err := ctx.Err(); err != nil {
396+ return err
397+ }
224398 if err := r.syncDirectory(dir); err != nil {
225399 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
226400 }
@@ -234,7 +408,7 @@ func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) e
234408 if *budget == 0 {
235409 return nil
236410 }
237- entries, err := os.ReadDir(dir)
411+ entries, err := readDirBatch(dir, maintenanceBatch)
238412 if errors.Is(err, os.ErrNotExist) {
239413 return nil
240414 }
@@ -255,6 +429,9 @@ func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) e
255429 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
256430 return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
257431 }
432+ if err := ctx.Err(); err != nil {
433+ return err
434+ }
258435 if err := r.syncDirectory(dir); err != nil {
259436 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
260437 }
@@ -263,15 +440,6 @@ func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) e
263440 return nil
264441 }
265442
266-func (r *Repository) removeTree(path string) error {
267- if r.hooks.remove != nil {
268- if err := r.hooks.remove(path); err != nil {
269- return err
270- }
271- }
272- return os.RemoveAll(path)
273-}
274-
275443 // maintenancePath avoids exposing the repository root while retaining a
276444 // path-safe, actionable location in errors.
277445 func maintenancePath(root, path string) string {
@@ -4,6 +4,7 @@ import (
4 "context"4 "context"
5 "errors"5 "errors"
6 "fmt"6 "fmt"
7+ "io"
7 "os"8 "os"
8 "path/filepath"9 "path/filepath"
9 "strings"10 "strings"
@@ -13,6 +14,9 @@ import (
13 codec "github.com/bnema/vev/internal/usecase/snapshot"14 codec "github.com/bnema/vev/internal/usecase/snapshot"
14 )15 )
15 16
17+// maintenanceBatch bounds both a single directory read and the number of
18+// namespace entries removed by one maintenance pass. A pass never materializes
19+// an unbounded directory in memory.
16 const maintenanceBatch = 6420 const maintenanceBatch = 64
17 21
18 // Delete makes a session unavailable by durably moving it out of the canonical22 // Delete makes a session unavailable by durably moving it out of the canonical
@@ -26,34 +30,86 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
26 lock := r.sessionLock(key)30 lock := r.sessionLock(key)
27 lock.Lock()31 lock.Lock()
28 defer lock.Unlock()32 defer lock.Unlock()
33+ if err := ctx.Err(); err != nil {
34+ return err
35+ }
29 36
30 canonical := r.sessionPath(key)37 canonical := r.sessionPath(key)
38+ sessions := filepath.Dir(canonical)
39+ // A prior rename may have succeeded while its parent sync failed. Complete
40+ // that durability boundary before considering a canonical directory: this
41+ // also leaves a newly recreated session untouched.
42+ pending, err := pendingQuarantine(sessions, key)
43+ if err != nil {
44+ return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
45+ }
46+ if pending {
47+ if err := ctx.Err(); err != nil {
48+ return err
49+ }
50+ if err := r.syncDirectory(sessions); err != nil {
51+ return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
52+ }
53+ return nil
54+ }
31 if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {55 if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {
32 return nil56 return nil
33 } else if err != nil {57 } else if err != nil {
34 return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))58 return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
35 }59 }
36- quarantine := filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))60+ quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
37 for attempt := 0; ; attempt++ {61 for attempt := 0; ; attempt++ {
38 if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {62 if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
39 break63 break
40 } else if err != nil {64 } else if err != nil {
41 return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))65 return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
42 }66 }
43- quarantine = filepath.Join(filepath.Dir(canonical), ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))67+ quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
68+ }
69+ if err := ctx.Err(); err != nil {
70+ return err
44 }71 }
45 if err := r.rename(canonical, quarantine); err != nil {72 if err := r.rename(canonical, quarantine); err != nil {
46 return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))73 return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
47 }74 }
48- if err := r.syncDirectory(filepath.Dir(canonical)); err != nil {75+ if err := ctx.Err(); err != nil {
76+ return err
77+ }
78+ if err := r.syncDirectory(sessions); err != nil {
49 return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))79 return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
50 }80 }
51 return nil81 return nil
52 }82 }
53 83
54-// Maintain reaps a fixed number of stale entries. It serializes each session84+func pendingQuarantine(dir, key string) (bool, error) {
55-// with publication and loading so objects written by an in-flight publication85+ f, err := os.Open(dir)
56-// cannot be collected.86+ if errors.Is(err, os.ErrNotExist) {
87+ return false, nil
88+ }
89+ if err != nil {
90+ return false, err
91+ }
92+ defer f.Close()
93+ prefix := ".deleting-" + key + "-"
94+ for {
95+ entries, err := f.ReadDir(maintenanceBatch)
96+ for _, entry := range entries {
97+ if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
98+ return true, nil
99+ }
100+ }
101+ if errors.Is(err, io.EOF) {
102+ return false, nil
103+ }
104+ if err != nil {
105+ return false, err
106+ }
107+ }
108+}
109+
110+// Maintain reaps a bounded amount of stale state. Every scan uses ReadDir with
111+// a fixed batch, and a quarantine tree is peeled one entry at a time so an
112+// interrupted pass can continue after another call or restart.
57 func (r *Repository) Maintain(ctx context.Context) error {113 func (r *Repository) Maintain(ctx context.Context) error {
58 if err := ctx.Err(); err != nil {114 if err := ctx.Err(); err != nil {
59 return err115 return err
@@ -62,8 +118,11 @@ func (r *Repository) Maintain(ctx context.Context) error {
62 if err := r.removeTemps(ctx, r.dir, &budget); err != nil {118 if err := r.removeTemps(ctx, r.dir, &budget); err != nil {
63 return err119 return err
64 }120 }
121+ if budget == 0 {
122+ return nil
123+ }
65 sessions := filepath.Join(r.dir, repositorySessionsDir)124 sessions := filepath.Join(r.dir, repositorySessionsDir)
66- entries, err := os.ReadDir(sessions)125+ entries, err := readDirBatch(sessions, maintenanceBatch)
67 if errors.Is(err, os.ErrNotExist) {126 if errors.Is(err, os.ErrNotExist) {
68 return nil127 return nil
69 }128 }
@@ -79,13 +138,18 @@ func (r *Repository) Maintain(ctx context.Context) error {
79 }138 }
80 path := filepath.Join(sessions, entry.Name())139 path := filepath.Join(sessions, entry.Name())
81 if isQuarantine(entry.Name()) {140 if isQuarantine(entry.Name()) {
82- if err := r.removeTree(path); err != nil && !errors.Is(err, os.ErrNotExist) {141+ changed, err := r.removeTreeBatch(ctx, path, &budget)
142+ if err != nil && !errors.Is(err, os.ErrNotExist) {
83 return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))143 return fmt.Errorf("remove snapshot quarantine %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
84 }144 }
85- if err := r.syncDirectory(sessions); err != nil {145+ if changed {
86- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))146+ if err := ctx.Err(); err != nil {
147+ return err
148+ }
149+ if err := r.syncDirectory(sessions); err != nil {
150+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
151+ }
87 }152 }
88- budget--
89 continue153 continue
90 }154 }
91 if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {155 if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {
@@ -93,6 +157,10 @@ func (r *Repository) Maintain(ctx context.Context) error {
93 }157 }
94 lock := r.sessionLock(entry.Name())158 lock := r.sessionLock(entry.Name())
95 lock.Lock()159 lock.Lock()
160+ if err := ctx.Err(); err != nil {
161+ lock.Unlock()
162+ return err
163+ }
96 err := r.maintainSession(ctx, entry.Name(), &budget)164 err := r.maintainSession(ctx, entry.Name(), &budget)
97 lock.Unlock()165 lock.Unlock()
98 if err != nil {166 if err != nil {
@@ -104,12 +172,81 @@ func (r *Repository) Maintain(ctx context.Context) error {
104 172
105 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }173 func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
106 174
175+func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
176+ f, err := os.Open(dir)
177+ if err != nil {
178+ return nil, err
179+ }
180+ entries, readErr := f.ReadDir(n)
181+ closeErr := f.Close()
182+ if readErr != nil && !errors.Is(readErr, io.EOF) {
183+ return nil, readErr
184+ }
185+ if closeErr != nil {
186+ return nil, closeErr
187+ }
188+ return entries, nil
189+}
190+
191+func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
192+ changed := false
193+ for *budget > 0 {
194+ did, err := r.removeTreeStep(ctx, path, budget)
195+ if err != nil || !did {
196+ return changed, err
197+ }
198+ changed = true
199+ }
200+ return changed, nil
201+}
202+
203+func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
204+ if *budget == 0 {
205+ return false, nil
206+ }
207+ info, err := os.Lstat(path)
208+ if errors.Is(err, os.ErrNotExist) {
209+ return false, nil
210+ }
211+ if err != nil {
212+ return false, err
213+ }
214+ if !info.IsDir() {
215+ if err := ctx.Err(); err != nil {
216+ return false, err
217+ }
218+ if err := r.remove(path); err != nil {
219+ return false, err
220+ }
221+ *budget--
222+ return true, nil
223+ }
224+ entries, err := readDirBatch(path, 1)
225+ if err != nil {
226+ return false, err
227+ }
228+ if len(entries) == 0 {
229+ if err := ctx.Err(); err != nil {
230+ return false, err
231+ }
232+ if err := r.remove(path); err != nil {
233+ return false, err
234+ }
235+ *budget--
236+ return true, nil
237+ }
238+ return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
239+}
240+
107 func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {241 func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {
242+ if err := ctx.Err(); err != nil {
243+ return err
244+ }
108 if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil {245 if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget); err != nil {
109 return err246 return err
110 }247 }
111 objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)248 objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
112- shards, err := os.ReadDir(objectRoot)249+ shards, err := readDirBatch(objectRoot, maintenanceBatch)
113 if err != nil && !errors.Is(err, os.ErrNotExist) {250 if err != nil && !errors.Is(err, os.ErrNotExist) {
114 return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))251 return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
115 }252 }
@@ -124,7 +261,7 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
124 }261 }
125 }262 }
126 263
127- generations, err := r.generationNumbers(key)264+ generations, err := r.generationNumbersBatch(key)
128 if err != nil {265 if err != nil {
129 return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))266 return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
130 }267 }
@@ -155,7 +292,7 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
155 }292 }
156 293
157 retained := make(map[uint64]bool, 2)294 retained := make(map[uint64]bool, 2)
158- for _, state := range states { // generationNumbers is newest first.295+ for _, state := range states {
159 if state.complete && complete < 2 {296 if state.complete && complete < 2 {
160 retained[state.generation] = true297 retained[state.generation] = true
161 complete++298 complete++
@@ -166,25 +303,59 @@ func (r *Repository) maintainSession(ctx context.Context, key string, budget *in
166 keep := retained[state.generation] || !state.complete303 keep := retained[state.generation] || !state.complete
167 if !keep && *budget > 0 {304 if !keep && *budget > 0 {
168 path := r.manifestPath(key, state.generation)305 path := r.manifestPath(key, state.generation)
306+ if err := ctx.Err(); err != nil {
307+ return err
308+ }
169 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {309 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
170 return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))310 return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
171 }311 }
312+ if err := ctx.Err(); err != nil {
313+ return err
314+ }
172 if err := r.syncDirectory(filepath.Dir(path)); err != nil {315 if err := r.syncDirectory(filepath.Dir(path)); err != nil {
173 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))316 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
174 }317 }
175 *budget--318 *budget--
176 continue319 continue
177 }320 }
178- for digest := range state.refs { // incomplete manifests protect in-flight objects.321+ for digest := range state.refs {
179 referenced[digest] = struct{}{}322 referenced[digest] = struct{}{}
180 }323 }
181 }324 }
182 return r.removeUnreferencedObjects(ctx, key, referenced, budget)325 return r.removeUnreferencedObjects(ctx, key, referenced, budget)
183 }326 }
184 327
328+func (r *Repository) generationNumbersBatch(key string) ([]uint64, error) {
329+ entries, err := readDirBatch(filepath.Join(r.sessionPath(key), repositoryGenerations), maintenanceBatch)
330+ if errors.Is(err, os.ErrNotExist) {
331+ return nil, nil
332+ }
333+ if err != nil {
334+ return nil, err
335+ }
336+ numbers := make([]uint64, 0, len(entries))
337+ for _, entry := range entries {
338+ if !entry.IsDir() {
339+ if generation, ok := parseGenerationFilename(entry.Name()); ok {
340+ numbers = append(numbers, generation)
341+ }
342+ }
343+ }
344+ // Generation file names are zero-padded, but keep the explicit comparison
345+ // in case the representation changes.
346+ for i := range numbers {
347+ for j := i + 1; j < len(numbers); j++ {
348+ if numbers[j] > numbers[i] {
349+ numbers[i], numbers[j] = numbers[j], numbers[i]
350+ }
351+ }
352+ }
353+ return numbers, nil
354+}
355+
185 func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error {356 func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string, referenced map[ports.SnapshotDigest]struct{}, budget *int) error {
186 root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)357 root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
187- shards, err := os.ReadDir(root)358+ shards, err := readDirBatch(root, maintenanceBatch)
188 if errors.Is(err, os.ErrNotExist) {359 if errors.Is(err, os.ErrNotExist) {
189 return nil360 return nil
190 }361 }
@@ -199,7 +370,7 @@ func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string,
199 continue370 continue
200 }371 }
201 dir := filepath.Join(root, shard.Name())372 dir := filepath.Join(root, shard.Name())
202- objects, err := os.ReadDir(dir)373+ objects, err := readDirBatch(dir, maintenanceBatch)
203 if err != nil {374 if err != nil {
204 return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))375 return fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
205 }376 }
@@ -221,6 +392,9 @@ func (r *Repository) removeUnreferencedObjects(ctx context.Context, key string,
221 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {392 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
222 return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))393 return fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
223 }394 }
395+ if err := ctx.Err(); err != nil {
396+ return err
397+ }
224 if err := r.syncDirectory(dir); err != nil {398 if err := r.syncDirectory(dir); err != nil {
225 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))399 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
226 }400 }
@@ -234,7 +408,7 @@ func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) e
234 if *budget == 0 {408 if *budget == 0 {
235 return nil409 return nil
236 }410 }
237- entries, err := os.ReadDir(dir)411+ entries, err := readDirBatch(dir, maintenanceBatch)
238 if errors.Is(err, os.ErrNotExist) {412 if errors.Is(err, os.ErrNotExist) {
239 return nil413 return nil
240 }414 }
@@ -255,6 +429,9 @@ func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) e
255 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {429 if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
256 return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))430 return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
257 }431 }
432+ if err := ctx.Err(); err != nil {
433+ return err
434+ }
258 if err := r.syncDirectory(dir); err != nil {435 if err := r.syncDirectory(dir); err != nil {
259 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))436 return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
260 }437 }
@@ -263,15 +440,6 @@ func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int) e
263 return nil440 return nil
264 }441 }
265 442
266-func (r *Repository) removeTree(path string) error {
267- if r.hooks.remove != nil {
268- if err := r.hooks.remove(path); err != nil {
269- return err
270- }
271- }
272- return os.RemoveAll(path)
273-}
274-
275 // maintenancePath avoids exposing the repository root while retaining a443 // maintenancePath avoids exposing the repository root while retaining a
276 // path-safe, actionable location in errors.444 // path-safe, actionable location in errors.
277 func maintenancePath(root, path string) string {445 func maintenancePath(root, path string) string {
M internal/adapters/snapshot/maintenance_test.go +128 -0
@@ -140,6 +140,134 @@ func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) {
140140 }
141141 }
142142
143+func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.T) {
144+ t.Run("publish", func(t *testing.T) {
145+ repo := NewRepository(privateDir(t))
146+ key := sessionKey("named")
147+ lock := repo.sessionLock(key)
148+ lock.Lock()
149+ ctx, cancel := context.WithCancel(context.Background())
150+ started := make(chan struct{})
151+ done := make(chan error, 1)
152+ go func() {
153+ close(started)
154+ done <- repo.Publish(ctx, repositoryPublication(t, "named", 1, []byte("state")))
155+ }()
156+ <-started
157+ cancel()
158+ lock.Unlock()
159+ if err := <-done; !errors.Is(err, context.Canceled) {
160+ t.Fatalf("Publish error = %v, want canceled", err)
161+ }
162+ if _, err := os.Lstat(repo.dir); !errors.Is(err, os.ErrNotExist) {
163+ t.Fatalf("Publish mutated canceled repository: %v", err)
164+ }
165+ })
166+ t.Run("load and delete", func(t *testing.T) {
167+ repo := NewRepository(privateDir(t))
168+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
169+ t.Fatal(err)
170+ }
171+ key := sessionKey("named")
172+ for _, operation := range []struct {
173+ name string
174+ run func(context.Context) error
175+ }{
176+ {"load", func(ctx context.Context) error { _, err := repo.Load(ctx, "named"); return err }},
177+ {"delete", func(ctx context.Context) error { return repo.Delete(ctx, "named") }},
178+ } {
179+ t.Run(operation.name, func(t *testing.T) {
180+ lock := repo.sessionLock(key)
181+ lock.Lock()
182+ ctx, cancel := context.WithCancel(context.Background())
183+ started := make(chan struct{})
184+ done := make(chan error, 1)
185+ go func() { close(started); done <- operation.run(ctx) }()
186+ <-started
187+ cancel()
188+ lock.Unlock()
189+ if err := <-done; !errors.Is(err, context.Canceled) {
190+ t.Fatalf("%s error = %v, want canceled", operation.name, err)
191+ }
192+ if _, err := repo.Load(context.Background(), "named"); err != nil {
193+ t.Fatalf("%s mutated session: %v", operation.name, err)
194+ }
195+ })
196+ }
197+ })
198+ t.Run("maintain", func(t *testing.T) {
199+ repo := NewRepository(privateDir(t))
200+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
201+ t.Fatal(err)
202+ }
203+ lock := repo.sessionLock(sessionKey("named"))
204+ lock.Lock()
205+ ctx, cancel := context.WithCancel(context.Background())
206+ started := make(chan struct{})
207+ done := make(chan error, 1)
208+ go func() { close(started); done <- repo.Maintain(ctx) }()
209+ <-started
210+ cancel()
211+ lock.Unlock()
212+ if err := <-done; !errors.Is(err, context.Canceled) {
213+ t.Fatalf("Maintain error = %v, want canceled", err)
214+ }
215+ if _, err := repo.Load(context.Background(), "named"); err != nil {
216+ t.Fatalf("Maintain mutated session: %v", err)
217+ }
218+ })
219+}
220+
221+func TestRepositoryDeleteRetriesPendingQuarantineSyncWithoutDeletingRecreatedSession(t *testing.T) {
222+ repo := NewRepository(privateDir(t))
223+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("old"))); err != nil {
224+ t.Fatal(err)
225+ }
226+ injected := errors.New("sessions sync")
227+ repo.hooks.syncDirectory = func(string) error { return injected }
228+ if err := repo.Delete(context.Background(), "named"); !errors.Is(err, injected) {
229+ t.Fatalf("Delete error = %v, want sync failure", err)
230+ }
231+ repo.hooks.syncDirectory = nil
232+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("new"))); err != nil {
233+ t.Fatal(err)
234+ }
235+ if err := repo.Delete(context.Background(), "named"); err != nil {
236+ t.Fatalf("retry Delete: %v", err)
237+ }
238+ if _, err := repo.Load(context.Background(), "named"); err != nil {
239+ t.Fatalf("recreated session was deleted: %v", err)
240+ }
241+}
242+
243+func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
244+ repo := NewRepository(privateDir(t))
245+ quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-named-test")
246+ for i := 0; i < maintenanceBatch+1; i++ {
247+ path := filepath.Join(quarantine, "nested", string(rune('a'+i)))
248+ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
249+ t.Fatal(err)
250+ }
251+ if err := os.WriteFile(path, []byte("stale"), 0o600); err != nil {
252+ t.Fatal(err)
253+ }
254+ }
255+ if err := repo.Maintain(context.Background()); err != nil {
256+ t.Fatal(err)
257+ }
258+ if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
259+ t.Fatalf("quarantine after one bounded call = %v, want remaining tree", err)
260+ }
261+ for i := 0; i < 4; i++ {
262+ if err := repo.Maintain(context.Background()); err != nil {
263+ t.Fatal(err)
264+ }
265+ }
266+ if _, err := os.Lstat(quarantine); !errors.Is(err, os.ErrNotExist) {
267+ t.Fatalf("quarantine after resumed maintenance = %v, want removed", err)
268+ }
269+}
270+
143271 func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {
144272 repo := NewRepository(privateDir(t))
145273 if err := os.Mkdir(repo.dir, 0o700); err != nil {
@@ -140,6 +140,134 @@ func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) {
140 }140 }
141 }141 }
142 142
143+func TestRepositoryCancellationAfterSessionLockWaitPreventsMutations(t *testing.T) {
144+ t.Run("publish", func(t *testing.T) {
145+ repo := NewRepository(privateDir(t))
146+ key := sessionKey("named")
147+ lock := repo.sessionLock(key)
148+ lock.Lock()
149+ ctx, cancel := context.WithCancel(context.Background())
150+ started := make(chan struct{})
151+ done := make(chan error, 1)
152+ go func() {
153+ close(started)
154+ done <- repo.Publish(ctx, repositoryPublication(t, "named", 1, []byte("state")))
155+ }()
156+ <-started
157+ cancel()
158+ lock.Unlock()
159+ if err := <-done; !errors.Is(err, context.Canceled) {
160+ t.Fatalf("Publish error = %v, want canceled", err)
161+ }
162+ if _, err := os.Lstat(repo.dir); !errors.Is(err, os.ErrNotExist) {
163+ t.Fatalf("Publish mutated canceled repository: %v", err)
164+ }
165+ })
166+ t.Run("load and delete", func(t *testing.T) {
167+ repo := NewRepository(privateDir(t))
168+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
169+ t.Fatal(err)
170+ }
171+ key := sessionKey("named")
172+ for _, operation := range []struct {
173+ name string
174+ run func(context.Context) error
175+ }{
176+ {"load", func(ctx context.Context) error { _, err := repo.Load(ctx, "named"); return err }},
177+ {"delete", func(ctx context.Context) error { return repo.Delete(ctx, "named") }},
178+ } {
179+ t.Run(operation.name, func(t *testing.T) {
180+ lock := repo.sessionLock(key)
181+ lock.Lock()
182+ ctx, cancel := context.WithCancel(context.Background())
183+ started := make(chan struct{})
184+ done := make(chan error, 1)
185+ go func() { close(started); done <- operation.run(ctx) }()
186+ <-started
187+ cancel()
188+ lock.Unlock()
189+ if err := <-done; !errors.Is(err, context.Canceled) {
190+ t.Fatalf("%s error = %v, want canceled", operation.name, err)
191+ }
192+ if _, err := repo.Load(context.Background(), "named"); err != nil {
193+ t.Fatalf("%s mutated session: %v", operation.name, err)
194+ }
195+ })
196+ }
197+ })
198+ t.Run("maintain", func(t *testing.T) {
199+ repo := NewRepository(privateDir(t))
200+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("state"))); err != nil {
201+ t.Fatal(err)
202+ }
203+ lock := repo.sessionLock(sessionKey("named"))
204+ lock.Lock()
205+ ctx, cancel := context.WithCancel(context.Background())
206+ started := make(chan struct{})
207+ done := make(chan error, 1)
208+ go func() { close(started); done <- repo.Maintain(ctx) }()
209+ <-started
210+ cancel()
211+ lock.Unlock()
212+ if err := <-done; !errors.Is(err, context.Canceled) {
213+ t.Fatalf("Maintain error = %v, want canceled", err)
214+ }
215+ if _, err := repo.Load(context.Background(), "named"); err != nil {
216+ t.Fatalf("Maintain mutated session: %v", err)
217+ }
218+ })
219+}
220+
221+func TestRepositoryDeleteRetriesPendingQuarantineSyncWithoutDeletingRecreatedSession(t *testing.T) {
222+ repo := NewRepository(privateDir(t))
223+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("old"))); err != nil {
224+ t.Fatal(err)
225+ }
226+ injected := errors.New("sessions sync")
227+ repo.hooks.syncDirectory = func(string) error { return injected }
228+ if err := repo.Delete(context.Background(), "named"); !errors.Is(err, injected) {
229+ t.Fatalf("Delete error = %v, want sync failure", err)
230+ }
231+ repo.hooks.syncDirectory = nil
232+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", 1, []byte("new"))); err != nil {
233+ t.Fatal(err)
234+ }
235+ if err := repo.Delete(context.Background(), "named"); err != nil {
236+ t.Fatalf("retry Delete: %v", err)
237+ }
238+ if _, err := repo.Load(context.Background(), "named"); err != nil {
239+ t.Fatalf("recreated session was deleted: %v", err)
240+ }
241+}
242+
243+func TestRepositoryMaintainResumesNestedQuarantineWithinBatch(t *testing.T) {
244+ repo := NewRepository(privateDir(t))
245+ quarantine := filepath.Join(repo.dir, repositorySessionsDir, ".deleting-named-test")
246+ for i := 0; i < maintenanceBatch+1; i++ {
247+ path := filepath.Join(quarantine, "nested", string(rune('a'+i)))
248+ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
249+ t.Fatal(err)
250+ }
251+ if err := os.WriteFile(path, []byte("stale"), 0o600); err != nil {
252+ t.Fatal(err)
253+ }
254+ }
255+ if err := repo.Maintain(context.Background()); err != nil {
256+ t.Fatal(err)
257+ }
258+ if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
259+ t.Fatalf("quarantine after one bounded call = %v, want remaining tree", err)
260+ }
261+ for i := 0; i < 4; i++ {
262+ if err := repo.Maintain(context.Background()); err != nil {
263+ t.Fatal(err)
264+ }
265+ }
266+ if _, err := os.Lstat(quarantine); !errors.Is(err, os.ErrNotExist) {
267+ t.Fatalf("quarantine after resumed maintenance = %v, want removed", err)
268+ }
269+}
270+
143 func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {271 func TestRepositoryMaintainUsesFixedRemovalBatch(t *testing.T) {
144 repo := NewRepository(privateDir(t))272 repo := NewRepository(privateDir(t))
145 if err := os.Mkdir(repo.dir, 0o700); err != nil {273 if err := os.Mkdir(repo.dir, 0o700); err != nil {
M internal/adapters/snapshot/publish.go +15 -0
@@ -17,11 +17,17 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
1717 lock := r.sessionLock(key)
1818 lock.Lock()
1919 defer lock.Unlock()
20+ if err := ctx.Err(); err != nil {
21+ return err
22+ }
2023
2124 refs, supplied, err := validatePublication(publication)
2225 if err != nil {
2326 return err
2427 }
28+ if err := ctx.Err(); err != nil {
29+ return err
30+ }
2531 if err := r.ensureSession(key); err != nil {
2632 return err
2733 }
@@ -61,6 +67,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
6167 return err
6268 }
6369 }
70+ if err := ctx.Err(); err != nil {
71+ return err
72+ }
6473 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {
6574 if sha256.Sum256(existing) != digest || !validObject(existing, ref) {
6675 return fmt.Errorf("existing immutable object is invalid")
@@ -86,6 +95,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
8695 return err
8796 }
8897 }
98+ if err := ctx.Err(); err != nil {
99+ return err
100+ }
89101 if err := r.writeImmutable(manifestPath, publication.Manifest, func(existing []byte) error {
90102 if !equalBytes(existing, publication.Manifest) {
91103 return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
@@ -100,6 +112,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
100112 return err
101113 }
102114 }
115+ if err := ctx.Err(); err != nil {
116+ return err
117+ }
103118 if err := r.writeMutable(r.headPath(key), marshalHead(publication.Generation, sha256.Sum256(publication.Manifest))); err != nil {
104119 return fmt.Errorf("write HEAD: %w", err)
105120 }
@@ -17,11 +17,17 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
17 lock := r.sessionLock(key)17 lock := r.sessionLock(key)
18 lock.Lock()18 lock.Lock()
19 defer lock.Unlock()19 defer lock.Unlock()
20+ if err := ctx.Err(); err != nil {
21+ return err
22+ }
20 23
21 refs, supplied, err := validatePublication(publication)24 refs, supplied, err := validatePublication(publication)
22 if err != nil {25 if err != nil {
23 return err26 return err
24 }27 }
28+ if err := ctx.Err(); err != nil {
29+ return err
30+ }
25 if err := r.ensureSession(key); err != nil {31 if err := r.ensureSession(key); err != nil {
26 return err32 return err
27 }33 }
@@ -61,6 +67,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
61 return err67 return err
62 }68 }
63 }69 }
70+ if err := ctx.Err(); err != nil {
71+ return err
72+ }
64 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {73 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {
65 if sha256.Sum256(existing) != digest || !validObject(existing, ref) {74 if sha256.Sum256(existing) != digest || !validObject(existing, ref) {
66 return fmt.Errorf("existing immutable object is invalid")75 return fmt.Errorf("existing immutable object is invalid")
@@ -86,6 +95,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
86 return err95 return err
87 }96 }
88 }97 }
98+ if err := ctx.Err(); err != nil {
99+ return err
100+ }
89 if err := r.writeImmutable(manifestPath, publication.Manifest, func(existing []byte) error {101 if err := r.writeImmutable(manifestPath, publication.Manifest, func(existing []byte) error {
90 if !equalBytes(existing, publication.Manifest) {102 if !equalBytes(existing, publication.Manifest) {
91 return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)103 return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
@@ -100,6 +112,9 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
100 return err112 return err
101 }113 }
102 }114 }
115+ if err := ctx.Err(); err != nil {
116+ return err
117+ }
103 if err := r.writeMutable(r.headPath(key), marshalHead(publication.Generation, sha256.Sum256(publication.Manifest))); err != nil {118 if err := r.writeMutable(r.headPath(key), marshalHead(publication.Generation, sha256.Sum256(publication.Manifest))); err != nil {
104 return fmt.Errorf("write HEAD: %w", err)119 return fmt.Errorf("write HEAD: %w", err)
105 }120 }
M internal/adapters/snapshot/repository.go +4 -0
@@ -13,6 +13,10 @@ const (
1313 repositoryHead = "HEAD"
1414 generationWidth = 20
1515 maxRepositoryRead = maxSnapshotFileSize
16+ // Legacy import is deliberately smaller than the per-file parser ceiling:
17+ // importing old snapshots must not accumulate an unbounded startup payload.
18+ maxLegacySnapshotFiles = 64
19+ maxLegacySnapshotBytes = 8 << 20
1620 )
1721
1822 // Repository is the crash-safe, content-addressed session snapshot store.
@@ -13,6 +13,10 @@ const (
13 repositoryHead = "HEAD"13 repositoryHead = "HEAD"
14 generationWidth = 2014 generationWidth = 20
15 maxRepositoryRead = maxSnapshotFileSize15 maxRepositoryRead = maxSnapshotFileSize
16+ // Legacy import is deliberately smaller than the per-file parser ceiling:
17+ // importing old snapshots must not accumulate an unbounded startup payload.
18+ maxLegacySnapshotFiles = 64
19+ maxLegacySnapshotBytes = 8 << 20
16 )20 )
17 21
18 // Repository is the crash-safe, content-addressed session snapshot store.22 // Repository is the crash-safe, content-addressed session snapshot store.

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