1package daemon
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "log/slog"
8 "sync"
9 "syscall"
10 "testing"
11
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
17 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
18)
19
20func TestRestoreAmbiguousBadVersionLoadFailurePreservesCheckpoint(t *testing.T) {
21 record := durableRecoveryRecord(0)
22 repository := &durableRecoveryRepository{
23 errors: map[string]error{record.Name: errors.Join(snapcodec.ErrBadVersion, syscall.ENOSPC)},
24 loads: make(map[string]int),
25 repairs: make(map[string]domain.CheckpointRef),
26 }
27 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
28
29 d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
30
31 persisted, ok, err := catalogue.Record(record.Name)
32 require.NoError(t, err)
33 require.True(t, ok)
34 require.Equal(t, record.IncarnationID, persisted.IncarnationID)
35 require.Equal(t, record.Committed, persisted.Committed)
36 require.Equal(t, "checkpoint load failed", persisted.DegradedReason)
37 require.Empty(t, repository.deleted)
38 d.mu.Lock()
39 entry := d.stopped[record.Name]
40 d.mu.Unlock()
41 require.Equal(t, ports.SessionBroken, entry.state)
42 require.Equal(t, persisted, entry.record)
43}
44
45func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
46 record := durableRecoveryRecord(0)
47 repository := &durableRecoveryRepository{
48 errors: map[string]error{record.Name: syscall.ENOSPC},
49 loads: make(map[string]int),
50 repairs: make(map[string]domain.CheckpointRef),
51 }
52 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
53
54 d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
55
56 persisted, ok, err := catalogue.Record(record.Name)
57 require.NoError(t, err)
58 require.True(t, ok)
59 require.Equal(t, "checkpoint load failed", persisted.DegradedReason)
60 d.mu.Lock()
61 entry := d.stopped[record.Name]
62 d.mu.Unlock()
63 require.Equal(t, ports.SessionBroken, entry.state)
64 require.Equal(t, persisted, entry.record)
65 require.Equal(t, ports.SessionBroken, listSessions(t, d).Sessions[0].State)
66 select {
67 case <-entry.restoreDone:
68 default:
69 t.Fatal("restore completion was not signaled")
70 }
71}
72
73type blockingDiscardRecoveryRepository struct {
74 *durableRecoveryRepository
75 deleteEntered chan struct{}
76 releaseDelete chan struct{}
77}
78
79func (r *blockingDiscardRecoveryRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
80 close(r.deleteEntered)
81 <-r.releaseDelete
82 return nil
83}
84
85func TestRestoreFailureCannotOverwriteDiscardReplacement(t *testing.T) {
86 record := durableRecoveryRecord(0)
87 record.DegradedReason = "previous restore failure"
88 repository := &blockingDiscardRecoveryRepository{
89 durableRecoveryRepository: &durableRecoveryRepository{},
90 deleteEntered: make(chan struct{}),
91 releaseDelete: make(chan struct{}),
92 }
93 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
94 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{7}, 16)))
95
96 discarded := make(chan error, 1)
97 go func() {
98 _, err := (controlExec{ctx: t.Context(), d: d, recoveryName: record.Name}).SessionRecovery("discard")
99 discarded <- err
100 }()
101 <-repository.deleteEntered // The fresh record is now authoritative behind Discard's mutation fence.
102
103 finished := make(chan struct{})
104 go func() {
105 d.finishRecordRestore(record, errors.New("late restore failure"), make(chan struct{}))
106 close(finished)
107 }()
108
109 close(repository.releaseDelete)
110 require.NoError(t, <-discarded)
111 <-finished
112
113 persisted, ok, err := catalogue.Record(record.Name)
114 require.NoError(t, err)
115 require.True(t, ok)
116 require.NotEqual(t, record.IncarnationID, persisted.IncarnationID)
117 require.Nil(t, persisted.Committed)
118 require.Empty(t, persisted.DegradedReason)
119 d.mu.Lock()
120 entry := d.stopped[record.Name]
121 d.mu.Unlock()
122 require.Equal(t, persisted, entry.record)
123 require.Equal(t, ports.SessionDown, entry.state)
124}
125
126func TestStaleIncompatibleRestoreCannotOverwriteNewerAuthority(t *testing.T) {
127 tests := []struct {
128 name string
129 replacement func(domain.CatalogueRecord) domain.CatalogueRecord
130 state ports.SessionState
131 }{
132 {
133 name: "newer checkpoint",
134 replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
135 ref := *record.Committed
136 ref.Generation++
137 ref.ManifestDigest[0]++
138 record.Committed = &ref
139 return record
140 },
141 state: ports.SessionDown,
142 },
143 {
144 name: "newer incarnation",
145 replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
146 record.IncarnationID = domain.IncarnationID{9}
147 return record
148 },
149 state: ports.SessionDown,
150 },
151 {
152 name: "degraded authority",
153 replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
154 record.DegradedReason = "newer restore decision"
155 return record
156 },
157 state: ports.SessionBroken,
158 },
159 }
160 for _, tt := range tests {
161 t.Run(tt.name, func(t *testing.T) {
162 record := durableRecoveryRecord(0)
163 replacement := tt.replacement(record)
164 repository := &durableRecoveryRepository{
165 errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
166 loads: make(map[string]int),
167 repairs: make(map[string]domain.CheckpointRef),
168 }
169 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
170 var logs bytes.Buffer
171 d.log = slog.New(slog.NewJSONHandler(&logs, nil))
172 repository.onLoad = func() {
173 require.NoError(t, catalogue.Replace(record.Name, replacement))
174 d.setStoppedRecovery(replacement, tt.state)
175 }
176
177 d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
178
179 persisted, ok, err := catalogue.Record(record.Name)
180 require.NoError(t, err)
181 require.True(t, ok)
182 require.Equal(t, replacement, persisted)
183 d.mu.Lock()
184 entry := d.stopped[record.Name]
185 d.mu.Unlock()
186 require.Equal(t, tt.state, entry.state)
187 require.Equal(t, replacement, entry.record)
188 require.Empty(t, repository.deleted)
189 daemonRequireEvent(t, daemonJSONLogs(t, logs.Bytes()), "snapshot_incompatible_reset_superseded")
190 })
191 }
192}
193
194func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {
195 d := newTestDaemon(t, nil, stubClock{})
196 done := make(chan struct{})
197
198 d.finishRecordRestore(domain.CatalogueRecord{Name: "removed"}, nil, done)
199
200 select {
201 case <-done:
202 default:
203 t.Fatal("restore completion was not signaled")
204 }
205}
206
207type blockingRecordCatalogue struct {
208 ports.Catalogue
209 record domain.CatalogueRecord
210 entered chan struct{}
211 release chan struct{}
212}
213
214func (c *blockingRecordCatalogue) Record(string) (domain.CatalogueRecord, bool, error) {
215 close(c.entered)
216 <-c.release
217 return c.record, true, nil
218}
219
220func TestCreateSessionRechecksShutdownAfterCatalogueRead(t *testing.T) {
221 record := durableRecoveryRecord(0)
222 catalogue := &blockingRecordCatalogue{record: record, entered: make(chan struct{}), release: make(chan struct{})}
223 d := newTestDaemon(t, nil, stubClock{})
224 d.catalogue = catalogue
225 d.persistEnabled = true
226 d.stopped[record.Name] = stoppedSessionFromRecord(record, ports.SessionDown, make(chan struct{}))
227
228 result := make(chan error, 1)
229 go func() {
230 _, err := createSessionForTest(d, record.Name, false, "/tmp", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, nil)
231 result <- err
232 }()
233 <-catalogue.entered
234 d.mu.Lock()
235 d.closing = true
236 d.mu.Unlock()
237 close(catalogue.release)
238
239 err := <-result
240 require.ErrorContains(t, err, "daemon is shutting down")
241 d.mu.Lock()
242 defer d.mu.Unlock()
243 require.Empty(t, d.sessions)
244}
245
246func TestSnapshotStopContextCancelIsIdempotent(t *testing.T) {
247 deadline := &snapshotShutdownDeadline{done: make(chan struct{})}
248 _, cancel := snapshotStopContext(deadline)
249
250 var wg sync.WaitGroup
251 for range 8 {
252 wg.Go(cancel)
253 }
254 wg.Wait()
255}