detached 80597c9

fix(recovery): bound migration maintenance io

commit
80597c958978f52b7b4003b40aca6f6648d95cf4
author
brice · · 2026-07-25 22:12 UTC
committer
brice · · 2026-07-25 22:15 UTC
parent
ea0c849
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
8 files · +277 -66
Unified
UnifiedSplit
M internal/adapters/recoveryfs/atomic.go +22 -5
@@ -8,7 +8,13 @@ import (
88 "github.com/bnema/vev/pkg/safedir"
99 )
1010
11-func atomicWrite(path string, data []byte) (retErr error) {
11+type journalHooks struct {
12+ syncFile func(string) error
13+ rename func(string, string) error
14+ syncDirectory func(string) error
15+}
16+
17+func (j *Journal) atomicWrite(path string, data []byte) (retErr error) {
1218 dir := filepath.Dir(path)
1319 if err := safedir.EnsurePrivate(dir); err != nil {
1420 return err
@@ -29,19 +35,30 @@ func atomicWrite(path string, data []byte) (retErr error) {
2935 if _, err := file.Write(data); err != nil {
3036 return errors.Join(err, file.Close())
3137 }
32- if err := file.Sync(); err != nil {
38+ if j.hooks.syncFile != nil {
39+ if err := j.hooks.syncFile(tmp); err != nil {
40+ return errors.Join(err, file.Close())
41+ }
42+ } else if err := file.Sync(); err != nil {
3343 return errors.Join(err, file.Close())
3444 }
3545 if err := file.Close(); err != nil {
3646 return err
3747 }
38- if err := os.Rename(tmp, path); err != nil {
48+ if j.hooks.rename != nil {
49+ if err := j.hooks.rename(tmp, path); err != nil {
50+ return err
51+ }
52+ } else if err := os.Rename(tmp, path); err != nil {
3953 return err
4054 }
41- return syncDir(dir)
55+ return j.syncDir(dir)
4256 }
4357
44-func syncDir(dir string) error {
58+func (j *Journal) syncDir(dir string) error {
59+ if j.hooks.syncDirectory != nil {
60+ return j.hooks.syncDirectory(dir)
61+ }
4562 file, err := os.Open(dir)
4663 if err != nil {
4764 return err
@@ -8,7 +8,13 @@ import (
8 "github.com/bnema/vev/pkg/safedir"8 "github.com/bnema/vev/pkg/safedir"
9 )9 )
10 10
11-func atomicWrite(path string, data []byte) (retErr error) {11+type journalHooks struct {
12+ syncFile func(string) error
13+ rename func(string, string) error
14+ syncDirectory func(string) error
15+}
16+
17+func (j *Journal) atomicWrite(path string, data []byte) (retErr error) {
12 dir := filepath.Dir(path)18 dir := filepath.Dir(path)
13 if err := safedir.EnsurePrivate(dir); err != nil {19 if err := safedir.EnsurePrivate(dir); err != nil {
14 return err20 return err
@@ -29,19 +35,30 @@ func atomicWrite(path string, data []byte) (retErr error) {
29 if _, err := file.Write(data); err != nil {35 if _, err := file.Write(data); err != nil {
30 return errors.Join(err, file.Close())36 return errors.Join(err, file.Close())
31 }37 }
32- if err := file.Sync(); err != nil {38+ if j.hooks.syncFile != nil {
39+ if err := j.hooks.syncFile(tmp); err != nil {
40+ return errors.Join(err, file.Close())
41+ }
42+ } else if err := file.Sync(); err != nil {
33 return errors.Join(err, file.Close())43 return errors.Join(err, file.Close())
34 }44 }
35 if err := file.Close(); err != nil {45 if err := file.Close(); err != nil {
36 return err46 return err
37 }47 }
38- if err := os.Rename(tmp, path); err != nil {48+ if j.hooks.rename != nil {
49+ if err := j.hooks.rename(tmp, path); err != nil {
50+ return err
51+ }
52+ } else if err := os.Rename(tmp, path); err != nil {
39 return err53 return err
40 }54 }
41- return syncDir(dir)55+ return j.syncDir(dir)
42 }56 }
43 57
44-func syncDir(dir string) error {58+func (j *Journal) syncDir(dir string) error {
59+ if j.hooks.syncDirectory != nil {
60+ return j.hooks.syncDirectory(dir)
61+ }
45 file, err := os.Open(dir)62 file, err := os.Open(dir)
46 if err != nil {63 if err != nil {
47 return err64 return err
M internal/adapters/recoveryfs/journal.go +47 -8
@@ -16,9 +16,16 @@ import (
1616 "github.com/bnema/vev/pkg/safedir"
1717 )
1818
19-const discardVersion = 1
19+const (
20+ discardVersion = 1
21+ maxDiscardIntents = 1024
22+ maxDiscardIntentSize = 64 << 10
23+)
2024
21-type Journal struct{ dir string }
25+type Journal struct {
26+ dir string
27+ hooks journalHooks
28+}
2229
2330 type discardFile struct {
2431 Version int `json:"version"`
@@ -76,25 +83,38 @@ func (j *Journal) SaveDiscard(ctx context.Context, intent domain.DiscardIntent)
7683 return err
7784 }
7885 path := j.path(intent.OldIncarnation)
79- if existing, err := os.ReadFile(path); err == nil {
86+ if existing, err := readFileBounded(path, maxDiscardIntentSize); err == nil {
8087 if !bytes.Equal(existing, data) {
8188 return errors.New("recovery journal: discard identity collision")
8289 }
83- return nil
90+ // Re-sync the parent on idempotent retry: the prior install may have
91+ // succeeded while its durability acknowledgement failed.
92+ return j.syncDir(filepath.Dir(path))
8493 } else if !errors.Is(err, os.ErrNotExist) {
8594 return err
8695 }
87- return atomicWrite(path, data)
96+ return j.atomicWrite(path, data)
8897 }
8998
9099 func (j *Journal) ListDiscards(ctx context.Context) ([]domain.DiscardIntent, error) {
91- entries, err := os.ReadDir(j.dir)
100+ dir, err := os.Open(j.dir)
92101 if errors.Is(err, os.ErrNotExist) {
93102 return nil, nil
94103 }
95104 if err != nil {
96105 return nil, err
97106 }
107+ entries, readErr := dir.ReadDir(maxDiscardIntents + 1)
108+ closeErr := dir.Close()
109+ if readErr != nil && !errors.Is(readErr, io.EOF) {
110+ return nil, readErr
111+ }
112+ if closeErr != nil {
113+ return nil, closeErr
114+ }
115+ if len(entries) > maxDiscardIntents {
116+ return nil, errors.New("recovery journal: transaction entry limit exceeded")
117+ }
98118 sort.Slice(entries, func(a, b int) bool { return entries[a].Name() < entries[b].Name() })
99119 out := make([]domain.DiscardIntent, 0, len(entries))
100120 seen := make(map[domain.IncarnationID]struct{}, len(entries))
@@ -105,7 +125,7 @@ func (j *Journal) ListDiscards(ctx context.Context) ([]domain.DiscardIntent, err
105125 if entry.IsDir() || !strings.HasPrefix(entry.Name(), "discard-") {
106126 return nil, errors.New("recovery journal: invalid transaction entry")
107127 }
108- data, err := os.ReadFile(filepath.Join(j.dir, entry.Name()))
128+ data, err := readFileBounded(filepath.Join(j.dir, entry.Name()), maxDiscardIntentSize)
109129 if err != nil {
110130 return nil, err
111131 }
@@ -138,5 +158,24 @@ func (j *Journal) DeleteDiscard(ctx context.Context, id domain.IncarnationID) er
138158 if err := os.Remove(j.path(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
139159 return err
140160 }
141- return syncDir(j.dir)
161+ return j.syncDir(j.dir)
162+}
163+
164+func readFileBounded(path string, limit int64) ([]byte, error) {
165+ file, err := os.Open(path)
166+ if err != nil {
167+ return nil, err
168+ }
169+ data, readErr := io.ReadAll(io.LimitReader(file, limit+1))
170+ closeErr := file.Close()
171+ if readErr != nil {
172+ return nil, readErr
173+ }
174+ if closeErr != nil {
175+ return nil, closeErr
176+ }
177+ if int64(len(data)) > limit {
178+ return nil, errors.New("recovery journal: intent size limit exceeded")
179+ }
180+ return data, nil
142181 }
@@ -16,9 +16,16 @@ import (
16 "github.com/bnema/vev/pkg/safedir"16 "github.com/bnema/vev/pkg/safedir"
17 )17 )
18 18
19-const discardVersion = 119+const (
20+ discardVersion = 1
21+ maxDiscardIntents = 1024
22+ maxDiscardIntentSize = 64 << 10
23+)
20 24
21-type Journal struct{ dir string }25+type Journal struct {
26+ dir string
27+ hooks journalHooks
28+}
22 29
23 type discardFile struct {30 type discardFile struct {
24 Version int `json:"version"`31 Version int `json:"version"`
@@ -76,25 +83,38 @@ func (j *Journal) SaveDiscard(ctx context.Context, intent domain.DiscardIntent)
76 return err83 return err
77 }84 }
78 path := j.path(intent.OldIncarnation)85 path := j.path(intent.OldIncarnation)
79- if existing, err := os.ReadFile(path); err == nil {86+ if existing, err := readFileBounded(path, maxDiscardIntentSize); err == nil {
80 if !bytes.Equal(existing, data) {87 if !bytes.Equal(existing, data) {
81 return errors.New("recovery journal: discard identity collision")88 return errors.New("recovery journal: discard identity collision")
82 }89 }
83- return nil90+ // Re-sync the parent on idempotent retry: the prior install may have
91+ // succeeded while its durability acknowledgement failed.
92+ return j.syncDir(filepath.Dir(path))
84 } else if !errors.Is(err, os.ErrNotExist) {93 } else if !errors.Is(err, os.ErrNotExist) {
85 return err94 return err
86 }95 }
87- return atomicWrite(path, data)96+ return j.atomicWrite(path, data)
88 }97 }
89 98
90 func (j *Journal) ListDiscards(ctx context.Context) ([]domain.DiscardIntent, error) {99 func (j *Journal) ListDiscards(ctx context.Context) ([]domain.DiscardIntent, error) {
91- entries, err := os.ReadDir(j.dir)100+ dir, err := os.Open(j.dir)
92 if errors.Is(err, os.ErrNotExist) {101 if errors.Is(err, os.ErrNotExist) {
93 return nil, nil102 return nil, nil
94 }103 }
95 if err != nil {104 if err != nil {
96 return nil, err105 return nil, err
97 }106 }
107+ entries, readErr := dir.ReadDir(maxDiscardIntents + 1)
108+ closeErr := dir.Close()
109+ if readErr != nil && !errors.Is(readErr, io.EOF) {
110+ return nil, readErr
111+ }
112+ if closeErr != nil {
113+ return nil, closeErr
114+ }
115+ if len(entries) > maxDiscardIntents {
116+ return nil, errors.New("recovery journal: transaction entry limit exceeded")
117+ }
98 sort.Slice(entries, func(a, b int) bool { return entries[a].Name() < entries[b].Name() })118 sort.Slice(entries, func(a, b int) bool { return entries[a].Name() < entries[b].Name() })
99 out := make([]domain.DiscardIntent, 0, len(entries))119 out := make([]domain.DiscardIntent, 0, len(entries))
100 seen := make(map[domain.IncarnationID]struct{}, len(entries))120 seen := make(map[domain.IncarnationID]struct{}, len(entries))
@@ -105,7 +125,7 @@ func (j *Journal) ListDiscards(ctx context.Context) ([]domain.DiscardIntent, err
105 if entry.IsDir() || !strings.HasPrefix(entry.Name(), "discard-") {125 if entry.IsDir() || !strings.HasPrefix(entry.Name(), "discard-") {
106 return nil, errors.New("recovery journal: invalid transaction entry")126 return nil, errors.New("recovery journal: invalid transaction entry")
107 }127 }
108- data, err := os.ReadFile(filepath.Join(j.dir, entry.Name()))128+ data, err := readFileBounded(filepath.Join(j.dir, entry.Name()), maxDiscardIntentSize)
109 if err != nil {129 if err != nil {
110 return nil, err130 return nil, err
111 }131 }
@@ -138,5 +158,24 @@ func (j *Journal) DeleteDiscard(ctx context.Context, id domain.IncarnationID) er
138 if err := os.Remove(j.path(id)); err != nil && !errors.Is(err, os.ErrNotExist) {158 if err := os.Remove(j.path(id)); err != nil && !errors.Is(err, os.ErrNotExist) {
139 return err159 return err
140 }160 }
141- return syncDir(j.dir)161+ return j.syncDir(j.dir)
162+}
163+
164+func readFileBounded(path string, limit int64) ([]byte, error) {
165+ file, err := os.Open(path)
166+ if err != nil {
167+ return nil, err
168+ }
169+ data, readErr := io.ReadAll(io.LimitReader(file, limit+1))
170+ closeErr := file.Close()
171+ if readErr != nil {
172+ return nil, readErr
173+ }
174+ if closeErr != nil {
175+ return nil, closeErr
176+ }
177+ if int64(len(data)) > limit {
178+ return nil, errors.New("recovery journal: intent size limit exceeded")
179+ }
180+ return data, nil
142 }181 }
M internal/adapters/recoveryfs/journal_test.go +48 -0
@@ -2,6 +2,9 @@ package recoveryfs
22
33 import (
44 "context"
5+ "errors"
6+ "os"
7+ "path/filepath"
58 "testing"
69
710 "github.com/bnema/vev/internal/domain"
@@ -30,3 +33,48 @@ func TestRecoveryJournalRejectsMalformedIntent(t *testing.T) {
3033 j := New(t.TempDir())
3134 require.Error(t, j.SaveDiscard(context.Background(), domain.DiscardIntent{}))
3235 }
36+
37+func TestRecoveryJournalBoundsEnumerationAndIntentReads(t *testing.T) {
38+ state := t.TempDir()
39+ j := New(state)
40+ require.NoError(t, os.MkdirAll(j.dir, 0o700))
41+ for i := 0; i <= maxDiscardIntents; i++ {
42+ require.NoError(t, os.WriteFile(filepath.Join(j.dir, "discard-extra-"+string(rune(0x1000+i))), nil, 0o600))
43+ }
44+ _, err := j.ListDiscards(context.Background())
45+ require.ErrorContains(t, err, "entry limit exceeded")
46+
47+ require.NoError(t, os.RemoveAll(j.dir))
48+ require.NoError(t, os.MkdirAll(j.dir, 0o700))
49+ require.NoError(t, os.WriteFile(filepath.Join(j.dir, "discard-oversized"), make([]byte, maxDiscardIntentSize+1), 0o600))
50+ _, err = j.ListDiscards(context.Background())
51+ require.ErrorContains(t, err, "size limit exceeded")
52+}
53+
54+func TestRecoveryJournalFailsClosedAtDurabilityBoundaries(t *testing.T) {
55+ for _, boundary := range []string{"file sync", "rename", "directory sync"} {
56+ t.Run(boundary, func(t *testing.T) {
57+ j := New(t.TempDir())
58+ old := domain.IncarnationID{1}
59+ intent := domain.DiscardIntent{
60+ OldRecord: domain.CatalogueRecord{Name: "work", IncarnationID: old, RecoveryState: domain.RecoveryDegraded, Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "invalid"},
61+ OldIncarnation: old, NewIncarnation: domain.IncarnationID{2}, SessionName: "work", Reason: "discard",
62+ }
63+ injected := errors.New("injected durability failure")
64+ switch boundary {
65+ case "file sync":
66+ j.hooks.syncFile = func(string) error { return injected }
67+ case "rename":
68+ j.hooks.rename = func(string, string) error { return injected }
69+ case "directory sync":
70+ j.hooks.syncDirectory = func(string) error { return injected }
71+ }
72+ require.ErrorIs(t, j.SaveDiscard(context.Background(), intent), injected)
73+ j.hooks = journalHooks{}
74+ require.NoError(t, j.SaveDiscard(context.Background(), intent))
75+ got, err := j.ListDiscards(context.Background())
76+ require.NoError(t, err)
77+ require.Equal(t, []domain.DiscardIntent{intent}, got)
78+ })
79+ }
80+}
@@ -2,6 +2,9 @@ package recoveryfs
2 2
3 import (3 import (
4 "context"4 "context"
5+ "errors"
6+ "os"
7+ "path/filepath"
5 "testing"8 "testing"
6 9
7 "github.com/bnema/vev/internal/domain"10 "github.com/bnema/vev/internal/domain"
@@ -30,3 +33,48 @@ func TestRecoveryJournalRejectsMalformedIntent(t *testing.T) {
30 j := New(t.TempDir())33 j := New(t.TempDir())
31 require.Error(t, j.SaveDiscard(context.Background(), domain.DiscardIntent{}))34 require.Error(t, j.SaveDiscard(context.Background(), domain.DiscardIntent{}))
32 }35 }
36+
37+func TestRecoveryJournalBoundsEnumerationAndIntentReads(t *testing.T) {
38+ state := t.TempDir()
39+ j := New(state)
40+ require.NoError(t, os.MkdirAll(j.dir, 0o700))
41+ for i := 0; i <= maxDiscardIntents; i++ {
42+ require.NoError(t, os.WriteFile(filepath.Join(j.dir, "discard-extra-"+string(rune(0x1000+i))), nil, 0o600))
43+ }
44+ _, err := j.ListDiscards(context.Background())
45+ require.ErrorContains(t, err, "entry limit exceeded")
46+
47+ require.NoError(t, os.RemoveAll(j.dir))
48+ require.NoError(t, os.MkdirAll(j.dir, 0o700))
49+ require.NoError(t, os.WriteFile(filepath.Join(j.dir, "discard-oversized"), make([]byte, maxDiscardIntentSize+1), 0o600))
50+ _, err = j.ListDiscards(context.Background())
51+ require.ErrorContains(t, err, "size limit exceeded")
52+}
53+
54+func TestRecoveryJournalFailsClosedAtDurabilityBoundaries(t *testing.T) {
55+ for _, boundary := range []string{"file sync", "rename", "directory sync"} {
56+ t.Run(boundary, func(t *testing.T) {
57+ j := New(t.TempDir())
58+ old := domain.IncarnationID{1}
59+ intent := domain.DiscardIntent{
60+ OldRecord: domain.CatalogueRecord{Name: "work", IncarnationID: old, RecoveryState: domain.RecoveryDegraded, Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "invalid"},
61+ OldIncarnation: old, NewIncarnation: domain.IncarnationID{2}, SessionName: "work", Reason: "discard",
62+ }
63+ injected := errors.New("injected durability failure")
64+ switch boundary {
65+ case "file sync":
66+ j.hooks.syncFile = func(string) error { return injected }
67+ case "rename":
68+ j.hooks.rename = func(string, string) error { return injected }
69+ case "directory sync":
70+ j.hooks.syncDirectory = func(string) error { return injected }
71+ }
72+ require.ErrorIs(t, j.SaveDiscard(context.Background(), intent), injected)
73+ j.hooks = journalHooks{}
74+ require.NoError(t, j.SaveDiscard(context.Background(), intent))
75+ got, err := j.ListDiscards(context.Background())
76+ require.NoError(t, err)
77+ require.Equal(t, []domain.DiscardIntent{intent}, got)
78+ })
79+ }
80+}
M internal/adapters/snapshot/maintenance_delete.go +78 -42
@@ -8,7 +8,6 @@ import (
88 "io"
99 "os"
1010 "path/filepath"
11- "sort"
1211 "strings"
1312
1413 "github.com/bnema/vev/internal/domain"
@@ -85,61 +84,98 @@ func (r *Repository) WriteDeletionTombstone(ctx context.Context, tombstone domai
8584 })
8685 }
8786
87+type deletionListingState struct {
88+ after string
89+ candidate maintenanceDirEntry
90+ found uint64
91+ scanDone bool
92+ token string
93+}
94+
8895 func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.DeletionTombstoneCursor, budget ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
8996 if budget.Entries == 0 || budget.Bytes == 0 {
9097 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
9198 }
9299 dir := filepath.Join(r.dir, deletionTombstonesDir)
93- entries, err := os.ReadDir(dir)
94- if errors.Is(err, os.ErrNotExist) {
95- return ports.DeletionTombstonePage{Done: true}, nil
96- }
97- if err != nil {
98- return ports.DeletionTombstonePage{}, err
99- }
100- sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
101- page := ports.DeletionTombstonePage{}
102- seen := make(map[domain.IncarnationID]struct{})
103- for _, entry := range entries {
104- if err := ctx.Err(); err != nil {
105- return ports.DeletionTombstonePage{}, err
106- }
107- if entry.Name() <= cursor.After {
108- continue
109- }
110- if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".tombstone") {
111- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
112- }
113- if uint64(len(page.Tombstones)) >= budget.Entries {
114- return page, nil
100+ r.maintenanceMu.Lock()
101+ defer r.maintenanceMu.Unlock()
102+
103+ state := r.deletionListing
104+ if state == nil || (cursor.After != state.token && cursor.After != state.after) {
105+ state = &deletionListingState{after: cursor.After}
106+ r.deletionListing = state
107+ delete(r.maintenanceCursors, maintenanceCursorID(dir, "deletion-tombstones"))
108+ }
109+ if !state.scanDone {
110+ entryLimit := budget.Entries
111+ if maxInt := uint64(^uint(0) >> 1); entryLimit > maxInt {
112+ entryLimit = maxInt
115113 }
116- data, err := r.readBounded(filepath.Join(dir, entry.Name()))
117- if err != nil {
118- return ports.DeletionTombstonePage{}, err
114+ entries, done, err := r.readMaintenanceDir(dir, int(entryLimit), "deletion-tombstones")
115+ if errors.Is(err, os.ErrNotExist) {
116+ r.deletionListing = nil
117+ return ports.DeletionTombstonePage{Done: true}, nil
119118 }
120- decoded, err := decodeDeletionTombstone(data)
121119 if err != nil {
120+ r.deletionListing = nil
122121 return ports.DeletionTombstonePage{}, err
123122 }
124- charge := uint64(len(entry.Name()) + len(data))
125- if charge > budget.Bytes {
126- if len(page.Tombstones) == 0 {
127- return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
123+ for _, entry := range entries {
124+ if err := ctx.Err(); err != nil {
125+ return ports.DeletionTombstonePage{}, err
126+ }
127+ if entry.isDir || !strings.HasSuffix(entry.name, ".tombstone") {
128+ r.deletionListing = nil
129+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
130+ }
131+ if entry.name > state.after {
132+ state.found++
133+ if state.candidate.name == "" || entry.name < state.candidate.name {
134+ state.candidate = entry
135+ }
128136 }
129- return page, nil
130- }
131- if entry.Name() != decoded.IncarnationID.String()+".tombstone" {
132- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")
133137 }
134- if _, duplicate := seen[decoded.IncarnationID]; duplicate {
135- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: duplicate deletion tombstone identity")
138+ state.scanDone = done
139+ if !done {
140+ r.deletionListingToken++
141+ state.token = fmt.Sprintf("@%d", r.deletionListingToken)
142+ return ports.DeletionTombstonePage{Next: ports.DeletionTombstoneCursor{After: state.token}}, nil
136143 }
137- seen[decoded.IncarnationID] = struct{}{}
138- page.Tombstones = append(page.Tombstones, decoded)
139- page.Next.After = entry.Name()
140- budget.Bytes -= charge
141144 }
142- page.Done = true
145+ if state.candidate.name == "" {
146+ r.deletionListing = nil
147+ return ports.DeletionTombstonePage{Done: true}, nil
148+ }
149+ path := filepath.Join(dir, state.candidate.name)
150+ info, err := r.stat(path)
151+ if err != nil {
152+ return ports.DeletionTombstonePage{}, err
153+ }
154+ charge := uint64(len(state.candidate.name))
155+ if info.Size() < 0 || uint64(info.Size()) > budget.Bytes || charge > budget.Bytes-uint64(info.Size()) {
156+ r.deletionListing = nil
157+ return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
158+ }
159+ if hook := r.hooks.beforeDeletionTombstoneRead; hook != nil {
160+ hook(path)
161+ }
162+ data, err := r.readBounded(path)
163+ if err != nil {
164+ return ports.DeletionTombstonePage{}, err
165+ }
166+ decoded, err := decodeDeletionTombstone(data)
167+ if err != nil {
168+ return ports.DeletionTombstonePage{}, err
169+ }
170+ if state.candidate.name != decoded.IncarnationID.String()+".tombstone" {
171+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")
172+ }
173+ page := ports.DeletionTombstonePage{Tombstones: []domain.DeletionTombstone{decoded}, Next: ports.DeletionTombstoneCursor{After: state.candidate.name}, Done: state.found == 1}
174+ if page.Done {
175+ r.deletionListing = nil
176+ } else {
177+ r.deletionListing = &deletionListingState{after: state.candidate.name}
178+ }
143179 return page, nil
144180 }
145181
@@ -8,7 +8,6 @@ import (
8 "io"8 "io"
9 "os"9 "os"
10 "path/filepath"10 "path/filepath"
11- "sort"
12 "strings"11 "strings"
13 12
14 "github.com/bnema/vev/internal/domain"13 "github.com/bnema/vev/internal/domain"
@@ -85,61 +84,98 @@ func (r *Repository) WriteDeletionTombstone(ctx context.Context, tombstone domai
85 })84 })
86 }85 }
87 86
87+type deletionListingState struct {
88+ after string
89+ candidate maintenanceDirEntry
90+ found uint64
91+ scanDone bool
92+ token string
93+}
94+
88 func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.DeletionTombstoneCursor, budget ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {95 func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.DeletionTombstoneCursor, budget ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
89 if budget.Entries == 0 || budget.Bytes == 0 {96 if budget.Entries == 0 || budget.Bytes == 0 {
90 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall97 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
91 }98 }
92 dir := filepath.Join(r.dir, deletionTombstonesDir)99 dir := filepath.Join(r.dir, deletionTombstonesDir)
93- entries, err := os.ReadDir(dir)100+ r.maintenanceMu.Lock()
94- if errors.Is(err, os.ErrNotExist) {101+ defer r.maintenanceMu.Unlock()
95- return ports.DeletionTombstonePage{Done: true}, nil102+
96- }103+ state := r.deletionListing
97- if err != nil {104+ if state == nil || (cursor.After != state.token && cursor.After != state.after) {
98- return ports.DeletionTombstonePage{}, err105+ state = &deletionListingState{after: cursor.After}
99- }106+ r.deletionListing = state
100- sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })107+ delete(r.maintenanceCursors, maintenanceCursorID(dir, "deletion-tombstones"))
101- page := ports.DeletionTombstonePage{}108+ }
102- seen := make(map[domain.IncarnationID]struct{})109+ if !state.scanDone {
103- for _, entry := range entries {110+ entryLimit := budget.Entries
104- if err := ctx.Err(); err != nil {111+ if maxInt := uint64(^uint(0) >> 1); entryLimit > maxInt {
105- return ports.DeletionTombstonePage{}, err112+ entryLimit = maxInt
106- }
107- if entry.Name() <= cursor.After {
108- continue
109- }
110- if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".tombstone") {
111- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
112- }
113- if uint64(len(page.Tombstones)) >= budget.Entries {
114- return page, nil
115 }113 }
116- data, err := r.readBounded(filepath.Join(dir, entry.Name()))114+ entries, done, err := r.readMaintenanceDir(dir, int(entryLimit), "deletion-tombstones")
117- if err != nil {115+ if errors.Is(err, os.ErrNotExist) {
118- return ports.DeletionTombstonePage{}, err116+ r.deletionListing = nil
117+ return ports.DeletionTombstonePage{Done: true}, nil
119 }118 }
120- decoded, err := decodeDeletionTombstone(data)
121 if err != nil {119 if err != nil {
120+ r.deletionListing = nil
122 return ports.DeletionTombstonePage{}, err121 return ports.DeletionTombstonePage{}, err
123 }122 }
124- charge := uint64(len(entry.Name()) + len(data))123+ for _, entry := range entries {
125- if charge > budget.Bytes {124+ if err := ctx.Err(); err != nil {
126- if len(page.Tombstones) == 0 {125+ return ports.DeletionTombstonePage{}, err
127- return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall126+ }
127+ if entry.isDir || !strings.HasSuffix(entry.name, ".tombstone") {
128+ r.deletionListing = nil
129+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
130+ }
131+ if entry.name > state.after {
132+ state.found++
133+ if state.candidate.name == "" || entry.name < state.candidate.name {
134+ state.candidate = entry
135+ }
128 }136 }
129- return page, nil
130- }
131- if entry.Name() != decoded.IncarnationID.String()+".tombstone" {
132- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")
133 }137 }
134- if _, duplicate := seen[decoded.IncarnationID]; duplicate {138+ state.scanDone = done
135- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: duplicate deletion tombstone identity")139+ if !done {
140+ r.deletionListingToken++
141+ state.token = fmt.Sprintf("@%d", r.deletionListingToken)
142+ return ports.DeletionTombstonePage{Next: ports.DeletionTombstoneCursor{After: state.token}}, nil
136 }143 }
137- seen[decoded.IncarnationID] = struct{}{}
138- page.Tombstones = append(page.Tombstones, decoded)
139- page.Next.After = entry.Name()
140- budget.Bytes -= charge
141 }144 }
142- page.Done = true145+ if state.candidate.name == "" {
146+ r.deletionListing = nil
147+ return ports.DeletionTombstonePage{Done: true}, nil
148+ }
149+ path := filepath.Join(dir, state.candidate.name)
150+ info, err := r.stat(path)
151+ if err != nil {
152+ return ports.DeletionTombstonePage{}, err
153+ }
154+ charge := uint64(len(state.candidate.name))
155+ if info.Size() < 0 || uint64(info.Size()) > budget.Bytes || charge > budget.Bytes-uint64(info.Size()) {
156+ r.deletionListing = nil
157+ return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
158+ }
159+ if hook := r.hooks.beforeDeletionTombstoneRead; hook != nil {
160+ hook(path)
161+ }
162+ data, err := r.readBounded(path)
163+ if err != nil {
164+ return ports.DeletionTombstonePage{}, err
165+ }
166+ decoded, err := decodeDeletionTombstone(data)
167+ if err != nil {
168+ return ports.DeletionTombstonePage{}, err
169+ }
170+ if state.candidate.name != decoded.IncarnationID.String()+".tombstone" {
171+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")
172+ }
173+ page := ports.DeletionTombstonePage{Tombstones: []domain.DeletionTombstone{decoded}, Next: ports.DeletionTombstoneCursor{After: state.candidate.name}, Done: state.found == 1}
174+ if page.Done {
175+ r.deletionListing = nil
176+ } else {
177+ r.deletionListing = &deletionListingState{after: state.candidate.name}
178+ }
143 return page, nil179 return page, nil
144 }180 }
145 181
M internal/adapters/snapshot/maintenance_delete_test.go +11 -4
@@ -79,24 +79,31 @@ func TestDeletionTombstoneListing(t *testing.T) {
7979 if err := repo.WriteDeletionTombstone(ctx, first); err != nil {
8080 t.Fatal(err)
8181 }
82- page, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})
82+ page, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 3, Bytes: 1024})
8383 if err != nil {
8484 t.Fatal(err)
8585 }
8686 if page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != first || page.Next.After == "" {
8787 t.Fatalf("first page = %#v", page)
8888 }
89- page, err = repo.ListDeletionTombstones(ctx, page.Next, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})
89+ page, err = repo.ListDeletionTombstones(ctx, page.Next, ports.MaintenanceBudget{Entries: 3, Bytes: 1024})
9090 if err != nil {
9191 t.Fatal(err)
9292 }
93- if !page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != second {
93+ if !page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != second || page.Next.After == "" {
9494 t.Fatalf("second page = %#v", page)
9595 }
96- if _, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 1, Bytes: 1}); err == nil {
96+ if _, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 3, Bytes: 1}); err == nil {
9797 t.Fatal("non-advancing budget accepted")
9898 }
9999
100+ reads := 0
101+ repo.hooks.beforeDeletionTombstoneRead = func(string) { reads++ }
102+ page, err = repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})
103+ if err != nil || reads != 0 || len(page.Tombstones) != 0 || page.Next.After == "" {
104+ t.Fatalf("bounded scan read %d objects: page=%#v err=%v", reads, page, err)
105+ }
106+
100107 injected := errors.New("injected directory sync failure")
101108 tombstoneDir := filepath.Join(repo.dir, deletionTombstonesDir)
102109 repo.hooks.syncDirectory = func(path string) error {
@@ -79,24 +79,31 @@ func TestDeletionTombstoneListing(t *testing.T) {
79 if err := repo.WriteDeletionTombstone(ctx, first); err != nil {79 if err := repo.WriteDeletionTombstone(ctx, first); err != nil {
80 t.Fatal(err)80 t.Fatal(err)
81 }81 }
82- page, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})82+ page, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 3, Bytes: 1024})
83 if err != nil {83 if err != nil {
84 t.Fatal(err)84 t.Fatal(err)
85 }85 }
86 if page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != first || page.Next.After == "" {86 if page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != first || page.Next.After == "" {
87 t.Fatalf("first page = %#v", page)87 t.Fatalf("first page = %#v", page)
88 }88 }
89- page, err = repo.ListDeletionTombstones(ctx, page.Next, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})89+ page, err = repo.ListDeletionTombstones(ctx, page.Next, ports.MaintenanceBudget{Entries: 3, Bytes: 1024})
90 if err != nil {90 if err != nil {
91 t.Fatal(err)91 t.Fatal(err)
92 }92 }
93- if !page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != second {93+ if !page.Done || len(page.Tombstones) != 1 || page.Tombstones[0] != second || page.Next.After == "" {
94 t.Fatalf("second page = %#v", page)94 t.Fatalf("second page = %#v", page)
95 }95 }
96- if _, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 1, Bytes: 1}); err == nil {96+ if _, err := repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 3, Bytes: 1}); err == nil {
97 t.Fatal("non-advancing budget accepted")97 t.Fatal("non-advancing budget accepted")
98 }98 }
99 99
100+ reads := 0
101+ repo.hooks.beforeDeletionTombstoneRead = func(string) { reads++ }
102+ page, err = repo.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})
103+ if err != nil || reads != 0 || len(page.Tombstones) != 0 || page.Next.After == "" {
104+ t.Fatalf("bounded scan read %d objects: page=%#v err=%v", reads, page, err)
105+ }
106+
100 injected := errors.New("injected directory sync failure")107 injected := errors.New("injected directory sync failure")
101 tombstoneDir := filepath.Join(repo.dir, deletionTombstonesDir)108 tombstoneDir := filepath.Join(repo.dir, deletionTombstonesDir)
102 repo.hooks.syncDirectory = func(path string) error {109 repo.hooks.syncDirectory = func(path string) error {
M internal/adapters/snapshot/migration_v0.go +7 -0
@@ -261,6 +261,13 @@ func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.Snapshot
261261 if sha256.Sum256(object) != ref.Digest || uint64(len(object)) != ref.Size {
262262 return domain.CheckpointRef{}, uncertainLegacyError("invalid object", nil)
263263 }
264+ kind, _, err := codec.UnmarshalObject(object)
265+ if err != nil {
266+ return domain.CheckpointRef{}, uncertainLegacyError("decode referenced object", err)
267+ }
268+ if uint8(kind) != ref.Kind {
269+ return domain.CheckpointRef{}, uncertainLegacyError("referenced object kind mismatch", nil)
270+ }
264271 if err := r.writeImmutable(r.objectPath(req.IncarnationID, ref.Digest), object, func(existing []byte) error {
265272 if sha256.Sum256(existing) != ref.Digest {
266273 return uncertainLegacyError("migrated object conflict", nil)
@@ -261,6 +261,13 @@ func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.Snapshot
261 if sha256.Sum256(object) != ref.Digest || uint64(len(object)) != ref.Size {261 if sha256.Sum256(object) != ref.Digest || uint64(len(object)) != ref.Size {
262 return domain.CheckpointRef{}, uncertainLegacyError("invalid object", nil)262 return domain.CheckpointRef{}, uncertainLegacyError("invalid object", nil)
263 }263 }
264+ kind, _, err := codec.UnmarshalObject(object)
265+ if err != nil {
266+ return domain.CheckpointRef{}, uncertainLegacyError("decode referenced object", err)
267+ }
268+ if uint8(kind) != ref.Kind {
269+ return domain.CheckpointRef{}, uncertainLegacyError("referenced object kind mismatch", nil)
270+ }
264 if err := r.writeImmutable(r.objectPath(req.IncarnationID, ref.Digest), object, func(existing []byte) error {271 if err := r.writeImmutable(r.objectPath(req.IncarnationID, ref.Digest), object, func(existing []byte) error {
265 if sha256.Sum256(existing) != ref.Digest {272 if sha256.Sum256(existing) != ref.Digest {
266 return uncertainLegacyError("migrated object conflict", nil)273 return uncertainLegacyError("migrated object conflict", nil)
M internal/adapters/snapshot/migration_v0_test.go +52 -0
@@ -76,6 +76,58 @@ func TestHasLegacyState(t *testing.T) {
7676 }
7777 }
7878
79+func TestMigrateV1CheckpointValidatesObjectEnvelopeAndKind(t *testing.T) {
80+ for _, tc := range []struct {
81+ name string
82+ object func(t *testing.T) ports.SnapshotObject
83+ }{
84+ {"malformed envelope", func(t *testing.T) ports.SnapshotObject {
85+ object, err := codec.MarshalObject(codec.HistoryTail, []byte("payload"))
86+ require.NoError(t, err)
87+ object.Data[0] = 'X'
88+ object.Digest = sha256.Sum256(object.Data)
89+ return object
90+ }},
91+ {"kind mismatch", func(t *testing.T) ports.SnapshotObject {
92+ object, err := codec.MarshalObject(codec.Visible, []byte("wrong kind"))
93+ require.NoError(t, err)
94+ return object
95+ }},
96+ } {
97+ t.Run(tc.name, func(t *testing.T) {
98+ dir := privateDir(t)
99+ repo := NewRepository(dir)
100+ object := tc.object(t)
101+ visible, err := codec.MarshalObject(codec.Visible, []byte("visible"))
102+ require.NoError(t, err)
103+ manifest := codec.Manifest{Generation: 1, IncarnationID: domain.IncarnationID{1}, Name: "work", Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: object.Digest, Size: uint32(len(object.Data))}, Visible: codec.ObjectRef{Kind: codec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))}}}}}}
104+ modern, err := codec.MarshalManifest(manifest)
105+ require.NoError(t, err)
106+ body := append([]byte(nil), modern[legacyManifestHeaderSize:]...)
107+ body = append(body[:8], body[25:]...)
108+ legacyManifest := make([]byte, legacyManifestHeaderSize, legacyManifestHeaderSize+len(body))
109+ copy(legacyManifest, "VEVM")
110+ binary.BigEndian.PutUint16(legacyManifest[4:6], 1)
111+ binary.BigEndian.PutUint32(legacyManifest[8:12], uint32(len(body)))
112+ binary.BigEndian.PutUint32(legacyManifest[12:16], crc32.ChecksumIEEE(body))
113+ legacyManifest = append(legacyManifest, body...)
114+ key := sessionKey("work")
115+ for _, stored := range []ports.SnapshotObject{object, visible} {
116+ require.NoError(t, os.MkdirAll(filepath.Dir(repo.legacyObjectPath(key, stored.Digest)), 0o700))
117+ require.NoError(t, os.WriteFile(repo.legacyObjectPath(key, stored.Digest), stored.Data, 0o600))
118+ }
119+ manifestPath := repo.legacyManifestPath(key, 1)
120+ require.NoError(t, os.MkdirAll(filepath.Dir(manifestPath), 0o700))
121+ require.NoError(t, os.WriteFile(manifestPath, legacyManifest, 0o600))
122+ legacyRef := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(legacyManifest)}
123+
124+ _, err = repo.MigrateV1Checkpoint(context.Background(), ports.SnapshotMigrationRequest{LegacyName: "work", IncarnationID: domain.IncarnationID{9}, LegacyRef: legacyRef})
125+ require.ErrorIs(t, err, ports.ErrLegacySnapshotUncertain)
126+ require.FileExists(t, manifestPath)
127+ })
128+ }
129+}
130+
79131 func TestMigrationOver4096(t *testing.T) {
80132 dir := t.TempDir()
81133 require.NoError(t, os.Chmod(dir, 0o700))
@@ -76,6 +76,58 @@ func TestHasLegacyState(t *testing.T) {
76 }76 }
77 }77 }
78 78
79+func TestMigrateV1CheckpointValidatesObjectEnvelopeAndKind(t *testing.T) {
80+ for _, tc := range []struct {
81+ name string
82+ object func(t *testing.T) ports.SnapshotObject
83+ }{
84+ {"malformed envelope", func(t *testing.T) ports.SnapshotObject {
85+ object, err := codec.MarshalObject(codec.HistoryTail, []byte("payload"))
86+ require.NoError(t, err)
87+ object.Data[0] = 'X'
88+ object.Digest = sha256.Sum256(object.Data)
89+ return object
90+ }},
91+ {"kind mismatch", func(t *testing.T) ports.SnapshotObject {
92+ object, err := codec.MarshalObject(codec.Visible, []byte("wrong kind"))
93+ require.NoError(t, err)
94+ return object
95+ }},
96+ } {
97+ t.Run(tc.name, func(t *testing.T) {
98+ dir := privateDir(t)
99+ repo := NewRepository(dir)
100+ object := tc.object(t)
101+ visible, err := codec.MarshalObject(codec.Visible, []byte("visible"))
102+ require.NoError(t, err)
103+ manifest := codec.Manifest{Generation: 1, IncarnationID: domain.IncarnationID{1}, Name: "work", Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: object.Digest, Size: uint32(len(object.Data))}, Visible: codec.ObjectRef{Kind: codec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))}}}}}}
104+ modern, err := codec.MarshalManifest(manifest)
105+ require.NoError(t, err)
106+ body := append([]byte(nil), modern[legacyManifestHeaderSize:]...)
107+ body = append(body[:8], body[25:]...)
108+ legacyManifest := make([]byte, legacyManifestHeaderSize, legacyManifestHeaderSize+len(body))
109+ copy(legacyManifest, "VEVM")
110+ binary.BigEndian.PutUint16(legacyManifest[4:6], 1)
111+ binary.BigEndian.PutUint32(legacyManifest[8:12], uint32(len(body)))
112+ binary.BigEndian.PutUint32(legacyManifest[12:16], crc32.ChecksumIEEE(body))
113+ legacyManifest = append(legacyManifest, body...)
114+ key := sessionKey("work")
115+ for _, stored := range []ports.SnapshotObject{object, visible} {
116+ require.NoError(t, os.MkdirAll(filepath.Dir(repo.legacyObjectPath(key, stored.Digest)), 0o700))
117+ require.NoError(t, os.WriteFile(repo.legacyObjectPath(key, stored.Digest), stored.Data, 0o600))
118+ }
119+ manifestPath := repo.legacyManifestPath(key, 1)
120+ require.NoError(t, os.MkdirAll(filepath.Dir(manifestPath), 0o700))
121+ require.NoError(t, os.WriteFile(manifestPath, legacyManifest, 0o600))
122+ legacyRef := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(legacyManifest)}
123+
124+ _, err = repo.MigrateV1Checkpoint(context.Background(), ports.SnapshotMigrationRequest{LegacyName: "work", IncarnationID: domain.IncarnationID{9}, LegacyRef: legacyRef})
125+ require.ErrorIs(t, err, ports.ErrLegacySnapshotUncertain)
126+ require.FileExists(t, manifestPath)
127+ })
128+ }
129+}
130+
79 func TestMigrationOver4096(t *testing.T) {131 func TestMigrationOver4096(t *testing.T) {
80 dir := t.TempDir()132 dir := t.TempDir()
81 require.NoError(t, os.Chmod(dir, 0o700))133 require.NoError(t, os.Chmod(dir, 0o700))
M internal/adapters/snapshot/repository.go +12 -7
@@ -40,6 +40,8 @@ type Repository struct {
4040 maintenanceCursors map[string]*maintenanceCursor
4141 maintenanceSessions map[string]*sessionMaintenance
4242 maintenanceQuarantine *quarantineMaintenance
43+ deletionListing *deletionListingState
44+ deletionListingToken uint64
4345
4446 // pendingLegacySync records an unlink whose root-directory sync failed.
4547 // It is keyed by the deterministic legacy filename and shares a per-file
@@ -73,10 +75,11 @@ type repositoryHooks struct {
7375 remove func(string) error
7476 // afterOpenRoot and closeRoot make descriptor race and close-error paths
7577 // deterministic in package tests. closeRoot never replaces the real close.
76- afterOpenRoot func()
77- closeRoot func() error
78- openMaintenanceDirectory func(string) (maintenanceDirectory, error)
79- beforeDirectoryRead func(string)
78+ afterOpenRoot func()
79+ closeRoot func() error
80+ openMaintenanceDirectory func(string) (maintenanceDirectory, error)
81+ beforeDirectoryRead func(string)
82+ beforeDeletionTombstoneRead func(string)
8083 // beforeMaintenanceWork observes each budgeted quarantine filesystem step.
8184 // It is test-only instrumentation for hostile traversal bounds.
8285 beforeMaintenanceWork func(string)
@@ -91,9 +94,11 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil)
9194 // until the first publication, so merely constructing it is side-effect free.
9295 func NewRepository(dir string) *Repository {
9396 return &Repository{
94- dir: dir,
95- locks: make(map[string]*sessionMutex),
96- storageEpochs: make(map[string]uint64),
97+ dir: dir,
98+ locks: make(map[string]*sessionMutex),
99+ storageEpochs: make(map[string]uint64),
100+ maintenanceCursors: make(map[string]*maintenanceCursor),
101+ maintenanceSessions: make(map[string]*sessionMaintenance),
97102 }
98103 }
99104
@@ -40,6 +40,8 @@ type Repository struct {
40 maintenanceCursors map[string]*maintenanceCursor40 maintenanceCursors map[string]*maintenanceCursor
41 maintenanceSessions map[string]*sessionMaintenance41 maintenanceSessions map[string]*sessionMaintenance
42 maintenanceQuarantine *quarantineMaintenance42 maintenanceQuarantine *quarantineMaintenance
43+ deletionListing *deletionListingState
44+ deletionListingToken uint64
43 45
44 // pendingLegacySync records an unlink whose root-directory sync failed.46 // pendingLegacySync records an unlink whose root-directory sync failed.
45 // It is keyed by the deterministic legacy filename and shares a per-file47 // It is keyed by the deterministic legacy filename and shares a per-file
@@ -73,10 +75,11 @@ type repositoryHooks struct {
73 remove func(string) error75 remove func(string) error
74 // afterOpenRoot and closeRoot make descriptor race and close-error paths76 // afterOpenRoot and closeRoot make descriptor race and close-error paths
75 // deterministic in package tests. closeRoot never replaces the real close.77 // deterministic in package tests. closeRoot never replaces the real close.
76- afterOpenRoot func()78+ afterOpenRoot func()
77- closeRoot func() error79+ closeRoot func() error
78- openMaintenanceDirectory func(string) (maintenanceDirectory, error)80+ openMaintenanceDirectory func(string) (maintenanceDirectory, error)
79- beforeDirectoryRead func(string)81+ beforeDirectoryRead func(string)
82+ beforeDeletionTombstoneRead func(string)
80 // beforeMaintenanceWork observes each budgeted quarantine filesystem step.83 // beforeMaintenanceWork observes each budgeted quarantine filesystem step.
81 // It is test-only instrumentation for hostile traversal bounds.84 // It is test-only instrumentation for hostile traversal bounds.
82 beforeMaintenanceWork func(string)85 beforeMaintenanceWork func(string)
@@ -91,9 +94,11 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil)
91 // until the first publication, so merely constructing it is side-effect free.94 // until the first publication, so merely constructing it is side-effect free.
92 func NewRepository(dir string) *Repository {95 func NewRepository(dir string) *Repository {
93 return &Repository{96 return &Repository{
94- dir: dir,97+ dir: dir,
95- locks: make(map[string]*sessionMutex),98+ locks: make(map[string]*sessionMutex),
96- storageEpochs: make(map[string]uint64),99+ storageEpochs: make(map[string]uint64),
100+ maintenanceCursors: make(map[string]*maintenanceCursor),
101+ maintenanceSessions: make(map[string]*sessionMaintenance),
97 }102 }
98 }103 }
99 104

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