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 TestResetProtocolIncompatibleKeepsOnlySessionName(t *testing.T) {
239 t.Parallel()
240 old := healthyTransactionRecord()
241 old.TabRecords = []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}}
242 old.DegradedReason = ""
243 catalogue := newTransactionCatalogue(old)
244 repository := &transactionRepository{}
245 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
246
247 fresh, committed, err := coordinator.ResetProtocolIncompatible(context.Background(), old.Name, old.IncarnationID)
248
249 require.NoError(t, err)
250 require.True(t, committed)
251 require.Equal(t, old.Name, fresh.Name)
252 require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{2}, 16)), fresh.IncarnationID)
253 require.Empty(t, fresh.Cwd)
254 require.Zero(t, fresh.CreatedAt)
255 require.Zero(t, fresh.UpdatedAt)
256 require.Zero(t, fresh.LastUsedSeq)
257 require.Empty(t, fresh.TabNames)
258 require.Empty(t, fresh.TabRecords)
259 require.Nil(t, fresh.Committed)
260 require.Empty(t, fresh.DegradedReason)
261 stored, ok, recordErr := catalogue.Record(old.Name)
262 require.NoError(t, recordErr)
263 require.True(t, ok)
264 require.True(t, fresh.Equal(stored))
265 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
266}
267
268func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
269 t.Parallel()
270 old := healthyTransactionRecord()
271 catalogue := newTransactionCatalogue(old)
272 repository := &transactionRepository{}
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.NoError(t, err)
278 require.True(t, committed)
279 require.NotZero(t, fresh.IncarnationID)
280 require.NotEqual(t, old.IncarnationID, fresh.IncarnationID)
281 require.Equal(t, old.Name, fresh.Name)
282 require.Equal(t, old.Cwd, fresh.Cwd)
283 require.Equal(t, old.CreatedAt, fresh.CreatedAt)
284 require.Equal(t, old.UpdatedAt, fresh.UpdatedAt)
285 require.Equal(t, old.LastUsedSeq, fresh.LastUsedSeq)
286 require.Empty(t, fresh.TabNames)
287 require.Nil(t, fresh.Committed)
288 require.Empty(t, fresh.DegradedReason)
289 stored, ok, recordErr := catalogue.Record(old.Name)
290 require.NoError(t, recordErr)
291 require.True(t, ok)
292 require.True(t, fresh.Equal(stored))
293 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
294 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs)
295}
296
297func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) {
298 t.Parallel()
299 old := healthyTransactionRecord()
300 cause := errors.New("repository delete failed")
301 catalogue := newTransactionCatalogue(old)
302 repository := &transactionRepository{deleteErrOnce: cause}
303 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
304
305 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
306
307 require.ErrorIs(t, err, cause)
308 require.True(t, committed)
309 require.NotEqual(t, old.IncarnationID, fresh.IncarnationID)
310 stored, ok, recordErr := catalogue.Record(old.Name)
311 require.NoError(t, recordErr)
312 require.True(t, ok)
313 require.True(t, fresh.Equal(stored))
314 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
315 require.Empty(t, repository.deletedIDs)
316}
317
318func TestResetIncompatibleReplaceFailurePreservesOldAuthorityAndRepository(t *testing.T) {
319 t.Parallel()
320 old := healthyTransactionRecord()
321 cause := errors.New("catalogue replace failed")
322 catalogue := newTransactionCatalogue(old)
323 catalogue.replaceErrOnce = cause
324 repository := &transactionRepository{}
325 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
326
327 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
328
329 require.ErrorIs(t, err, cause)
330 require.False(t, committed)
331 require.Equal(t, domain.CatalogueRecord{}, fresh)
332 stored, ok, recordErr := catalogue.Record(old.Name)
333 require.NoError(t, recordErr)
334 require.True(t, ok)
335 require.True(t, old.Equal(stored))
336 require.Empty(t, repository.deleteCalls)
337}
338
339func TestResetIncompatibleStaleAuthorityIsNoOp(t *testing.T) {
340 t.Parallel()
341 old := healthyTransactionRecord()
342 otherGeneration := *old.Committed
343 otherGeneration.Generation++
344 otherDigest := *old.Committed
345 otherDigest.ManifestDigest[0]++
346 uncommitted := old
347 uncommitted.Committed = nil
348 tests := []struct {
349 name string
350 record *domain.CatalogueRecord
351 expected domain.IncarnationID
352 ref domain.CheckpointRef
353 }{
354 {name: "missing record", expected: old.IncarnationID, ref: *old.Committed},
355 {name: "different incarnation", record: &old, expected: domain.IncarnationID{9}, ref: *old.Committed},
356 {name: "missing checkpoint", record: &uncommitted, expected: old.IncarnationID, ref: *old.Committed},
357 {name: "different checkpoint generation", record: &old, expected: old.IncarnationID, ref: otherGeneration},
358 {name: "different checkpoint digest", record: &old, expected: old.IncarnationID, ref: otherDigest},
359 {name: "degraded record", record: func() *domain.CatalogueRecord {
360 degraded := old
361 degraded.DegradedReason = "restore failed"
362 return °raded
363 }(), expected: old.IncarnationID, ref: *old.Committed},
364 }
365 for _, tt := range tests {
366 t.Run(tt.name, func(t *testing.T) {
367 t.Parallel()
368 catalogue := newTransactionCatalogue()
369 if tt.record != nil {
370 catalogue.records[tt.record.Name] = *tt.record
371 }
372 repository := &transactionRepository{}
373 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
374
375 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, tt.expected, tt.ref)
376
377 require.NoError(t, err)
378 require.False(t, committed)
379 require.Equal(t, domain.CatalogueRecord{}, fresh)
380 require.Empty(t, catalogue.events)
381 require.Empty(t, repository.deleteCalls)
382 })
383 }
384}
385
386func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) {
387 t.Parallel()
388
389 t.Run("a stale keep snapshot cannot remove a newly created incarnation", func(t *testing.T) {
390 t.Parallel()
391 catalogue := newTransactionCatalogue()
392 repository := &garbageCollectionFenceRepository{
393 collectionEntered: make(chan struct{}),
394 releaseCollection: make(chan struct{}),
395 }
396 catalogue.onCreate = repository.addIncarnation
397 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
398
399 collected := make(chan error, 1)
400 go func() {
401 _, err := coordinator.CollectGarbage(context.Background())
402 collected <- err
403 }()
404 <-repository.collectionEntered
405
406 created := make(chan error, 1)
407 go func() {
408 _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
409 created <- err
410 }()
411 close(repository.releaseCollection)
412
413 require.NoError(t, <-collected)
414 require.NoError(t, <-created)
415 got, ok, err := catalogue.Record("new")
416 require.NoError(t, err)
417 require.True(t, ok)
418 require.NotZero(t, got.IncarnationID)
419 require.True(t, repository.hasIncarnation(got.IncarnationID), "GC's stale keep snapshot must finish before the new incarnation is published")
420 events, _ := repository.eventLog()
421 require.Equal(t, []string{"collect"}, events)
422 })
423
424 t.Run("a stale keep snapshot cannot prune a newly committed generation", func(t *testing.T) {
425 t.Parallel()
426 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
427 record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
428 catalogue := newTransactionCatalogue(record)
429 repository := &garbageCollectionFenceRepository{
430 collectionEntered: make(chan struct{}),
431 releaseCollection: make(chan struct{}),
432 generations: map[domain.IncarnationID]map[uint64]struct{}{
433 record.IncarnationID: {1: {}},
434 },
435 }
436 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
437
438 collected := make(chan error, 1)
439 go func() {
440 _, err := coordinator.CollectGarbage(context.Background())
441 collected <- err
442 }()
443 <-repository.collectionEntered
444
445 published := make(chan error, 1)
446 go func() {
447 _, err := coordinator.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
448 Name: "work",
449 IncarnationID: record.IncarnationID,
450 Generation: 2,
451 ParentCheckpoint: &committed,
452 Manifest: []byte("generation two"),
453 })
454 published <- err
455 }()
456 close(repository.releaseCollection)
457
458 require.NoError(t, <-collected)
459 require.NoError(t, <-published)
460 events, publishedBeforeGC := repository.eventLog()
461 require.Equal(t, []string{"collect", "publish"}, events)
462 require.False(t, publishedBeforeGC)
463 require.True(t, repository.hasGeneration(record.IncarnationID, 2), "GC's stale keep snapshot must finish before generation two is published")
464 })
465}
466
467func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
468 t.Parallel()
469 old := degradedTransactionRecord()
470 catalogue := newTransactionCatalogue(old)
471 repository := &transactionRepository{deleteEntered: make(chan struct{}), releaseDelete: make(chan struct{})}
472 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
473
474 discarded := make(chan error, 1)
475 go func() { discarded <- coordinator.Discard(t.Context(), old.Name) }()
476 <-repository.deleteEntered // Replace committed the fresh incarnation while Discard retains the mutation fence.
477
478 marked := make(chan struct {
479 marked bool
480 err error
481 }, 1)
482 go func() {
483 _, markedSession, err := coordinator.MarkBroken(t.Context(), old.Name, old.IncarnationID, "late restore failure")
484 marked <- struct {
485 marked bool
486 err error
487 }{markedSession, err}
488 }()
489
490 close(repository.releaseDelete)
491 require.NoError(t, <-discarded)
492 result := <-marked
493 require.NoError(t, result.err)
494 require.False(t, result.marked, "a late restore must not degrade the replacement incarnation")
495
496 got, ok, err := catalogue.Record(old.Name)
497 require.NoError(t, err)
498 require.True(t, ok)
499 require.NotEqual(t, old.IncarnationID, got.IncarnationID)
500 require.Nil(t, got.Committed)
501 require.Empty(t, got.DegradedReason)
502}
503
504func TestDiscardPublicPreconditions(t *testing.T) {
505 t.Parallel()
506 uncommitted := degradedTransactionRecord()
507 uncommitted.Name = "work"
508 uncommitted.Committed = nil
509 uncommitted.DegradedReason = ""
510 healthy := healthyTransactionRecord()
511 tests := []struct {
512 name string
513 record *domain.CatalogueRecord
514 incomplete bool
515 cancelled bool
516 wantErr error
517 }{
518 {name: "incomplete dependencies", incomplete: true, wantErr: errors.New("recovery: incomplete discard dependencies")},
519 {name: "cancelled context", record: &healthy, cancelled: true, wantErr: context.Canceled},
520 {name: "missing record", wantErr: ErrRecoveryRecordNotFound},
521 {name: "already fresh incarnation", record: &uncommitted},
522 {name: "healthy checkpoint", record: &healthy, wantErr: ErrSessionNotBroken},
523 }
524 for _, tt := range tests {
525 t.Run(tt.name, func(t *testing.T) {
526 t.Parallel()
527 ctx := context.Background()
528 if tt.cancelled {
529 cancelled, cancel := context.WithCancel(ctx)
530 cancel()
531 ctx = cancelled
532 }
533 var coordinator *Coordinator
534 var catalogue *transactionCatalogue
535 var repository *transactionRepository
536 if tt.incomplete {
537 coordinator = &Coordinator{}
538 } else {
539 catalogue = newTransactionCatalogue()
540 if tt.record != nil {
541 catalogue.records[tt.record.Name] = *tt.record
542 }
543 repository = &transactionRepository{}
544 coordinator = NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
545 }
546
547 err := coordinator.Discard(ctx, "work")
548
549 if tt.wantErr == nil {
550 require.NoError(t, err)
551 } else if tt.incomplete {
552 require.EqualError(t, err, tt.wantErr.Error())
553 } else {
554 require.ErrorIs(t, err, tt.wantErr)
555 }
556 if !tt.incomplete {
557 require.Empty(t, catalogue.events)
558 require.Empty(t, repository.deleteCalls)
559 }
560 })
561 }
562}
563
564func TestDiscardIsRetryIdempotent(t *testing.T) {
565 t.Parallel()
566 failure := errors.New("injected discard failure")
567 tests := []struct {
568 name string
569 failBefore bool
570 }{
571 {name: "retry from old catalogue record", failBefore: true},
572 {name: "retry from fresh catalogue record"},
573 }
574 for _, tt := range tests {
575 t.Run(tt.name, func(t *testing.T) {
576 t.Parallel()
577 old := degradedTransactionRecord()
578 old.TabNames = []string{"shell"}
579 catalogue := newTransactionCatalogue(old)
580 repository := &transactionRepository{}
581 if tt.failBefore {
582 catalogue.replaceErrOnce = failure
583 } else {
584 repository.deleteErrOnce = failure
585 }
586 ids := append(bytes.Repeat([]byte{2}, 16), bytes.Repeat([]byte{3}, 16)...)
587 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(ids))
588
589 err := coordinator.Discard(context.Background(), old.Name)
590 require.ErrorIs(t, err, failure)
591 require.NoError(t, coordinator.Discard(context.Background(), old.Name))
592
593 got, ok, err := catalogue.Record(old.Name)
594 require.NoError(t, err)
595 require.True(t, ok)
596 require.Nil(t, got.Committed)
597 require.NotEqual(t, old.IncarnationID, got.IncarnationID)
598 require.Equal(t, old.TabNames, got.TabNames)
599 if tt.failBefore {
600 require.Contains(t, repository.deletedIDs, old.IncarnationID)
601 } else {
602 require.Empty(t, repository.deletedIDs, "startup garbage collection owns the orphaned old directory")
603 }
604 })
605 }
606}