1package recovery
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "sync"
8 "testing"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/stretchr/testify/require"
13)
14
15type transactionCatalogue struct {
16 mu sync.Mutex
17 records map[string]domain.CatalogueRecord
18 events []string
19 replaceErrOnce error
20 onCreate func(domain.CatalogueRecord)
21}
22
23func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {
24 catalogue := &transactionCatalogue{records: make(map[string]domain.CatalogueRecord)}
25 for _, record := range records {
26 catalogue.records[record.Name] = record
27 }
28 return catalogue
29}
30
31func (c *transactionCatalogue) Records() ([]domain.CatalogueRecord, error) {
32 c.mu.Lock()
33 defer c.mu.Unlock()
34 out := make([]domain.CatalogueRecord, 0, len(c.records))
35 for _, record := range c.records {
36 out = append(out, record)
37 }
38 return out, nil
39}
40
41func (c *transactionCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
42 c.mu.Lock()
43 defer c.mu.Unlock()
44 record, ok := c.records[name]
45 return record, ok, nil
46}
47
48func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
49 c.mu.Lock()
50 defer c.mu.Unlock()
51 if _, exists := c.records[record.Name]; exists {
52 return errors.New("record exists")
53 }
54 c.records[record.Name] = record
55 c.events = append(c.events, "create")
56 if c.onCreate != nil {
57 c.onCreate(record)
58 }
59 return nil
60}
61
62func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }
63func (c *transactionCatalogue) Sync() error { return nil }
64
65func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {
66 c.mu.Lock()
67 defer c.mu.Unlock()
68 if c.replaceErrOnce != nil {
69 err := c.replaceErrOnce
70 c.replaceErrOnce = nil
71 return err
72 }
73 if _, exists := c.records[name]; !exists {
74 return errors.New("record missing")
75 }
76 c.records[name] = record
77 c.events = append(c.events, "replace:"+name)
78 return nil
79}
80
81func (c *transactionCatalogue) Rename(oldName string, record domain.CatalogueRecord) error {
82 c.mu.Lock()
83 defer c.mu.Unlock()
84 if _, exists := c.records[oldName]; !exists {
85 return errors.New("record missing")
86 }
87 if _, exists := c.records[record.Name]; exists && record.Name != oldName {
88 return errors.New("record exists")
89 }
90 delete(c.records, oldName)
91 c.records[record.Name] = record
92 c.events = append(c.events, "rename:"+oldName+":"+record.Name)
93 return nil
94}
95
96func (c *transactionCatalogue) Delete(name string) error {
97 c.mu.Lock()
98 defer c.mu.Unlock()
99 delete(c.records, name)
100 c.events = append(c.events, "delete:"+name)
101 return nil
102}
103
104func (c *transactionCatalogue) Close() error { return nil }
105
106type transactionRepository struct {
107 ports.SnapshotRepository
108 mu sync.Mutex
109 deleteErrOnce error
110 deleteCalls []domain.IncarnationID
111 deletedIDs []domain.IncarnationID
112 deleteEntered chan struct{}
113 releaseDelete chan struct{}
114}
115
116type garbageCollectionFenceRepository struct {
117 ports.SnapshotRepository
118 mu sync.Mutex
119 events []string
120 collectionEntered chan struct{}
121 releaseCollection chan struct{}
122 collectionFinished bool
123 publishBeforeGC bool
124 incarnations map[domain.IncarnationID]struct{}
125 generations map[domain.IncarnationID]map[uint64]struct{}
126}
127
128func (r *garbageCollectionFenceRepository) CollectGarbage(_ context.Context, keep map[domain.IncarnationID]domain.CheckpointRef) error {
129 close(r.collectionEntered)
130 <-r.releaseCollection
131 r.mu.Lock()
132 defer r.mu.Unlock()
133 for id := range r.incarnations {
134 if _, known := keep[id]; !known {
135 delete(r.incarnations, id)
136 }
137 }
138 for id, generations := range r.generations {
139 committed, known := keep[id]
140 if !known {
141 delete(r.generations, id)
142 continue
143 }
144 for generation := range generations {
145 if generation != committed.Generation && generation+1 != committed.Generation {
146 delete(generations, generation)
147 }
148 }
149 }
150 r.collectionFinished = true
151 r.events = append(r.events, "collect")
152 return nil
153}
154
155func (r *garbageCollectionFenceRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
156 r.mu.Lock()
157 defer r.mu.Unlock()
158 if !r.collectionFinished {
159 r.publishBeforeGC = true
160 }
161 if r.generations == nil {
162 r.generations = make(map[domain.IncarnationID]map[uint64]struct{})
163 }
164 if r.generations[publication.IncarnationID] == nil {
165 r.generations[publication.IncarnationID] = make(map[uint64]struct{})
166 }
167 r.generations[publication.IncarnationID][publication.Generation] = struct{}{}
168 r.events = append(r.events, "publish")
169 return nil
170}
171
172func (r *garbageCollectionFenceRepository) addIncarnation(record domain.CatalogueRecord) {
173 r.mu.Lock()
174 defer r.mu.Unlock()
175 if r.incarnations == nil {
176 r.incarnations = make(map[domain.IncarnationID]struct{})
177 }
178 r.incarnations[record.IncarnationID] = struct{}{}
179}
180
181func (r *garbageCollectionFenceRepository) hasIncarnation(id domain.IncarnationID) bool {
182 r.mu.Lock()
183 defer r.mu.Unlock()
184 _, ok := r.incarnations[id]
185 return ok
186}
187
188func (r *garbageCollectionFenceRepository) hasGeneration(id domain.IncarnationID, generation uint64) bool {
189 r.mu.Lock()
190 defer r.mu.Unlock()
191 _, ok := r.generations[id][generation]
192 return ok
193}
194
195func (r *garbageCollectionFenceRepository) eventLog() ([]string, bool) {
196 r.mu.Lock()
197 defer r.mu.Unlock()
198 return append([]string(nil), r.events...), r.publishBeforeGC
199}
200
201func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
202 if r.deleteEntered != nil {
203 close(r.deleteEntered)
204 <-r.releaseDelete
205 }
206 r.mu.Lock()
207 defer r.mu.Unlock()
208 r.deleteCalls = append(r.deleteCalls, id)
209 if r.deleteErrOnce != nil {
210 err := r.deleteErrOnce
211 r.deleteErrOnce = nil
212 return err
213 }
214 r.deletedIDs = append(r.deletedIDs, id)
215 return nil
216}
217
218func degradedTransactionRecord() domain.CatalogueRecord {
219 return domain.CatalogueRecord{
220 Name: "broken", IncarnationID: domain.IncarnationID{1}, Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
221 DegradedReason: "checkpoint unreadable",
222 }
223}
224
225func healthyTransactionRecord() domain.CatalogueRecord {
226 return domain.CatalogueRecord{
227 Name: "work",
228 IncarnationID: domain.IncarnationID{1},
229 Cwd: "/workspace",
230 CreatedAt: 11,
231 UpdatedAt: 22,
232 LastUsedSeq: 33,
233 TabNames: []string{"shell", "logs"},
234 Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
235 }
236}
237
238func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
239 t.Parallel()
240 old := healthyTransactionRecord()
241 catalogue := newTransactionCatalogue(old)
242 repository := &transactionRepository{}
243 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
244
245 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
246
247 require.NoError(t, err)
248 require.True(t, committed)
249 require.NotZero(t, fresh.IncarnationID)
250 require.NotEqual(t, old.IncarnationID, fresh.IncarnationID)
251 require.Equal(t, old.Name, fresh.Name)
252 require.Equal(t, old.Cwd, fresh.Cwd)
253 require.Equal(t, old.CreatedAt, fresh.CreatedAt)
254 require.Equal(t, old.UpdatedAt, fresh.UpdatedAt)
255 require.Equal(t, old.LastUsedSeq, fresh.LastUsedSeq)
256 require.Empty(t, fresh.TabNames)
257 require.Nil(t, fresh.Committed)
258 require.Empty(t, fresh.DegradedReason)
259 stored, ok, recordErr := catalogue.Record(old.Name)
260 require.NoError(t, recordErr)
261 require.True(t, ok)
262 require.True(t, fresh.Equal(stored))
263 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
264 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs)
265}
266
267func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) {
268 t.Parallel()
269 old := healthyTransactionRecord()
270 cause := errors.New("repository delete failed")
271 catalogue := newTransactionCatalogue(old)
272 repository := &transactionRepository{deleteErrOnce: cause}
273 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
274
275 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
276
277 require.ErrorIs(t, err, cause)
278 require.True(t, committed)
279 require.NotEqual(t, old.IncarnationID, fresh.IncarnationID)
280 stored, ok, recordErr := catalogue.Record(old.Name)
281 require.NoError(t, recordErr)
282 require.True(t, ok)
283 require.True(t, fresh.Equal(stored))
284 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
285 require.Empty(t, repository.deletedIDs)
286}
287
288func TestResetIncompatibleReplaceFailurePreservesOldAuthorityAndRepository(t *testing.T) {
289 t.Parallel()
290 old := healthyTransactionRecord()
291 cause := errors.New("catalogue replace failed")
292 catalogue := newTransactionCatalogue(old)
293 catalogue.replaceErrOnce = cause
294 repository := &transactionRepository{}
295 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
296
297 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
298
299 require.ErrorIs(t, err, cause)
300 require.False(t, committed)
301 require.Equal(t, domain.CatalogueRecord{}, fresh)
302 stored, ok, recordErr := catalogue.Record(old.Name)
303 require.NoError(t, recordErr)
304 require.True(t, ok)
305 require.True(t, old.Equal(stored))
306 require.Empty(t, repository.deleteCalls)
307}
308
309func TestResetIncompatibleStaleAuthorityIsNoOp(t *testing.T) {
310 t.Parallel()
311 old := healthyTransactionRecord()
312 otherGeneration := *old.Committed
313 otherGeneration.Generation++
314 otherDigest := *old.Committed
315 otherDigest.ManifestDigest[0]++
316 uncommitted := old
317 uncommitted.Committed = nil
318 tests := []struct {
319 name string
320 record *domain.CatalogueRecord
321 expected domain.IncarnationID
322 ref domain.CheckpointRef
323 }{
324 {name: "missing record", expected: old.IncarnationID, ref: *old.Committed},
325 {name: "different incarnation", record: &old, expected: domain.IncarnationID{9}, ref: *old.Committed},
326 {name: "missing checkpoint", record: &uncommitted, expected: old.IncarnationID, ref: *old.Committed},
327 {name: "different checkpoint generation", record: &old, expected: old.IncarnationID, ref: otherGeneration},
328 {name: "different checkpoint digest", record: &old, expected: old.IncarnationID, ref: otherDigest},
329 {name: "degraded record", record: func() *domain.CatalogueRecord {
330 degraded := old
331 degraded.DegradedReason = "restore failed"
332 return °raded
333 }(), expected: old.IncarnationID, ref: *old.Committed},
334 }
335 for _, tt := range tests {
336 t.Run(tt.name, func(t *testing.T) {
337 t.Parallel()
338 catalogue := newTransactionCatalogue()
339 if tt.record != nil {
340 catalogue.records[tt.record.Name] = *tt.record
341 }
342 repository := &transactionRepository{}
343 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
344
345 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, tt.expected, tt.ref)
346
347 require.NoError(t, err)
348 require.False(t, committed)
349 require.Equal(t, domain.CatalogueRecord{}, fresh)
350 require.Empty(t, catalogue.events)
351 require.Empty(t, repository.deleteCalls)
352 })
353 }
354}
355
356func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) {
357 t.Parallel()
358
359 t.Run("a stale keep snapshot cannot remove a newly created incarnation", func(t *testing.T) {
360 t.Parallel()
361 catalogue := newTransactionCatalogue()
362 repository := &garbageCollectionFenceRepository{
363 collectionEntered: make(chan struct{}),
364 releaseCollection: make(chan struct{}),
365 }
366 catalogue.onCreate = repository.addIncarnation
367 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
368
369 collected := make(chan error, 1)
370 go func() {
371 _, err := coordinator.CollectGarbage(context.Background())
372 collected <- err
373 }()
374 <-repository.collectionEntered
375
376 created := make(chan error, 1)
377 go func() {
378 _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
379 created <- err
380 }()
381 close(repository.releaseCollection)
382
383 require.NoError(t, <-collected)
384 require.NoError(t, <-created)
385 got, ok, err := catalogue.Record("new")
386 require.NoError(t, err)
387 require.True(t, ok)
388 require.NotZero(t, got.IncarnationID)
389 require.True(t, repository.hasIncarnation(got.IncarnationID), "GC's stale keep snapshot must finish before the new incarnation is published")
390 events, _ := repository.eventLog()
391 require.Equal(t, []string{"collect"}, events)
392 })
393
394 t.Run("a stale keep snapshot cannot prune a newly committed generation", func(t *testing.T) {
395 t.Parallel()
396 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
397 record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
398 catalogue := newTransactionCatalogue(record)
399 repository := &garbageCollectionFenceRepository{
400 collectionEntered: make(chan struct{}),
401 releaseCollection: make(chan struct{}),
402 generations: map[domain.IncarnationID]map[uint64]struct{}{
403 record.IncarnationID: {1: {}},
404 },
405 }
406 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
407
408 collected := make(chan error, 1)
409 go func() {
410 _, err := coordinator.CollectGarbage(context.Background())
411 collected <- err
412 }()
413 <-repository.collectionEntered
414
415 published := make(chan error, 1)
416 go func() {
417 _, err := coordinator.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
418 Name: "work",
419 IncarnationID: record.IncarnationID,
420 Generation: 2,
421 ParentCheckpoint: &committed,
422 Manifest: []byte("generation two"),
423 })
424 published <- err
425 }()
426 close(repository.releaseCollection)
427
428 require.NoError(t, <-collected)
429 require.NoError(t, <-published)
430 events, publishedBeforeGC := repository.eventLog()
431 require.Equal(t, []string{"collect", "publish"}, events)
432 require.False(t, publishedBeforeGC)
433 require.True(t, repository.hasGeneration(record.IncarnationID, 2), "GC's stale keep snapshot must finish before generation two is published")
434 })
435}
436
437func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
438 t.Parallel()
439 old := degradedTransactionRecord()
440 catalogue := newTransactionCatalogue(old)
441 repository := &transactionRepository{deleteEntered: make(chan struct{}), releaseDelete: make(chan struct{})}
442 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
443
444 discarded := make(chan error, 1)
445 go func() { discarded <- coordinator.Discard(t.Context(), old.Name) }()
446 <-repository.deleteEntered // Replace committed the fresh incarnation while Discard retains the mutation fence.
447
448 marked := make(chan struct {
449 marked bool
450 err error
451 }, 1)
452 go func() {
453 _, markedSession, err := coordinator.MarkBroken(t.Context(), old.Name, old.IncarnationID, "late restore failure")
454 marked <- struct {
455 marked bool
456 err error
457 }{markedSession, err}
458 }()
459
460 close(repository.releaseDelete)
461 require.NoError(t, <-discarded)
462 result := <-marked
463 require.NoError(t, result.err)
464 require.False(t, result.marked, "a late restore must not degrade the replacement incarnation")
465
466 got, ok, err := catalogue.Record(old.Name)
467 require.NoError(t, err)
468 require.True(t, ok)
469 require.NotEqual(t, old.IncarnationID, got.IncarnationID)
470 require.Nil(t, got.Committed)
471 require.Empty(t, got.DegradedReason)
472}
473
474func TestDiscardPublicPreconditions(t *testing.T) {
475 t.Parallel()
476 uncommitted := degradedTransactionRecord()
477 uncommitted.Name = "work"
478 uncommitted.Committed = nil
479 uncommitted.DegradedReason = ""
480 healthy := healthyTransactionRecord()
481 tests := []struct {
482 name string
483 record *domain.CatalogueRecord
484 incomplete bool
485 cancelled bool
486 wantErr error
487 }{
488 {name: "incomplete dependencies", incomplete: true, wantErr: errors.New("recovery: incomplete discard dependencies")},
489 {name: "cancelled context", record: &healthy, cancelled: true, wantErr: context.Canceled},
490 {name: "missing record", wantErr: ErrRecoveryRecordNotFound},
491 {name: "already fresh incarnation", record: &uncommitted},
492 {name: "healthy checkpoint", record: &healthy, wantErr: ErrSessionNotBroken},
493 }
494 for _, tt := range tests {
495 t.Run(tt.name, func(t *testing.T) {
496 t.Parallel()
497 ctx := context.Background()
498 if tt.cancelled {
499 cancelled, cancel := context.WithCancel(ctx)
500 cancel()
501 ctx = cancelled
502 }
503 var coordinator *Coordinator
504 var catalogue *transactionCatalogue
505 var repository *transactionRepository
506 if tt.incomplete {
507 coordinator = &Coordinator{}
508 } else {
509 catalogue = newTransactionCatalogue()
510 if tt.record != nil {
511 catalogue.records[tt.record.Name] = *tt.record
512 }
513 repository = &transactionRepository{}
514 coordinator = NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
515 }
516
517 err := coordinator.Discard(ctx, "work")
518
519 if tt.wantErr == nil {
520 require.NoError(t, err)
521 } else if tt.incomplete {
522 require.EqualError(t, err, tt.wantErr.Error())
523 } else {
524 require.ErrorIs(t, err, tt.wantErr)
525 }
526 if !tt.incomplete {
527 require.Empty(t, catalogue.events)
528 require.Empty(t, repository.deleteCalls)
529 }
530 })
531 }
532}
533
534func TestDiscardIsRetryIdempotent(t *testing.T) {
535 t.Parallel()
536 failure := errors.New("injected discard failure")
537 tests := []struct {
538 name string
539 failBefore bool
540 }{
541 {name: "retry from old catalogue record", failBefore: true},
542 {name: "retry from fresh catalogue record"},
543 }
544 for _, tt := range tests {
545 t.Run(tt.name, func(t *testing.T) {
546 t.Parallel()
547 old := degradedTransactionRecord()
548 old.TabNames = []string{"shell"}
549 catalogue := newTransactionCatalogue(old)
550 repository := &transactionRepository{}
551 if tt.failBefore {
552 catalogue.replaceErrOnce = failure
553 } else {
554 repository.deleteErrOnce = failure
555 }
556 ids := append(bytes.Repeat([]byte{2}, 16), bytes.Repeat([]byte{3}, 16)...)
557 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(ids))
558
559 err := coordinator.Discard(context.Background(), old.Name)
560 require.ErrorIs(t, err, failure)
561 require.NoError(t, coordinator.Discard(context.Background(), old.Name))
562
563 got, ok, err := catalogue.Record(old.Name)
564 require.NoError(t, err)
565 require.True(t, ok)
566 require.Nil(t, got.Committed)
567 require.NotEqual(t, old.IncarnationID, got.IncarnationID)
568 require.Equal(t, old.TabNames, got.TabNames)
569 if tt.failBefore {
570 require.Contains(t, repository.deletedIDs, old.IncarnationID)
571 } else {
572 require.Empty(t, repository.deletedIDs, "startup garbage collection owns the orphaned old directory")
573 }
574 })
575 }
576}