1package recovery
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "slices"
8 "sync"
9 "testing"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/stretchr/testify/require"
14)
15
16type transactionCatalogue struct {
17 mu sync.Mutex
18 records map[string]domain.CatalogueRecord
19 events []string
20 replaceErrOnce error
21 onCreate func(domain.CatalogueRecord)
22}
23
24func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {
25 catalogue := &transactionCatalogue{records: make(map[string]domain.CatalogueRecord)}
26 for _, record := range records {
27 catalogue.records[record.Name] = record
28 }
29 return catalogue
30}
31
32func (c *transactionCatalogue) Records() ([]domain.CatalogueRecord, error) {
33 c.mu.Lock()
34 defer c.mu.Unlock()
35 out := make([]domain.CatalogueRecord, 0, len(c.records))
36 for _, record := range c.records {
37 out = append(out, record)
38 }
39 return out, nil
40}
41
42func (c *transactionCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
43 c.mu.Lock()
44 defer c.mu.Unlock()
45 record, ok := c.records[name]
46 return record, ok, nil
47}
48
49func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
50 c.mu.Lock()
51 defer c.mu.Unlock()
52 if _, exists := c.records[record.Name]; exists {
53 return errors.New("record exists")
54 }
55 c.records[record.Name] = record
56 c.events = append(c.events, "create")
57 if c.onCreate != nil {
58 c.onCreate(record)
59 }
60 return nil
61}
62
63func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }
64func (c *transactionCatalogue) Sync() error { return nil }
65
66func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {
67 c.mu.Lock()
68 defer c.mu.Unlock()
69 if c.replaceErrOnce != nil {
70 err := c.replaceErrOnce
71 c.replaceErrOnce = nil
72 return err
73 }
74 if _, exists := c.records[name]; !exists {
75 return errors.New("record missing")
76 }
77 c.records[name] = record
78 c.events = append(c.events, "replace:"+name)
79 return nil
80}
81
82func (c *transactionCatalogue) Rename(oldName string, record domain.CatalogueRecord) error {
83 c.mu.Lock()
84 defer c.mu.Unlock()
85 if _, exists := c.records[oldName]; !exists {
86 return errors.New("record missing")
87 }
88 if _, exists := c.records[record.Name]; exists && record.Name != oldName {
89 return errors.New("record exists")
90 }
91 delete(c.records, oldName)
92 c.records[record.Name] = record
93 c.events = append(c.events, "rename:"+oldName+":"+record.Name)
94 return nil
95}
96
97func (c *transactionCatalogue) Delete(name string) error {
98 c.mu.Lock()
99 defer c.mu.Unlock()
100 delete(c.records, name)
101 c.events = append(c.events, "delete:"+name)
102 return nil
103}
104
105func (c *transactionCatalogue) Close() error { return nil }
106
107type transactionRepository struct {
108 ports.SnapshotRepository
109 mu sync.Mutex
110 deleteErrOnce error
111 deleteCalls []domain.IncarnationID
112 deletedIDs []domain.IncarnationID
113 deleteEntered chan struct{}
114 releaseDelete chan struct{}
115}
116
117// garbageCollectionFenceRepository models per-incarnation retention. The first
118// CollectIncarnationGarbage call blocks until released, so a test can race a
119// mutation against a step that already read its keep decision.
120type garbageCollectionFenceRepository struct {
121 ports.SnapshotRepository
122 mu sync.Mutex
123 events []string
124 collectionEntered chan struct{}
125 releaseCollection chan struct{}
126 blocked bool
127 incarnations map[domain.IncarnationID]struct{}
128 generations map[domain.IncarnationID]map[uint64]struct{}
129}
130
131func (r *garbageCollectionFenceRepository) SnapshotIncarnations(context.Context) ([]domain.IncarnationID, error) {
132 r.mu.Lock()
133 defer r.mu.Unlock()
134 seen := make(map[domain.IncarnationID]struct{})
135 for id := range r.incarnations {
136 seen[id] = struct{}{}
137 }
138 for id := range r.generations {
139 seen[id] = struct{}{}
140 }
141 ids := make([]domain.IncarnationID, 0, len(seen))
142 for id := range seen {
143 ids = append(ids, id)
144 }
145 // Deterministic order lets tests pause between known steps.
146 slices.SortFunc(ids, func(a, b domain.IncarnationID) int { return bytes.Compare(a[:], b[:]) })
147 return ids, nil
148}
149
150func (r *garbageCollectionFenceRepository) CollectIncarnationGarbage(_ context.Context, id domain.IncarnationID, keep *domain.CheckpointRef) error {
151 r.mu.Lock()
152 block := !r.blocked
153 r.blocked = true
154 r.mu.Unlock()
155 if block {
156 close(r.collectionEntered)
157 <-r.releaseCollection
158 }
159 r.mu.Lock()
160 defer r.mu.Unlock()
161 if keep == nil {
162 delete(r.incarnations, id)
163 delete(r.generations, id)
164 } else {
165 for generation := range r.generations[id] {
166 if generation != keep.Generation && generation+1 != keep.Generation {
167 delete(r.generations[id], generation)
168 }
169 }
170 }
171 r.events = append(r.events, "collect")
172 return nil
173}
174
175func (r *garbageCollectionFenceRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
176 r.mu.Lock()
177 defer r.mu.Unlock()
178 delete(r.incarnations, id)
179 delete(r.generations, id)
180 return nil
181}
182
183func (r *garbageCollectionFenceRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
184 r.mu.Lock()
185 defer r.mu.Unlock()
186 if r.generations == nil {
187 r.generations = make(map[domain.IncarnationID]map[uint64]struct{})
188 }
189 if r.generations[publication.IncarnationID] == nil {
190 r.generations[publication.IncarnationID] = make(map[uint64]struct{})
191 }
192 r.generations[publication.IncarnationID][publication.Generation] = struct{}{}
193 r.events = append(r.events, "publish")
194 return nil
195}
196
197func (r *garbageCollectionFenceRepository) addIncarnation(record domain.CatalogueRecord) {
198 r.mu.Lock()
199 defer r.mu.Unlock()
200 if r.incarnations == nil {
201 r.incarnations = make(map[domain.IncarnationID]struct{})
202 }
203 r.incarnations[record.IncarnationID] = struct{}{}
204}
205
206func (r *garbageCollectionFenceRepository) hasIncarnation(id domain.IncarnationID) bool {
207 r.mu.Lock()
208 defer r.mu.Unlock()
209 _, ok := r.incarnations[id]
210 return ok
211}
212
213func (r *garbageCollectionFenceRepository) hasGeneration(id domain.IncarnationID, generation uint64) bool {
214 r.mu.Lock()
215 defer r.mu.Unlock()
216 _, ok := r.generations[id][generation]
217 return ok
218}
219
220func (r *garbageCollectionFenceRepository) eventLog() []string {
221 r.mu.Lock()
222 defer r.mu.Unlock()
223 return append([]string(nil), r.events...)
224}
225
226func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
227 if r.deleteEntered != nil {
228 close(r.deleteEntered)
229 <-r.releaseDelete
230 }
231 r.mu.Lock()
232 defer r.mu.Unlock()
233 r.deleteCalls = append(r.deleteCalls, id)
234 if r.deleteErrOnce != nil {
235 err := r.deleteErrOnce
236 r.deleteErrOnce = nil
237 return err
238 }
239 r.deletedIDs = append(r.deletedIDs, id)
240 return nil
241}
242
243func degradedTransactionRecord() domain.CatalogueRecord {
244 return domain.CatalogueRecord{
245 Name: "broken", IncarnationID: domain.IncarnationID{1}, Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
246 DegradedReason: "checkpoint unreadable",
247 }
248}
249
250func healthyTransactionRecord() domain.CatalogueRecord {
251 return domain.CatalogueRecord{
252 Name: "work",
253 IncarnationID: domain.IncarnationID{1},
254 Cwd: "/workspace",
255 CreatedAt: 11,
256 UpdatedAt: 22,
257 LastUsedSeq: 33,
258 TabNames: []string{"shell", "logs"},
259 Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
260 }
261}
262
263func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
264 t.Parallel()
265 old := healthyTransactionRecord()
266 catalogue := newTransactionCatalogue(old)
267 repository := &transactionRepository{}
268 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
269
270 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
271
272 require.NoError(t, err)
273 require.True(t, committed)
274 require.NotZero(t, fresh.IncarnationID)
275 require.NotEqual(t, old.IncarnationID, fresh.IncarnationID)
276 require.Equal(t, old.Name, fresh.Name)
277 require.Equal(t, old.Cwd, fresh.Cwd)
278 require.Equal(t, old.CreatedAt, fresh.CreatedAt)
279 require.Equal(t, old.UpdatedAt, fresh.UpdatedAt)
280 require.Equal(t, old.LastUsedSeq, fresh.LastUsedSeq)
281 require.Empty(t, fresh.TabNames)
282 require.Nil(t, fresh.Committed)
283 require.Empty(t, fresh.DegradedReason)
284 stored, ok, recordErr := catalogue.Record(old.Name)
285 require.NoError(t, recordErr)
286 require.True(t, ok)
287 require.True(t, fresh.Equal(stored))
288 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
289 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deletedIDs)
290}
291
292func TestDeleteExactDoesNotDeleteSameNameReplacement(t *testing.T) {
293 oldID := domain.IncarnationID{1}
294 newID := domain.IncarnationID{2}
295 catalogue := newTransactionCatalogue(domain.CatalogueRecord{Name: "work", IncarnationID: newID, Cwd: "/tmp", CreatedAt: 22, UpdatedAt: 22})
296 repository := &transactionRepository{}
297 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
298
299 require.NoError(t, coordinator.DeleteExact(context.Background(), "work", oldID, 11))
300 record, ok, err := catalogue.Record("work")
301 require.NoError(t, err)
302 require.True(t, ok)
303 require.Equal(t, newID, record.IncarnationID)
304 require.Empty(t, repository.deleteCalls)
305}
306
307func TestDeleteExactDeletesMatchingLifecycle(t *testing.T) {
308 id := domain.IncarnationID{1}
309 catalogue := newTransactionCatalogue(domain.CatalogueRecord{Name: "work", IncarnationID: id, Cwd: "/tmp", CreatedAt: 11, UpdatedAt: 11})
310 repository := &transactionRepository{}
311 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
312
313 require.NoError(t, coordinator.DeleteExact(context.Background(), "work", id, 11))
314 _, ok, err := catalogue.Record("work")
315 require.NoError(t, err)
316 require.False(t, ok)
317 require.Equal(t, []domain.IncarnationID{id}, repository.deleteCalls)
318}
319
320func TestResetIncompatibleDeleteFailureReturnsCommittedFreshAuthority(t *testing.T) {
321 t.Parallel()
322 old := healthyTransactionRecord()
323 cause := errors.New("repository delete failed")
324 catalogue := newTransactionCatalogue(old)
325 repository := &transactionRepository{deleteErrOnce: cause}
326 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
327
328 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
329
330 require.ErrorIs(t, err, cause)
331 require.True(t, committed)
332 require.NotEqual(t, old.IncarnationID, fresh.IncarnationID)
333 stored, ok, recordErr := catalogue.Record(old.Name)
334 require.NoError(t, recordErr)
335 require.True(t, ok)
336 require.True(t, fresh.Equal(stored))
337 require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
338 require.Empty(t, repository.deletedIDs)
339}
340
341func TestResetIncompatibleReplaceFailurePreservesOldAuthorityAndRepository(t *testing.T) {
342 t.Parallel()
343 old := healthyTransactionRecord()
344 cause := errors.New("catalogue replace failed")
345 catalogue := newTransactionCatalogue(old)
346 catalogue.replaceErrOnce = cause
347 repository := &transactionRepository{}
348 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
349
350 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, old.IncarnationID, *old.Committed)
351
352 require.ErrorIs(t, err, cause)
353 require.False(t, committed)
354 require.Equal(t, domain.CatalogueRecord{}, fresh)
355 stored, ok, recordErr := catalogue.Record(old.Name)
356 require.NoError(t, recordErr)
357 require.True(t, ok)
358 require.True(t, old.Equal(stored))
359 require.Empty(t, repository.deleteCalls)
360}
361
362func TestResetIncompatibleStaleAuthorityIsNoOp(t *testing.T) {
363 t.Parallel()
364 old := healthyTransactionRecord()
365 otherGeneration := *old.Committed
366 otherGeneration.Generation++
367 otherDigest := *old.Committed
368 otherDigest.ManifestDigest[0]++
369 uncommitted := old
370 uncommitted.Committed = nil
371 tests := []struct {
372 name string
373 record *domain.CatalogueRecord
374 expected domain.IncarnationID
375 ref domain.CheckpointRef
376 }{
377 {name: "missing record", expected: old.IncarnationID, ref: *old.Committed},
378 {name: "different incarnation", record: &old, expected: domain.IncarnationID{9}, ref: *old.Committed},
379 {name: "missing checkpoint", record: &uncommitted, expected: old.IncarnationID, ref: *old.Committed},
380 {name: "different checkpoint generation", record: &old, expected: old.IncarnationID, ref: otherGeneration},
381 {name: "different checkpoint digest", record: &old, expected: old.IncarnationID, ref: otherDigest},
382 {name: "degraded record", record: func() *domain.CatalogueRecord {
383 degraded := old
384 degraded.DegradedReason = "restore failed"
385 return °raded
386 }(), expected: old.IncarnationID, ref: *old.Committed},
387 }
388 for _, tt := range tests {
389 t.Run(tt.name, func(t *testing.T) {
390 t.Parallel()
391 catalogue := newTransactionCatalogue()
392 if tt.record != nil {
393 catalogue.records[tt.record.Name] = *tt.record
394 }
395 repository := &transactionRepository{}
396 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
397
398 fresh, committed, err := coordinator.ResetIncompatible(context.Background(), old.Name, tt.expected, tt.ref)
399
400 require.NoError(t, err)
401 require.False(t, committed)
402 require.Equal(t, domain.CatalogueRecord{}, fresh)
403 require.Empty(t, catalogue.events)
404 require.Empty(t, repository.deleteCalls)
405 })
406 }
407}
408
409func TestGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(t *testing.T) {
410 t.Parallel()
411
412 t.Run("a keep decision is re-read under the fence for every incarnation", func(t *testing.T) {
413 t.Parallel()
414 // The pass lists incarnation 9 (an orphan) and pauses after step 1;
415 // "new" is then created with ID 9, so step 9 must re-read the catalogue
416 // and keep it. A keep decision read once per pass would remove it.
417 reused := domain.IncarnationID{9}
418 catalogue := newTransactionCatalogue()
419 repository := &garbageCollectionFenceRepository{
420 blocked: true,
421 incarnations: map[domain.IncarnationID]struct{}{{1}: {}, reused: {}},
422 }
423 catalogue.onCreate = repository.addIncarnation
424 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
425 ctx := &betweenStepsContext{Context: context.Background(), repository: repository, reached: make(chan struct{}), release: make(chan struct{})}
426
427 collected := make(chan error, 1)
428 go func() {
429 _, err := coordinator.CollectGarbage(ctx)
430 collected <- err
431 }()
432 <-ctx.reached
433 require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "step 1 ran first")
434
435 _, err := coordinator.Create(context.Background(), domain.CatalogueRecord{Name: "new", IncarnationID: reused})
436 require.NoError(t, err)
437 close(ctx.release)
438
439 require.NoError(t, <-collected)
440 require.True(t, repository.hasIncarnation(reused), "a step after the creation must keep the new incarnation")
441 require.False(t, repository.hasIncarnation(domain.IncarnationID{1}), "a true orphan is still removed")
442 })
443
444 t.Run("a stale keep decision cannot prune a newly committed generation", func(t *testing.T) {
445 t.Parallel()
446 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
447 record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
448 catalogue := newTransactionCatalogue(record)
449 repository := &garbageCollectionFenceRepository{
450 collectionEntered: make(chan struct{}),
451 releaseCollection: make(chan struct{}),
452 generations: map[domain.IncarnationID]map[uint64]struct{}{
453 record.IncarnationID: {1: {}},
454 },
455 }
456 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(nil))
457
458 collected := make(chan error, 1)
459 go func() {
460 _, err := coordinator.CollectGarbage(context.Background())
461 collected <- err
462 }()
463 <-repository.collectionEntered
464
465 published := make(chan error, 1)
466 go func() {
467 _, err := coordinator.PublishCheckpoint(context.Background(), "work", ports.SnapshotPublication{
468 Name: "work",
469 IncarnationID: record.IncarnationID,
470 Generation: 2,
471 ParentCheckpoint: &committed,
472 Manifest: []byte("generation two"),
473 })
474 published <- err
475 }()
476 close(repository.releaseCollection)
477
478 require.NoError(t, <-collected)
479 require.NoError(t, <-published)
480 require.Equal(t, []string{"collect", "publish"}, repository.eventLog())
481 require.True(t, repository.hasGeneration(record.IncarnationID, 2), "the fenced step must finish before generation two is published")
482 })
483}
484
485// betweenStepsContext pauses a GC pass at its first cancellation check after
486// one incarnation step has completed, outside any fenced step.
487type betweenStepsContext struct {
488 context.Context
489 repository *garbageCollectionFenceRepository
490 once sync.Once
491 reached chan struct{}
492 release chan struct{}
493}
494
495func (c *betweenStepsContext) Err() error {
496 if len(c.repository.eventLog()) == 1 {
497 c.once.Do(func() {
498 close(c.reached)
499 <-c.release
500 })
501 }
502 return c.Context.Err()
503}
504
505// TestDurableMutationWaitsForAtMostOneIncarnation proves the fence is released
506// between incarnation steps, so every mutation entry point can run while a
507// pass is still in progress instead of waiting for the whole pass.
508func TestDurableMutationWaitsForAtMostOneIncarnation(t *testing.T) {
509 t.Parallel()
510 committed := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
511 work := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}
512 broken := degradedTransactionRecord()
513 broken.IncarnationID = domain.IncarnationID{2}
514 tests := []struct {
515 name string
516 mutate func(*Coordinator) error
517 }{
518 {name: "create", mutate: func(c *Coordinator) error {
519 _, err := c.Create(context.Background(), domain.CatalogueRecord{Name: "new"})
520 return err
521 }},
522 {name: "rename", mutate: func(c *Coordinator) error {
523 _, err := c.Rename(context.Background(), work.Name, "renamed")
524 return err
525 }},
526 {name: "delete", mutate: func(c *Coordinator) error { return c.Delete(context.Background(), work.Name) }},
527 {name: "exact delete", mutate: func(c *Coordinator) error {
528 return c.DeleteExact(context.Background(), work.Name, work.IncarnationID, work.CreatedAt)
529 }},
530 {name: "mark broken", mutate: func(c *Coordinator) error {
531 _, _, err := c.MarkBroken(context.Background(), work.Name, work.IncarnationID, "restore failed")
532 return err
533 }},
534 {name: "discard", mutate: func(c *Coordinator) error { return c.Discard(context.Background(), broken.Name) }},
535 {name: "reset incompatible", mutate: func(c *Coordinator) error {
536 _, _, err := c.ResetIncompatible(context.Background(), work.Name, work.IncarnationID, committed)
537 return err
538 }},
539 {name: "checkpoint publication", mutate: func(c *Coordinator) error {
540 _, err := c.PublishCheckpoint(context.Background(), work.Name, ports.SnapshotPublication{
541 Name: work.Name, IncarnationID: work.IncarnationID, Generation: 2,
542 ParentCheckpoint: &committed, Manifest: []byte("generation two"),
543 })
544 return err
545 }},
546 }
547 for _, tt := range tests {
548 t.Run(tt.name, func(t *testing.T) {
549 t.Parallel()
550 // Two orphan incarnations give the pass two fenced steps.
551 repository := &garbageCollectionFenceRepository{
552 blocked: true,
553 incarnations: map[domain.IncarnationID]struct{}{{7}: {}, {8}: {}},
554 }
555 coordinator := NewCoordinator(newTransactionCatalogue(work, broken), repository, bytes.NewReader(bytes.Repeat([]byte{2}, 32)))
556 ctx := &betweenStepsContext{Context: context.Background(), repository: repository, reached: make(chan struct{}), release: make(chan struct{})}
557
558 collected := make(chan error, 1)
559 go func() {
560 _, err := coordinator.CollectGarbage(ctx)
561 collected <- err
562 }()
563 <-ctx.reached
564
565 // TryLock first so a regression that holds the fence across the pass
566 // fails here instead of deadlocking the mutation.
567 require.True(t, coordinator.mutationMu.TryLock(), "the fence must be free between incarnation steps")
568 coordinator.mutationMu.Unlock()
569 require.NoError(t, tt.mutate(coordinator))
570
571 close(ctx.release)
572 require.NoError(t, <-collected)
573 steps := 0
574 for _, event := range repository.eventLog() {
575 if event == "collect" {
576 steps++
577 }
578 }
579 require.Equal(t, 2, steps, "both incarnation steps ran")
580 })
581 }
582}
583
584func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
585 t.Parallel()
586 old := degradedTransactionRecord()
587 catalogue := newTransactionCatalogue(old)
588 repository := &transactionRepository{deleteEntered: make(chan struct{}), releaseDelete: make(chan struct{})}
589 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
590
591 discarded := make(chan error, 1)
592 go func() { discarded <- coordinator.Discard(t.Context(), old.Name) }()
593 <-repository.deleteEntered // Replace committed the fresh incarnation while Discard retains the mutation fence.
594
595 marked := make(chan struct {
596 marked bool
597 err error
598 }, 1)
599 go func() {
600 _, markedSession, err := coordinator.MarkBroken(t.Context(), old.Name, old.IncarnationID, "late restore failure")
601 marked <- struct {
602 marked bool
603 err error
604 }{markedSession, err}
605 }()
606
607 close(repository.releaseDelete)
608 require.NoError(t, <-discarded)
609 result := <-marked
610 require.NoError(t, result.err)
611 require.False(t, result.marked, "a late restore must not degrade the replacement incarnation")
612
613 got, ok, err := catalogue.Record(old.Name)
614 require.NoError(t, err)
615 require.True(t, ok)
616 require.NotEqual(t, old.IncarnationID, got.IncarnationID)
617 require.Nil(t, got.Committed)
618 require.Empty(t, got.DegradedReason)
619}
620
621func TestDiscardPublicPreconditions(t *testing.T) {
622 t.Parallel()
623 uncommitted := degradedTransactionRecord()
624 uncommitted.Name = "work"
625 uncommitted.Committed = nil
626 uncommitted.DegradedReason = ""
627 healthy := healthyTransactionRecord()
628 tests := []struct {
629 name string
630 record *domain.CatalogueRecord
631 incomplete bool
632 cancelled bool
633 wantErr error
634 }{
635 {name: "incomplete dependencies", incomplete: true, wantErr: errors.New("recovery: incomplete discard dependencies")},
636 {name: "cancelled context", record: &healthy, cancelled: true, wantErr: context.Canceled},
637 {name: "missing record", wantErr: ErrRecoveryRecordNotFound},
638 {name: "already fresh incarnation", record: &uncommitted},
639 {name: "healthy checkpoint", record: &healthy, wantErr: ErrSessionNotBroken},
640 }
641 for _, tt := range tests {
642 t.Run(tt.name, func(t *testing.T) {
643 t.Parallel()
644 ctx := context.Background()
645 if tt.cancelled {
646 cancelled, cancel := context.WithCancel(ctx)
647 cancel()
648 ctx = cancelled
649 }
650 var coordinator *Coordinator
651 var catalogue *transactionCatalogue
652 var repository *transactionRepository
653 if tt.incomplete {
654 coordinator = &Coordinator{}
655 } else {
656 catalogue = newTransactionCatalogue()
657 if tt.record != nil {
658 catalogue.records[tt.record.Name] = *tt.record
659 }
660 repository = &transactionRepository{}
661 coordinator = NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
662 }
663
664 err := coordinator.Discard(ctx, "work")
665
666 if tt.wantErr == nil {
667 require.NoError(t, err)
668 } else if tt.incomplete {
669 require.EqualError(t, err, tt.wantErr.Error())
670 } else {
671 require.ErrorIs(t, err, tt.wantErr)
672 }
673 if !tt.incomplete {
674 require.Empty(t, catalogue.events)
675 require.Empty(t, repository.deleteCalls)
676 }
677 })
678 }
679}
680
681func TestDiscardIsRetryIdempotent(t *testing.T) {
682 t.Parallel()
683 failure := errors.New("injected discard failure")
684 tests := []struct {
685 name string
686 failBefore bool
687 }{
688 {name: "retry from old catalogue record", failBefore: true},
689 {name: "retry from fresh catalogue record"},
690 }
691 for _, tt := range tests {
692 t.Run(tt.name, func(t *testing.T) {
693 t.Parallel()
694 old := degradedTransactionRecord()
695 old.TabNames = []string{"shell"}
696 catalogue := newTransactionCatalogue(old)
697 repository := &transactionRepository{}
698 if tt.failBefore {
699 catalogue.replaceErrOnce = failure
700 } else {
701 repository.deleteErrOnce = failure
702 }
703 ids := append(bytes.Repeat([]byte{2}, 16), bytes.Repeat([]byte{3}, 16)...)
704 coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(ids))
705
706 err := coordinator.Discard(context.Background(), old.Name)
707 require.ErrorIs(t, err, failure)
708 require.NoError(t, coordinator.Discard(context.Background(), old.Name))
709
710 got, ok, err := catalogue.Record(old.Name)
711 require.NoError(t, err)
712 require.True(t, ok)
713 require.Nil(t, got.Committed)
714 require.NotEqual(t, old.IncarnationID, got.IncarnationID)
715 require.Equal(t, old.TabNames, got.TabNames)
716 if tt.failBefore {
717 require.Contains(t, repository.deletedIDs, old.IncarnationID)
718 } else {
719 require.Empty(t, repository.deletedIDs, "background garbage collection owns the orphaned old directory")
720 }
721 })
722 }
723}