1package daemon
2
3import (
4 "context"
5 "errors"
6 "testing"
7 "testing/synctest"
8 "time"
9
10 "github.com/bnema/vev/internal/domain"
11 portsmocks "github.com/bnema/vev/internal/ports/mocks"
12 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
13 "github.com/stretchr/testify/mock"
14 "github.com/stretchr/testify/require"
15)
16
17func TestSnapshotGarbageCollectionRunsAfterRestorationThenPeriodically(t *testing.T) {
18 committed := domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}
19 records := []domain.CatalogueRecord{{Name: "work", IncarnationID: domain.IncarnationID{1}, Committed: &committed}}
20 id := records[0].IncarnationID
21
22 tests := []struct {
23 name string
24 recordsErr error
25 wantCollect bool
26 }{
27 {name: "collects after restoration and on every tick", wantCollect: true},
28 {name: "unreadable catalogue never collects", recordsErr: errors.New("catalogue read failed")},
29 }
30 for _, tt := range tests {
31 t.Run(tt.name, func(t *testing.T) {
32 synctest.Test(t, func(t *testing.T) {
33 const passes = 2
34 catalogue := portsmocks.NewMockCatalogue(t)
35 repository := portsmocks.NewMockSnapshotRepository(t)
36 catalogue.EXPECT().Records().Return(records, tt.recordsErr).Times(passes)
37 repository.EXPECT().SnapshotIncarnations(mock.Anything).Return([]domain.IncarnationID{id}, nil).Times(passes)
38 collected := make(chan struct{}, passes)
39 if tt.wantCollect {
40 repository.EXPECT().CollectIncarnationGarbage(mock.Anything, id, &committed).
41 Run(func(context.Context, domain.IncarnationID, *domain.CheckpointRef) { collected <- struct{}{} }).
42 Return(nil).Times(passes)
43 }
44
45 clk := &signalClock{timers: make(chan *signalTimer, 1)}
46 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clk)
47 WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
48 WithSnapshotGarbageCollection()(d)
49 d.startDurableMaintenance()
50
51 synctest.Wait()
52 require.Empty(t, clk.timers, "GC must wait for restoration to finish")
53 d.closeRestoreDone()
54
55 for pass := range passes {
56 synctest.Wait()
57 timer := <-clk.timers
58 require.Equal(t, snapshotGarbageCollectionInterval, timer.duration)
59 if tt.wantCollect {
60 require.Len(t, collected, 1)
61 <-collected
62 }
63 if pass < passes-1 {
64 timer.ch <- time.Time{}
65 }
66 }
67
68 d.serveCancel()
69 d.WaitDurableWriters()
70 })
71 })
72 }
73}
74
75func TestSnapshotGarbageCollectionRequiresExplicitOption(t *testing.T) {
76 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
77 coordinator := recoveryusecase.NewCoordinator(portsmocks.NewMockCatalogue(t), portsmocks.NewMockSnapshotRepository(t), nil)
78 WithRecoveryCoordinator(coordinator)(d)
79 d.startDurableMaintenance()
80
81 d.snapshots.mu.Lock()
82 defer d.snapshots.mu.Unlock()
83 require.Nil(t, d.snapshots.maintenanceWorkerDone)
84}