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 portsmocks "github.com/bnema/vev/internal/ports/mocks"
17 "github.com/bnema/vev/internal/protocol"
18 snapcodec "github.com/bnema/vev/internal/snapshotcodec"
19 "github.com/bnema/vev/internal/usecase/picker"
20 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
21)
22
23func TestRestoreAmbiguousBadVersionLoadFailurePreservesCheckpoint(t *testing.T) {
24 record := durableRecoveryRecord(0)
25 repository := &durableRecoveryRepository{
26 errors: map[string]error{record.Name: errors.Join(snapcodec.ErrBadVersion, syscall.ENOSPC)},
27 loads: make(map[string]int),
28 repairs: make(map[string]domain.CheckpointRef),
29 }
30 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
31
32 d.restoreIncrementalSnapshots(context.Background())
33
34 persisted, ok, err := catalogue.Record(record.Name)
35 require.NoError(t, err)
36 require.True(t, ok)
37 require.Equal(t, record.IncarnationID, persisted.IncarnationID)
38 require.Equal(t, record.Committed, persisted.Committed)
39 require.Equal(t, "checkpoint load failed", persisted.DegradedReason)
40 require.Empty(t, repository.deleted)
41 d.mu.Lock()
42 entry := d.inactive[record.Name]
43 d.mu.Unlock()
44 require.Equal(t, protocol.SessionBroken, entry.state)
45 require.Equal(t, persisted, entry.record)
46}
47
48func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
49 record := durableRecoveryRecord(0)
50 repository := &durableRecoveryRepository{
51 errors: map[string]error{record.Name: syscall.ENOSPC},
52 loads: make(map[string]int),
53 repairs: make(map[string]domain.CheckpointRef),
54 }
55 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
56
57 d.restoreIncrementalSnapshots(context.Background())
58
59 persisted, ok, err := catalogue.Record(record.Name)
60 require.NoError(t, err)
61 require.True(t, ok)
62 require.Equal(t, "checkpoint load failed", persisted.DegradedReason)
63 d.mu.Lock()
64 entry := d.inactive[record.Name]
65 d.mu.Unlock()
66 require.Equal(t, protocol.SessionBroken, entry.state)
67 require.Equal(t, persisted, entry.record)
68 require.Equal(t, protocol.SessionBroken, listSessions(t, d).Sessions[0].State)
69 select {
70 case <-entry.restoreDone:
71 default:
72 t.Fatal("restore completion was not signaled")
73 }
74}
75
76type blockingDiscardRecoveryRepository struct {
77 *durableRecoveryRepository
78 deleteEntered chan struct{}
79 releaseDelete chan struct{}
80}
81
82func (r *blockingDiscardRecoveryRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
83 close(r.deleteEntered)
84 <-r.releaseDelete
85 return nil
86}
87
88func TestRestoreFailureCannotOverwriteDiscardReplacement(t *testing.T) {
89 record := durableRecoveryRecord(0)
90 record.DegradedReason = "previous restore failure"
91 repository := &blockingDiscardRecoveryRepository{
92 durableRecoveryRepository: &durableRecoveryRepository{},
93 deleteEntered: make(chan struct{}),
94 releaseDelete: make(chan struct{}),
95 }
96 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
97 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{7}, 16)))
98
99 discarded := make(chan error, 1)
100 go func() {
101 _, err := (controlExec{ctx: t.Context(), d: d, recoveryName: record.Name}).SessionRecovery("discard")
102 discarded <- err
103 }()
104 <-repository.deleteEntered // The fresh record is now authoritative behind Discard's mutation fence.
105
106 finished := make(chan struct{})
107 go func() {
108 d.finishRecordRestore(record, errors.New("late restore failure"), make(chan struct{}))
109 close(finished)
110 }()
111
112 close(repository.releaseDelete)
113 require.NoError(t, <-discarded)
114 <-finished
115
116 persisted, ok, err := catalogue.Record(record.Name)
117 require.NoError(t, err)
118 require.True(t, ok)
119 require.NotEqual(t, record.IncarnationID, persisted.IncarnationID)
120 require.Nil(t, persisted.Committed)
121 require.Empty(t, persisted.DegradedReason)
122 d.mu.Lock()
123 entry := d.inactive[record.Name]
124 d.mu.Unlock()
125 require.Equal(t, persisted, entry.record)
126 require.Equal(t, protocol.SessionDown, entry.state)
127}
128
129func TestStaleIncompatibleRestoreCannotOverwriteNewerAuthority(t *testing.T) {
130 tests := []struct {
131 name string
132 replacement func(domain.CatalogueRecord) domain.CatalogueRecord
133 state protocol.SessionState
134 }{
135 {
136 name: "newer checkpoint",
137 replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
138 ref := *record.Committed
139 ref.Generation++
140 ref.ManifestDigest[0]++
141 record.Committed = &ref
142 return record
143 },
144 state: protocol.SessionDown,
145 },
146 {
147 name: "newer incarnation",
148 replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
149 record.IncarnationID = domain.IncarnationID{9}
150 return record
151 },
152 state: protocol.SessionDown,
153 },
154 {
155 name: "degraded authority",
156 replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
157 record.DegradedReason = "newer restore decision"
158 return record
159 },
160 state: protocol.SessionBroken,
161 },
162 }
163 for _, tt := range tests {
164 t.Run(tt.name, func(t *testing.T) {
165 record := durableRecoveryRecord(0)
166 replacement := tt.replacement(record)
167 repository := &durableRecoveryRepository{
168 errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
169 loads: make(map[string]int),
170 repairs: make(map[string]domain.CheckpointRef),
171 }
172 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
173 var logs bytes.Buffer
174 d.log = slog.New(slog.NewJSONHandler(&logs, nil))
175 repository.onLoad = func() {
176 require.NoError(t, catalogue.Replace(record.Name, replacement))
177 d.setStoppedRecovery(replacement, tt.state)
178 }
179
180 d.restoreIncrementalSnapshots(context.Background())
181
182 persisted, ok, err := catalogue.Record(record.Name)
183 require.NoError(t, err)
184 require.True(t, ok)
185 require.Equal(t, replacement, persisted)
186 d.mu.Lock()
187 entry := d.inactive[record.Name]
188 d.mu.Unlock()
189 require.Equal(t, tt.state, entry.state)
190 require.Equal(t, replacement, entry.record)
191 require.Empty(t, repository.deleted)
192 daemonRequireEvent(t, daemonJSONLogs(t, logs.Bytes()), "snapshot_incompatible_reset_superseded")
193 })
194 }
195}
196
197func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {
198 d := newTestDaemon(t, nil, stubClock{})
199 done := make(chan struct{})
200
201 d.finishRecordRestore(domain.CatalogueRecord{Name: "removed"}, nil, done)
202
203 select {
204 case <-done:
205 default:
206 t.Fatal("restore completion was not signaled")
207 }
208}
209
210type blockingRecordCatalogue struct {
211 ports.Catalogue
212 record domain.CatalogueRecord
213 entered chan struct{}
214 release chan struct{}
215}
216
217func (c *blockingRecordCatalogue) Record(string) (domain.CatalogueRecord, bool, error) {
218 close(c.entered)
219 <-c.release
220 return c.record, true, nil
221}
222
223func TestCreateSessionRechecksShutdownAfterCatalogueRead(t *testing.T) {
224 record := durableRecoveryRecord(0)
225 catalogue := &blockingRecordCatalogue{record: record, entered: make(chan struct{}), release: make(chan struct{})}
226 d := newTestDaemon(t, nil, stubClock{})
227 d.catalogue = catalogue
228 d.persistEnabled = true
229 d.inactive[record.Name] = inactiveSessionFromRecord(record, protocol.SessionDown, make(chan struct{}))
230
231 result := make(chan error, 1)
232 go func() {
233 _, err := createSessionForTest(d, record.Name, false, "/tmp", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, nil)
234 result <- err
235 }()
236 <-catalogue.entered
237 d.mu.Lock()
238 d.closing = true
239 d.mu.Unlock()
240 close(catalogue.release)
241
242 err := <-result
243 require.ErrorContains(t, err, "daemon is shutting down")
244 d.mu.Lock()
245 defer d.mu.Unlock()
246 require.Empty(t, d.sessions)
247}
248
249type blockingResumeCatalogue struct {
250 *durableRecoveryCatalogue
251 entered chan struct{}
252 release chan struct{}
253 once sync.Once
254}
255
256func (c *blockingResumeCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
257 c.once.Do(func() { close(c.entered) })
258 <-c.release
259 return c.durableRecoveryCatalogue.Record(name)
260}
261
262func TestResumeStoppedSessionRejectsReplacementDuringCatalogueRead(t *testing.T) {
263 sourcePTY, releaseSource := newBlockingPTY(t)
264 defer releaseSource()
265 targetPTY, releaseTarget := newBlockingPTY(t)
266 defer releaseTarget()
267 d, from, ac, _ := newManualSessionWithPTYs(t, sourcePTY)
268 d.ptys = newFactorySeq(t, targetPTY)
269
270 record := durableRecoveryRecord(1)
271 record.Name = "stopped"
272 catalogue := &blockingResumeCatalogue{
273 durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}),
274 entered: make(chan struct{}),
275 release: make(chan struct{}),
276 }
277 d.catalogue = catalogue
278 d.persistEnabled = true
279 d.recovery = recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil)
280 d.inactive[record.Name] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
281
282 target := picker.Target{
283 Name: record.Name,
284 Stopped: true,
285 Incarnation: record.IncarnationID,
286 ExpectedCreatedAt: &record.CreatedAt,
287 }
288 result := make(chan bool, 1)
289 go func() { result <- d.resumeStoppedAndSwitch(from, ac, target) }()
290
291 <-catalogue.entered
292 replacement := record
293 replacement.IncarnationID = domain.IncarnationID{9}
294 replacement.CreatedAt++
295 replacement.UpdatedAt = replacement.CreatedAt
296 require.NoError(t, catalogue.Replace(replacement.Name, replacement))
297 d.mu.Lock()
298 d.inactive[replacement.Name] = inactiveSessionFromRecord(replacement, protocol.SessionDown, nil)
299 d.mu.Unlock()
300 close(catalogue.release)
301
302 require.False(t, <-result)
303 require.Same(t, from, ac.currentSession())
304 require.Empty(t, catalogue.MetadataUpdates())
305 d.mu.Lock()
306 defer d.mu.Unlock()
307 require.Nil(t, d.findByNameLocked(record.Name))
308 require.Equal(t, replacement.IncarnationID, d.inactive[record.Name].incarnation)
309}
310
311func TestSessionAttachTargetRejectsInactiveLifecycleReplacementDuringCatalogueRead(t *testing.T) {
312 d := newTestDaemon(t, newFactory(t, newQuietPTY()), stubClock{})
313 record := durableRecoveryRecord(1)
314 record.Name = "stopped"
315 record.Committed = nil
316 record.TabNames = []string{"main"}
317 record.TabRecords = []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "main"}}
318 catalogue := &blockingResumeCatalogue{
319 durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}),
320 entered: make(chan struct{}),
321 release: make(chan struct{}),
322 }
323 d.catalogue = catalogue
324 d.persistEnabled = true
325 d.recovery = recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil)
326 d.inactive[record.Name] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
327
328 target := protocol.SessionAttachTarget{
329 LifecycleID: record.IncarnationID, SessionName: record.Name,
330 TabID: "tab-1", TabIndex: protocol.NoTabIndex, Stopped: true,
331 }
332 transport, _ := newCapturingTransport(t)
333 result := make(chan error, 1)
334 go func() {
335 _, _, err := d.routeWithContext(context.Background(), protocol.Hello{
336 Version: protocol.Version, Intent: protocol.IntentAttach, Name: record.Name,
337 Size: domain.Size{Cols: 80, Rows: 24}, SessionTarget: ptrSessionAttachTarget(target),
338 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
339 }, transport)
340 result <- err
341 }()
342
343 <-catalogue.entered
344 replacement := record
345 replacement.IncarnationID = domain.IncarnationID{9}
346 replacement.CreatedAt++
347 replacement.UpdatedAt = replacement.CreatedAt
348 require.NoError(t, catalogue.Replace(replacement.Name, replacement))
349 d.mu.Lock()
350 d.inactive[replacement.Name] = inactiveSessionFromRecord(replacement, protocol.SessionDown, nil)
351 d.mu.Unlock()
352 close(catalogue.release)
353
354 var protocolErr *protoErr
355 require.ErrorAs(t, <-result, &protocolErr)
356 require.Equal(t, protocol.ErrNoSuchTarget, protocolErr.code)
357 require.Empty(t, catalogue.MetadataUpdates())
358 d.mu.Lock()
359 defer d.mu.Unlock()
360 require.Nil(t, d.findByNameLocked(record.Name))
361 require.Equal(t, replacement.IncarnationID, d.inactive[record.Name].incarnation)
362}
363
364func TestSessionAttachTargetRejectsTabAuthorityChangeDuringCatalogueRead(t *testing.T) {
365 for _, test := range []struct {
366 name string
367 record domain.CatalogueRecord
368 target domain.RemoteSessionTarget
369 mutate func(*domain.CatalogueRecord)
370 updateMap bool
371 }{
372 {
373 name: "stable selector replacement",
374 record: domain.CatalogueRecord{TabNames: []string{"main"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "main"}}},
375 target: domain.RemoteSessionTarget{Stopped: true, StoppedTab: domain.NewStableTabSelector("tab-1")},
376 mutate: func(record *domain.CatalogueRecord) {
377 record.TabRecords[0].StableID = "tab-2"
378 },
379 updateMap: true,
380 },
381 {
382 name: "ordinal selector replacement",
383 record: domain.CatalogueRecord{TabNames: []string{"main"}, TabRecords: []domain.CatalogueTabRecord{{Name: "main"}}},
384 target: domain.RemoteSessionTarget{Stopped: true, StoppedTab: domain.NewOrdinalTabSelector(0, "main", 1)},
385 mutate: func(record *domain.CatalogueRecord) {
386 record.TabNames[0] = "other"
387 record.TabRecords[0].Name = "other"
388 },
389 updateMap: true,
390 },
391 {
392 name: "catalogue-only metadata change",
393 record: domain.CatalogueRecord{TabNames: []string{"main"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "main"}}},
394 target: domain.RemoteSessionTarget{Stopped: true, StoppedTab: domain.NewStableTabSelector("tab-1")},
395 mutate: func(record *domain.CatalogueRecord) {
396 record.TabRecords[0].StableID = "tab-2"
397 },
398 },
399 } {
400 t.Run(test.name, func(t *testing.T) {
401 record := durableRecoveryRecord(1)
402 record.Name = "stopped"
403 record.Committed = nil
404 record.TabNames = append([]string(nil), test.record.TabNames...)
405 record.TabRecords = append([]domain.CatalogueTabRecord(nil), test.record.TabRecords...)
406 catalogue := &blockingResumeCatalogue{
407 durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}),
408 entered: make(chan struct{}),
409 release: make(chan struct{}),
410 }
411 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
412 d.catalogue = catalogue
413 d.persistEnabled = true
414 d.recovery = recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil)
415 d.inactive[record.Name] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
416
417 target := test.target
418 target.Endpoint = "remote"
419 target.DisplayOrigin = "remote"
420 target.LifecycleID = record.IncarnationID
421 target.SessionName = record.Name
422 transport, _ := newCapturingTransport(t)
423 result := make(chan error, 1)
424 go func() {
425 _, _, err := d.routeWithContext(context.Background(), protocol.Hello{
426 Version: protocol.Version, Intent: protocol.IntentAttach, Name: record.Name,
427 Size: domain.Size{Cols: 80, Rows: 24}, SessionTarget: ptrSessionAttachTarget(protocol.SessionAttachTargetFromRemote(target)),
428 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
429 }, transport)
430 result <- err
431 }()
432
433 <-catalogue.entered
434 replacement := record
435 replacement.TabNames = append([]string(nil), record.TabNames...)
436 replacement.TabRecords = append([]domain.CatalogueTabRecord(nil), record.TabRecords...)
437 test.mutate(&replacement)
438 replacement.UpdatedAt++
439 require.NoError(t, catalogue.Replace(replacement.Name, replacement))
440 if test.updateMap {
441 d.mu.Lock()
442 d.inactive[replacement.Name] = inactiveSessionFromRecord(replacement, protocol.SessionDown, nil)
443 d.mu.Unlock()
444 }
445 close(catalogue.release)
446
447 var protocolErr *protoErr
448 require.ErrorAs(t, <-result, &protocolErr)
449 require.Equal(t, protocol.ErrNoSuchTarget, protocolErr.code)
450 require.Empty(t, catalogue.MetadataUpdates())
451 d.mu.Lock()
452 defer d.mu.Unlock()
453 require.Nil(t, d.findByNameLocked(record.Name))
454 })
455 }
456}
457
458func TestSnapshotStopContextCancelIsIdempotent(t *testing.T) {
459 deadline := &snapshotShutdownDeadline{done: make(chan struct{})}
460 _, cancel := snapshotStopContext(deadline)
461
462 var wg sync.WaitGroup
463 for range 8 {
464 wg.Go(cancel)
465 }
466 wg.Wait()
467}