detached c8509d8

fix(snapshot): persist tombstone scan position

commit
c8509d8947f106b8135d0d40d63c3c6f6d997537
author
brice · · 2026-07-25 22:37 UTC
parent
80597c9
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +251 -53
Unified
UnifiedSplit
M internal/adapters/snapshot/maintenance_delete.go +173 -51
@@ -1,8 +1,11 @@
11 package snapshot
22
33 import (
4+ "bytes"
45 "context"
6+ "encoding/base64"
57 "encoding/binary"
8+ "encoding/json"
69 "errors"
710 "fmt"
811 "io"
@@ -84,76 +87,193 @@ func (r *Repository) WriteDeletionTombstone(ctx context.Context, tombstone domai
8487 })
8588 }
8689
90+const (
91+ deletionListingCursorPrefix = "dt1:"
92+ maxDeletionListingCursor = 64 << 10
93+)
94+
8795 type deletionListingState struct {
88- after string
89- candidate maintenanceDirEntry
90- found uint64
91- scanDone bool
92- token string
96+ After string `json:"after,omitempty"`
97+ Offset int64 `json:"offset,omitempty"`
98+ Pending []maintenanceDirEntry `json:"pending,omitempty"`
99+ PendingAt int `json:"pending_at,omitempty"`
100+ ScanDone bool `json:"scan_done,omitempty"`
101+ Candidate string `json:"candidate,omitempty"`
102+ Found uint8 `json:"found,omitempty"`
103+}
104+
105+// MarshalJSON keeps maintenanceDirEntry private while making a continuation
106+// self-contained. Pending entries are bounded by one directory read buffer.
107+func (entry maintenanceDirEntry) MarshalJSON() ([]byte, error) {
108+ return json.Marshal(struct {
109+ Name string `json:"name"`
110+ IsDir bool `json:"is_dir,omitempty"`
111+ }{Name: entry.name, IsDir: entry.isDir})
112+}
113+
114+func (entry *maintenanceDirEntry) UnmarshalJSON(encoded []byte) error {
115+ var wire struct {
116+ Name string `json:"name"`
117+ IsDir bool `json:"is_dir,omitempty"`
118+ }
119+ decoder := json.NewDecoder(bytes.NewReader(encoded))
120+ decoder.DisallowUnknownFields()
121+ if err := decoder.Decode(&wire); err != nil {
122+ return err
123+ }
124+ entry.name, entry.isDir = wire.Name, wire.IsDir
125+ return nil
126+}
127+
128+func canonicalDeletionTombstoneFilename(name string) bool {
129+ if len(name) != 32+len(".tombstone") || !strings.HasSuffix(name, ".tombstone") {
130+ return false
131+ }
132+ var id domain.IncarnationID
133+ key := strings.TrimSuffix(name, ".tombstone")
134+ return id.UnmarshalText([]byte(key)) == nil && id.String()+".tombstone" == name
135+}
136+
137+func decodeDeletionListingCursor(cursor ports.DeletionTombstoneCursor) (deletionListingState, error) {
138+ if cursor.After == "" {
139+ return deletionListingState{}, nil
140+ }
141+ if len(cursor.After) > maxDeletionListingCursor || !strings.HasPrefix(cursor.After, deletionListingCursorPrefix) {
142+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation")
143+ }
144+ encoded, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(cursor.After, deletionListingCursorPrefix))
145+ if err != nil {
146+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation: %w", err)
147+ }
148+ var state deletionListingState
149+ decoder := json.NewDecoder(bytes.NewReader(encoded))
150+ decoder.DisallowUnknownFields()
151+ if err := decoder.Decode(&state); err != nil {
152+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation: %w", err)
153+ }
154+ if decoder.Decode(&struct{}{}) != io.EOF {
155+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation")
156+ }
157+ if err := validateDeletionListingState(state); err != nil {
158+ return deletionListingState{}, err
159+ }
160+ return state, nil
161+}
162+
163+func validateDeletionListingState(state deletionListingState) error {
164+ if state.After != "" && !canonicalDeletionTombstoneFilename(state.After) {
165+ return fmt.Errorf("snapshot: non-canonical deletion tombstone continuation key")
166+ }
167+ if state.Offset < 0 || state.PendingAt < 0 || state.PendingAt > len(state.Pending) || state.Found > 2 {
168+ return fmt.Errorf("snapshot: malformed deletion tombstone continuation")
169+ }
170+ if state.After == "" && state.Offset == 0 && len(state.Pending) == 0 && !state.ScanDone && state.Candidate == "" {
171+ return fmt.Errorf("snapshot: non-advancing deletion tombstone continuation")
172+ }
173+ if (state.Candidate == "") != (state.Found == 0) || (state.Candidate != "" && (!canonicalDeletionTombstoneFilename(state.Candidate) || state.Candidate <= state.After)) {
174+ return fmt.Errorf("snapshot: non-advancing deletion tombstone continuation")
175+ }
176+ if len(state.Pending) > maintenanceDirentBufferSize {
177+ return fmt.Errorf("snapshot: malformed deletion tombstone continuation")
178+ }
179+ for _, entry := range state.Pending {
180+ if entry.isDir || !canonicalDeletionTombstoneFilename(entry.name) {
181+ return fmt.Errorf("snapshot: non-canonical deletion tombstone continuation entry")
182+ }
183+ }
184+ return nil
185+}
186+
187+func encodeDeletionListingCursor(state deletionListingState) (ports.DeletionTombstoneCursor, error) {
188+ if err := validateDeletionListingState(state); err != nil {
189+ return ports.DeletionTombstoneCursor{}, err
190+ }
191+ encoded, err := json.Marshal(state)
192+ if err != nil {
193+ return ports.DeletionTombstoneCursor{}, err
194+ }
195+ return ports.DeletionTombstoneCursor{After: deletionListingCursorPrefix + base64.RawURLEncoding.EncodeToString(encoded)}, nil
196+}
197+
198+func (r *Repository) readDeletionListingEntries(dir string, limit int, state *deletionListingState) ([]maintenanceDirEntry, bool, error) {
199+ if state.PendingAt < len(state.Pending) {
200+ end := min(state.PendingAt+limit, len(state.Pending))
201+ entries := state.Pending[state.PendingAt:end]
202+ state.PendingAt = end
203+ done := state.ScanDone && end == len(state.Pending)
204+ return entries, done, nil
205+ }
206+ cursor := &maintenanceCursor{offset: state.Offset}
207+ entries, done, err := r.readMaintenanceDirent(dir, limit, cursor)
208+ state.Offset = cursor.offset
209+ state.Pending = cursor.pending
210+ state.PendingAt = 0
211+ state.ScanDone = done || cursor.done
212+ return entries, done && len(cursor.pending) == 0, err
93213 }
94214
95215 func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.DeletionTombstoneCursor, budget ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
96216 if budget.Entries == 0 || budget.Bytes == 0 {
97217 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
98218 }
219+ if err := ctx.Err(); err != nil {
220+ return ports.DeletionTombstonePage{}, err
221+ }
222+ state, err := decodeDeletionListingCursor(cursor)
223+ if err != nil {
224+ return ports.DeletionTombstonePage{}, err
225+ }
99226 dir := filepath.Join(r.dir, deletionTombstonesDir)
100227 r.maintenanceMu.Lock()
101228 defer r.maintenanceMu.Unlock()
102229
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"))
230+ entryLimit := budget.Entries
231+ if maxInt := uint64(^uint(0) >> 1); entryLimit > maxInt {
232+ entryLimit = maxInt
108233 }
109- if !state.scanDone {
110- entryLimit := budget.Entries
111- if maxInt := uint64(^uint(0) >> 1); entryLimit > maxInt {
112- entryLimit = maxInt
113- }
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
118- }
119- if err != nil {
120- r.deletionListing = nil
234+ entries, done, err := r.readDeletionListingEntries(dir, int(entryLimit), &state)
235+ if errors.Is(err, os.ErrNotExist) {
236+ return ports.DeletionTombstonePage{Done: true}, nil
237+ }
238+ if err != nil {
239+ return ports.DeletionTombstonePage{}, err
240+ }
241+ for _, entry := range entries {
242+ if err := ctx.Err(); err != nil {
121243 return ports.DeletionTombstonePage{}, err
122244 }
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")
245+ if entry.isDir || !canonicalDeletionTombstoneFilename(entry.name) {
246+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
247+ }
248+ if entry.name > state.After {
249+ if state.Found < 2 {
250+ state.Found++
130251 }
131- if entry.name > state.after {
132- state.found++
133- if state.candidate.name == "" || entry.name < state.candidate.name {
134- state.candidate = entry
135- }
252+ if state.Candidate == "" || entry.name < state.Candidate {
253+ state.Candidate = entry.name
136254 }
137255 }
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
256+ }
257+ if !done {
258+ next, err := encodeDeletionListingCursor(state)
259+ if err != nil {
260+ return ports.DeletionTombstonePage{}, err
261+ }
262+ if next.After == cursor.After {
263+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: deletion tombstone continuation did not advance")
143264 }
265+ return ports.DeletionTombstonePage{Next: next}, nil
144266 }
145- if state.candidate.name == "" {
146- r.deletionListing = nil
267+ if state.Candidate == "" {
147268 return ports.DeletionTombstonePage{Done: true}, nil
148269 }
149- path := filepath.Join(dir, state.candidate.name)
270+ path := filepath.Join(dir, state.Candidate)
150271 info, err := r.stat(path)
151272 if err != nil {
152273 return ports.DeletionTombstonePage{}, err
153274 }
154- charge := uint64(len(state.candidate.name))
275+ charge := uint64(len(state.Candidate))
155276 if info.Size() < 0 || uint64(info.Size()) > budget.Bytes || charge > budget.Bytes-uint64(info.Size()) {
156- r.deletionListing = nil
157277 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
158278 }
159279 if hook := r.hooks.beforeDeletionTombstoneRead; hook != nil {
@@ -167,16 +287,18 @@ func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.De
167287 if err != nil {
168288 return ports.DeletionTombstonePage{}, err
169289 }
170- if state.candidate.name != decoded.IncarnationID.String()+".tombstone" {
290+ if state.Candidate != decoded.IncarnationID.String()+".tombstone" {
171291 return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")
172292 }
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}
293+ next, err := encodeDeletionListingCursor(deletionListingState{After: state.Candidate})
294+ if err != nil {
295+ return ports.DeletionTombstonePage{}, err
178296 }
179- return page, nil
297+ return ports.DeletionTombstonePage{
298+ Tombstones: []domain.DeletionTombstone{decoded},
299+ Next: next,
300+ Done: state.Found == 1,
301+ }, nil
180302 }
181303
182304 func (r *Repository) DeleteDeletionTombstone(ctx context.Context, id domain.IncarnationID) error {
@@ -1,8 +1,11 @@
1 package snapshot1 package snapshot
2 2
3 import (3 import (
4+ "bytes"
4 "context"5 "context"
6+ "encoding/base64"
5 "encoding/binary"7 "encoding/binary"
8+ "encoding/json"
6 "errors"9 "errors"
7 "fmt"10 "fmt"
8 "io"11 "io"
@@ -84,76 +87,193 @@ func (r *Repository) WriteDeletionTombstone(ctx context.Context, tombstone domai
84 })87 })
85 }88 }
86 89
90+const (
91+ deletionListingCursorPrefix = "dt1:"
92+ maxDeletionListingCursor = 64 << 10
93+)
94+
87 type deletionListingState struct {95 type deletionListingState struct {
88- after string96+ After string `json:"after,omitempty"`
89- candidate maintenanceDirEntry97+ Offset int64 `json:"offset,omitempty"`
90- found uint6498+ Pending []maintenanceDirEntry `json:"pending,omitempty"`
91- scanDone bool99+ PendingAt int `json:"pending_at,omitempty"`
92- token string100+ ScanDone bool `json:"scan_done,omitempty"`
101+ Candidate string `json:"candidate,omitempty"`
102+ Found uint8 `json:"found,omitempty"`
103+}
104+
105+// MarshalJSON keeps maintenanceDirEntry private while making a continuation
106+// self-contained. Pending entries are bounded by one directory read buffer.
107+func (entry maintenanceDirEntry) MarshalJSON() ([]byte, error) {
108+ return json.Marshal(struct {
109+ Name string `json:"name"`
110+ IsDir bool `json:"is_dir,omitempty"`
111+ }{Name: entry.name, IsDir: entry.isDir})
112+}
113+
114+func (entry *maintenanceDirEntry) UnmarshalJSON(encoded []byte) error {
115+ var wire struct {
116+ Name string `json:"name"`
117+ IsDir bool `json:"is_dir,omitempty"`
118+ }
119+ decoder := json.NewDecoder(bytes.NewReader(encoded))
120+ decoder.DisallowUnknownFields()
121+ if err := decoder.Decode(&wire); err != nil {
122+ return err
123+ }
124+ entry.name, entry.isDir = wire.Name, wire.IsDir
125+ return nil
126+}
127+
128+func canonicalDeletionTombstoneFilename(name string) bool {
129+ if len(name) != 32+len(".tombstone") || !strings.HasSuffix(name, ".tombstone") {
130+ return false
131+ }
132+ var id domain.IncarnationID
133+ key := strings.TrimSuffix(name, ".tombstone")
134+ return id.UnmarshalText([]byte(key)) == nil && id.String()+".tombstone" == name
135+}
136+
137+func decodeDeletionListingCursor(cursor ports.DeletionTombstoneCursor) (deletionListingState, error) {
138+ if cursor.After == "" {
139+ return deletionListingState{}, nil
140+ }
141+ if len(cursor.After) > maxDeletionListingCursor || !strings.HasPrefix(cursor.After, deletionListingCursorPrefix) {
142+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation")
143+ }
144+ encoded, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(cursor.After, deletionListingCursorPrefix))
145+ if err != nil {
146+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation: %w", err)
147+ }
148+ var state deletionListingState
149+ decoder := json.NewDecoder(bytes.NewReader(encoded))
150+ decoder.DisallowUnknownFields()
151+ if err := decoder.Decode(&state); err != nil {
152+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation: %w", err)
153+ }
154+ if decoder.Decode(&struct{}{}) != io.EOF {
155+ return deletionListingState{}, fmt.Errorf("snapshot: malformed deletion tombstone continuation")
156+ }
157+ if err := validateDeletionListingState(state); err != nil {
158+ return deletionListingState{}, err
159+ }
160+ return state, nil
161+}
162+
163+func validateDeletionListingState(state deletionListingState) error {
164+ if state.After != "" && !canonicalDeletionTombstoneFilename(state.After) {
165+ return fmt.Errorf("snapshot: non-canonical deletion tombstone continuation key")
166+ }
167+ if state.Offset < 0 || state.PendingAt < 0 || state.PendingAt > len(state.Pending) || state.Found > 2 {
168+ return fmt.Errorf("snapshot: malformed deletion tombstone continuation")
169+ }
170+ if state.After == "" && state.Offset == 0 && len(state.Pending) == 0 && !state.ScanDone && state.Candidate == "" {
171+ return fmt.Errorf("snapshot: non-advancing deletion tombstone continuation")
172+ }
173+ if (state.Candidate == "") != (state.Found == 0) || (state.Candidate != "" && (!canonicalDeletionTombstoneFilename(state.Candidate) || state.Candidate <= state.After)) {
174+ return fmt.Errorf("snapshot: non-advancing deletion tombstone continuation")
175+ }
176+ if len(state.Pending) > maintenanceDirentBufferSize {
177+ return fmt.Errorf("snapshot: malformed deletion tombstone continuation")
178+ }
179+ for _, entry := range state.Pending {
180+ if entry.isDir || !canonicalDeletionTombstoneFilename(entry.name) {
181+ return fmt.Errorf("snapshot: non-canonical deletion tombstone continuation entry")
182+ }
183+ }
184+ return nil
185+}
186+
187+func encodeDeletionListingCursor(state deletionListingState) (ports.DeletionTombstoneCursor, error) {
188+ if err := validateDeletionListingState(state); err != nil {
189+ return ports.DeletionTombstoneCursor{}, err
190+ }
191+ encoded, err := json.Marshal(state)
192+ if err != nil {
193+ return ports.DeletionTombstoneCursor{}, err
194+ }
195+ return ports.DeletionTombstoneCursor{After: deletionListingCursorPrefix + base64.RawURLEncoding.EncodeToString(encoded)}, nil
196+}
197+
198+func (r *Repository) readDeletionListingEntries(dir string, limit int, state *deletionListingState) ([]maintenanceDirEntry, bool, error) {
199+ if state.PendingAt < len(state.Pending) {
200+ end := min(state.PendingAt+limit, len(state.Pending))
201+ entries := state.Pending[state.PendingAt:end]
202+ state.PendingAt = end
203+ done := state.ScanDone && end == len(state.Pending)
204+ return entries, done, nil
205+ }
206+ cursor := &maintenanceCursor{offset: state.Offset}
207+ entries, done, err := r.readMaintenanceDirent(dir, limit, cursor)
208+ state.Offset = cursor.offset
209+ state.Pending = cursor.pending
210+ state.PendingAt = 0
211+ state.ScanDone = done || cursor.done
212+ return entries, done && len(cursor.pending) == 0, err
93 }213 }
94 214
95 func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.DeletionTombstoneCursor, budget ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {215 func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.DeletionTombstoneCursor, budget ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
96 if budget.Entries == 0 || budget.Bytes == 0 {216 if budget.Entries == 0 || budget.Bytes == 0 {
97 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall217 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
98 }218 }
219+ if err := ctx.Err(); err != nil {
220+ return ports.DeletionTombstonePage{}, err
221+ }
222+ state, err := decodeDeletionListingCursor(cursor)
223+ if err != nil {
224+ return ports.DeletionTombstonePage{}, err
225+ }
99 dir := filepath.Join(r.dir, deletionTombstonesDir)226 dir := filepath.Join(r.dir, deletionTombstonesDir)
100 r.maintenanceMu.Lock()227 r.maintenanceMu.Lock()
101 defer r.maintenanceMu.Unlock()228 defer r.maintenanceMu.Unlock()
102 229
103- state := r.deletionListing230+ entryLimit := budget.Entries
104- if state == nil || (cursor.After != state.token && cursor.After != state.after) {231+ if maxInt := uint64(^uint(0) >> 1); entryLimit > maxInt {
105- state = &deletionListingState{after: cursor.After}232+ entryLimit = maxInt
106- r.deletionListing = state
107- delete(r.maintenanceCursors, maintenanceCursorID(dir, "deletion-tombstones"))
108 }233 }
109- if !state.scanDone {234+ entries, done, err := r.readDeletionListingEntries(dir, int(entryLimit), &state)
110- entryLimit := budget.Entries235+ if errors.Is(err, os.ErrNotExist) {
111- if maxInt := uint64(^uint(0) >> 1); entryLimit > maxInt {236+ return ports.DeletionTombstonePage{Done: true}, nil
112- entryLimit = maxInt237+ }
113- }238+ if err != nil {
114- entries, done, err := r.readMaintenanceDir(dir, int(entryLimit), "deletion-tombstones")239+ return ports.DeletionTombstonePage{}, err
115- if errors.Is(err, os.ErrNotExist) {240+ }
116- r.deletionListing = nil241+ for _, entry := range entries {
117- return ports.DeletionTombstonePage{Done: true}, nil242+ if err := ctx.Err(); err != nil {
118- }
119- if err != nil {
120- r.deletionListing = nil
121 return ports.DeletionTombstonePage{}, err243 return ports.DeletionTombstonePage{}, err
122 }244 }
123- for _, entry := range entries {245+ if entry.isDir || !canonicalDeletionTombstoneFilename(entry.name) {
124- if err := ctx.Err(); err != nil {246+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
125- return ports.DeletionTombstonePage{}, err247+ }
126- }248+ if entry.name > state.After {
127- if entry.isDir || !strings.HasSuffix(entry.name, ".tombstone") {249+ if state.Found < 2 {
128- r.deletionListing = nil250+ state.Found++
129- return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: invalid deletion tombstone entry")
130 }251 }
131- if entry.name > state.after {252+ if state.Candidate == "" || entry.name < state.Candidate {
132- state.found++253+ state.Candidate = entry.name
133- if state.candidate.name == "" || entry.name < state.candidate.name {
134- state.candidate = entry
135- }
136 }254 }
137 }255 }
138- state.scanDone = done256+ }
139- if !done {257+ if !done {
140- r.deletionListingToken++258+ next, err := encodeDeletionListingCursor(state)
141- state.token = fmt.Sprintf("@%d", r.deletionListingToken)259+ if err != nil {
142- return ports.DeletionTombstonePage{Next: ports.DeletionTombstoneCursor{After: state.token}}, nil260+ return ports.DeletionTombstonePage{}, err
261+ }
262+ if next.After == cursor.After {
263+ return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: deletion tombstone continuation did not advance")
143 }264 }
265+ return ports.DeletionTombstonePage{Next: next}, nil
144 }266 }
145- if state.candidate.name == "" {267+ if state.Candidate == "" {
146- r.deletionListing = nil
147 return ports.DeletionTombstonePage{Done: true}, nil268 return ports.DeletionTombstonePage{Done: true}, nil
148 }269 }
149- path := filepath.Join(dir, state.candidate.name)270+ path := filepath.Join(dir, state.Candidate)
150 info, err := r.stat(path)271 info, err := r.stat(path)
151 if err != nil {272 if err != nil {
152 return ports.DeletionTombstonePage{}, err273 return ports.DeletionTombstonePage{}, err
153 }274 }
154- charge := uint64(len(state.candidate.name))275+ charge := uint64(len(state.Candidate))
155 if info.Size() < 0 || uint64(info.Size()) > budget.Bytes || charge > budget.Bytes-uint64(info.Size()) {276 if info.Size() < 0 || uint64(info.Size()) > budget.Bytes || charge > budget.Bytes-uint64(info.Size()) {
156- r.deletionListing = nil
157 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall277 return ports.DeletionTombstonePage{}, ErrMaintenanceBudgetTooSmall
158 }278 }
159 if hook := r.hooks.beforeDeletionTombstoneRead; hook != nil {279 if hook := r.hooks.beforeDeletionTombstoneRead; hook != nil {
@@ -167,16 +287,18 @@ func (r *Repository) ListDeletionTombstones(ctx context.Context, cursor ports.De
167 if err != nil {287 if err != nil {
168 return ports.DeletionTombstonePage{}, err288 return ports.DeletionTombstonePage{}, err
169 }289 }
170- if state.candidate.name != decoded.IncarnationID.String()+".tombstone" {290+ if state.Candidate != decoded.IncarnationID.String()+".tombstone" {
171 return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")291 return ports.DeletionTombstonePage{}, fmt.Errorf("snapshot: non-canonical deletion tombstone filename")
172 }292 }
173- page := ports.DeletionTombstonePage{Tombstones: []domain.DeletionTombstone{decoded}, Next: ports.DeletionTombstoneCursor{After: state.candidate.name}, Done: state.found == 1}293+ next, err := encodeDeletionListingCursor(deletionListingState{After: state.Candidate})
174- if page.Done {294+ if err != nil {
175- r.deletionListing = nil295+ return ports.DeletionTombstonePage{}, err
176- } else {
177- r.deletionListing = &deletionListingState{after: state.candidate.name}
178 }296 }
179- return page, nil297+ return ports.DeletionTombstonePage{
298+ Tombstones: []domain.DeletionTombstone{decoded},
299+ Next: next,
300+ Done: state.Found == 1,
301+ }, nil
180 }302 }
181 303
182 func (r *Repository) DeleteDeletionTombstone(ctx context.Context, id domain.IncarnationID) error {304 func (r *Repository) DeleteDeletionTombstone(ctx context.Context, id domain.IncarnationID) error {
M internal/adapters/snapshot/maintenance_delete_test.go +78 -0
@@ -2,9 +2,11 @@ package snapshot
22
33 import (
44 "context"
5+ "encoding/base64"
56 "encoding/hex"
67 "errors"
78 "path/filepath"
9+ "slices"
810 "testing"
911
1012 "github.com/bnema/vev/internal/domain"
@@ -68,6 +70,82 @@ func TestDeletionTombstoneCodec(t *testing.T) {
6870 }
6971 }
7072
73+func TestDeletionTombstoneListingResumesAcrossRepositoryRecreation(t *testing.T) {
74+ dir := privateDir(t)
75+ ctx := context.Background()
76+ want := []domain.DeletionTombstone{
77+ {Name: "first", IncarnationID: domain.IncarnationID{15: 1}},
78+ {Name: "second", IncarnationID: domain.IncarnationID{15: 2}},
79+ {Name: "third", IncarnationID: domain.IncarnationID{15: 3}},
80+ }
81+ repo := NewRepository(dir)
82+ for _, index := range []int{2, 0, 1} {
83+ if err := repo.WriteDeletionTombstone(ctx, want[index]); err != nil {
84+ t.Fatal(err)
85+ }
86+ }
87+
88+ var got []domain.DeletionTombstone
89+ cursor := ports.DeletionTombstoneCursor{}
90+ reads := 0
91+ for calls := 0; ; calls++ {
92+ if calls == 32 {
93+ t.Fatal("listing did not complete")
94+ }
95+ // A new Repository has no in-memory scan state. Every continuation must
96+ // therefore carry everything needed to resume the bounded directory scan.
97+ repo = NewRepository(dir)
98+ repo.hooks.beforeDeletionTombstoneRead = func(string) { reads++ }
99+ page, err := repo.ListDeletionTombstones(ctx, cursor, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})
100+ if err != nil {
101+ t.Fatal(err)
102+ }
103+ got = append(got, page.Tombstones...)
104+ if page.Done {
105+ break
106+ }
107+ if page.Next.After == "" || page.Next.After == cursor.After {
108+ t.Fatalf("continuation did not advance: before=%q after=%q", cursor.After, page.Next.After)
109+ }
110+ cursor = page.Next
111+ }
112+ if !slices.Equal(got, want) {
113+ t.Fatalf("tombstones = %#v, want %#v", got, want)
114+ }
115+ if reads != len(want) {
116+ t.Fatalf("tombstone object reads = %d, want %d", reads, len(want))
117+ }
118+}
119+
120+func TestDeletionTombstoneListingRejectsMalformedContinuation(t *testing.T) {
121+ repo := NewRepository(privateDir(t))
122+ ctx := context.Background()
123+ after := domain.IncarnationID{15: 2}.String() + ".tombstone"
124+
125+ tests := []struct {
126+ name string
127+ cursor ports.DeletionTombstoneCursor
128+ }{
129+ {name: "repository-memory token", cursor: ports.DeletionTombstoneCursor{After: "@1"}},
130+ {name: "non-advancing empty state", cursor: rawDeletionListingCursor(t, `{}`)},
131+ {name: "non-canonical key", cursor: rawDeletionListingCursor(t, `{"after":"ABC.tombstone"}`)},
132+ {name: "candidate does not advance", cursor: rawDeletionListingCursor(t, `{"after":"`+after+`","candidate":"`+after+`","found":1}`)},
133+ {name: "unknown field", cursor: rawDeletionListingCursor(t, `{"unknown":1}`)},
134+ }
135+ for _, test := range tests {
136+ t.Run(test.name, func(t *testing.T) {
137+ if _, err := repo.ListDeletionTombstones(ctx, test.cursor, ports.MaintenanceBudget{Entries: 1, Bytes: 1024}); err == nil {
138+ t.Fatal("malformed continuation accepted")
139+ }
140+ })
141+ }
142+}
143+
144+func rawDeletionListingCursor(t *testing.T, state string) ports.DeletionTombstoneCursor {
145+ t.Helper()
146+ return ports.DeletionTombstoneCursor{After: deletionListingCursorPrefix + base64.RawURLEncoding.EncodeToString([]byte(state))}
147+}
148+
71149 func TestDeletionTombstoneListing(t *testing.T) {
72150 repo := NewRepository(privateDir(t))
73151 ctx := context.Background()
@@ -2,9 +2,11 @@ package snapshot
2 2
3 import (3 import (
4 "context"4 "context"
5+ "encoding/base64"
5 "encoding/hex"6 "encoding/hex"
6 "errors"7 "errors"
7 "path/filepath"8 "path/filepath"
9+ "slices"
8 "testing"10 "testing"
9 11
10 "github.com/bnema/vev/internal/domain"12 "github.com/bnema/vev/internal/domain"
@@ -68,6 +70,82 @@ func TestDeletionTombstoneCodec(t *testing.T) {
68 }70 }
69 }71 }
70 72
73+func TestDeletionTombstoneListingResumesAcrossRepositoryRecreation(t *testing.T) {
74+ dir := privateDir(t)
75+ ctx := context.Background()
76+ want := []domain.DeletionTombstone{
77+ {Name: "first", IncarnationID: domain.IncarnationID{15: 1}},
78+ {Name: "second", IncarnationID: domain.IncarnationID{15: 2}},
79+ {Name: "third", IncarnationID: domain.IncarnationID{15: 3}},
80+ }
81+ repo := NewRepository(dir)
82+ for _, index := range []int{2, 0, 1} {
83+ if err := repo.WriteDeletionTombstone(ctx, want[index]); err != nil {
84+ t.Fatal(err)
85+ }
86+ }
87+
88+ var got []domain.DeletionTombstone
89+ cursor := ports.DeletionTombstoneCursor{}
90+ reads := 0
91+ for calls := 0; ; calls++ {
92+ if calls == 32 {
93+ t.Fatal("listing did not complete")
94+ }
95+ // A new Repository has no in-memory scan state. Every continuation must
96+ // therefore carry everything needed to resume the bounded directory scan.
97+ repo = NewRepository(dir)
98+ repo.hooks.beforeDeletionTombstoneRead = func(string) { reads++ }
99+ page, err := repo.ListDeletionTombstones(ctx, cursor, ports.MaintenanceBudget{Entries: 1, Bytes: 1024})
100+ if err != nil {
101+ t.Fatal(err)
102+ }
103+ got = append(got, page.Tombstones...)
104+ if page.Done {
105+ break
106+ }
107+ if page.Next.After == "" || page.Next.After == cursor.After {
108+ t.Fatalf("continuation did not advance: before=%q after=%q", cursor.After, page.Next.After)
109+ }
110+ cursor = page.Next
111+ }
112+ if !slices.Equal(got, want) {
113+ t.Fatalf("tombstones = %#v, want %#v", got, want)
114+ }
115+ if reads != len(want) {
116+ t.Fatalf("tombstone object reads = %d, want %d", reads, len(want))
117+ }
118+}
119+
120+func TestDeletionTombstoneListingRejectsMalformedContinuation(t *testing.T) {
121+ repo := NewRepository(privateDir(t))
122+ ctx := context.Background()
123+ after := domain.IncarnationID{15: 2}.String() + ".tombstone"
124+
125+ tests := []struct {
126+ name string
127+ cursor ports.DeletionTombstoneCursor
128+ }{
129+ {name: "repository-memory token", cursor: ports.DeletionTombstoneCursor{After: "@1"}},
130+ {name: "non-advancing empty state", cursor: rawDeletionListingCursor(t, `{}`)},
131+ {name: "non-canonical key", cursor: rawDeletionListingCursor(t, `{"after":"ABC.tombstone"}`)},
132+ {name: "candidate does not advance", cursor: rawDeletionListingCursor(t, `{"after":"`+after+`","candidate":"`+after+`","found":1}`)},
133+ {name: "unknown field", cursor: rawDeletionListingCursor(t, `{"unknown":1}`)},
134+ }
135+ for _, test := range tests {
136+ t.Run(test.name, func(t *testing.T) {
137+ if _, err := repo.ListDeletionTombstones(ctx, test.cursor, ports.MaintenanceBudget{Entries: 1, Bytes: 1024}); err == nil {
138+ t.Fatal("malformed continuation accepted")
139+ }
140+ })
141+ }
142+}
143+
144+func rawDeletionListingCursor(t *testing.T, state string) ports.DeletionTombstoneCursor {
145+ t.Helper()
146+ return ports.DeletionTombstoneCursor{After: deletionListingCursorPrefix + base64.RawURLEncoding.EncodeToString([]byte(state))}
147+}
148+
71 func TestDeletionTombstoneListing(t *testing.T) {149 func TestDeletionTombstoneListing(t *testing.T) {
72 repo := NewRepository(privateDir(t))150 repo := NewRepository(privateDir(t))
73 ctx := context.Background()151 ctx := context.Background()
M internal/adapters/snapshot/repository.go +0 -2
@@ -40,8 +40,6 @@ type Repository struct {
4040 maintenanceCursors map[string]*maintenanceCursor
4141 maintenanceSessions map[string]*sessionMaintenance
4242 maintenanceQuarantine *quarantineMaintenance
43- deletionListing *deletionListingState
44- deletionListingToken uint64
4543
4644 // pendingLegacySync records an unlink whose root-directory sync failed.
4745 // It is keyed by the deterministic legacy filename and shares a per-file
@@ -40,8 +40,6 @@ 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
45 43
46 // pendingLegacySync records an unlink whose root-directory sync failed.44 // pendingLegacySync records an unlink whose root-directory sync failed.
47 // It is keyed by the deterministic legacy filename and shares a per-file45 // It is keyed by the deterministic legacy filename and shares a per-file

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