detached 058bbc8

refactor(snapshot): bound repository maintenance

commit
058bbc82690171b4b8a83c24d1212631238e656f
author
brice · · 2026-07-22 02:23 UTC
parent
d6a5676
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
7 files · +1100 -842
M internal/adapters/snapshot/load.go +148 -84
@@ -6,6 +6,7 @@ import (
66 "encoding/binary"
77 "errors"
88 "fmt"
9+ "io"
910 "os"
1011 "path/filepath"
1112 "sort"
@@ -14,52 +15,40 @@ import (
1415 codec "github.com/bnema/vev/internal/usecase/snapshot"
1516 )
1617
18+// Directory scans are deliberately finite. Snapshot directories are attacker
19+// controlled state, so callers get a retryable error instead of retaining an
20+// arbitrary number of names or spending an arbitrary amount of work.
21+const (
22+ directoryTraversalBatch = 64
23+ maxDirectoryTraversalEntries = 4096
24+)
25+
26+var ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
27+
1728 func (r *Repository) List(ctx context.Context) ([]string, error) {
1829 if err := ctx.Err(); err != nil {
1930 return nil, err
2031 }
21- entries, err := os.ReadDir(filepath.Join(r.dir, repositorySessionsDir))
32+ budget := maxDirectoryTraversalEntries
33+ out := make([]string, 0)
34+ sessions := filepath.Join(r.dir, repositorySessionsDir)
35+ err := walkDirectory(ctx, sessions, &budget, func(entry os.DirEntry) error {
36+ if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {
37+ return nil
38+ }
39+ generation, ok, err := r.loadNewestGeneration(ctx, entry.Name(), &budget)
40+ if err != nil || !ok {
41+ return err
42+ }
43+ out = append(out, generation.Name)
44+ return nil
45+ })
2246 if errors.Is(err, os.ErrNotExist) {
2347 return nil, nil
2448 }
2549 if err != nil {
2650 return nil, fmt.Errorf("read snapshot sessions: %w", err)
2751 }
28- out := make([]string, 0, len(entries))
29- for _, entry := range entries {
30- if err := ctx.Err(); err != nil {
31- return nil, err
32- }
33- if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {
34- continue
35- }
36- numbers, err := r.generationNumbers(entry.Name())
37- if err != nil {
38- continue
39- }
40- for _, number := range numbers {
41- data, err := readBounded(r.manifestPath(entry.Name(), number))
42- if err != nil {
43- continue
44- }
45- manifest, err := codec.UnmarshalManifest(data)
46- if err != nil || sessionKey(manifest.Name) != entry.Name() {
47- continue
48- }
49- killed, tombstoneErr := r.tombstoned(manifest.Name)
50- if tombstoneErr != nil {
51- return nil, fmt.Errorf("read killed session marker: %w", tombstoneErr)
52- }
53- if killed {
54- break
55- }
56- generation, err := r.loadGeneration(ctx, manifest.Name, entry.Name(), number)
57- if err == nil {
58- out = append(out, generation.Name)
59- break
60- }
61- }
62- }
6352 sort.Strings(out)
6453 return out, nil
6554 }
@@ -90,24 +79,32 @@ func (r *Repository) load(ctx context.Context, name, key string) (ports.Snapshot
9079 if err != nil && !errors.Is(err, os.ErrNotExist) {
9180 preferred = 0
9281 }
93- candidates, err := r.generationNumbers(key)
94- if err != nil {
95- return ports.SnapshotGeneration{}, err
96- }
82+ skipped := false
9783 if preferred != 0 {
98- candidates = append([]uint64{preferred}, removeGeneration(candidates, preferred)...)
84+ got, err := r.loadGeneration(ctx, name, key, preferred)
85+ if err == nil && sha256.Sum256(got.Manifest) == preferredDigest {
86+ return got, nil
87+ }
88+ skipped = true
9989 }
100- skipped := false
101- for _, generation := range candidates {
102- if err := ctx.Err(); err != nil {
90+
91+ budget := maxDirectoryTraversalEntries
92+ before := uint64(0) // zero means no upper bound.
93+ for {
94+ generation, ok, err := r.nextGeneration(ctx, key, before, preferred, &budget)
95+ if err != nil {
10396 return ports.SnapshotGeneration{}, err
10497 }
98+ if !ok {
99+ break
100+ }
101+ before = generation
105102 got, err := r.loadGeneration(ctx, name, key, generation)
106- if err != nil || (generation == preferred && sha256.Sum256(got.Manifest) != preferredDigest) {
103+ if err != nil {
107104 skipped = true
108105 continue
109106 }
110- got.Fallback = skipped || (preferred != 0 && generation != preferred)
107+ got.Fallback = skipped || preferred != 0
111108 return got, nil
112109 }
113110 return ports.SnapshotGeneration{}, fmt.Errorf("no complete snapshot generation for %q", name)
@@ -121,16 +118,115 @@ func (r *Repository) currentGeneration(ctx context.Context, name, key string) (u
121118 return generation, got.Manifest, nil
122119 }
123120 }
124- candidates, err := r.generationNumbers(key)
121+ budget := maxDirectoryTraversalEntries
122+ before := uint64(0)
123+ for {
124+ generation, ok, err := r.nextGeneration(ctx, key, before, 0, &budget)
125+ if err != nil || !ok {
126+ return 0, nil, err
127+ }
128+ before = generation
129+ got, err := r.loadGeneration(ctx, name, key, generation)
130+ if err == nil {
131+ return generation, got.Manifest, nil
132+ }
133+ }
134+}
135+
136+// loadNewestGeneration validates candidates in descending generation order
137+// without materializing the generations directory. It is used by List, whose
138+// name is discovered from each manifest rather than supplied by a caller.
139+func (r *Repository) loadNewestGeneration(ctx context.Context, key string, budget *int) (ports.SnapshotGeneration, bool, error) {
140+ before := uint64(0)
141+ for {
142+ generation, ok, err := r.nextGeneration(ctx, key, before, 0, budget)
143+ if err != nil || !ok {
144+ return ports.SnapshotGeneration{}, false, err
145+ }
146+ before = generation
147+ data, err := readBounded(r.manifestPath(key, generation))
148+ if err != nil {
149+ continue
150+ }
151+ manifest, err := codec.UnmarshalManifest(data)
152+ if err != nil || sessionKey(manifest.Name) != key {
153+ continue
154+ }
155+ killed, err := r.tombstoned(manifest.Name)
156+ if err != nil {
157+ return ports.SnapshotGeneration{}, false, fmt.Errorf("read killed session marker: %w", err)
158+ }
159+ if killed {
160+ return ports.SnapshotGeneration{}, false, nil
161+ }
162+ got, err := r.loadGeneration(ctx, manifest.Name, key, generation)
163+ if err == nil {
164+ return got, true, nil
165+ }
166+ }
167+}
168+
169+// nextGeneration finds one candidate in a complete, batched directory cursor
170+// pass. Repeating it with the returned number as before preserves strict
171+// newest-to-oldest fallback while work remains bounded by budget.
172+func (r *Repository) nextGeneration(ctx context.Context, key string, before, exclude uint64, budget *int) (uint64, bool, error) {
173+ var newest uint64
174+ err := walkDirectory(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, func(entry os.DirEntry) error {
175+ if entry.IsDir() {
176+ return nil
177+ }
178+ generation, ok := parseGenerationFilename(entry.Name())
179+ if !ok || generation == exclude || (before != 0 && generation >= before) {
180+ return nil
181+ }
182+ if generation > newest {
183+ newest = generation
184+ }
185+ return nil
186+ })
187+ if errors.Is(err, os.ErrNotExist) {
188+ return 0, false, nil
189+ }
125190 if err != nil {
126- return 0, nil, err
191+ return 0, false, err
127192 }
128- for _, candidate := range candidates {
129- if got, err := r.loadGeneration(ctx, name, key, candidate); err == nil {
130- return candidate, got.Manifest, nil
193+ return newest, newest != 0, nil
194+}
195+
196+// walkDirectory is the common finite directory traversal primitive for reads.
197+// File.ReadDir maintains a cursor on one descriptor and returns at most one
198+// small batch, so no full directory enumeration is retained in memory.
199+func walkDirectory(ctx context.Context, dir string, budget *int, visit func(os.DirEntry) error) (err error) {
200+ file, err := os.Open(dir)
201+ if err != nil {
202+ return err
203+ }
204+ defer func() {
205+ if closeErr := file.Close(); err == nil && closeErr != nil {
206+ err = closeErr
207+ }
208+ }()
209+ for {
210+ if err := ctx.Err(); err != nil {
211+ return err
212+ }
213+ entries, readErr := file.ReadDir(directoryTraversalBatch)
214+ for _, entry := range entries {
215+ if *budget == 0 {
216+ return ErrDirectoryTraversalBudget
217+ }
218+ *budget--
219+ if err := visit(entry); err != nil {
220+ return err
221+ }
222+ }
223+ if errors.Is(readErr, io.EOF) {
224+ return nil
225+ }
226+ if readErr != nil {
227+ return readErr
131228 }
132229 }
133- return 0, nil, nil
134230 }
135231
136232 func (r *Repository) loadGeneration(ctx context.Context, name, key string, generation uint64) (ports.SnapshotGeneration, error) {
@@ -158,43 +254,11 @@ func (r *Repository) loadGeneration(ctx context.Context, name, key string, gener
158254 if sha256.Sum256(object) != digest || !validObject(object, ref) {
159255 return ports.SnapshotGeneration{}, fmt.Errorf("invalid object")
160256 }
161- // readBounded returns a fresh caller-owned buffer, so returning it directly
162- // avoids a second allocation without retaining it in the repository.
163257 objects[digest] = object
164258 }
165259 return ports.SnapshotGeneration{Name: manifest.Name, Generation: generation, Manifest: data, Objects: objects}, nil
166260 }
167261
168-func (r *Repository) generationNumbers(key string) ([]uint64, error) {
169- entries, err := os.ReadDir(filepath.Join(r.sessionPath(key), repositoryGenerations))
170- if errors.Is(err, os.ErrNotExist) {
171- return nil, nil
172- }
173- if err != nil {
174- return nil, err
175- }
176- numbers := make([]uint64, 0, len(entries))
177- for _, entry := range entries {
178- if entry.IsDir() {
179- continue
180- }
181- if generation, ok := parseGenerationFilename(entry.Name()); ok {
182- numbers = append(numbers, generation)
183- }
184- }
185- sort.Slice(numbers, func(i, j int) bool { return numbers[i] > numbers[j] })
186- return numbers, nil
187-}
188-func removeGeneration(numbers []uint64, generation uint64) []uint64 {
189- out := numbers[:0]
190- for _, n := range numbers {
191- if n != generation {
192- out = append(out, n)
193- }
194- }
195- return out
196-}
197-
198262 func marshalHead(generation uint64, digest ports.SnapshotDigest) []byte {
199263 out := make([]byte, 4+8+sha256.Size)
200264 copy(out, "VEVH")
@@ -6,6 +6,7 @@ import (
6 "encoding/binary"6 "encoding/binary"
7 "errors"7 "errors"
8 "fmt"8 "fmt"
9+ "io"
9 "os"10 "os"
10 "path/filepath"11 "path/filepath"
11 "sort"12 "sort"
@@ -14,52 +15,40 @@ import (
14 codec "github.com/bnema/vev/internal/usecase/snapshot"15 codec "github.com/bnema/vev/internal/usecase/snapshot"
15 )16 )
16 17
18+// Directory scans are deliberately finite. Snapshot directories are attacker
19+// controlled state, so callers get a retryable error instead of retaining an
20+// arbitrary number of names or spending an arbitrary amount of work.
21+const (
22+ directoryTraversalBatch = 64
23+ maxDirectoryTraversalEntries = 4096
24+)
25+
26+var ErrDirectoryTraversalBudget = errors.New("snapshot directory traversal budget exceeded")
27+
17 func (r *Repository) List(ctx context.Context) ([]string, error) {28 func (r *Repository) List(ctx context.Context) ([]string, error) {
18 if err := ctx.Err(); err != nil {29 if err := ctx.Err(); err != nil {
19 return nil, err30 return nil, err
20 }31 }
21- entries, err := os.ReadDir(filepath.Join(r.dir, repositorySessionsDir))32+ budget := maxDirectoryTraversalEntries
33+ out := make([]string, 0)
34+ sessions := filepath.Join(r.dir, repositorySessionsDir)
35+ err := walkDirectory(ctx, sessions, &budget, func(entry os.DirEntry) error {
36+ if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {
37+ return nil
38+ }
39+ generation, ok, err := r.loadNewestGeneration(ctx, entry.Name(), &budget)
40+ if err != nil || !ok {
41+ return err
42+ }
43+ out = append(out, generation.Name)
44+ return nil
45+ })
22 if errors.Is(err, os.ErrNotExist) {46 if errors.Is(err, os.ErrNotExist) {
23 return nil, nil47 return nil, nil
24 }48 }
25 if err != nil {49 if err != nil {
26 return nil, fmt.Errorf("read snapshot sessions: %w", err)50 return nil, fmt.Errorf("read snapshot sessions: %w", err)
27 }51 }
28- out := make([]string, 0, len(entries))
29- for _, entry := range entries {
30- if err := ctx.Err(); err != nil {
31- return nil, err
32- }
33- if !entry.IsDir() || !canonicalSessionKey(entry.Name()) {
34- continue
35- }
36- numbers, err := r.generationNumbers(entry.Name())
37- if err != nil {
38- continue
39- }
40- for _, number := range numbers {
41- data, err := readBounded(r.manifestPath(entry.Name(), number))
42- if err != nil {
43- continue
44- }
45- manifest, err := codec.UnmarshalManifest(data)
46- if err != nil || sessionKey(manifest.Name) != entry.Name() {
47- continue
48- }
49- killed, tombstoneErr := r.tombstoned(manifest.Name)
50- if tombstoneErr != nil {
51- return nil, fmt.Errorf("read killed session marker: %w", tombstoneErr)
52- }
53- if killed {
54- break
55- }
56- generation, err := r.loadGeneration(ctx, manifest.Name, entry.Name(), number)
57- if err == nil {
58- out = append(out, generation.Name)
59- break
60- }
61- }
62- }
63 sort.Strings(out)52 sort.Strings(out)
64 return out, nil53 return out, nil
65 }54 }
@@ -90,24 +79,32 @@ func (r *Repository) load(ctx context.Context, name, key string) (ports.Snapshot
90 if err != nil && !errors.Is(err, os.ErrNotExist) {79 if err != nil && !errors.Is(err, os.ErrNotExist) {
91 preferred = 080 preferred = 0
92 }81 }
93- candidates, err := r.generationNumbers(key)82+ skipped := false
94- if err != nil {
95- return ports.SnapshotGeneration{}, err
96- }
97 if preferred != 0 {83 if preferred != 0 {
98- candidates = append([]uint64{preferred}, removeGeneration(candidates, preferred)...)84+ got, err := r.loadGeneration(ctx, name, key, preferred)
85+ if err == nil && sha256.Sum256(got.Manifest) == preferredDigest {
86+ return got, nil
87+ }
88+ skipped = true
99 }89 }
100- skipped := false90+
101- for _, generation := range candidates {91+ budget := maxDirectoryTraversalEntries
102- if err := ctx.Err(); err != nil {92+ before := uint64(0) // zero means no upper bound.
93+ for {
94+ generation, ok, err := r.nextGeneration(ctx, key, before, preferred, &budget)
95+ if err != nil {
103 return ports.SnapshotGeneration{}, err96 return ports.SnapshotGeneration{}, err
104 }97 }
98+ if !ok {
99+ break
100+ }
101+ before = generation
105 got, err := r.loadGeneration(ctx, name, key, generation)102 got, err := r.loadGeneration(ctx, name, key, generation)
106- if err != nil || (generation == preferred && sha256.Sum256(got.Manifest) != preferredDigest) {103+ if err != nil {
107 skipped = true104 skipped = true
108 continue105 continue
109 }106 }
110- got.Fallback = skipped || (preferred != 0 && generation != preferred)107+ got.Fallback = skipped || preferred != 0
111 return got, nil108 return got, nil
112 }109 }
113 return ports.SnapshotGeneration{}, fmt.Errorf("no complete snapshot generation for %q", name)110 return ports.SnapshotGeneration{}, fmt.Errorf("no complete snapshot generation for %q", name)
@@ -121,16 +118,115 @@ func (r *Repository) currentGeneration(ctx context.Context, name, key string) (u
121 return generation, got.Manifest, nil118 return generation, got.Manifest, nil
122 }119 }
123 }120 }
124- candidates, err := r.generationNumbers(key)121+ budget := maxDirectoryTraversalEntries
122+ before := uint64(0)
123+ for {
124+ generation, ok, err := r.nextGeneration(ctx, key, before, 0, &budget)
125+ if err != nil || !ok {
126+ return 0, nil, err
127+ }
128+ before = generation
129+ got, err := r.loadGeneration(ctx, name, key, generation)
130+ if err == nil {
131+ return generation, got.Manifest, nil
132+ }
133+ }
134+}
135+
136+// loadNewestGeneration validates candidates in descending generation order
137+// without materializing the generations directory. It is used by List, whose
138+// name is discovered from each manifest rather than supplied by a caller.
139+func (r *Repository) loadNewestGeneration(ctx context.Context, key string, budget *int) (ports.SnapshotGeneration, bool, error) {
140+ before := uint64(0)
141+ for {
142+ generation, ok, err := r.nextGeneration(ctx, key, before, 0, budget)
143+ if err != nil || !ok {
144+ return ports.SnapshotGeneration{}, false, err
145+ }
146+ before = generation
147+ data, err := readBounded(r.manifestPath(key, generation))
148+ if err != nil {
149+ continue
150+ }
151+ manifest, err := codec.UnmarshalManifest(data)
152+ if err != nil || sessionKey(manifest.Name) != key {
153+ continue
154+ }
155+ killed, err := r.tombstoned(manifest.Name)
156+ if err != nil {
157+ return ports.SnapshotGeneration{}, false, fmt.Errorf("read killed session marker: %w", err)
158+ }
159+ if killed {
160+ return ports.SnapshotGeneration{}, false, nil
161+ }
162+ got, err := r.loadGeneration(ctx, manifest.Name, key, generation)
163+ if err == nil {
164+ return got, true, nil
165+ }
166+ }
167+}
168+
169+// nextGeneration finds one candidate in a complete, batched directory cursor
170+// pass. Repeating it with the returned number as before preserves strict
171+// newest-to-oldest fallback while work remains bounded by budget.
172+func (r *Repository) nextGeneration(ctx context.Context, key string, before, exclude uint64, budget *int) (uint64, bool, error) {
173+ var newest uint64
174+ err := walkDirectory(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, func(entry os.DirEntry) error {
175+ if entry.IsDir() {
176+ return nil
177+ }
178+ generation, ok := parseGenerationFilename(entry.Name())
179+ if !ok || generation == exclude || (before != 0 && generation >= before) {
180+ return nil
181+ }
182+ if generation > newest {
183+ newest = generation
184+ }
185+ return nil
186+ })
187+ if errors.Is(err, os.ErrNotExist) {
188+ return 0, false, nil
189+ }
125 if err != nil {190 if err != nil {
126- return 0, nil, err191+ return 0, false, err
127 }192 }
128- for _, candidate := range candidates {193+ return newest, newest != 0, nil
129- if got, err := r.loadGeneration(ctx, name, key, candidate); err == nil {194+}
130- return candidate, got.Manifest, nil195+
196+// walkDirectory is the common finite directory traversal primitive for reads.
197+// File.ReadDir maintains a cursor on one descriptor and returns at most one
198+// small batch, so no full directory enumeration is retained in memory.
199+func walkDirectory(ctx context.Context, dir string, budget *int, visit func(os.DirEntry) error) (err error) {
200+ file, err := os.Open(dir)
201+ if err != nil {
202+ return err
203+ }
204+ defer func() {
205+ if closeErr := file.Close(); err == nil && closeErr != nil {
206+ err = closeErr
207+ }
208+ }()
209+ for {
210+ if err := ctx.Err(); err != nil {
211+ return err
212+ }
213+ entries, readErr := file.ReadDir(directoryTraversalBatch)
214+ for _, entry := range entries {
215+ if *budget == 0 {
216+ return ErrDirectoryTraversalBudget
217+ }
218+ *budget--
219+ if err := visit(entry); err != nil {
220+ return err
221+ }
222+ }
223+ if errors.Is(readErr, io.EOF) {
224+ return nil
225+ }
226+ if readErr != nil {
227+ return readErr
131 }228 }
132 }229 }
133- return 0, nil, nil
134 }230 }
135 231
136 func (r *Repository) loadGeneration(ctx context.Context, name, key string, generation uint64) (ports.SnapshotGeneration, error) {232 func (r *Repository) loadGeneration(ctx context.Context, name, key string, generation uint64) (ports.SnapshotGeneration, error) {
@@ -158,43 +254,11 @@ func (r *Repository) loadGeneration(ctx context.Context, name, key string, gener
158 if sha256.Sum256(object) != digest || !validObject(object, ref) {254 if sha256.Sum256(object) != digest || !validObject(object, ref) {
159 return ports.SnapshotGeneration{}, fmt.Errorf("invalid object")255 return ports.SnapshotGeneration{}, fmt.Errorf("invalid object")
160 }256 }
161- // readBounded returns a fresh caller-owned buffer, so returning it directly
162- // avoids a second allocation without retaining it in the repository.
163 objects[digest] = object257 objects[digest] = object
164 }258 }
165 return ports.SnapshotGeneration{Name: manifest.Name, Generation: generation, Manifest: data, Objects: objects}, nil259 return ports.SnapshotGeneration{Name: manifest.Name, Generation: generation, Manifest: data, Objects: objects}, nil
166 }260 }
167 261
168-func (r *Repository) generationNumbers(key string) ([]uint64, error) {
169- entries, err := os.ReadDir(filepath.Join(r.sessionPath(key), repositoryGenerations))
170- if errors.Is(err, os.ErrNotExist) {
171- return nil, nil
172- }
173- if err != nil {
174- return nil, err
175- }
176- numbers := make([]uint64, 0, len(entries))
177- for _, entry := range entries {
178- if entry.IsDir() {
179- continue
180- }
181- if generation, ok := parseGenerationFilename(entry.Name()); ok {
182- numbers = append(numbers, generation)
183- }
184- }
185- sort.Slice(numbers, func(i, j int) bool { return numbers[i] > numbers[j] })
186- return numbers, nil
187-}
188-func removeGeneration(numbers []uint64, generation uint64) []uint64 {
189- out := numbers[:0]
190- for _, n := range numbers {
191- if n != generation {
192- out = append(out, n)
193- }
194- }
195- return out
196-}
197-
198 func marshalHead(generation uint64, digest ports.SnapshotDigest) []byte {262 func marshalHead(generation uint64, digest ports.SnapshotDigest) []byte {
199 out := make([]byte, 4+8+sha256.Size)263 out := make([]byte, 4+8+sha256.Size)
200 copy(out, "VEVH")264 copy(out, "VEVH")
A internal/adapters/snapshot/load_bounds_test.go +106 -0
@@ -0,0 +1,106 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "os"
8+ "path/filepath"
9+ "testing"
10+)
11+
12+func TestRepositoryListStopsAtTraversalBudget(t *testing.T) {
13+ repo := NewRepository(privateDir(t))
14+ sessions := filepath.Join(repo.dir, repositorySessionsDir)
15+ if err := os.MkdirAll(sessions, 0o700); err != nil {
16+ t.Fatal(err)
17+ }
18+ for i := 0; i < maxDirectoryTraversalEntries+1; i++ {
19+ if err := os.Mkdir(filepath.Join(sessions, fmt.Sprintf("hostile-%05d", i)), 0o700); err != nil {
20+ t.Fatal(err)
21+ }
22+ }
23+
24+ _, err := repo.List(context.Background())
25+ if !errors.Is(err, ErrDirectoryTraversalBudget) {
26+ t.Fatalf("List error = %v, want directory traversal budget", err)
27+ }
28+}
29+
30+func TestRepositoryLoadStopsAtGenerationTraversalBudget(t *testing.T) {
31+ repo := NewRepository(privateDir(t))
32+ key := sessionKey("named")
33+ generations := filepath.Join(repo.sessionPath(key), repositoryGenerations)
34+ if err := os.MkdirAll(generations, 0o700); err != nil {
35+ t.Fatal(err)
36+ }
37+ for generation := uint64(1); generation <= uint64(maxDirectoryTraversalEntries+1); generation++ {
38+ if err := os.WriteFile(filepath.Join(generations, generationFilename(generation)), nil, 0o600); err != nil {
39+ t.Fatal(err)
40+ }
41+ }
42+
43+ _, err := repo.Load(context.Background(), "named")
44+ if !errors.Is(err, ErrDirectoryTraversalBudget) {
45+ t.Fatalf("Load error = %v, want directory traversal budget", err)
46+ }
47+}
48+
49+func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *testing.T) {
50+ repo := NewRepository(privateDir(t))
51+ for generation := uint64(1); generation <= 3; generation++ {
52+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil {
53+ t.Fatal(err)
54+ }
55+ }
56+ key := sessionKey("named")
57+ publication := repositoryPublication(t, "named", 3, []byte("state-3"))
58+ if err := os.Remove(repo.objectPath(key, publication.Objects[0].Digest)); err != nil {
59+ t.Fatal(err)
60+ }
61+
62+ got, err := repo.Load(context.Background(), "named")
63+ if err != nil {
64+ t.Fatal(err)
65+ }
66+ if got.Generation != 2 || !got.Fallback {
67+ t.Fatalf("Load = generation %d fallback %v, want generation 2 fallback true", got.Generation, got.Fallback)
68+ }
69+}
70+
71+func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing.T) {
72+ repo := NewRepository(privateDir(t))
73+ for generation := uint64(1); generation <= uint64(maxMaintenanceMarkedGenerations+1); generation++ {
74+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil {
75+ t.Fatal(err)
76+ }
77+ }
78+ for pass := 0; pass < 3; pass++ {
79+ if err := repo.Maintain(context.Background()); err != nil {
80+ t.Fatal(err)
81+ }
82+ }
83+ if _, err := os.Lstat(repo.manifestPath(sessionKey("named"), 1)); err != nil {
84+ t.Fatalf("old manifest collected after mark budget overflow: %v", err)
85+ }
86+ if state := repo.maintenanceSessions[sessionKey("named")]; state != nil {
87+ t.Fatalf("overflow maintenance state retained = %#v, want reset", state)
88+ }
89+}
90+
91+func TestMaintenanceMarkStateIsCapped(t *testing.T) {
92+ references := &sessionMaintenance{marked: make(map[uint64]manifestMaintenance)}
93+ if !references.canRetainManifest(maxMaintenanceReferences) {
94+ t.Fatal("state rejected references at its documented ceiling")
95+ }
96+ if references.canRetainManifest(1) {
97+ t.Fatal("state accepted references beyond its documented ceiling")
98+ }
99+ generations := &sessionMaintenance{marked: make(map[uint64]manifestMaintenance)}
100+ for generation := uint64(1); generation <= maxMaintenanceMarkedGenerations; generation++ {
101+ generations.marked[generation] = manifestMaintenance{}
102+ }
103+ if generations.canRetainManifest(0) {
104+ t.Fatal("state accepted generations beyond its documented ceiling")
105+ }
106+}
@@ -0,0 +1,106 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "os"
8+ "path/filepath"
9+ "testing"
10+)
11+
12+func TestRepositoryListStopsAtTraversalBudget(t *testing.T) {
13+ repo := NewRepository(privateDir(t))
14+ sessions := filepath.Join(repo.dir, repositorySessionsDir)
15+ if err := os.MkdirAll(sessions, 0o700); err != nil {
16+ t.Fatal(err)
17+ }
18+ for i := 0; i < maxDirectoryTraversalEntries+1; i++ {
19+ if err := os.Mkdir(filepath.Join(sessions, fmt.Sprintf("hostile-%05d", i)), 0o700); err != nil {
20+ t.Fatal(err)
21+ }
22+ }
23+
24+ _, err := repo.List(context.Background())
25+ if !errors.Is(err, ErrDirectoryTraversalBudget) {
26+ t.Fatalf("List error = %v, want directory traversal budget", err)
27+ }
28+}
29+
30+func TestRepositoryLoadStopsAtGenerationTraversalBudget(t *testing.T) {
31+ repo := NewRepository(privateDir(t))
32+ key := sessionKey("named")
33+ generations := filepath.Join(repo.sessionPath(key), repositoryGenerations)
34+ if err := os.MkdirAll(generations, 0o700); err != nil {
35+ t.Fatal(err)
36+ }
37+ for generation := uint64(1); generation <= uint64(maxDirectoryTraversalEntries+1); generation++ {
38+ if err := os.WriteFile(filepath.Join(generations, generationFilename(generation)), nil, 0o600); err != nil {
39+ t.Fatal(err)
40+ }
41+ }
42+
43+ _, err := repo.Load(context.Background(), "named")
44+ if !errors.Is(err, ErrDirectoryTraversalBudget) {
45+ t.Fatalf("Load error = %v, want directory traversal budget", err)
46+ }
47+}
48+
49+func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *testing.T) {
50+ repo := NewRepository(privateDir(t))
51+ for generation := uint64(1); generation <= 3; generation++ {
52+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil {
53+ t.Fatal(err)
54+ }
55+ }
56+ key := sessionKey("named")
57+ publication := repositoryPublication(t, "named", 3, []byte("state-3"))
58+ if err := os.Remove(repo.objectPath(key, publication.Objects[0].Digest)); err != nil {
59+ t.Fatal(err)
60+ }
61+
62+ got, err := repo.Load(context.Background(), "named")
63+ if err != nil {
64+ t.Fatal(err)
65+ }
66+ if got.Generation != 2 || !got.Fallback {
67+ t.Fatalf("Load = generation %d fallback %v, want generation 2 fallback true", got.Generation, got.Fallback)
68+ }
69+}
70+
71+func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing.T) {
72+ repo := NewRepository(privateDir(t))
73+ for generation := uint64(1); generation <= uint64(maxMaintenanceMarkedGenerations+1); generation++ {
74+ if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil {
75+ t.Fatal(err)
76+ }
77+ }
78+ for pass := 0; pass < 3; pass++ {
79+ if err := repo.Maintain(context.Background()); err != nil {
80+ t.Fatal(err)
81+ }
82+ }
83+ if _, err := os.Lstat(repo.manifestPath(sessionKey("named"), 1)); err != nil {
84+ t.Fatalf("old manifest collected after mark budget overflow: %v", err)
85+ }
86+ if state := repo.maintenanceSessions[sessionKey("named")]; state != nil {
87+ t.Fatalf("overflow maintenance state retained = %#v, want reset", state)
88+ }
89+}
90+
91+func TestMaintenanceMarkStateIsCapped(t *testing.T) {
92+ references := &sessionMaintenance{marked: make(map[uint64]manifestMaintenance)}
93+ if !references.canRetainManifest(maxMaintenanceReferences) {
94+ t.Fatal("state rejected references at its documented ceiling")
95+ }
96+ if references.canRetainManifest(1) {
97+ t.Fatal("state accepted references beyond its documented ceiling")
98+ }
99+ generations := &sessionMaintenance{marked: make(map[uint64]manifestMaintenance)}
100+ for generation := uint64(1); generation <= maxMaintenanceMarkedGenerations; generation++ {
101+ generations.marked[generation] = manifestMaintenance{}
102+ }
103+ if generations.canRetainManifest(0) {
104+ t.Fatal("state accepted generations beyond its documented ceiling")
105+ }
106+}
M internal/adapters/snapshot/maintenance.go +8 -758
@@ -2,125 +2,25 @@ package snapshot
22
33 import (
44 "context"
5- "crypto/sha256"
65 "errors"
76 "fmt"
8- "io"
97 "os"
108 "path/filepath"
11- "strings"
12- "syscall"
13- "time"
14- "unsafe"
15-
16- "github.com/bnema/vev/internal/ports"
17- codec "github.com/bnema/vev/internal/usecase/snapshot"
189 )
1910
20-// maintenanceBatch bounds both a single directory read and the number of
21-// namespace entries removed by one maintenance pass. A pass never materializes
22-// an unbounded directory in memory.
2311 const maintenanceBatch = 64
2412
25-// Delete makes a session unavailable by durably moving it out of the canonical
26-// namespace. Maintain reaps the private quarantine later; Delete never restores
27-// a quarantined directory.
28-func (r *Repository) Delete(ctx context.Context, name string) error {
29- if err := ctx.Err(); err != nil {
30- return err
31- }
32- key := sessionKey(name)
33- lock := r.sessionLock(key)
34- lock.Lock()
35- defer lock.Unlock()
36- if err := ctx.Err(); err != nil {
37- return err
38- }
39-
40- canonical := r.sessionPath(key)
41- sessions := filepath.Dir(canonical)
42- // A prior rename may have succeeded while its parent sync failed. Complete
43- // that durability boundary before considering a canonical directory: this
44- // also leaves a newly recreated session untouched.
45- pending, err := pendingQuarantine(sessions, key)
46- if err != nil {
47- return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
48- }
49- if pending {
50- if err := ctx.Err(); err != nil {
51- return err
52- }
53- if err := r.syncDirectory(sessions); err != nil {
54- return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
55- }
56- return nil
57- }
58- if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {
59- return nil
60- } else if err != nil {
61- return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
62- }
63- quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
64- for attempt := 0; ; attempt++ {
65- if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
66- break
67- } else if err != nil {
68- return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
69- }
70- quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
71- }
72- if err := ctx.Err(); err != nil {
73- return err
74- }
75- // A renamed session can be replaced by a later publication. Invalidate any
76- // mark made for the old namespace before that replacement becomes possible.
77- r.invalidateStorageEpoch(key)
78- if err := r.rename(canonical, quarantine); err != nil {
79- return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
80- }
81- if err := ctx.Err(); err != nil {
82- return err
83- }
84- if err := r.syncDirectory(sessions); err != nil {
85- return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
86- }
87- return nil
88-}
89-
90-func pendingQuarantine(dir, key string) (pending bool, err error) {
91- f, err := os.Open(dir)
92- if errors.Is(err, os.ErrNotExist) {
93- return false, nil
94- }
95- if err != nil {
96- return false, err
97- }
98- defer func() {
99- if closeErr := f.Close(); closeErr != nil {
100- err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
101- }
102- }()
103- prefix := ".deleting-" + key + "-"
104- for {
105- entries, err := f.ReadDir(maintenanceBatch)
106- for _, entry := range entries {
107- if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
108- return true, nil
109- }
110- }
111- if errors.Is(err, io.EOF) {
112- return false, nil
113- }
114- if err != nil {
115- return false, err
116- }
117- }
118-}
13+// These caps bound metadata retained across resumable GC passes. If a session
14+// exceeds either cap, marking remains conservative and no destructive sweep or
15+// generation collection occurs in that cycle.
16+const (
17+ maxMaintenanceMarkedGenerations = 128
18+ maxMaintenanceReferences = 8192
19+)
11920
12021 // Maintain reaps a bounded amount of stale state. Continuation handles are
12122 // repository-owned, so a stable prefix cannot starve entries later in a large
122-// directory. Cancellation and errors discard those handles: a restarted
123-// repository (and the next fresh cycle) always begins at a safe boundary.
23+// directory. Cancellation and errors discard those handles.
12424 func (r *Repository) Maintain(ctx context.Context) (err error) {
12525 r.maintenanceMu.Lock()
12626 defer r.maintenanceMu.Unlock()
@@ -194,653 +94,3 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
19494 // cannot be used here: its 8 KiB internal getdents buffer advances the kernel
19595 // offset beyond a short ReadDir result (see os/dir_unix.go), so saving SeekCur
19696 // after ReadDir would skip buffered entries.
197-type maintenanceCursor struct {
198- offset int64
199- pending []maintenanceDirEntry
200- done bool
201-}
202-
203-type maintenanceDirEntry struct {
204- name string
205- isDir bool
206-}
207-
208-type manifestMaintenance struct {
209- refs map[ports.SnapshotDigest]codec.ObjectRef
210- complete bool
211-}
212-
213-type sessionMaintenance struct {
214- token string
215- epoch uint64
216- marked map[uint64]manifestMaintenance
217- uncertain bool
218- conservative bool
219- markDone bool
220- manifestQueue []uint64
221- sweepQueue []string
222- sweepRootDone bool
223-}
224-
225-func (r *Repository) resetMaintenance() {
226- r.maintenanceCursors = nil
227- r.maintenanceSessions = nil
228-}
229-
230-func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
231-
232-// readMaintenanceDir advances one named directory cursor. Its continuation is
233-// only a Linux seek cookie and at most one call's worth of pending names. The
234-// descriptor used to read a call is always closed before this method returns.
235-// Callers must return entries they could not process with
236-// requeueMaintenanceEntries before yielding for budget exhaustion.
237-func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
238- id := maintenanceCursorID(dir, purpose)
239- cursor := r.maintenanceCursors[id]
240- if cursor != nil && len(cursor.pending) != 0 {
241- entries := cursor.pending
242- cursor.pending = nil
243- if cursor.done {
244- delete(r.maintenanceCursors, id)
245- }
246- return entries, cursor.done, nil
247- }
248- if cursor == nil {
249- cursor = &maintenanceCursor{}
250- r.maintenanceCursors[id] = cursor
251- }
252-
253- entries, done, err := r.readMaintenanceDirent(dir, n, cursor)
254- if err != nil {
255- delete(r.maintenanceCursors, id)
256- return nil, false, err
257- }
258- if done {
259- delete(r.maintenanceCursors, id)
260- }
261- return entries, done, nil
262-}
263-
264-// maintenanceDirectory is the small syscall surface needed to resume a
265-// bounded getdents cursor. It permits deterministic fault injection without
266-// retaining directory descriptors between maintenance passes.
267-type maintenanceDirectory interface {
268- Seek(int64, int) (int64, error)
269- ReadDirent([]byte) (int, error)
270- Close() error
271-}
272-
273-type osMaintenanceDirectory struct{ file *os.File }
274-
275-func (d osMaintenanceDirectory) Seek(offset int64, whence int) (int64, error) {
276- return syscall.Seek(int(d.file.Fd()), offset, whence)
277-}
278-
279-func (d osMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
280- return syscall.ReadDirent(int(d.file.Fd()), buffer)
281-}
282-
283-func (d osMaintenanceDirectory) Close() error { return d.file.Close() }
284-
285-func (r *Repository) openMaintenanceDirectory(dir string) (maintenanceDirectory, error) {
286- if open := r.hooks.openMaintenanceDirectory; open != nil {
287- return open(dir)
288- }
289- file, err := os.Open(dir)
290- if err != nil {
291- return nil, err
292- }
293- return osMaintenanceDirectory{file: file}, nil
294-}
295-
296-// maintenanceDirectoryError adds stable operation context and removes any
297-// filesystem path while retaining the underlying cause for errors.Is.
298-func maintenanceDirectoryError(operation string, err error) error {
299- return fmt.Errorf("%s: %w", operation, safeFilesystemError(err))
300-}
301-
302-// readMaintenanceDirent uses the Linux getdents seek cookie from each record.
303-// syscall.Dirent's Off field is the d_off member of linux_dirent64 and
304-// syscall.Seek is lseek(2), as defined by Go's syscall Linux sources.
305-func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
306- file, err := r.openMaintenanceDirectory(dir)
307- if err != nil {
308- return nil, false, maintenanceDirectoryError("open maintenance directory", err)
309- }
310- defer func() {
311- if closeErr := file.Close(); closeErr != nil {
312- closeErr = maintenanceDirectoryError("close maintenance directory", closeErr)
313- if err != nil {
314- err = errors.Join(err, closeErr)
315- } else {
316- entries = nil
317- done = false
318- err = closeErr
319- }
320- }
321- }()
322- if cursor.offset != 0 {
323- if _, err := file.Seek(cursor.offset, io.SeekStart); err != nil {
324- return nil, false, maintenanceDirectoryError("seek maintenance directory", err)
325- }
326- }
327-
328- entries = make([]maintenanceDirEntry, 0, limit)
329- buffer := make([]byte, 8192)
330- for len(entries) < limit {
331- count, err := file.ReadDirent(buffer)
332- if err != nil {
333- return nil, false, maintenanceDirectoryError("read maintenance directory", err)
334- }
335- if count == 0 {
336- return entries, true, nil
337- }
338- for data := buffer[:count]; len(data) != 0 && len(entries) < limit; {
339- if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) {
340- return nil, false, syscall.EIO
341- }
342- record := (*syscall.Dirent)(unsafe.Pointer(&data[0]))
343- reclen := int(record.Reclen)
344- if reclen <= 0 || reclen > len(data) {
345- return nil, false, syscall.EIO
346- }
347- data = data[reclen:]
348- // Advance past every raw record, including dot and disappeared entries.
349- cursor.offset = record.Off
350- nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-int(unsafe.Offsetof(syscall.Dirent{}.Name)))
351- if end := strings.IndexByte(string(nameBytes), 0); end >= 0 {
352- nameBytes = nameBytes[:end]
353- }
354- name := string(nameBytes)
355- if name == "." || name == ".." || name == "" {
356- continue
357- }
358- info, err := os.Lstat(filepath.Join(dir, name))
359- if errors.Is(err, os.ErrNotExist) {
360- continue
361- }
362- if err != nil {
363- return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)
364- }
365- entries = append(entries, maintenanceDirEntry{name: name, isDir: info.IsDir()})
366- }
367- }
368- return entries, false, nil
369-}
370-
371-func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []maintenanceDirEntry) {
372- if len(entries) == 0 {
373- return
374- }
375- id := maintenanceCursorID(dir, purpose)
376- cursor := r.maintenanceCursors[id]
377- if cursor == nil {
378- cursor = &maintenanceCursor{done: true}
379- r.maintenanceCursors[id] = cursor
380- }
381- cursor.pending = append(entries, cursor.pending...)
382-}
383-
384-func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
385-
386-// readDirBatch is used only for a mutating quarantine walk, where every
387-// successful step removes the entry it selected. General maintenance scans use
388-// repository-owned cursors above.
389-func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
390- f, err := os.Open(dir)
391- if err != nil {
392- return nil, err
393- }
394- entries, readErr := f.ReadDir(n)
395- closeErr := f.Close()
396- if readErr != nil && !errors.Is(readErr, io.EOF) {
397- return nil, readErr
398- }
399- if closeErr != nil {
400- return nil, closeErr
401- }
402- return entries, nil
403-}
404-
405-func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
406- changed := false
407- for *budget > 0 {
408- did, err := r.removeTreeStep(ctx, path, budget)
409- if err != nil || !did {
410- return changed, err
411- }
412- changed = true
413- }
414- return changed, nil
415-}
416-
417-func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
418- if *budget == 0 {
419- return false, nil
420- }
421- info, err := os.Lstat(path)
422- if errors.Is(err, os.ErrNotExist) {
423- return false, nil
424- }
425- if err != nil {
426- return false, err
427- }
428- if !info.IsDir() {
429- if err := ctx.Err(); err != nil {
430- return false, err
431- }
432- if err := r.remove(path); err != nil {
433- return false, err
434- }
435- *budget--
436- return true, nil
437- }
438- // A quarantine tree contains no live entries. Reopening from its start is
439- // safe here because each successful step removes that first descendant.
440- entries, err := readDirBatch(path, 1)
441- if err != nil {
442- return false, err
443- }
444- if len(entries) == 0 {
445- if err := ctx.Err(); err != nil {
446- return false, err
447- }
448- if err := r.remove(path); err != nil {
449- return false, err
450- }
451- *budget--
452- return true, nil
453- }
454- return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
455-}
456-
457-func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {
458- if err := ctx.Err(); err != nil {
459- return err
460- }
461- state, err := r.sessionMaintenanceState(key)
462- if err != nil {
463- return err
464- }
465- if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil || *budget == 0 {
466- return err
467- }
468-
469- objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
470- shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
471- if err != nil && !errors.Is(err, os.ErrNotExist) {
472- return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
473- }
474- for i, shard := range shards {
475- if *budget == 0 {
476- r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
477- return nil
478- }
479- if shard.isDir {
480- if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
481- return err
482- }
483- }
484- }
485- if *budget == 0 {
486- return nil
487- }
488- if !state.markDone {
489- if err := r.markSession(ctx, key, state); err != nil {
490- return err
491- }
492- if !state.markDone {
493- return nil
494- }
495- }
496- if state.epoch != r.storageEpoch(key) {
497- // A publication may have left an unpublished manifest after this mark.
498- // Discard all mark and sweep cursors; the next pass starts from it.
499- r.clearSessionMaintenance(key)
500- return nil
501- }
502- if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
503- return err
504- }
505- if state.uncertain {
506- // An unreadable or invalid manifest may reference any existing blob. Do
507- // not sweep in this cycle; retry marking from a fresh directory pass.
508- r.clearSessionMaintenance(key)
509- return nil
510- }
511- if err := r.sweepSession(ctx, key, state, budget); err != nil {
512- return err
513- }
514- return nil
515-}
516-
517-func (r *Repository) sessionMaintenanceState(key string) (*sessionMaintenance, error) {
518- token, conservative, err := r.maintenanceToken(key)
519- if err != nil {
520- return nil, err
521- }
522- state := r.maintenanceSessions[key]
523- if state == nil || state.token != token || state.epoch != r.storageEpoch(key) {
524- r.clearSessionMaintenance(key)
525- state = &sessionMaintenance{
526- token: token,
527- epoch: r.storageEpoch(key),
528- conservative: conservative,
529- marked: make(map[uint64]manifestMaintenance),
530- }
531- r.maintenanceSessions[key] = state
532- }
533- return state, nil
534-}
535-
536-// maintenanceToken is the publication boundary. A missing or corrupt HEAD is
537-// itself stable maintenance state: we retain every classified reference until a
538-// valid publication changes that state, rather than restarting the mark pass.
539-func (r *Repository) maintenanceToken(key string) (string, bool, error) {
540- data, exists, err := readOptionalBounded(r.headPath(key))
541- if err != nil {
542- return "", false, err
543- }
544- if !exists {
545- return "missing", true, nil
546- }
547- sum := sha256.Sum256(data)
548- if _, _, err := r.readHead(key); err != nil {
549- return fmt.Sprintf("invalid:%x", sum), true, nil
550- }
551- return fmt.Sprintf("valid:%x", sum), false, nil
552-}
553-
554-func (r *Repository) clearSessionMaintenance(key string) {
555- delete(r.maintenanceSessions, key)
556- prefix := "\x00" + filepath.Clean(r.sessionPath(key))
557- for id := range r.maintenanceCursors {
558- if strings.HasSuffix(id, prefix) || strings.Contains(id, ":"+key+":") || strings.Contains(id, ":"+key+"\x00") {
559- delete(r.maintenanceCursors, id)
560- }
561- }
562-}
563-
564-func (r *Repository) markSession(ctx context.Context, key string, state *sessionMaintenance) error {
565- dir := filepath.Join(r.sessionPath(key), repositoryGenerations)
566- entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "mark-generations:"+key)
567- if errors.Is(err, os.ErrNotExist) {
568- state.markDone = true
569- return nil
570- }
571- if err != nil {
572- return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
573- }
574- for _, entry := range entries {
575- if err := ctx.Err(); err != nil {
576- return err
577- }
578- if entry.isDir {
579- continue
580- }
581- generation, ok := parseGenerationFilename(entry.name)
582- if !ok {
583- continue
584- }
585- item := manifestMaintenance{}
586- data, err := readBounded(r.manifestPath(key, generation))
587- if err != nil {
588- state.uncertain = true
589- state.marked[generation] = item
590- continue
591- }
592- manifest, err := codec.UnmarshalManifest(data)
593- if err != nil || manifest.Generation != generation || sessionKey(manifest.Name) != key {
594- state.uncertain = true
595- state.marked[generation] = item
596- continue
597- }
598- item.refs = manifestRefs(manifest)
599- if item.refs == nil || !withinGenerationBudget(len(data), item.refs) {
600- state.uncertain = true
601- state.marked[generation] = item
602- continue
603- }
604- // Valid but incomplete generations are marked too. This protects blobs
605- // written before a failed or in-flight publication reaches HEAD.
606- _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
607- item.complete = loadErr == nil
608- state.marked[generation] = item
609- }
610- if done {
611- state.markDone = true
612- if !state.conservative {
613- state.manifestQueue = retainedManifestQueue(state.marked)
614- }
615- }
616- return nil
617-}
618-
619-func retainedManifestQueue(marked map[uint64]manifestMaintenance) []uint64 {
620- complete := make([]uint64, 0, len(marked))
621- for generation, item := range marked {
622- if item.complete {
623- complete = append(complete, generation)
624- }
625- }
626- for i := range complete {
627- for j := i + 1; j < len(complete); j++ {
628- if complete[j] > complete[i] {
629- complete[i], complete[j] = complete[j], complete[i]
630- }
631- }
632- }
633- if len(complete) <= 2 {
634- return nil
635- }
636- return complete[2:]
637-}
638-
639-func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
640- for len(state.manifestQueue) != 0 && *budget > 0 {
641- if err := ctx.Err(); err != nil {
642- return err
643- }
644- generation := state.manifestQueue[0]
645- path := r.manifestPath(key, generation)
646- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
647- return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
648- }
649- if err := ctx.Err(); err != nil {
650- return err
651- }
652- if err := r.syncDirectory(filepath.Dir(path)); err != nil {
653- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
654- }
655- state.manifestQueue = state.manifestQueue[1:]
656- *budget--
657- }
658- return nil
659-}
660-
661-func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
662- // Every bounded sweep batch is tied to the mark's storage epoch. Do not use
663- // stale references after any publication or replacement.
664- if state.epoch != r.storageEpoch(key) {
665- r.clearSessionMaintenance(key)
666- return nil
667- }
668- referenced := retainedReferences(state.marked, state.conservative)
669- root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
670- // Do not keep accumulating shard names while draining earlier work. This
671- // bounds retained names to one directory batch even for a session with many
672- // object shards.
673- if !state.sweepRootDone && len(state.sweepQueue) == 0 {
674- entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
675- if errors.Is(err, os.ErrNotExist) {
676- state.sweepRootDone = true
677- } else if err != nil {
678- return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
679- } else {
680- for _, entry := range entries {
681- if entry.isDir {
682- state.sweepQueue = append(state.sweepQueue, entry.name)
683- }
684- }
685- state.sweepRootDone = done
686- }
687- }
688- if len(state.sweepQueue) != 0 && *budget > 0 {
689- shard := state.sweepQueue[0]
690- done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
691- if err != nil {
692- return err
693- }
694- if done {
695- state.sweepQueue = state.sweepQueue[1:]
696- }
697- }
698- if state.sweepRootDone && len(state.sweepQueue) == 0 {
699- r.clearSessionMaintenance(key)
700- }
701- return nil
702-}
703-
704-func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
705- if conservative {
706- references := make(map[ports.SnapshotDigest]struct{})
707- for _, item := range marked {
708- for digest := range item.refs {
709- references[digest] = struct{}{}
710- }
711- }
712- return references
713- }
714- complete := make([]uint64, 0, len(marked))
715- for generation, item := range marked {
716- if item.complete {
717- complete = append(complete, generation)
718- }
719- }
720- for i := range complete {
721- for j := i + 1; j < len(complete); j++ {
722- if complete[j] > complete[i] {
723- complete[i], complete[j] = complete[j], complete[i]
724- }
725- }
726- }
727- keep := make(map[uint64]bool, 2)
728- for _, generation := range complete {
729- if len(keep) == 2 {
730- break
731- }
732- keep[generation] = true
733- }
734- references := make(map[ports.SnapshotDigest]struct{})
735- for generation, item := range marked {
736- if !item.complete || keep[generation] {
737- for digest := range item.refs {
738- references[digest] = struct{}{}
739- }
740- }
741- }
742- return references
743-}
744-
745-func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[ports.SnapshotDigest]struct{}, budget *int, key, shard string) (bool, error) {
746- entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "sweep-objects:"+key+":"+shard)
747- if errors.Is(err, os.ErrNotExist) {
748- return true, nil
749- }
750- if err != nil {
751- return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
752- }
753- for i, object := range entries {
754- if *budget == 0 {
755- r.requeueMaintenanceEntries(dir, "sweep-objects:"+key+":"+shard, entries[i:])
756- return false, nil
757- }
758- if err := ctx.Err(); err != nil {
759- return false, err
760- }
761- digest, ok := parseObjectDigest(object.name)
762- if !ok || object.isDir {
763- continue
764- }
765- if _, used := referenced[digest]; used {
766- continue
767- }
768- path := filepath.Join(dir, object.name)
769- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
770- return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
771- }
772- if err := ctx.Err(); err != nil {
773- return false, err
774- }
775- if err := r.syncDirectory(dir); err != nil {
776- return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
777- }
778- *budget--
779- }
780- return done, nil
781-}
782-
783-func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
784- if *budget == 0 {
785- return nil
786- }
787- entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
788- if errors.Is(err, os.ErrNotExist) {
789- return nil
790- }
791- if err != nil {
792- return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
793- }
794- for i, entry := range entries {
795- if *budget == 0 {
796- r.requeueMaintenanceEntries(dir, purpose, entries[i:])
797- return nil
798- }
799- if err := ctx.Err(); err != nil {
800- return err
801- }
802- if entry.isDir || !strings.HasPrefix(entry.name, ".tmp-") {
803- continue
804- }
805- path := filepath.Join(dir, entry.name)
806- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
807- return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
808- }
809- if err := ctx.Err(); err != nil {
810- return err
811- }
812- if err := r.syncDirectory(dir); err != nil {
813- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
814- }
815- *budget--
816- }
817- return nil
818-}
819-
820-// maintenancePath avoids exposing the repository root while retaining a
821-// path-safe, actionable location in errors.
822-func maintenancePath(root, path string) string {
823- rel, err := filepath.Rel(root, path)
824- if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." {
825- return filepath.Base(path)
826- }
827- return rel
828-}
829-
830-type safePathError struct {
831- op string
832- cause error
833-}
834-
835-func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() }
836-func (e safePathError) Unwrap() error { return e.cause }
837-
838-// safeFilesystemError preserves error matching without including a filesystem
839-// path from terminal-facing error text.
840-func safeFilesystemError(err error) error {
841- var pathError *os.PathError
842- if errors.As(err, &pathError) {
843- return safePathError{op: pathError.Op, cause: pathError.Err}
844- }
845- return err
846-}
@@ -2,125 +2,25 @@ package snapshot
2 2
3 import (3 import (
4 "context"4 "context"
5- "crypto/sha256"
6 "errors"5 "errors"
7 "fmt"6 "fmt"
8- "io"
9 "os"7 "os"
10 "path/filepath"8 "path/filepath"
11- "strings"
12- "syscall"
13- "time"
14- "unsafe"
15-
16- "github.com/bnema/vev/internal/ports"
17- codec "github.com/bnema/vev/internal/usecase/snapshot"
18 )9 )
19 10
20-// maintenanceBatch bounds both a single directory read and the number of
21-// namespace entries removed by one maintenance pass. A pass never materializes
22-// an unbounded directory in memory.
23 const maintenanceBatch = 6411 const maintenanceBatch = 64
24 12
25-// Delete makes a session unavailable by durably moving it out of the canonical13+// These caps bound metadata retained across resumable GC passes. If a session
26-// namespace. Maintain reaps the private quarantine later; Delete never restores14+// exceeds either cap, marking remains conservative and no destructive sweep or
27-// a quarantined directory.15+// generation collection occurs in that cycle.
28-func (r *Repository) Delete(ctx context.Context, name string) error {16+const (
29- if err := ctx.Err(); err != nil {17+ maxMaintenanceMarkedGenerations = 128
30- return err18+ maxMaintenanceReferences = 8192
31- }19+)
32- key := sessionKey(name)
33- lock := r.sessionLock(key)
34- lock.Lock()
35- defer lock.Unlock()
36- if err := ctx.Err(); err != nil {
37- return err
38- }
39-
40- canonical := r.sessionPath(key)
41- sessions := filepath.Dir(canonical)
42- // A prior rename may have succeeded while its parent sync failed. Complete
43- // that durability boundary before considering a canonical directory: this
44- // also leaves a newly recreated session untouched.
45- pending, err := pendingQuarantine(sessions, key)
46- if err != nil {
47- return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
48- }
49- if pending {
50- if err := ctx.Err(); err != nil {
51- return err
52- }
53- if err := r.syncDirectory(sessions); err != nil {
54- return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
55- }
56- return nil
57- }
58- if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {
59- return nil
60- } else if err != nil {
61- return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
62- }
63- quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
64- for attempt := 0; ; attempt++ {
65- if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
66- break
67- } else if err != nil {
68- return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
69- }
70- quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
71- }
72- if err := ctx.Err(); err != nil {
73- return err
74- }
75- // A renamed session can be replaced by a later publication. Invalidate any
76- // mark made for the old namespace before that replacement becomes possible.
77- r.invalidateStorageEpoch(key)
78- if err := r.rename(canonical, quarantine); err != nil {
79- return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
80- }
81- if err := ctx.Err(); err != nil {
82- return err
83- }
84- if err := r.syncDirectory(sessions); err != nil {
85- return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
86- }
87- return nil
88-}
89-
90-func pendingQuarantine(dir, key string) (pending bool, err error) {
91- f, err := os.Open(dir)
92- if errors.Is(err, os.ErrNotExist) {
93- return false, nil
94- }
95- if err != nil {
96- return false, err
97- }
98- defer func() {
99- if closeErr := f.Close(); closeErr != nil {
100- err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
101- }
102- }()
103- prefix := ".deleting-" + key + "-"
104- for {
105- entries, err := f.ReadDir(maintenanceBatch)
106- for _, entry := range entries {
107- if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
108- return true, nil
109- }
110- }
111- if errors.Is(err, io.EOF) {
112- return false, nil
113- }
114- if err != nil {
115- return false, err
116- }
117- }
118-}
119 20
120 // Maintain reaps a bounded amount of stale state. Continuation handles are21 // Maintain reaps a bounded amount of stale state. Continuation handles are
121 // repository-owned, so a stable prefix cannot starve entries later in a large22 // repository-owned, so a stable prefix cannot starve entries later in a large
122-// directory. Cancellation and errors discard those handles: a restarted23+// directory. Cancellation and errors discard those handles.
123-// repository (and the next fresh cycle) always begins at a safe boundary.
124 func (r *Repository) Maintain(ctx context.Context) (err error) {24 func (r *Repository) Maintain(ctx context.Context) (err error) {
125 r.maintenanceMu.Lock()25 r.maintenanceMu.Lock()
126 defer r.maintenanceMu.Unlock()26 defer r.maintenanceMu.Unlock()
@@ -194,653 +94,3 @@ func (r *Repository) Maintain(ctx context.Context) (err error) {
194 // cannot be used here: its 8 KiB internal getdents buffer advances the kernel94 // cannot be used here: its 8 KiB internal getdents buffer advances the kernel
195 // offset beyond a short ReadDir result (see os/dir_unix.go), so saving SeekCur95 // offset beyond a short ReadDir result (see os/dir_unix.go), so saving SeekCur
196 // after ReadDir would skip buffered entries.96 // after ReadDir would skip buffered entries.
197-type maintenanceCursor struct {
198- offset int64
199- pending []maintenanceDirEntry
200- done bool
201-}
202-
203-type maintenanceDirEntry struct {
204- name string
205- isDir bool
206-}
207-
208-type manifestMaintenance struct {
209- refs map[ports.SnapshotDigest]codec.ObjectRef
210- complete bool
211-}
212-
213-type sessionMaintenance struct {
214- token string
215- epoch uint64
216- marked map[uint64]manifestMaintenance
217- uncertain bool
218- conservative bool
219- markDone bool
220- manifestQueue []uint64
221- sweepQueue []string
222- sweepRootDone bool
223-}
224-
225-func (r *Repository) resetMaintenance() {
226- r.maintenanceCursors = nil
227- r.maintenanceSessions = nil
228-}
229-
230-func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
231-
232-// readMaintenanceDir advances one named directory cursor. Its continuation is
233-// only a Linux seek cookie and at most one call's worth of pending names. The
234-// descriptor used to read a call is always closed before this method returns.
235-// Callers must return entries they could not process with
236-// requeueMaintenanceEntries before yielding for budget exhaustion.
237-func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
238- id := maintenanceCursorID(dir, purpose)
239- cursor := r.maintenanceCursors[id]
240- if cursor != nil && len(cursor.pending) != 0 {
241- entries := cursor.pending
242- cursor.pending = nil
243- if cursor.done {
244- delete(r.maintenanceCursors, id)
245- }
246- return entries, cursor.done, nil
247- }
248- if cursor == nil {
249- cursor = &maintenanceCursor{}
250- r.maintenanceCursors[id] = cursor
251- }
252-
253- entries, done, err := r.readMaintenanceDirent(dir, n, cursor)
254- if err != nil {
255- delete(r.maintenanceCursors, id)
256- return nil, false, err
257- }
258- if done {
259- delete(r.maintenanceCursors, id)
260- }
261- return entries, done, nil
262-}
263-
264-// maintenanceDirectory is the small syscall surface needed to resume a
265-// bounded getdents cursor. It permits deterministic fault injection without
266-// retaining directory descriptors between maintenance passes.
267-type maintenanceDirectory interface {
268- Seek(int64, int) (int64, error)
269- ReadDirent([]byte) (int, error)
270- Close() error
271-}
272-
273-type osMaintenanceDirectory struct{ file *os.File }
274-
275-func (d osMaintenanceDirectory) Seek(offset int64, whence int) (int64, error) {
276- return syscall.Seek(int(d.file.Fd()), offset, whence)
277-}
278-
279-func (d osMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
280- return syscall.ReadDirent(int(d.file.Fd()), buffer)
281-}
282-
283-func (d osMaintenanceDirectory) Close() error { return d.file.Close() }
284-
285-func (r *Repository) openMaintenanceDirectory(dir string) (maintenanceDirectory, error) {
286- if open := r.hooks.openMaintenanceDirectory; open != nil {
287- return open(dir)
288- }
289- file, err := os.Open(dir)
290- if err != nil {
291- return nil, err
292- }
293- return osMaintenanceDirectory{file: file}, nil
294-}
295-
296-// maintenanceDirectoryError adds stable operation context and removes any
297-// filesystem path while retaining the underlying cause for errors.Is.
298-func maintenanceDirectoryError(operation string, err error) error {
299- return fmt.Errorf("%s: %w", operation, safeFilesystemError(err))
300-}
301-
302-// readMaintenanceDirent uses the Linux getdents seek cookie from each record.
303-// syscall.Dirent's Off field is the d_off member of linux_dirent64 and
304-// syscall.Seek is lseek(2), as defined by Go's syscall Linux sources.
305-func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
306- file, err := r.openMaintenanceDirectory(dir)
307- if err != nil {
308- return nil, false, maintenanceDirectoryError("open maintenance directory", err)
309- }
310- defer func() {
311- if closeErr := file.Close(); closeErr != nil {
312- closeErr = maintenanceDirectoryError("close maintenance directory", closeErr)
313- if err != nil {
314- err = errors.Join(err, closeErr)
315- } else {
316- entries = nil
317- done = false
318- err = closeErr
319- }
320- }
321- }()
322- if cursor.offset != 0 {
323- if _, err := file.Seek(cursor.offset, io.SeekStart); err != nil {
324- return nil, false, maintenanceDirectoryError("seek maintenance directory", err)
325- }
326- }
327-
328- entries = make([]maintenanceDirEntry, 0, limit)
329- buffer := make([]byte, 8192)
330- for len(entries) < limit {
331- count, err := file.ReadDirent(buffer)
332- if err != nil {
333- return nil, false, maintenanceDirectoryError("read maintenance directory", err)
334- }
335- if count == 0 {
336- return entries, true, nil
337- }
338- for data := buffer[:count]; len(data) != 0 && len(entries) < limit; {
339- if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) {
340- return nil, false, syscall.EIO
341- }
342- record := (*syscall.Dirent)(unsafe.Pointer(&data[0]))
343- reclen := int(record.Reclen)
344- if reclen <= 0 || reclen > len(data) {
345- return nil, false, syscall.EIO
346- }
347- data = data[reclen:]
348- // Advance past every raw record, including dot and disappeared entries.
349- cursor.offset = record.Off
350- nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-int(unsafe.Offsetof(syscall.Dirent{}.Name)))
351- if end := strings.IndexByte(string(nameBytes), 0); end >= 0 {
352- nameBytes = nameBytes[:end]
353- }
354- name := string(nameBytes)
355- if name == "." || name == ".." || name == "" {
356- continue
357- }
358- info, err := os.Lstat(filepath.Join(dir, name))
359- if errors.Is(err, os.ErrNotExist) {
360- continue
361- }
362- if err != nil {
363- return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)
364- }
365- entries = append(entries, maintenanceDirEntry{name: name, isDir: info.IsDir()})
366- }
367- }
368- return entries, false, nil
369-}
370-
371-func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []maintenanceDirEntry) {
372- if len(entries) == 0 {
373- return
374- }
375- id := maintenanceCursorID(dir, purpose)
376- cursor := r.maintenanceCursors[id]
377- if cursor == nil {
378- cursor = &maintenanceCursor{done: true}
379- r.maintenanceCursors[id] = cursor
380- }
381- cursor.pending = append(entries, cursor.pending...)
382-}
383-
384-func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
385-
386-// readDirBatch is used only for a mutating quarantine walk, where every
387-// successful step removes the entry it selected. General maintenance scans use
388-// repository-owned cursors above.
389-func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
390- f, err := os.Open(dir)
391- if err != nil {
392- return nil, err
393- }
394- entries, readErr := f.ReadDir(n)
395- closeErr := f.Close()
396- if readErr != nil && !errors.Is(readErr, io.EOF) {
397- return nil, readErr
398- }
399- if closeErr != nil {
400- return nil, closeErr
401- }
402- return entries, nil
403-}
404-
405-func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
406- changed := false
407- for *budget > 0 {
408- did, err := r.removeTreeStep(ctx, path, budget)
409- if err != nil || !did {
410- return changed, err
411- }
412- changed = true
413- }
414- return changed, nil
415-}
416-
417-func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
418- if *budget == 0 {
419- return false, nil
420- }
421- info, err := os.Lstat(path)
422- if errors.Is(err, os.ErrNotExist) {
423- return false, nil
424- }
425- if err != nil {
426- return false, err
427- }
428- if !info.IsDir() {
429- if err := ctx.Err(); err != nil {
430- return false, err
431- }
432- if err := r.remove(path); err != nil {
433- return false, err
434- }
435- *budget--
436- return true, nil
437- }
438- // A quarantine tree contains no live entries. Reopening from its start is
439- // safe here because each successful step removes that first descendant.
440- entries, err := readDirBatch(path, 1)
441- if err != nil {
442- return false, err
443- }
444- if len(entries) == 0 {
445- if err := ctx.Err(); err != nil {
446- return false, err
447- }
448- if err := r.remove(path); err != nil {
449- return false, err
450- }
451- *budget--
452- return true, nil
453- }
454- return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
455-}
456-
457-func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {
458- if err := ctx.Err(); err != nil {
459- return err
460- }
461- state, err := r.sessionMaintenanceState(key)
462- if err != nil {
463- return err
464- }
465- if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil || *budget == 0 {
466- return err
467- }
468-
469- objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
470- shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
471- if err != nil && !errors.Is(err, os.ErrNotExist) {
472- return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
473- }
474- for i, shard := range shards {
475- if *budget == 0 {
476- r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
477- return nil
478- }
479- if shard.isDir {
480- if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
481- return err
482- }
483- }
484- }
485- if *budget == 0 {
486- return nil
487- }
488- if !state.markDone {
489- if err := r.markSession(ctx, key, state); err != nil {
490- return err
491- }
492- if !state.markDone {
493- return nil
494- }
495- }
496- if state.epoch != r.storageEpoch(key) {
497- // A publication may have left an unpublished manifest after this mark.
498- // Discard all mark and sweep cursors; the next pass starts from it.
499- r.clearSessionMaintenance(key)
500- return nil
501- }
502- if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
503- return err
504- }
505- if state.uncertain {
506- // An unreadable or invalid manifest may reference any existing blob. Do
507- // not sweep in this cycle; retry marking from a fresh directory pass.
508- r.clearSessionMaintenance(key)
509- return nil
510- }
511- if err := r.sweepSession(ctx, key, state, budget); err != nil {
512- return err
513- }
514- return nil
515-}
516-
517-func (r *Repository) sessionMaintenanceState(key string) (*sessionMaintenance, error) {
518- token, conservative, err := r.maintenanceToken(key)
519- if err != nil {
520- return nil, err
521- }
522- state := r.maintenanceSessions[key]
523- if state == nil || state.token != token || state.epoch != r.storageEpoch(key) {
524- r.clearSessionMaintenance(key)
525- state = &sessionMaintenance{
526- token: token,
527- epoch: r.storageEpoch(key),
528- conservative: conservative,
529- marked: make(map[uint64]manifestMaintenance),
530- }
531- r.maintenanceSessions[key] = state
532- }
533- return state, nil
534-}
535-
536-// maintenanceToken is the publication boundary. A missing or corrupt HEAD is
537-// itself stable maintenance state: we retain every classified reference until a
538-// valid publication changes that state, rather than restarting the mark pass.
539-func (r *Repository) maintenanceToken(key string) (string, bool, error) {
540- data, exists, err := readOptionalBounded(r.headPath(key))
541- if err != nil {
542- return "", false, err
543- }
544- if !exists {
545- return "missing", true, nil
546- }
547- sum := sha256.Sum256(data)
548- if _, _, err := r.readHead(key); err != nil {
549- return fmt.Sprintf("invalid:%x", sum), true, nil
550- }
551- return fmt.Sprintf("valid:%x", sum), false, nil
552-}
553-
554-func (r *Repository) clearSessionMaintenance(key string) {
555- delete(r.maintenanceSessions, key)
556- prefix := "\x00" + filepath.Clean(r.sessionPath(key))
557- for id := range r.maintenanceCursors {
558- if strings.HasSuffix(id, prefix) || strings.Contains(id, ":"+key+":") || strings.Contains(id, ":"+key+"\x00") {
559- delete(r.maintenanceCursors, id)
560- }
561- }
562-}
563-
564-func (r *Repository) markSession(ctx context.Context, key string, state *sessionMaintenance) error {
565- dir := filepath.Join(r.sessionPath(key), repositoryGenerations)
566- entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "mark-generations:"+key)
567- if errors.Is(err, os.ErrNotExist) {
568- state.markDone = true
569- return nil
570- }
571- if err != nil {
572- return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
573- }
574- for _, entry := range entries {
575- if err := ctx.Err(); err != nil {
576- return err
577- }
578- if entry.isDir {
579- continue
580- }
581- generation, ok := parseGenerationFilename(entry.name)
582- if !ok {
583- continue
584- }
585- item := manifestMaintenance{}
586- data, err := readBounded(r.manifestPath(key, generation))
587- if err != nil {
588- state.uncertain = true
589- state.marked[generation] = item
590- continue
591- }
592- manifest, err := codec.UnmarshalManifest(data)
593- if err != nil || manifest.Generation != generation || sessionKey(manifest.Name) != key {
594- state.uncertain = true
595- state.marked[generation] = item
596- continue
597- }
598- item.refs = manifestRefs(manifest)
599- if item.refs == nil || !withinGenerationBudget(len(data), item.refs) {
600- state.uncertain = true
601- state.marked[generation] = item
602- continue
603- }
604- // Valid but incomplete generations are marked too. This protects blobs
605- // written before a failed or in-flight publication reaches HEAD.
606- _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
607- item.complete = loadErr == nil
608- state.marked[generation] = item
609- }
610- if done {
611- state.markDone = true
612- if !state.conservative {
613- state.manifestQueue = retainedManifestQueue(state.marked)
614- }
615- }
616- return nil
617-}
618-
619-func retainedManifestQueue(marked map[uint64]manifestMaintenance) []uint64 {
620- complete := make([]uint64, 0, len(marked))
621- for generation, item := range marked {
622- if item.complete {
623- complete = append(complete, generation)
624- }
625- }
626- for i := range complete {
627- for j := i + 1; j < len(complete); j++ {
628- if complete[j] > complete[i] {
629- complete[i], complete[j] = complete[j], complete[i]
630- }
631- }
632- }
633- if len(complete) <= 2 {
634- return nil
635- }
636- return complete[2:]
637-}
638-
639-func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
640- for len(state.manifestQueue) != 0 && *budget > 0 {
641- if err := ctx.Err(); err != nil {
642- return err
643- }
644- generation := state.manifestQueue[0]
645- path := r.manifestPath(key, generation)
646- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
647- return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
648- }
649- if err := ctx.Err(); err != nil {
650- return err
651- }
652- if err := r.syncDirectory(filepath.Dir(path)); err != nil {
653- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
654- }
655- state.manifestQueue = state.manifestQueue[1:]
656- *budget--
657- }
658- return nil
659-}
660-
661-func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
662- // Every bounded sweep batch is tied to the mark's storage epoch. Do not use
663- // stale references after any publication or replacement.
664- if state.epoch != r.storageEpoch(key) {
665- r.clearSessionMaintenance(key)
666- return nil
667- }
668- referenced := retainedReferences(state.marked, state.conservative)
669- root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
670- // Do not keep accumulating shard names while draining earlier work. This
671- // bounds retained names to one directory batch even for a session with many
672- // object shards.
673- if !state.sweepRootDone && len(state.sweepQueue) == 0 {
674- entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
675- if errors.Is(err, os.ErrNotExist) {
676- state.sweepRootDone = true
677- } else if err != nil {
678- return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
679- } else {
680- for _, entry := range entries {
681- if entry.isDir {
682- state.sweepQueue = append(state.sweepQueue, entry.name)
683- }
684- }
685- state.sweepRootDone = done
686- }
687- }
688- if len(state.sweepQueue) != 0 && *budget > 0 {
689- shard := state.sweepQueue[0]
690- done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
691- if err != nil {
692- return err
693- }
694- if done {
695- state.sweepQueue = state.sweepQueue[1:]
696- }
697- }
698- if state.sweepRootDone && len(state.sweepQueue) == 0 {
699- r.clearSessionMaintenance(key)
700- }
701- return nil
702-}
703-
704-func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
705- if conservative {
706- references := make(map[ports.SnapshotDigest]struct{})
707- for _, item := range marked {
708- for digest := range item.refs {
709- references[digest] = struct{}{}
710- }
711- }
712- return references
713- }
714- complete := make([]uint64, 0, len(marked))
715- for generation, item := range marked {
716- if item.complete {
717- complete = append(complete, generation)
718- }
719- }
720- for i := range complete {
721- for j := i + 1; j < len(complete); j++ {
722- if complete[j] > complete[i] {
723- complete[i], complete[j] = complete[j], complete[i]
724- }
725- }
726- }
727- keep := make(map[uint64]bool, 2)
728- for _, generation := range complete {
729- if len(keep) == 2 {
730- break
731- }
732- keep[generation] = true
733- }
734- references := make(map[ports.SnapshotDigest]struct{})
735- for generation, item := range marked {
736- if !item.complete || keep[generation] {
737- for digest := range item.refs {
738- references[digest] = struct{}{}
739- }
740- }
741- }
742- return references
743-}
744-
745-func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[ports.SnapshotDigest]struct{}, budget *int, key, shard string) (bool, error) {
746- entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "sweep-objects:"+key+":"+shard)
747- if errors.Is(err, os.ErrNotExist) {
748- return true, nil
749- }
750- if err != nil {
751- return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
752- }
753- for i, object := range entries {
754- if *budget == 0 {
755- r.requeueMaintenanceEntries(dir, "sweep-objects:"+key+":"+shard, entries[i:])
756- return false, nil
757- }
758- if err := ctx.Err(); err != nil {
759- return false, err
760- }
761- digest, ok := parseObjectDigest(object.name)
762- if !ok || object.isDir {
763- continue
764- }
765- if _, used := referenced[digest]; used {
766- continue
767- }
768- path := filepath.Join(dir, object.name)
769- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
770- return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
771- }
772- if err := ctx.Err(); err != nil {
773- return false, err
774- }
775- if err := r.syncDirectory(dir); err != nil {
776- return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
777- }
778- *budget--
779- }
780- return done, nil
781-}
782-
783-func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
784- if *budget == 0 {
785- return nil
786- }
787- entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
788- if errors.Is(err, os.ErrNotExist) {
789- return nil
790- }
791- if err != nil {
792- return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
793- }
794- for i, entry := range entries {
795- if *budget == 0 {
796- r.requeueMaintenanceEntries(dir, purpose, entries[i:])
797- return nil
798- }
799- if err := ctx.Err(); err != nil {
800- return err
801- }
802- if entry.isDir || !strings.HasPrefix(entry.name, ".tmp-") {
803- continue
804- }
805- path := filepath.Join(dir, entry.name)
806- if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
807- return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
808- }
809- if err := ctx.Err(); err != nil {
810- return err
811- }
812- if err := r.syncDirectory(dir); err != nil {
813- return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
814- }
815- *budget--
816- }
817- return nil
818-}
819-
820-// maintenancePath avoids exposing the repository root while retaining a
821-// path-safe, actionable location in errors.
822-func maintenancePath(root, path string) string {
823- rel, err := filepath.Rel(root, path)
824- if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." {
825- return filepath.Base(path)
826- }
827- return rel
828-}
829-
830-type safePathError struct {
831- op string
832- cause error
833-}
834-
835-func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() }
836-func (e safePathError) Unwrap() error { return e.cause }
837-
838-// safeFilesystemError preserves error matching without including a filesystem
839-// path from terminal-facing error text.
840-func safeFilesystemError(err error) error {
841- var pathError *os.PathError
842- if errors.As(err, &pathError) {
843- return safePathError{op: pathError.Op, cause: pathError.Err}
844- }
845- return err
846-}
A internal/adapters/snapshot/maintenance_cursor.go +222 -0
@@ -0,0 +1,222 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+ "syscall"
12+ "unsafe"
13+)
14+
15+type maintenanceCursor struct {
16+ offset int64
17+ pending []maintenanceDirEntry
18+ done bool
19+}
20+
21+type maintenanceDirEntry struct {
22+ name string
23+ isDir bool
24+}
25+
26+func (r *Repository) resetMaintenance() {
27+ r.maintenanceCursors = nil
28+ r.maintenanceSessions = nil
29+}
30+
31+func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
32+
33+// readMaintenanceDir advances one named directory cursor. Its continuation is
34+// only a Linux seek cookie and at most one call's worth of pending names. The
35+// descriptor used to read a call is always closed before this method returns.
36+// Callers must return entries they could not process with
37+// requeueMaintenanceEntries before yielding for budget exhaustion.
38+func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
39+ id := maintenanceCursorID(dir, purpose)
40+ cursor := r.maintenanceCursors[id]
41+ if cursor != nil && len(cursor.pending) != 0 {
42+ entries := cursor.pending
43+ cursor.pending = nil
44+ if cursor.done {
45+ delete(r.maintenanceCursors, id)
46+ }
47+ return entries, cursor.done, nil
48+ }
49+ if cursor == nil {
50+ cursor = &maintenanceCursor{}
51+ r.maintenanceCursors[id] = cursor
52+ }
53+
54+ entries, done, err := r.readMaintenanceDirent(dir, n, cursor)
55+ if err != nil {
56+ delete(r.maintenanceCursors, id)
57+ return nil, false, err
58+ }
59+ if done {
60+ delete(r.maintenanceCursors, id)
61+ }
62+ return entries, done, nil
63+}
64+
65+// maintenanceDirectory is the small syscall surface needed to resume a
66+// bounded getdents cursor. It permits deterministic fault injection without
67+// retaining directory descriptors between maintenance passes.
68+type maintenanceDirectory interface {
69+ Seek(int64, int) (int64, error)
70+ ReadDirent([]byte) (int, error)
71+ Close() error
72+}
73+
74+type osMaintenanceDirectory struct{ file *os.File }
75+
76+func (d osMaintenanceDirectory) Seek(offset int64, whence int) (int64, error) {
77+ return syscall.Seek(int(d.file.Fd()), offset, whence)
78+}
79+
80+func (d osMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
81+ return syscall.ReadDirent(int(d.file.Fd()), buffer)
82+}
83+
84+func (d osMaintenanceDirectory) Close() error { return d.file.Close() }
85+
86+func (r *Repository) openMaintenanceDirectory(dir string) (maintenanceDirectory, error) {
87+ if open := r.hooks.openMaintenanceDirectory; open != nil {
88+ return open(dir)
89+ }
90+ file, err := os.Open(dir)
91+ if err != nil {
92+ return nil, err
93+ }
94+ return osMaintenanceDirectory{file: file}, nil
95+}
96+
97+// maintenanceDirectoryError adds stable operation context and removes any
98+// filesystem path while retaining the underlying cause for errors.Is.
99+func maintenanceDirectoryError(operation string, err error) error {
100+ return fmt.Errorf("%s: %w", operation, safeFilesystemError(err))
101+}
102+
103+// readMaintenanceDirent uses the Linux getdents seek cookie from each record.
104+// syscall.Dirent's Off field is the d_off member of linux_dirent64 and
105+// syscall.Seek is lseek(2), as defined by Go's syscall Linux sources.
106+func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
107+ file, err := r.openMaintenanceDirectory(dir)
108+ if err != nil {
109+ return nil, false, maintenanceDirectoryError("open maintenance directory", err)
110+ }
111+ defer func() {
112+ if closeErr := file.Close(); closeErr != nil {
113+ closeErr = maintenanceDirectoryError("close maintenance directory", closeErr)
114+ if err != nil {
115+ err = errors.Join(err, closeErr)
116+ } else {
117+ entries = nil
118+ done = false
119+ err = closeErr
120+ }
121+ }
122+ }()
123+ if cursor.offset != 0 {
124+ if _, err := file.Seek(cursor.offset, io.SeekStart); err != nil {
125+ return nil, false, maintenanceDirectoryError("seek maintenance directory", err)
126+ }
127+ }
128+
129+ entries = make([]maintenanceDirEntry, 0, limit)
130+ buffer := make([]byte, 8192)
131+ for len(entries) < limit {
132+ count, err := file.ReadDirent(buffer)
133+ if err != nil {
134+ return nil, false, maintenanceDirectoryError("read maintenance directory", err)
135+ }
136+ if count == 0 {
137+ return entries, true, nil
138+ }
139+ for data := buffer[:count]; len(data) != 0 && len(entries) < limit; {
140+ if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) {
141+ return nil, false, syscall.EIO
142+ }
143+ record := (*syscall.Dirent)(unsafe.Pointer(&data[0]))
144+ reclen := int(record.Reclen)
145+ if reclen <= 0 || reclen > len(data) {
146+ return nil, false, syscall.EIO
147+ }
148+ data = data[reclen:]
149+ // Advance past every raw record, including dot and disappeared entries.
150+ cursor.offset = record.Off
151+ nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-int(unsafe.Offsetof(syscall.Dirent{}.Name)))
152+ if end := strings.IndexByte(string(nameBytes), 0); end >= 0 {
153+ nameBytes = nameBytes[:end]
154+ }
155+ name := string(nameBytes)
156+ if name == "." || name == ".." || name == "" {
157+ continue
158+ }
159+ info, err := os.Lstat(filepath.Join(dir, name))
160+ if errors.Is(err, os.ErrNotExist) {
161+ continue
162+ }
163+ if err != nil {
164+ return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)
165+ }
166+ entries = append(entries, maintenanceDirEntry{name: name, isDir: info.IsDir()})
167+ }
168+ }
169+ return entries, false, nil
170+}
171+
172+func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []maintenanceDirEntry) {
173+ if len(entries) == 0 {
174+ return
175+ }
176+ id := maintenanceCursorID(dir, purpose)
177+ cursor := r.maintenanceCursors[id]
178+ if cursor == nil {
179+ cursor = &maintenanceCursor{done: true}
180+ r.maintenanceCursors[id] = cursor
181+ }
182+ cursor.pending = append(entries, cursor.pending...)
183+}
184+
185+func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
186+ if *budget == 0 {
187+ return nil
188+ }
189+ entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
190+ if errors.Is(err, os.ErrNotExist) {
191+ return nil
192+ }
193+ if err != nil {
194+ return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
195+ }
196+ for i, entry := range entries {
197+ if *budget == 0 {
198+ r.requeueMaintenanceEntries(dir, purpose, entries[i:])
199+ return nil
200+ }
201+ if err := ctx.Err(); err != nil {
202+ return err
203+ }
204+ if entry.isDir || !strings.HasPrefix(entry.name, ".tmp-") {
205+ continue
206+ }
207+ path := filepath.Join(dir, entry.name)
208+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
209+ return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
210+ }
211+ if err := ctx.Err(); err != nil {
212+ return err
213+ }
214+ if err := r.syncDirectory(dir); err != nil {
215+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
216+ }
217+ *budget--
218+ }
219+ return nil
220+}
221+
222+// maintenancePath avoids exposing the repository root while retaining a
@@ -0,0 +1,222 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+ "syscall"
12+ "unsafe"
13+)
14+
15+type maintenanceCursor struct {
16+ offset int64
17+ pending []maintenanceDirEntry
18+ done bool
19+}
20+
21+type maintenanceDirEntry struct {
22+ name string
23+ isDir bool
24+}
25+
26+func (r *Repository) resetMaintenance() {
27+ r.maintenanceCursors = nil
28+ r.maintenanceSessions = nil
29+}
30+
31+func maintenanceCursorID(dir, purpose string) string { return purpose + "\x00" + dir }
32+
33+// readMaintenanceDir advances one named directory cursor. Its continuation is
34+// only a Linux seek cookie and at most one call's worth of pending names. The
35+// descriptor used to read a call is always closed before this method returns.
36+// Callers must return entries they could not process with
37+// requeueMaintenanceEntries before yielding for budget exhaustion.
38+func (r *Repository) readMaintenanceDir(dir string, n int, purpose string) ([]maintenanceDirEntry, bool, error) {
39+ id := maintenanceCursorID(dir, purpose)
40+ cursor := r.maintenanceCursors[id]
41+ if cursor != nil && len(cursor.pending) != 0 {
42+ entries := cursor.pending
43+ cursor.pending = nil
44+ if cursor.done {
45+ delete(r.maintenanceCursors, id)
46+ }
47+ return entries, cursor.done, nil
48+ }
49+ if cursor == nil {
50+ cursor = &maintenanceCursor{}
51+ r.maintenanceCursors[id] = cursor
52+ }
53+
54+ entries, done, err := r.readMaintenanceDirent(dir, n, cursor)
55+ if err != nil {
56+ delete(r.maintenanceCursors, id)
57+ return nil, false, err
58+ }
59+ if done {
60+ delete(r.maintenanceCursors, id)
61+ }
62+ return entries, done, nil
63+}
64+
65+// maintenanceDirectory is the small syscall surface needed to resume a
66+// bounded getdents cursor. It permits deterministic fault injection without
67+// retaining directory descriptors between maintenance passes.
68+type maintenanceDirectory interface {
69+ Seek(int64, int) (int64, error)
70+ ReadDirent([]byte) (int, error)
71+ Close() error
72+}
73+
74+type osMaintenanceDirectory struct{ file *os.File }
75+
76+func (d osMaintenanceDirectory) Seek(offset int64, whence int) (int64, error) {
77+ return syscall.Seek(int(d.file.Fd()), offset, whence)
78+}
79+
80+func (d osMaintenanceDirectory) ReadDirent(buffer []byte) (int, error) {
81+ return syscall.ReadDirent(int(d.file.Fd()), buffer)
82+}
83+
84+func (d osMaintenanceDirectory) Close() error { return d.file.Close() }
85+
86+func (r *Repository) openMaintenanceDirectory(dir string) (maintenanceDirectory, error) {
87+ if open := r.hooks.openMaintenanceDirectory; open != nil {
88+ return open(dir)
89+ }
90+ file, err := os.Open(dir)
91+ if err != nil {
92+ return nil, err
93+ }
94+ return osMaintenanceDirectory{file: file}, nil
95+}
96+
97+// maintenanceDirectoryError adds stable operation context and removes any
98+// filesystem path while retaining the underlying cause for errors.Is.
99+func maintenanceDirectoryError(operation string, err error) error {
100+ return fmt.Errorf("%s: %w", operation, safeFilesystemError(err))
101+}
102+
103+// readMaintenanceDirent uses the Linux getdents seek cookie from each record.
104+// syscall.Dirent's Off field is the d_off member of linux_dirent64 and
105+// syscall.Seek is lseek(2), as defined by Go's syscall Linux sources.
106+func (r *Repository) readMaintenanceDirent(dir string, limit int, cursor *maintenanceCursor) (entries []maintenanceDirEntry, done bool, err error) {
107+ file, err := r.openMaintenanceDirectory(dir)
108+ if err != nil {
109+ return nil, false, maintenanceDirectoryError("open maintenance directory", err)
110+ }
111+ defer func() {
112+ if closeErr := file.Close(); closeErr != nil {
113+ closeErr = maintenanceDirectoryError("close maintenance directory", closeErr)
114+ if err != nil {
115+ err = errors.Join(err, closeErr)
116+ } else {
117+ entries = nil
118+ done = false
119+ err = closeErr
120+ }
121+ }
122+ }()
123+ if cursor.offset != 0 {
124+ if _, err := file.Seek(cursor.offset, io.SeekStart); err != nil {
125+ return nil, false, maintenanceDirectoryError("seek maintenance directory", err)
126+ }
127+ }
128+
129+ entries = make([]maintenanceDirEntry, 0, limit)
130+ buffer := make([]byte, 8192)
131+ for len(entries) < limit {
132+ count, err := file.ReadDirent(buffer)
133+ if err != nil {
134+ return nil, false, maintenanceDirectoryError("read maintenance directory", err)
135+ }
136+ if count == 0 {
137+ return entries, true, nil
138+ }
139+ for data := buffer[:count]; len(data) != 0 && len(entries) < limit; {
140+ if len(data) < int(unsafe.Offsetof(syscall.Dirent{}.Name)) {
141+ return nil, false, syscall.EIO
142+ }
143+ record := (*syscall.Dirent)(unsafe.Pointer(&data[0]))
144+ reclen := int(record.Reclen)
145+ if reclen <= 0 || reclen > len(data) {
146+ return nil, false, syscall.EIO
147+ }
148+ data = data[reclen:]
149+ // Advance past every raw record, including dot and disappeared entries.
150+ cursor.offset = record.Off
151+ nameBytes := unsafe.Slice((*byte)(unsafe.Pointer(&record.Name[0])), reclen-int(unsafe.Offsetof(syscall.Dirent{}.Name)))
152+ if end := strings.IndexByte(string(nameBytes), 0); end >= 0 {
153+ nameBytes = nameBytes[:end]
154+ }
155+ name := string(nameBytes)
156+ if name == "." || name == ".." || name == "" {
157+ continue
158+ }
159+ info, err := os.Lstat(filepath.Join(dir, name))
160+ if errors.Is(err, os.ErrNotExist) {
161+ continue
162+ }
163+ if err != nil {
164+ return nil, false, maintenanceDirectoryError("stat maintenance directory entry", err)
165+ }
166+ entries = append(entries, maintenanceDirEntry{name: name, isDir: info.IsDir()})
167+ }
168+ }
169+ return entries, false, nil
170+}
171+
172+func (r *Repository) requeueMaintenanceEntries(dir, purpose string, entries []maintenanceDirEntry) {
173+ if len(entries) == 0 {
174+ return
175+ }
176+ id := maintenanceCursorID(dir, purpose)
177+ cursor := r.maintenanceCursors[id]
178+ if cursor == nil {
179+ cursor = &maintenanceCursor{done: true}
180+ r.maintenanceCursors[id] = cursor
181+ }
182+ cursor.pending = append(entries, cursor.pending...)
183+}
184+
185+func (r *Repository) removeTemps(ctx context.Context, dir string, budget *int, purpose string) error {
186+ if *budget == 0 {
187+ return nil
188+ }
189+ entries, _, err := r.readMaintenanceDir(dir, maintenanceBatch, purpose)
190+ if errors.Is(err, os.ErrNotExist) {
191+ return nil
192+ }
193+ if err != nil {
194+ return fmt.Errorf("read snapshot maintenance directory: %w", safeFilesystemError(err))
195+ }
196+ for i, entry := range entries {
197+ if *budget == 0 {
198+ r.requeueMaintenanceEntries(dir, purpose, entries[i:])
199+ return nil
200+ }
201+ if err := ctx.Err(); err != nil {
202+ return err
203+ }
204+ if entry.isDir || !strings.HasPrefix(entry.name, ".tmp-") {
205+ continue
206+ }
207+ path := filepath.Join(dir, entry.name)
208+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
209+ return fmt.Errorf("remove snapshot maintenance file %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
210+ }
211+ if err := ctx.Err(); err != nil {
212+ return err
213+ }
214+ if err := r.syncDirectory(dir); err != nil {
215+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
216+ }
217+ *budget--
218+ }
219+ return nil
220+}
221+
222+// maintenancePath avoids exposing the repository root while retaining a
A internal/adapters/snapshot/maintenance_delete.go +184 -0
@@ -0,0 +1,184 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+ "time"
12+)
13+
14+// Delete makes a session unavailable by durably moving it out of the canonical
15+// namespace. Maintain reaps the private quarantine later; Delete never restores
16+// a quarantined directory.
17+func (r *Repository) Delete(ctx context.Context, name string) error {
18+ if err := ctx.Err(); err != nil {
19+ return err
20+ }
21+ key := sessionKey(name)
22+ lock := r.sessionLock(key)
23+ lock.Lock()
24+ defer lock.Unlock()
25+ if err := ctx.Err(); err != nil {
26+ return err
27+ }
28+
29+ canonical := r.sessionPath(key)
30+ sessions := filepath.Dir(canonical)
31+ // A prior rename may have succeeded while its parent sync failed. Complete
32+ // that durability boundary before considering a canonical directory: this
33+ // also leaves a newly recreated session untouched.
34+ pending, err := pendingQuarantine(sessions, key)
35+ if err != nil {
36+ return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
37+ }
38+ if pending {
39+ if err := ctx.Err(); err != nil {
40+ return err
41+ }
42+ if err := r.syncDirectory(sessions); err != nil {
43+ return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
44+ }
45+ return nil
46+ }
47+ if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {
48+ return nil
49+ } else if err != nil {
50+ return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
51+ }
52+ quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
53+ for attempt := 0; ; attempt++ {
54+ if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
55+ break
56+ } else if err != nil {
57+ return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
58+ }
59+ quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
60+ }
61+ if err := ctx.Err(); err != nil {
62+ return err
63+ }
64+ // A renamed session can be replaced by a later publication. Invalidate any
65+ // mark made for the old namespace before that replacement becomes possible.
66+ r.invalidateStorageEpoch(key)
67+ if err := r.rename(canonical, quarantine); err != nil {
68+ return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
69+ }
70+ if err := ctx.Err(); err != nil {
71+ return err
72+ }
73+ if err := r.syncDirectory(sessions); err != nil {
74+ return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
75+ }
76+ return nil
77+}
78+
79+func pendingQuarantine(dir, key string) (pending bool, err error) {
80+ f, err := os.Open(dir)
81+ if errors.Is(err, os.ErrNotExist) {
82+ return false, nil
83+ }
84+ if err != nil {
85+ return false, err
86+ }
87+ defer func() {
88+ if closeErr := f.Close(); closeErr != nil {
89+ err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
90+ }
91+ }()
92+ prefix := ".deleting-" + key + "-"
93+ for {
94+ entries, err := f.ReadDir(maintenanceBatch)
95+ for _, entry := range entries {
96+ if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
97+ return true, nil
98+ }
99+ }
100+ if errors.Is(err, io.EOF) {
101+ return false, nil
102+ }
103+ if err != nil {
104+ return false, err
105+ }
106+ }
107+}
108+
109+// Maintain reaps a bounded amount of stale state. Continuation handles are
110+// repository-owned, so a stable prefix cannot starve entries later in a large
111+// directory. Cancellation and errors discard those handles: a restarted
112+// repository (and the next fresh cycle) always begins at a safe boundary.
113+func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
114+
115+// readDirBatch is used only for a mutating quarantine walk, where every
116+// successful step removes the entry it selected. General maintenance scans use
117+// repository-owned cursors above.
118+func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
119+ f, err := os.Open(dir)
120+ if err != nil {
121+ return nil, err
122+ }
123+ entries, readErr := f.ReadDir(n)
124+ closeErr := f.Close()
125+ if readErr != nil && !errors.Is(readErr, io.EOF) {
126+ return nil, readErr
127+ }
128+ if closeErr != nil {
129+ return nil, closeErr
130+ }
131+ return entries, nil
132+}
133+
134+func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
135+ changed := false
136+ for *budget > 0 {
137+ did, err := r.removeTreeStep(ctx, path, budget)
138+ if err != nil || !did {
139+ return changed, err
140+ }
141+ changed = true
142+ }
143+ return changed, nil
144+}
145+
146+func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
147+ if *budget == 0 {
148+ return false, nil
149+ }
150+ info, err := os.Lstat(path)
151+ if errors.Is(err, os.ErrNotExist) {
152+ return false, nil
153+ }
154+ if err != nil {
155+ return false, err
156+ }
157+ if !info.IsDir() {
158+ if err := ctx.Err(); err != nil {
159+ return false, err
160+ }
161+ if err := r.remove(path); err != nil {
162+ return false, err
163+ }
164+ *budget--
165+ return true, nil
166+ }
167+ // A quarantine tree contains no live entries. Reopening from its start is
168+ // safe here because each successful step removes that first descendant.
169+ entries, err := readDirBatch(path, 1)
170+ if err != nil {
171+ return false, err
172+ }
173+ if len(entries) == 0 {
174+ if err := ctx.Err(); err != nil {
175+ return false, err
176+ }
177+ if err := r.remove(path); err != nil {
178+ return false, err
179+ }
180+ *budget--
181+ return true, nil
182+ }
183+ return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
184+}
@@ -0,0 +1,184 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+ "time"
12+)
13+
14+// Delete makes a session unavailable by durably moving it out of the canonical
15+// namespace. Maintain reaps the private quarantine later; Delete never restores
16+// a quarantined directory.
17+func (r *Repository) Delete(ctx context.Context, name string) error {
18+ if err := ctx.Err(); err != nil {
19+ return err
20+ }
21+ key := sessionKey(name)
22+ lock := r.sessionLock(key)
23+ lock.Lock()
24+ defer lock.Unlock()
25+ if err := ctx.Err(); err != nil {
26+ return err
27+ }
28+
29+ canonical := r.sessionPath(key)
30+ sessions := filepath.Dir(canonical)
31+ // A prior rename may have succeeded while its parent sync failed. Complete
32+ // that durability boundary before considering a canonical directory: this
33+ // also leaves a newly recreated session untouched.
34+ pending, err := pendingQuarantine(sessions, key)
35+ if err != nil {
36+ return fmt.Errorf("read deleting snapshot session %q: %w", key, safeFilesystemError(err))
37+ }
38+ if pending {
39+ if err := ctx.Err(); err != nil {
40+ return err
41+ }
42+ if err := r.syncDirectory(sessions); err != nil {
43+ return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
44+ }
45+ return nil
46+ }
47+ if _, err := os.Lstat(canonical); errors.Is(err, os.ErrNotExist) {
48+ return nil
49+ } else if err != nil {
50+ return fmt.Errorf("stat snapshot session %q: %w", key, safeFilesystemError(err))
51+ }
52+ quarantine := filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d", time.Now().UnixNano()))
53+ for attempt := 0; ; attempt++ {
54+ if _, err := os.Lstat(quarantine); errors.Is(err, os.ErrNotExist) {
55+ break
56+ } else if err != nil {
57+ return fmt.Errorf("stat deleting snapshot session %q: %w", key, safeFilesystemError(err))
58+ }
59+ quarantine = filepath.Join(sessions, ".deleting-"+key+"-"+fmt.Sprintf("%d-%d", time.Now().UnixNano(), attempt))
60+ }
61+ if err := ctx.Err(); err != nil {
62+ return err
63+ }
64+ // A renamed session can be replaced by a later publication. Invalidate any
65+ // mark made for the old namespace before that replacement becomes possible.
66+ r.invalidateStorageEpoch(key)
67+ if err := r.rename(canonical, quarantine); err != nil {
68+ return fmt.Errorf("quarantine snapshot session %q: %w", key, safeFilesystemError(err))
69+ }
70+ if err := ctx.Err(); err != nil {
71+ return err
72+ }
73+ if err := r.syncDirectory(sessions); err != nil {
74+ return fmt.Errorf("sync deleted snapshot session directory %q: %w", key, safeFilesystemError(err))
75+ }
76+ return nil
77+}
78+
79+func pendingQuarantine(dir, key string) (pending bool, err error) {
80+ f, err := os.Open(dir)
81+ if errors.Is(err, os.ErrNotExist) {
82+ return false, nil
83+ }
84+ if err != nil {
85+ return false, err
86+ }
87+ defer func() {
88+ if closeErr := f.Close(); closeErr != nil {
89+ err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
90+ }
91+ }()
92+ prefix := ".deleting-" + key + "-"
93+ for {
94+ entries, err := f.ReadDir(maintenanceBatch)
95+ for _, entry := range entries {
96+ if entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) {
97+ return true, nil
98+ }
99+ }
100+ if errors.Is(err, io.EOF) {
101+ return false, nil
102+ }
103+ if err != nil {
104+ return false, err
105+ }
106+ }
107+}
108+
109+// Maintain reaps a bounded amount of stale state. Continuation handles are
110+// repository-owned, so a stable prefix cannot starve entries later in a large
111+// directory. Cancellation and errors discard those handles: a restarted
112+// repository (and the next fresh cycle) always begins at a safe boundary.
113+func isQuarantine(name string) bool { return strings.HasPrefix(name, ".deleting-") }
114+
115+// readDirBatch is used only for a mutating quarantine walk, where every
116+// successful step removes the entry it selected. General maintenance scans use
117+// repository-owned cursors above.
118+func readDirBatch(dir string, n int) ([]os.DirEntry, error) {
119+ f, err := os.Open(dir)
120+ if err != nil {
121+ return nil, err
122+ }
123+ entries, readErr := f.ReadDir(n)
124+ closeErr := f.Close()
125+ if readErr != nil && !errors.Is(readErr, io.EOF) {
126+ return nil, readErr
127+ }
128+ if closeErr != nil {
129+ return nil, closeErr
130+ }
131+ return entries, nil
132+}
133+
134+func (r *Repository) removeTreeBatch(ctx context.Context, path string, budget *int) (bool, error) {
135+ changed := false
136+ for *budget > 0 {
137+ did, err := r.removeTreeStep(ctx, path, budget)
138+ if err != nil || !did {
139+ return changed, err
140+ }
141+ changed = true
142+ }
143+ return changed, nil
144+}
145+
146+func (r *Repository) removeTreeStep(ctx context.Context, path string, budget *int) (bool, error) {
147+ if *budget == 0 {
148+ return false, nil
149+ }
150+ info, err := os.Lstat(path)
151+ if errors.Is(err, os.ErrNotExist) {
152+ return false, nil
153+ }
154+ if err != nil {
155+ return false, err
156+ }
157+ if !info.IsDir() {
158+ if err := ctx.Err(); err != nil {
159+ return false, err
160+ }
161+ if err := r.remove(path); err != nil {
162+ return false, err
163+ }
164+ *budget--
165+ return true, nil
166+ }
167+ // A quarantine tree contains no live entries. Reopening from its start is
168+ // safe here because each successful step removes that first descendant.
169+ entries, err := readDirBatch(path, 1)
170+ if err != nil {
171+ return false, err
172+ }
173+ if len(entries) == 0 {
174+ if err := ctx.Err(); err != nil {
175+ return false, err
176+ }
177+ if err := r.remove(path); err != nil {
178+ return false, err
179+ }
180+ *budget--
181+ return true, nil
182+ }
183+ return r.removeTreeStep(ctx, filepath.Join(path, entries[0].Name()), budget)
184+}
A internal/adapters/snapshot/maintenance_error.go +34 -0
@@ -0,0 +1,34 @@
1+package snapshot
2+
3+import (
4+ "errors"
5+ "os"
6+ "path/filepath"
7+ "strings"
8+)
9+
10+func maintenancePath(root, path string) string {
11+ rel, err := filepath.Rel(root, path)
12+ if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." {
13+ return filepath.Base(path)
14+ }
15+ return rel
16+}
17+
18+type safePathError struct {
19+ op string
20+ cause error
21+}
22+
23+func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() }
24+func (e safePathError) Unwrap() error { return e.cause }
25+
26+// safeFilesystemError preserves error matching without including a filesystem
27+// path from terminal-facing error text.
28+func safeFilesystemError(err error) error {
29+ var pathError *os.PathError
30+ if errors.As(err, &pathError) {
31+ return safePathError{op: pathError.Op, cause: pathError.Err}
32+ }
33+ return err
34+}
@@ -0,0 +1,34 @@
1+package snapshot
2+
3+import (
4+ "errors"
5+ "os"
6+ "path/filepath"
7+ "strings"
8+)
9+
10+func maintenancePath(root, path string) string {
11+ rel, err := filepath.Rel(root, path)
12+ if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || rel == ".." {
13+ return filepath.Base(path)
14+ }
15+ return rel
16+}
17+
18+type safePathError struct {
19+ op string
20+ cause error
21+}
22+
23+func (e safePathError) Error() string { return e.op + ": " + e.cause.Error() }
24+func (e safePathError) Unwrap() error { return e.cause }
25+
26+// safeFilesystemError preserves error matching without including a filesystem
27+// path from terminal-facing error text.
28+func safeFilesystemError(err error) error {
29+ var pathError *os.PathError
30+ if errors.As(err, &pathError) {
31+ return safePathError{op: pathError.Op, cause: pathError.Err}
32+ }
33+ return err
34+}
A internal/adapters/snapshot/maintenance_gc.go +398 -0
@@ -0,0 +1,398 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+
12+ "github.com/bnema/vev/internal/ports"
13+ codec "github.com/bnema/vev/internal/usecase/snapshot"
14+)
15+
16+type manifestMaintenance struct {
17+ refs map[ports.SnapshotDigest]codec.ObjectRef
18+ complete bool
19+}
20+
21+type sessionMaintenance struct {
22+ token string
23+ epoch uint64
24+ marked map[uint64]manifestMaintenance
25+ referenceCount int
26+ uncertain bool
27+ conservative bool
28+ markDone bool
29+ manifestQueue []uint64
30+ sweepQueue []string
31+ sweepRootDone bool
32+}
33+
34+// canRetainManifest reserves bounded state before retaining a manifest's
35+// reference map. The caller must leave the map unretained when it returns
36+// false, and make the cycle conservative instead.
37+func (s *sessionMaintenance) canRetainManifest(referenceCount int) bool {
38+ if referenceCount < 0 || len(s.marked) >= maxMaintenanceMarkedGenerations {
39+ return false
40+ }
41+ if referenceCount > maxMaintenanceReferences-s.referenceCount {
42+ return false
43+ }
44+ s.referenceCount += referenceCount
45+ return true
46+}
47+
48+func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {
49+ if err := ctx.Err(); err != nil {
50+ return err
51+ }
52+ state, err := r.sessionMaintenanceState(key)
53+ if err != nil {
54+ return err
55+ }
56+ if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil || *budget == 0 {
57+ return err
58+ }
59+
60+ objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
61+ shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
62+ if err != nil && !errors.Is(err, os.ErrNotExist) {
63+ return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
64+ }
65+ for i, shard := range shards {
66+ if *budget == 0 {
67+ r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
68+ return nil
69+ }
70+ if shard.isDir {
71+ if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
72+ return err
73+ }
74+ }
75+ }
76+ if *budget == 0 {
77+ return nil
78+ }
79+ if !state.markDone {
80+ if err := r.markSession(ctx, key, state); err != nil {
81+ return err
82+ }
83+ if !state.markDone {
84+ return nil
85+ }
86+ }
87+ if state.epoch != r.storageEpoch(key) {
88+ // A publication may have left an unpublished manifest after this mark.
89+ // Discard all mark and sweep cursors; the next pass starts from it.
90+ r.clearSessionMaintenance(key)
91+ return nil
92+ }
93+ if state.uncertain {
94+ // An unreadable, invalid, or unretained manifest may reference any
95+ // existing blob. Do not collect manifests or sweep in this cycle; retry
96+ // marking from a fresh directory pass.
97+ r.clearSessionMaintenance(key)
98+ return nil
99+ }
100+ if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
101+ return err
102+ }
103+ if err := r.sweepSession(ctx, key, state, budget); err != nil {
104+ return err
105+ }
106+ return nil
107+}
108+
109+func (r *Repository) sessionMaintenanceState(key string) (*sessionMaintenance, error) {
110+ token, conservative, err := r.maintenanceToken(key)
111+ if err != nil {
112+ return nil, err
113+ }
114+ state := r.maintenanceSessions[key]
115+ if state == nil || state.token != token || state.epoch != r.storageEpoch(key) {
116+ r.clearSessionMaintenance(key)
117+ state = &sessionMaintenance{
118+ token: token,
119+ epoch: r.storageEpoch(key),
120+ conservative: conservative,
121+ marked: make(map[uint64]manifestMaintenance),
122+ }
123+ r.maintenanceSessions[key] = state
124+ }
125+ return state, nil
126+}
127+
128+// maintenanceToken is the publication boundary. A missing or corrupt HEAD is
129+// itself stable maintenance state: we retain every classified reference until a
130+// valid publication changes that state, rather than restarting the mark pass.
131+func (r *Repository) maintenanceToken(key string) (string, bool, error) {
132+ data, exists, err := readOptionalBounded(r.headPath(key))
133+ if err != nil {
134+ return "", false, err
135+ }
136+ if !exists {
137+ return "missing", true, nil
138+ }
139+ sum := sha256.Sum256(data)
140+ if _, _, err := r.readHead(key); err != nil {
141+ return fmt.Sprintf("invalid:%x", sum), true, nil
142+ }
143+ return fmt.Sprintf("valid:%x", sum), false, nil
144+}
145+
146+func (r *Repository) clearSessionMaintenance(key string) {
147+ delete(r.maintenanceSessions, key)
148+ prefix := "\x00" + filepath.Clean(r.sessionPath(key))
149+ for id := range r.maintenanceCursors {
150+ if strings.HasSuffix(id, prefix) || strings.Contains(id, ":"+key+":") || strings.Contains(id, ":"+key+"\x00") {
151+ delete(r.maintenanceCursors, id)
152+ }
153+ }
154+}
155+
156+func (r *Repository) markSession(ctx context.Context, key string, state *sessionMaintenance) error {
157+ dir := filepath.Join(r.sessionPath(key), repositoryGenerations)
158+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "mark-generations:"+key)
159+ if errors.Is(err, os.ErrNotExist) {
160+ state.markDone = true
161+ return nil
162+ }
163+ if err != nil {
164+ return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
165+ }
166+ for _, entry := range entries {
167+ if err := ctx.Err(); err != nil {
168+ return err
169+ }
170+ if entry.isDir {
171+ continue
172+ }
173+ generation, ok := parseGenerationFilename(entry.name)
174+ if !ok {
175+ continue
176+ }
177+ data, err := readBounded(r.manifestPath(key, generation))
178+ if err != nil {
179+ state.uncertain = true
180+ continue
181+ }
182+ manifest, err := codec.UnmarshalManifest(data)
183+ if err != nil || manifest.Generation != generation || sessionKey(manifest.Name) != key {
184+ state.uncertain = true
185+ continue
186+ }
187+ refs, validRefs := maintenanceManifestRefs(manifest)
188+ if !validRefs || !withinGenerationBudget(len(data), refs) || !state.canRetainManifest(len(refs)) {
189+ state.uncertain = true
190+ continue
191+ }
192+ // Valid but incomplete generations are marked too. This protects blobs
193+ // written before a failed or in-flight publication reaches HEAD.
194+ _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
195+ state.marked[generation] = manifestMaintenance{refs: refs, complete: loadErr == nil}
196+ }
197+ if done {
198+ state.markDone = true
199+ if !state.conservative {
200+ state.manifestQueue = retainedManifestQueue(state.marked)
201+ }
202+ }
203+ return nil
204+}
205+
206+// maintenanceManifestRefs applies the GC reference ceiling while building its
207+// map, rather than first materializing a hostile manifest's entire reference
208+// set. Conflicting references are invalid just as they are for publication.
209+func maintenanceManifestRefs(manifest codec.Manifest) (map[ports.SnapshotDigest]codec.ObjectRef, bool) {
210+ refs := make(map[ports.SnapshotDigest]codec.ObjectRef)
211+ add := func(ref codec.ObjectRef) bool {
212+ if old, ok := refs[ref.Digest]; ok {
213+ return old == ref
214+ }
215+ if len(refs) == maxMaintenanceReferences {
216+ return false
217+ }
218+ refs[ref.Digest] = ref
219+ return true
220+ }
221+ for _, tab := range manifest.Tabs {
222+ for _, pane := range tab.Panes {
223+ for _, ref := range pane.Sealed {
224+ if !add(ref) {
225+ return nil, false
226+ }
227+ }
228+ if !add(pane.Tail) || !add(pane.Visible) {
229+ return nil, false
230+ }
231+ }
232+ }
233+ return refs, true
234+}
235+
236+func retainedManifestQueue(marked map[uint64]manifestMaintenance) []uint64 {
237+ complete := make([]uint64, 0, len(marked))
238+ for generation, item := range marked {
239+ if item.complete {
240+ complete = append(complete, generation)
241+ }
242+ }
243+ for i := range complete {
244+ for j := i + 1; j < len(complete); j++ {
245+ if complete[j] > complete[i] {
246+ complete[i], complete[j] = complete[j], complete[i]
247+ }
248+ }
249+ }
250+ if len(complete) <= 2 {
251+ return nil
252+ }
253+ return complete[2:]
254+}
255+
256+func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
257+ for len(state.manifestQueue) != 0 && *budget > 0 {
258+ if err := ctx.Err(); err != nil {
259+ return err
260+ }
261+ generation := state.manifestQueue[0]
262+ path := r.manifestPath(key, generation)
263+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
264+ return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
265+ }
266+ if err := ctx.Err(); err != nil {
267+ return err
268+ }
269+ if err := r.syncDirectory(filepath.Dir(path)); err != nil {
270+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
271+ }
272+ state.manifestQueue = state.manifestQueue[1:]
273+ *budget--
274+ }
275+ return nil
276+}
277+
278+func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
279+ // Every bounded sweep batch is tied to the mark's storage epoch. Do not use
280+ // stale references after any publication or replacement.
281+ if state.epoch != r.storageEpoch(key) {
282+ r.clearSessionMaintenance(key)
283+ return nil
284+ }
285+ referenced := retainedReferences(state.marked, state.conservative)
286+ root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
287+ // Do not keep accumulating shard names while draining earlier work. This
288+ // bounds retained names to one directory batch even for a session with many
289+ // object shards.
290+ if !state.sweepRootDone && len(state.sweepQueue) == 0 {
291+ entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
292+ if errors.Is(err, os.ErrNotExist) {
293+ state.sweepRootDone = true
294+ } else if err != nil {
295+ return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
296+ } else {
297+ for _, entry := range entries {
298+ if entry.isDir {
299+ state.sweepQueue = append(state.sweepQueue, entry.name)
300+ }
301+ }
302+ state.sweepRootDone = done
303+ }
304+ }
305+ if len(state.sweepQueue) != 0 && *budget > 0 {
306+ shard := state.sweepQueue[0]
307+ done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
308+ if err != nil {
309+ return err
310+ }
311+ if done {
312+ state.sweepQueue = state.sweepQueue[1:]
313+ }
314+ }
315+ if state.sweepRootDone && len(state.sweepQueue) == 0 {
316+ r.clearSessionMaintenance(key)
317+ }
318+ return nil
319+}
320+
321+func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
322+ if conservative {
323+ references := make(map[ports.SnapshotDigest]struct{})
324+ for _, item := range marked {
325+ for digest := range item.refs {
326+ references[digest] = struct{}{}
327+ }
328+ }
329+ return references
330+ }
331+ complete := make([]uint64, 0, len(marked))
332+ for generation, item := range marked {
333+ if item.complete {
334+ complete = append(complete, generation)
335+ }
336+ }
337+ for i := range complete {
338+ for j := i + 1; j < len(complete); j++ {
339+ if complete[j] > complete[i] {
340+ complete[i], complete[j] = complete[j], complete[i]
341+ }
342+ }
343+ }
344+ keep := make(map[uint64]bool, 2)
345+ for _, generation := range complete {
346+ if len(keep) == 2 {
347+ break
348+ }
349+ keep[generation] = true
350+ }
351+ references := make(map[ports.SnapshotDigest]struct{})
352+ for generation, item := range marked {
353+ if !item.complete || keep[generation] {
354+ for digest := range item.refs {
355+ references[digest] = struct{}{}
356+ }
357+ }
358+ }
359+ return references
360+}
361+
362+func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[ports.SnapshotDigest]struct{}, budget *int, key, shard string) (bool, error) {
363+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "sweep-objects:"+key+":"+shard)
364+ if errors.Is(err, os.ErrNotExist) {
365+ return true, nil
366+ }
367+ if err != nil {
368+ return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
369+ }
370+ for i, object := range entries {
371+ if *budget == 0 {
372+ r.requeueMaintenanceEntries(dir, "sweep-objects:"+key+":"+shard, entries[i:])
373+ return false, nil
374+ }
375+ if err := ctx.Err(); err != nil {
376+ return false, err
377+ }
378+ digest, ok := parseObjectDigest(object.name)
379+ if !ok || object.isDir {
380+ continue
381+ }
382+ if _, used := referenced[digest]; used {
383+ continue
384+ }
385+ path := filepath.Join(dir, object.name)
386+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
387+ return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
388+ }
389+ if err := ctx.Err(); err != nil {
390+ return false, err
391+ }
392+ if err := r.syncDirectory(dir); err != nil {
393+ return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
394+ }
395+ *budget--
396+ }
397+ return done, nil
398+}
@@ -0,0 +1,398 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+
12+ "github.com/bnema/vev/internal/ports"
13+ codec "github.com/bnema/vev/internal/usecase/snapshot"
14+)
15+
16+type manifestMaintenance struct {
17+ refs map[ports.SnapshotDigest]codec.ObjectRef
18+ complete bool
19+}
20+
21+type sessionMaintenance struct {
22+ token string
23+ epoch uint64
24+ marked map[uint64]manifestMaintenance
25+ referenceCount int
26+ uncertain bool
27+ conservative bool
28+ markDone bool
29+ manifestQueue []uint64
30+ sweepQueue []string
31+ sweepRootDone bool
32+}
33+
34+// canRetainManifest reserves bounded state before retaining a manifest's
35+// reference map. The caller must leave the map unretained when it returns
36+// false, and make the cycle conservative instead.
37+func (s *sessionMaintenance) canRetainManifest(referenceCount int) bool {
38+ if referenceCount < 0 || len(s.marked) >= maxMaintenanceMarkedGenerations {
39+ return false
40+ }
41+ if referenceCount > maxMaintenanceReferences-s.referenceCount {
42+ return false
43+ }
44+ s.referenceCount += referenceCount
45+ return true
46+}
47+
48+func (r *Repository) maintainSession(ctx context.Context, key string, budget *int) error {
49+ if err := ctx.Err(); err != nil {
50+ return err
51+ }
52+ state, err := r.sessionMaintenanceState(key)
53+ if err != nil {
54+ return err
55+ }
56+ if err := r.removeTemps(ctx, filepath.Join(r.sessionPath(key), repositoryGenerations), budget, "generation-temps:"+key); err != nil || *budget == 0 {
57+ return err
58+ }
59+
60+ objectRoot := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
61+ shards, _, err := r.readMaintenanceDir(objectRoot, maintenanceBatch, "object-temps-shards:"+key)
62+ if err != nil && !errors.Is(err, os.ErrNotExist) {
63+ return fmt.Errorf("read snapshot object shards: %w", safeFilesystemError(err))
64+ }
65+ for i, shard := range shards {
66+ if *budget == 0 {
67+ r.requeueMaintenanceEntries(objectRoot, "object-temps-shards:"+key, shards[i:])
68+ return nil
69+ }
70+ if shard.isDir {
71+ if err := r.removeTemps(ctx, filepath.Join(objectRoot, shard.name), budget, "object-temps:"+key+":"+shard.name); err != nil {
72+ return err
73+ }
74+ }
75+ }
76+ if *budget == 0 {
77+ return nil
78+ }
79+ if !state.markDone {
80+ if err := r.markSession(ctx, key, state); err != nil {
81+ return err
82+ }
83+ if !state.markDone {
84+ return nil
85+ }
86+ }
87+ if state.epoch != r.storageEpoch(key) {
88+ // A publication may have left an unpublished manifest after this mark.
89+ // Discard all mark and sweep cursors; the next pass starts from it.
90+ r.clearSessionMaintenance(key)
91+ return nil
92+ }
93+ if state.uncertain {
94+ // An unreadable, invalid, or unretained manifest may reference any
95+ // existing blob. Do not collect manifests or sweep in this cycle; retry
96+ // marking from a fresh directory pass.
97+ r.clearSessionMaintenance(key)
98+ return nil
99+ }
100+ if err := r.removeObsoleteManifests(ctx, key, state, budget); err != nil || *budget == 0 {
101+ return err
102+ }
103+ if err := r.sweepSession(ctx, key, state, budget); err != nil {
104+ return err
105+ }
106+ return nil
107+}
108+
109+func (r *Repository) sessionMaintenanceState(key string) (*sessionMaintenance, error) {
110+ token, conservative, err := r.maintenanceToken(key)
111+ if err != nil {
112+ return nil, err
113+ }
114+ state := r.maintenanceSessions[key]
115+ if state == nil || state.token != token || state.epoch != r.storageEpoch(key) {
116+ r.clearSessionMaintenance(key)
117+ state = &sessionMaintenance{
118+ token: token,
119+ epoch: r.storageEpoch(key),
120+ conservative: conservative,
121+ marked: make(map[uint64]manifestMaintenance),
122+ }
123+ r.maintenanceSessions[key] = state
124+ }
125+ return state, nil
126+}
127+
128+// maintenanceToken is the publication boundary. A missing or corrupt HEAD is
129+// itself stable maintenance state: we retain every classified reference until a
130+// valid publication changes that state, rather than restarting the mark pass.
131+func (r *Repository) maintenanceToken(key string) (string, bool, error) {
132+ data, exists, err := readOptionalBounded(r.headPath(key))
133+ if err != nil {
134+ return "", false, err
135+ }
136+ if !exists {
137+ return "missing", true, nil
138+ }
139+ sum := sha256.Sum256(data)
140+ if _, _, err := r.readHead(key); err != nil {
141+ return fmt.Sprintf("invalid:%x", sum), true, nil
142+ }
143+ return fmt.Sprintf("valid:%x", sum), false, nil
144+}
145+
146+func (r *Repository) clearSessionMaintenance(key string) {
147+ delete(r.maintenanceSessions, key)
148+ prefix := "\x00" + filepath.Clean(r.sessionPath(key))
149+ for id := range r.maintenanceCursors {
150+ if strings.HasSuffix(id, prefix) || strings.Contains(id, ":"+key+":") || strings.Contains(id, ":"+key+"\x00") {
151+ delete(r.maintenanceCursors, id)
152+ }
153+ }
154+}
155+
156+func (r *Repository) markSession(ctx context.Context, key string, state *sessionMaintenance) error {
157+ dir := filepath.Join(r.sessionPath(key), repositoryGenerations)
158+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "mark-generations:"+key)
159+ if errors.Is(err, os.ErrNotExist) {
160+ state.markDone = true
161+ return nil
162+ }
163+ if err != nil {
164+ return fmt.Errorf("read snapshot generations: %w", safeFilesystemError(err))
165+ }
166+ for _, entry := range entries {
167+ if err := ctx.Err(); err != nil {
168+ return err
169+ }
170+ if entry.isDir {
171+ continue
172+ }
173+ generation, ok := parseGenerationFilename(entry.name)
174+ if !ok {
175+ continue
176+ }
177+ data, err := readBounded(r.manifestPath(key, generation))
178+ if err != nil {
179+ state.uncertain = true
180+ continue
181+ }
182+ manifest, err := codec.UnmarshalManifest(data)
183+ if err != nil || manifest.Generation != generation || sessionKey(manifest.Name) != key {
184+ state.uncertain = true
185+ continue
186+ }
187+ refs, validRefs := maintenanceManifestRefs(manifest)
188+ if !validRefs || !withinGenerationBudget(len(data), refs) || !state.canRetainManifest(len(refs)) {
189+ state.uncertain = true
190+ continue
191+ }
192+ // Valid but incomplete generations are marked too. This protects blobs
193+ // written before a failed or in-flight publication reaches HEAD.
194+ _, loadErr := r.loadGeneration(ctx, manifest.Name, key, generation)
195+ state.marked[generation] = manifestMaintenance{refs: refs, complete: loadErr == nil}
196+ }
197+ if done {
198+ state.markDone = true
199+ if !state.conservative {
200+ state.manifestQueue = retainedManifestQueue(state.marked)
201+ }
202+ }
203+ return nil
204+}
205+
206+// maintenanceManifestRefs applies the GC reference ceiling while building its
207+// map, rather than first materializing a hostile manifest's entire reference
208+// set. Conflicting references are invalid just as they are for publication.
209+func maintenanceManifestRefs(manifest codec.Manifest) (map[ports.SnapshotDigest]codec.ObjectRef, bool) {
210+ refs := make(map[ports.SnapshotDigest]codec.ObjectRef)
211+ add := func(ref codec.ObjectRef) bool {
212+ if old, ok := refs[ref.Digest]; ok {
213+ return old == ref
214+ }
215+ if len(refs) == maxMaintenanceReferences {
216+ return false
217+ }
218+ refs[ref.Digest] = ref
219+ return true
220+ }
221+ for _, tab := range manifest.Tabs {
222+ for _, pane := range tab.Panes {
223+ for _, ref := range pane.Sealed {
224+ if !add(ref) {
225+ return nil, false
226+ }
227+ }
228+ if !add(pane.Tail) || !add(pane.Visible) {
229+ return nil, false
230+ }
231+ }
232+ }
233+ return refs, true
234+}
235+
236+func retainedManifestQueue(marked map[uint64]manifestMaintenance) []uint64 {
237+ complete := make([]uint64, 0, len(marked))
238+ for generation, item := range marked {
239+ if item.complete {
240+ complete = append(complete, generation)
241+ }
242+ }
243+ for i := range complete {
244+ for j := i + 1; j < len(complete); j++ {
245+ if complete[j] > complete[i] {
246+ complete[i], complete[j] = complete[j], complete[i]
247+ }
248+ }
249+ }
250+ if len(complete) <= 2 {
251+ return nil
252+ }
253+ return complete[2:]
254+}
255+
256+func (r *Repository) removeObsoleteManifests(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
257+ for len(state.manifestQueue) != 0 && *budget > 0 {
258+ if err := ctx.Err(); err != nil {
259+ return err
260+ }
261+ generation := state.manifestQueue[0]
262+ path := r.manifestPath(key, generation)
263+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
264+ return fmt.Errorf("remove snapshot manifest %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
265+ }
266+ if err := ctx.Err(); err != nil {
267+ return err
268+ }
269+ if err := r.syncDirectory(filepath.Dir(path)); err != nil {
270+ return fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
271+ }
272+ state.manifestQueue = state.manifestQueue[1:]
273+ *budget--
274+ }
275+ return nil
276+}
277+
278+func (r *Repository) sweepSession(ctx context.Context, key string, state *sessionMaintenance, budget *int) error {
279+ // Every bounded sweep batch is tied to the mark's storage epoch. Do not use
280+ // stale references after any publication or replacement.
281+ if state.epoch != r.storageEpoch(key) {
282+ r.clearSessionMaintenance(key)
283+ return nil
284+ }
285+ referenced := retainedReferences(state.marked, state.conservative)
286+ root := filepath.Join(r.sessionPath(key), repositoryObjectsDir)
287+ // Do not keep accumulating shard names while draining earlier work. This
288+ // bounds retained names to one directory batch even for a session with many
289+ // object shards.
290+ if !state.sweepRootDone && len(state.sweepQueue) == 0 {
291+ entries, done, err := r.readMaintenanceDir(root, maintenanceBatch, "sweep-shards:"+key)
292+ if errors.Is(err, os.ErrNotExist) {
293+ state.sweepRootDone = true
294+ } else if err != nil {
295+ return fmt.Errorf("read snapshot objects: %w", safeFilesystemError(err))
296+ } else {
297+ for _, entry := range entries {
298+ if entry.isDir {
299+ state.sweepQueue = append(state.sweepQueue, entry.name)
300+ }
301+ }
302+ state.sweepRootDone = done
303+ }
304+ }
305+ if len(state.sweepQueue) != 0 && *budget > 0 {
306+ shard := state.sweepQueue[0]
307+ done, err := r.sweepShard(ctx, filepath.Join(root, shard), referenced, budget, key, shard)
308+ if err != nil {
309+ return err
310+ }
311+ if done {
312+ state.sweepQueue = state.sweepQueue[1:]
313+ }
314+ }
315+ if state.sweepRootDone && len(state.sweepQueue) == 0 {
316+ r.clearSessionMaintenance(key)
317+ }
318+ return nil
319+}
320+
321+func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
322+ if conservative {
323+ references := make(map[ports.SnapshotDigest]struct{})
324+ for _, item := range marked {
325+ for digest := range item.refs {
326+ references[digest] = struct{}{}
327+ }
328+ }
329+ return references
330+ }
331+ complete := make([]uint64, 0, len(marked))
332+ for generation, item := range marked {
333+ if item.complete {
334+ complete = append(complete, generation)
335+ }
336+ }
337+ for i := range complete {
338+ for j := i + 1; j < len(complete); j++ {
339+ if complete[j] > complete[i] {
340+ complete[i], complete[j] = complete[j], complete[i]
341+ }
342+ }
343+ }
344+ keep := make(map[uint64]bool, 2)
345+ for _, generation := range complete {
346+ if len(keep) == 2 {
347+ break
348+ }
349+ keep[generation] = true
350+ }
351+ references := make(map[ports.SnapshotDigest]struct{})
352+ for generation, item := range marked {
353+ if !item.complete || keep[generation] {
354+ for digest := range item.refs {
355+ references[digest] = struct{}{}
356+ }
357+ }
358+ }
359+ return references
360+}
361+
362+func (r *Repository) sweepShard(ctx context.Context, dir string, referenced map[ports.SnapshotDigest]struct{}, budget *int, key, shard string) (bool, error) {
363+ entries, done, err := r.readMaintenanceDir(dir, maintenanceBatch, "sweep-objects:"+key+":"+shard)
364+ if errors.Is(err, os.ErrNotExist) {
365+ return true, nil
366+ }
367+ if err != nil {
368+ return false, fmt.Errorf("read snapshot object shard: %w", safeFilesystemError(err))
369+ }
370+ for i, object := range entries {
371+ if *budget == 0 {
372+ r.requeueMaintenanceEntries(dir, "sweep-objects:"+key+":"+shard, entries[i:])
373+ return false, nil
374+ }
375+ if err := ctx.Err(); err != nil {
376+ return false, err
377+ }
378+ digest, ok := parseObjectDigest(object.name)
379+ if !ok || object.isDir {
380+ continue
381+ }
382+ if _, used := referenced[digest]; used {
383+ continue
384+ }
385+ path := filepath.Join(dir, object.name)
386+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
387+ return false, fmt.Errorf("remove snapshot object %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
388+ }
389+ if err := ctx.Err(); err != nil {
390+ return false, err
391+ }
392+ if err := r.syncDirectory(dir); err != nil {
393+ return false, fmt.Errorf("sync snapshot maintenance directory for %q: %w", maintenancePath(r.dir, path), safeFilesystemError(err))
394+ }
395+ *budget--
396+ }
397+ return done, nil
398+}

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