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 the startup GC snapshot and collection with every
89 // catalogue or checkpoint mutation. Startup normally provides this fence by
90 // running before socket publication; retaining it here also makes the
91 // coordinator safe when used directly by a composition root.
92 mutationMu sync.Mutex
93 random io.Reader
94}
95
96func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, random io.Reader) *Coordinator {
97 return &Coordinator{catalogue: catalogue, repository: repository, locks: NewKeyLocks(), random: random}
98}
99
100// Create commits fresh catalogue metadata before the caller creates or exposes
101// any runtime session.
102func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord) (domain.CatalogueRecord, error) {
103 if c == nil || c.catalogue == nil || c.locks == nil || c.random == nil {
104 return domain.CatalogueRecord{}, errors.New("recovery: incomplete create dependencies")
105 }
106 if err := ctx.Err(); err != nil {
107 return domain.CatalogueRecord{}, err
108 }
109 c.mutationMu.Lock()
110 defer c.mutationMu.Unlock()
111 unlock := c.locks.Lock([]string{record.Name})
112 defer unlock()
113 if _, exists, err := c.catalogue.Record(record.Name); err != nil {
114 return domain.CatalogueRecord{}, err
115 } else if exists {
116 return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", record.Name)
117 }
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 record.Committed = nil
124 record.DegradedReason = ""
125 if err := record.Validate(); err != nil {
126 return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fresh record: %w", err)
127 }
128 if err := c.catalogue.Create(record); err != nil {
129 return domain.CatalogueRecord{}, err
130 }
131 return record, nil
132}
133
134// Rename atomically moves one catalogue key while retaining the immutable
135// incarnation and all checkpoint references.
136func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (domain.CatalogueRecord, error) {
137 if c == nil || c.catalogue == nil || c.locks == nil {
138 return domain.CatalogueRecord{}, errors.New("recovery: incomplete rename dependencies")
139 }
140 if err := ctx.Err(); err != nil {
141 return domain.CatalogueRecord{}, err
142 }
143 c.mutationMu.Lock()
144 defer c.mutationMu.Unlock()
145 unlock := c.locks.Lock([]string{oldName, newName})
146 defer unlock()
147 record, ok, err := c.catalogue.Record(oldName)
148 if err != nil {
149 return domain.CatalogueRecord{}, err
150 }
151 if !ok {
152 return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q not found", oldName)
153 }
154 if existing, exists, err := c.catalogue.Record(newName); err != nil {
155 return domain.CatalogueRecord{}, err
156 } else if exists && (newName != oldName || existing.IncarnationID != record.IncarnationID) {
157 return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", newName)
158 }
159 record.Name = newName
160 if err := record.Validate(); err != nil {
161 return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid rename: %w", err)
162 }
163 if oldName != newName {
164 if err := c.catalogue.Rename(oldName, record); err != nil {
165 return domain.CatalogueRecord{}, err
166 }
167 }
168 return record, nil
169}
170
171// Delete removes the catalogue record before deleting its incarnation. A crash
172// between those steps leaves an orphan that startup garbage collection removes.
173func (c *Coordinator) Delete(ctx context.Context, name string) error {
174 if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
175 return errors.New("recovery: incomplete delete dependencies")
176 }
177 if err := ctx.Err(); err != nil {
178 return err
179 }
180 c.mutationMu.Lock()
181 defer c.mutationMu.Unlock()
182 unlock := c.locks.Lock([]string{name})
183 defer unlock()
184 record, ok, err := c.catalogue.Record(name)
185 if err != nil {
186 return err
187 }
188 if !ok {
189 return nil
190 }
191 if err := c.catalogue.Delete(record.Name); err != nil {
192 return err
193 }
194 return c.repository.DeleteIncarnation(ctx, record.IncarnationID)
195}
196
197// CollectGarbage takes a catalogue snapshot and applies retention while all
198// catalogue mutations and checkpoint publications are fenced. Its keep map is
199// therefore never stale relative to an operation that creates an incarnation
200// or commits a new generation.
201func (c *Coordinator) CollectGarbage(ctx context.Context) (int, error) {
202 if c == nil || c.catalogue == nil || c.repository == nil {
203 return 0, errors.New("recovery: incomplete garbage collection dependencies")
204 }
205 if err := ctx.Err(); err != nil {
206 return 0, err
207 }
208 c.mutationMu.Lock()
209 defer c.mutationMu.Unlock()
210
211 records, err := c.catalogue.Records()
212 if err != nil {
213 return 0, err
214 }
215 keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
216 for _, record := range records {
217 if record.Committed != nil {
218 keep[record.IncarnationID] = *record.Committed
219 continue
220 }
221 keep[record.IncarnationID] = domain.CheckpointRef{}
222 }
223 if err := c.repository.CollectGarbage(ctx, keep); err != nil {
224 return len(keep), err
225 }
226 return len(keep), nil
227}