1// Package recovery coordinates idempotent operations across the catalogue and
2// snapshot repository without depending on filesystem adapters.
3package recovery
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "sort"
11 "sync"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15)
16
17type KeyLocks struct {
18 mu sync.Mutex
19 refs map[string]*keyLockRef
20}
21
22type keyLockRef struct {
23 mu sync.Mutex
24 refs int
25}
26
27func NewKeyLocks() *KeyLocks {
28 return &KeyLocks{refs: make(map[string]*keyLockRef)}
29}
30
31// Lock acquires each distinct stable key in lexical order. The returned
32// function releases in reverse order and removes unused lock entries.
33func (k *KeyLocks) Lock(names []string) func() {
34 if k == nil {
35 return func() {}
36 }
37 ordered := append([]string(nil), names...)
38 sort.Strings(ordered)
39 ordered = compactStrings(ordered)
40
41 refs := make([]*keyLockRef, len(ordered))
42 k.mu.Lock()
43 for i, name := range ordered {
44 ref := k.refs[name]
45 if ref == nil {
46 ref = &keyLockRef{}
47 k.refs[name] = ref
48 }
49 ref.refs++
50 refs[i] = ref
51 }
52 k.mu.Unlock()
53 for _, ref := range refs {
54 ref.mu.Lock()
55 }
56 return func() {
57 for i := len(refs) - 1; i >= 0; i-- {
58 refs[i].mu.Unlock()
59 }
60 k.mu.Lock()
61 for i, name := range ordered {
62 refs[i].refs--
63 if refs[i].refs == 0 {
64 delete(k.refs, name)
65 }
66 }
67 k.mu.Unlock()
68 }
69}
70
71func compactStrings(values []string) []string {
72 if len(values) == 0 {
73 return values
74 }
75 out := values[:1]
76 for _, value := range values[1:] {
77 if value != out[len(out)-1] {
78 out = append(out, value)
79 }
80 }
81 return out
82}
83
84type Coordinator struct {
85 catalogue ports.Catalogue
86 repository ports.SnapshotRepository
87 locks *KeyLocks
88 // mutationMu serializes every catalogue or checkpoint mutation with each
89 // per-incarnation GC step. It is the only fence: GC runs while the daemon
90 // serves, so a step holds it for one incarnation at a time.
91 mutationMu sync.Mutex
92 random io.Reader
93}
94
95func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, random io.Reader) *Coordinator {
96 return &Coordinator{catalogue: catalogue, repository: repository, locks: NewKeyLocks(), random: random}
97}
98
99// Create commits fresh catalogue metadata before the caller creates or exposes
100// any runtime session.
101func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord) (domain.CatalogueRecord, error) {
102 if c == nil || c.catalogue == nil || c.locks == nil || c.random == nil {
103 return domain.CatalogueRecord{}, errors.New("recovery: incomplete create dependencies")
104 }
105 if err := ctx.Err(); err != nil {
106 return domain.CatalogueRecord{}, err
107 }
108 c.mutationMu.Lock()
109 defer c.mutationMu.Unlock()
110 unlock := c.locks.Lock([]string{record.Name})
111 defer unlock()
112 if _, exists, err := c.catalogue.Record(record.Name); err != nil {
113 return domain.CatalogueRecord{}, err
114 } else if exists {
115 return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", record.Name)
116 }
117 if record.IncarnationID == (domain.IncarnationID{}) {
118 id, err := domain.NewIncarnationID(c.random)
119 if err != nil {
120 return domain.CatalogueRecord{}, fmt.Errorf("recovery: generate incarnation: %w", err)
121 }
122 record.IncarnationID = id
123 }
124 record.Committed = nil
125 record.DegradedReason = ""
126 if err := record.Validate(); err != nil {
127 return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fresh record: %w", err)
128 }
129 if err := c.catalogue.Create(record); err != nil {
130 return domain.CatalogueRecord{}, err
131 }
132 return record, nil
133}
134
135// Rename atomically moves one catalogue key while retaining the immutable
136// incarnation and all checkpoint references.
137func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (domain.CatalogueRecord, error) {
138 if c == nil || c.catalogue == nil || c.locks == nil {
139 return domain.CatalogueRecord{}, errors.New("recovery: incomplete rename dependencies")
140 }
141 if err := ctx.Err(); err != nil {
142 return domain.CatalogueRecord{}, err
143 }
144 c.mutationMu.Lock()
145 defer c.mutationMu.Unlock()
146 unlock := c.locks.Lock([]string{oldName, newName})
147 defer unlock()
148 record, ok, err := c.catalogue.Record(oldName)
149 if err != nil {
150 return domain.CatalogueRecord{}, err
151 }
152 if !ok {
153 return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q not found", oldName)
154 }
155 if existing, exists, err := c.catalogue.Record(newName); err != nil {
156 return domain.CatalogueRecord{}, err
157 } else if exists && (newName != oldName || existing.IncarnationID != record.IncarnationID) {
158 return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", newName)
159 }
160 record.Name = newName
161 if err := record.Validate(); err != nil {
162 return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid rename: %w", err)
163 }
164 if oldName != newName {
165 if err := c.catalogue.Rename(oldName, record); err != nil {
166 return domain.CatalogueRecord{}, err
167 }
168 }
169 return record, nil
170}
171
172// Delete removes the current catalogue record before deleting its incarnation.
173// Callers that captured lifecycle identity must use DeleteExact.
174func (c *Coordinator) Delete(ctx context.Context, name string) error {
175 if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
176 return errors.New("recovery: incomplete delete dependencies")
177 }
178 if err := ctx.Err(); err != nil {
179 return err
180 }
181 c.mutationMu.Lock()
182 defer c.mutationMu.Unlock()
183 unlock := c.locks.Lock([]string{name})
184 defer unlock()
185 record, ok, err := c.catalogue.Record(name)
186 if err != nil {
187 return err
188 }
189 if !ok {
190 return nil
191 }
192 if err := c.catalogue.Delete(record.Name); err != nil {
193 return err
194 }
195 return c.repository.DeleteIncarnation(ctx, record.IncarnationID)
196}
197
198// DeleteExact removes a catalogue record only while it still identifies the
199// lifecycle captured by the caller. A stale teardown cannot delete a same-name
200// replacement created after that capture.
201func (c *Coordinator) DeleteExact(ctx context.Context, name string, incarnation domain.IncarnationID, createdAt int64) error {
202 if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
203 return errors.New("recovery: incomplete exact delete dependencies")
204 }
205 if err := ctx.Err(); err != nil {
206 return err
207 }
208 c.mutationMu.Lock()
209 defer c.mutationMu.Unlock()
210 unlock := c.locks.Lock([]string{name})
211 defer unlock()
212 record, ok, err := c.catalogue.Record(name)
213 if err != nil {
214 return err
215 }
216 if !ok || record.IncarnationID != incarnation || record.CreatedAt != createdAt {
217 return nil
218 }
219 if err := c.catalogue.Delete(name); err != nil {
220 return err
221 }
222 return c.repository.DeleteIncarnation(ctx, incarnation)
223}
224
225// errGarbageCollectionCatalogue stops a pass: without a readable catalogue no
226// incarnation can be classified.
227var errGarbageCollectionCatalogue = errors.New("recovery: read catalogue for garbage collection")
228
229// CollectGarbage applies retention to every repository incarnation. Each
230// incarnation is collected in its own fenced step that re-reads the catalogue
231// under mutationMu, so its keep decision is never stale relative to an
232// operation that creates an incarnation or commits a generation, and a
233// mutation waits for at most one incarnation instead of a whole pass. Failures
234// are joined so one bad incarnation does not stop the rest.
235func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
236 if c == nil || c.catalogue == nil || c.repository == nil {
237 return 0, errors.New("recovery: incomplete garbage collection dependencies")
238 }
239 if err := ctx.Err(); err != nil {
240 return 0, err
241 }
242 ids, err := c.repository.SnapshotIncarnations(ctx)
243 if err != nil {
244 return 0, err
245 }
246 var collected error
247 for _, id := range ids {
248 if err := ctx.Err(); err != nil {
249 return len(ids), errors.Join(collected, err)
250 }
251 if err := c.collectIncarnation(ctx, id); errors.Is(err, errGarbageCollectionCatalogue) {
252 return len(ids), errors.Join(collected, err)
253 } else if err != nil {
254 collected = errors.Join(collected, err)
255 }
256 }
257 return len(ids), collected
258}
259
260func (c *Coordinator) collectIncarnation(ctx context.Context, id domain.IncarnationID) error {
261 c.mutationMu.Lock()
262 defer c.mutationMu.Unlock()
263 // A catalogue read failure never produces a keep decision, so destructive
264 // collection cannot run without known-good catalogue state.
265 records, err := c.catalogue.Records()
266 if err != nil {
267 return fmt.Errorf("%w: %w", errGarbageCollectionCatalogue, err)
268 }
269 var keep *domain.CheckpointRef
270 for _, record := range records {
271 if record.IncarnationID != id {
272 continue
273 }
274 keep = &domain.CheckpointRef{}
275 if record.Committed != nil {
276 *keep = *record.Committed
277 }
278 break
279 }
280 return c.repository.CollectIncarnationGarbage(ctx, id, keep)
281}