1package daemon
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "log/slog"
10 "os"
11 "sync"
12 "sync/atomic"
13 "syscall"
14 "testing"
15 "time"
16
17 "github.com/stretchr/testify/mock"
18 "github.com/stretchr/testify/require"
19
20 "github.com/bnema/vev/internal/domain"
21 "github.com/bnema/vev/internal/ports"
22 portsmocks "github.com/bnema/vev/internal/ports/mocks"
23 "github.com/bnema/vev/internal/protocol"
24 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
25 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
26)
27
28type durableRecoveryCatalogue struct {
29 ports.Catalogue
30 mu sync.Mutex
31 records map[string]domain.CatalogueRecord
32 metadataUpdates []domain.CatalogueMetadataUpdate
33 replaceErr error
34}
35
36type recordErrorCatalogue struct {
37 *durableRecoveryCatalogue
38 err error
39}
40
41func (c recordErrorCatalogue) Record(string) (domain.CatalogueRecord, bool, error) {
42 return domain.CatalogueRecord{}, false, c.err
43}
44
45func newDurableRecoveryCatalogue(records []domain.CatalogueRecord) *durableRecoveryCatalogue {
46 catalogue := &durableRecoveryCatalogue{records: make(map[string]domain.CatalogueRecord, len(records))}
47 for _, record := range records {
48 catalogue.records[record.Name] = record
49 }
50 return catalogue
51}
52
53func (c *durableRecoveryCatalogue) Records() ([]domain.CatalogueRecord, error) {
54 c.mu.Lock()
55 defer c.mu.Unlock()
56 records := make([]domain.CatalogueRecord, 0, len(c.records))
57 for _, record := range c.records {
58 records = append(records, record)
59 }
60 return records, nil
61}
62
63func (c *durableRecoveryCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
64 c.mu.Lock()
65 defer c.mu.Unlock()
66 record, ok := c.records[name]
67 return record, ok, nil
68}
69
70func mustDurableRecords(t *testing.T, c *durableRecoveryCatalogue) []domain.CatalogueRecord {
71 t.Helper()
72 records, err := c.Records()
73 require.NoError(t, err)
74 return records
75}
76
77func (c *durableRecoveryCatalogue) Create(record domain.CatalogueRecord) error {
78 c.mu.Lock()
79 defer c.mu.Unlock()
80 if _, exists := c.records[record.Name]; exists {
81 return errors.New("catalogue record already exists")
82 }
83 c.records[record.Name] = record
84 return nil
85}
86
87func (c *durableRecoveryCatalogue) Delete(name string) error {
88 c.mu.Lock()
89 defer c.mu.Unlock()
90 delete(c.records, name)
91 return nil
92}
93
94func (c *durableRecoveryCatalogue) Replace(name string, record domain.CatalogueRecord) error {
95 c.mu.Lock()
96 defer c.mu.Unlock()
97 if c.replaceErr != nil {
98 return c.replaceErr
99 }
100 if name != record.Name {
101 return errors.New("catalogue key mismatch")
102 }
103 c.records[name] = record
104 return nil
105}
106
107func (c *durableRecoveryCatalogue) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {
108 c.mu.Lock()
109 defer c.mu.Unlock()
110 record, ok := c.records[update.Name]
111 if !ok {
112 return errors.New("catalogue record not found")
113 }
114 if update.IncarnationID == (domain.IncarnationID{}) || record.IncarnationID != update.IncarnationID {
115 return errors.New("catalogue incarnation changed")
116 }
117 if update.Cwd != nil {
118 record.Cwd = *update.Cwd
119 }
120 if update.UpdatedAt != nil {
121 record.UpdatedAt = *update.UpdatedAt
122 }
123 if update.LastUsedSeq != nil {
124 record.LastUsedSeq = *update.LastUsedSeq
125 }
126 if update.TabNames != nil {
127 record.TabNames = append([]string(nil), (*update.TabNames)...)
128 }
129 c.records[update.Name] = record
130 c.metadataUpdates = append(c.metadataUpdates, update)
131 return nil
132}
133
134func (c *durableRecoveryCatalogue) MetadataUpdates() []domain.CatalogueMetadataUpdate {
135 c.mu.Lock()
136 defer c.mu.Unlock()
137 return append([]domain.CatalogueMetadataUpdate(nil), c.metadataUpdates...)
138}
139
140func (c *durableRecoveryCatalogue) Sync() error { return nil }
141func (c *durableRecoveryCatalogue) Close() error { return nil }
142
143type durableRecoveryRepository struct {
144 ports.SnapshotRepository
145 mu sync.Mutex
146 generations map[string]ports.SnapshotGeneration
147 errors map[string]error
148 loads map[string]int
149 repairs map[string]domain.CheckpointRef
150 deleted []domain.IncarnationID
151 onLoad func()
152 repairErr error
153 deleteErr error
154 repairCalls int
155}
156
157func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
158 r.mu.Lock()
159 defer r.mu.Unlock()
160 if r.loads == nil {
161 r.loads = make(map[string]int)
162 }
163 r.loads[name]++
164 if r.onLoad != nil {
165 r.onLoad()
166 }
167 if err := r.errors[name]; err != nil {
168 return ports.SnapshotGeneration{}, err
169 }
170 generation, ok := r.generations[name]
171 if !ok || generation.Generation != ref.Generation {
172 return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
173 }
174 return generation, nil
175}
176
177func (r *durableRecoveryRepository) ReconcileCheckpoint(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
178 r.mu.Lock()
179 defer r.mu.Unlock()
180 r.repairCalls++
181 if r.repairErr != nil {
182 return r.repairErr
183 }
184 if r.repairs == nil {
185 r.repairs = make(map[string]domain.CheckpointRef)
186 }
187 r.repairs[id.String()] = ref
188 return nil
189}
190
191func (r *durableRecoveryRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
192 r.mu.Lock()
193 defer r.mu.Unlock()
194 r.deleted = append(r.deleted, id)
195 return r.deleteErr
196}
197
198func durableRecoveryRecord(index int) domain.CatalogueRecord {
199 id := domain.IncarnationID{0: byte(index>>8 + 1), 1: byte(index + 1)}
200 ref := domain.CheckpointRef{Generation: uint64(index + 1), ManifestDigest: [32]byte{byte(index%255 + 1)}}
201 return domain.CatalogueRecord{
202 Name: fmt.Sprintf("session-%04d", index),
203 IncarnationID: id,
204 Cwd: "/tmp",
205 CreatedAt: int64(index + 1),
206 UpdatedAt: int64(index + 1),
207 Committed: &ref,
208 }
209}
210
211func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.SnapshotGeneration {
212 t.Helper()
213 manifest, err := snapcodec.MarshalManifest(snapcodec.Manifest{
214 Generation: record.Committed.Generation,
215 IncarnationID: record.IncarnationID,
216 Name: record.Name,
217 CreatedAt: uint64(record.CreatedAt),
218 })
219 require.NoError(t, err)
220 return ports.SnapshotGeneration{
221 IncarnationID: record.IncarnationID,
222 Name: record.Name,
223 Generation: record.Committed.Generation,
224 Manifest: manifest,
225 Objects: map[ports.SnapshotDigest][]byte{},
226 }
227}
228
229func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {
230 t.Helper()
231 catalogue := newDurableRecoveryCatalogue(records)
232 coordinator := recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{0xa5}, 16)))
233 d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository), WithRecoveryCoordinator(coordinator))
234 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
235 t.Cleanup(d.serveCancel)
236 return d, catalogue
237}
238
239func TestFinalCheckpointFailurePreservesCatalogue(t *testing.T) {
240 priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
241 prior := domain.CatalogueRecord{
242 Name: "work",
243 IncarnationID: domain.IncarnationID{1},
244 Cwd: "/work",
245 CreatedAt: 42,
246 Committed: &priorRef,
247 }
248 catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{prior})
249 repository := &durableRecoveryRepository{}
250 coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
251 d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, []domain.CatalogueRecord{prior}), WithSnapshotRepository(repository), WithRecoveryCoordinator(coordinator))
252 startSnapshotEncodeWorker(t, d)
253
254 sess := newSnapshotTestSession(t, prior.Name, false, prior.Cwd)
255 expired, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
256 defer cancel()
257 sess.snapshotMu.Lock()
258 sess.snapshotPublishedGeneration = priorRef.Generation
259 sess.snapshotPublishedCheckpoint = &priorRef
260 sess.snapshotGeneration = priorRef.Generation + 1
261 sess.snapshotPublicationContext = expired
262 sess.snapDirty.Store(true)
263 sess.snapshotMu.Unlock()
264
265 require.True(t, d.scheduleFinalSnapshot(sess))
266 awaitSnapshotIdle(t, sess)
267 got, ok, _ := catalogue.Record(prior.Name)
268 require.True(t, ok)
269 require.Equal(t, prior, got, "a timed-out final checkpoint must not change authoritative bytes")
270 sess.snapshotMu.Lock()
271 published := sess.snapshotPublishedCheckpoint
272 sess.snapshotMu.Unlock()
273 require.Equal(t, priorRef, *published)
274}
275
276func TestCatalogueRestoreIndependent(t *testing.T) {
277 t.Run("empty catalogue restoration", func(t *testing.T) {
278 emptyDaemon, _ := newDurableRecoveryDaemon(t, nil, &durableRecoveryRepository{})
279 emptyDone := make(chan struct{})
280 go func() {
281 emptyDaemon.restoreIncrementalSnapshots(context.Background())
282 close(emptyDone)
283 }()
284 select {
285 case <-emptyDone:
286 case <-time.After(time.Second):
287 t.Fatal("empty catalogue restoration blocked")
288 }
289 })
290
291 t.Run("healthy and broken restoration", func(t *testing.T) {
292 healthy := durableRecoveryRecord(0)
293 broken := durableRecoveryRecord(1)
294 repository := &durableRecoveryRepository{
295 generations: map[string]ports.SnapshotGeneration{healthy.Name: validGeneration(t, healthy)},
296 errors: map[string]error{broken.Name: errors.New("corrupt checkpoint")},
297 loads: make(map[string]int),
298 repairs: make(map[string]domain.CheckpointRef),
299 }
300 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{healthy, broken}, repository)
301
302 d.restoreIncrementalSnapshots(context.Background())
303
304 d.mu.Lock()
305 healthyEntry := d.inactive[healthy.Name]
306 brokenEntry := d.inactive[broken.Name]
307 d.mu.Unlock()
308 require.Equal(t, protocol.SessionDown, healthyEntry.state)
309 require.Equal(t, protocol.SessionBroken, brokenEntry.state)
310 for name, done := range map[string]<-chan struct{}{healthy.Name: healthyEntry.restoreDone, broken.Name: brokenEntry.restoreDone} {
311 select {
312 case <-done:
313 default:
314 t.Fatalf("restore completion for %q was not closed", name)
315 }
316 }
317 brokenRecord, ok, _ := catalogue.Record(broken.Name)
318 require.True(t, ok)
319 require.Equal(t, "checkpoint validation failed", brokenRecord.DegradedReason)
320 require.Equal(t, 1, repository.loads[healthy.Name])
321 require.Equal(t, 1, repository.loads[broken.Name])
322 require.Empty(t, d.sessions, "a corrupt record must never open a replacement runtime")
323 })
324
325}
326
327func TestCatalogueRestoreResetsOnlyIncompatibleCheckpoint(t *testing.T) {
328 incompatible := durableRecoveryRecord(0)
329 healthy := durableRecoveryRecord(1)
330 repository := &durableRecoveryRepository{
331 generations: map[string]ports.SnapshotGeneration{healthy.Name: validGeneration(t, healthy)},
332 errors: map[string]error{incompatible.Name: fmt.Errorf("load envelope: %w", snapcodec.ErrBadVersion)},
333 loads: make(map[string]int),
334 repairs: make(map[string]domain.CheckpointRef),
335 }
336 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{incompatible, healthy}, repository)
337
338 d.restoreIncrementalSnapshots(context.Background())
339
340 fresh, ok, err := catalogue.Record(incompatible.Name)
341 require.NoError(t, err)
342 require.True(t, ok)
343 require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{0xa5}, 16)), fresh.IncarnationID)
344 require.Nil(t, fresh.Committed)
345 require.Empty(t, fresh.DegradedReason)
346 require.Empty(t, fresh.TabNames)
347 require.Equal(t, []domain.IncarnationID{incompatible.IncarnationID}, repository.deleted)
348 require.Equal(t, protocol.SessionDown, d.inactive[incompatible.Name].state)
349 require.Equal(t, fresh, d.inactive[incompatible.Name].record)
350
351 restoredSibling, ok, err := catalogue.Record(healthy.Name)
352 require.NoError(t, err)
353 require.True(t, ok)
354 require.Equal(t, healthy, restoredSibling)
355 require.Equal(t, protocol.SessionDown, d.inactive[healthy.Name].state)
356 require.Equal(t, 1, repository.loads[healthy.Name])
357}
358
359func TestIncompatibleCheckpointDeleteFailurePublishesFreshStoppedAuthority(t *testing.T) {
360 record := durableRecoveryRecord(0)
361 deleteErr := errors.New("delete old incarnation")
362 repository := &durableRecoveryRepository{
363 errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
364 loads: make(map[string]int),
365 repairs: make(map[string]domain.CheckpointRef),
366 deleteErr: deleteErr,
367 }
368 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
369 d.mu.Lock()
370 restoreDone := d.inactive[record.Name].restoreDone
371 d.mu.Unlock()
372
373 restoreErr := d.restoreRecord(context.Background(), record, d.purgeEpochSnapshot())
374 require.NoError(t, restoreErr, "post-commit cleanup failure must not fail the authority transition")
375
376 fresh, ok, err := catalogue.Record(record.Name)
377 require.NoError(t, err)
378 require.True(t, ok)
379 require.NotEqual(t, record.IncarnationID, fresh.IncarnationID)
380 require.Nil(t, fresh.Committed)
381 require.Empty(t, fresh.DegradedReason)
382 d.mu.Lock()
383 entry := d.inactive[record.Name]
384 d.mu.Unlock()
385 require.Equal(t, protocol.SessionDown, entry.state)
386 require.Equal(t, fresh, entry.record)
387 require.Equal(t, restoreDone, entry.restoreDone)
388 select {
389 case <-entry.restoreDone:
390 t.Fatal("direct restore closed its coordinator-owned completion barrier")
391 default:
392 }
393
394 d.finishRecordRestore(record, restoreErr, restoreDone)
395 d.mu.Lock()
396 entry = d.inactive[record.Name]
397 d.mu.Unlock()
398 require.Equal(t, protocol.SessionDown, entry.state)
399 require.Equal(t, fresh, entry.record)
400 select {
401 case <-entry.restoreDone:
402 default:
403 t.Fatal("restore completion was not signaled")
404 }
405 require.Equal(t, []domain.IncarnationID{record.IncarnationID}, repository.deleted)
406}
407
408func TestIncompatibleCheckpointPreCommitFailureReturnsError(t *testing.T) {
409 record := durableRecoveryRecord(0)
410 cause := errors.New("catalogue replace failed")
411 repository := &durableRecoveryRepository{
412 errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
413 loads: make(map[string]int),
414 repairs: make(map[string]domain.CheckpointRef),
415 }
416 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
417 catalogue.replaceErr = cause
418
419 err := d.restoreRecord(context.Background(), record, d.purgeEpochSnapshot())
420
421 require.ErrorIs(t, err, cause)
422 persisted, ok, recordErr := catalogue.Record(record.Name)
423 require.NoError(t, recordErr)
424 require.True(t, ok)
425 require.Equal(t, record, persisted)
426 require.Empty(t, repository.deleted)
427}
428
429func TestRestoreLoadFailuresRetainCheckpointWithoutDeletion(t *testing.T) {
430 tests := []struct {
431 name string
432 err error
433 }{
434 {name: "corruption", err: errors.New("corrupt checkpoint")},
435 {name: "generic", err: errors.New("load failed")},
436 {name: "io", err: &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.EACCES}},
437 }
438 for _, tt := range tests {
439 t.Run(tt.name, func(t *testing.T) {
440 record := durableRecoveryRecord(0)
441 repository := &durableRecoveryRepository{
442 errors: map[string]error{record.Name: tt.err},
443 loads: make(map[string]int),
444 repairs: make(map[string]domain.CheckpointRef),
445 }
446 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
447
448 d.restoreIncrementalSnapshots(context.Background())
449
450 persisted, ok, err := catalogue.Record(record.Name)
451 require.NoError(t, err)
452 require.True(t, ok)
453 require.Equal(t, record.IncarnationID, persisted.IncarnationID)
454 require.Equal(t, record.Committed, persisted.Committed)
455 require.Equal(t, "checkpoint validation failed", persisted.DegradedReason)
456 require.Equal(t, protocol.SessionBroken, d.inactive[record.Name].state)
457 require.Empty(t, repository.deleted)
458 })
459 }
460}
461
462func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
463 const count = 4200
464 records := make([]domain.CatalogueRecord, count)
465 repository := &durableRecoveryRepository{generations: make(map[string]ports.SnapshotGeneration, count), errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
466 for i := range records {
467 records[i] = durableRecoveryRecord(i)
468 repository.generations[records[i].Name] = validGeneration(t, records[i])
469 }
470 d, _ := newDurableRecoveryDaemon(t, records, repository)
471
472 d.restoreIncrementalSnapshots(context.Background())
473
474 require.Len(t, d.inactive, count)
475 for _, record := range records {
476 require.Equal(t, protocol.SessionDown, d.inactive[record.Name].state)
477 require.Equal(t, 1, repository.loads[record.Name])
478 }
479}
480
481func TestCatalogueRestoreSingleSessionOver4096(t *testing.T) {
482 record := durableRecoveryRecord(0)
483 repository := &durableRecoveryRepository{generations: map[string]ports.SnapshotGeneration{record.Name: validGeneration(t, record)}, errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
484 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
485
486 // The repository may contain any number of stale entries; restoration has no
487 // discovery API and therefore performs exactly one direct catalogue lookup.
488 d.restoreIncrementalSnapshots(context.Background())
489
490 require.Equal(t, protocol.SessionDown, d.inactive[record.Name].state)
491 require.Equal(t, 1, repository.loads[record.Name])
492}
493
494func TestCatalogueRestoreIncarnationMismatch(t *testing.T) {
495 record := durableRecoveryRecord(0)
496 generation := validGeneration(t, record)
497 generation.IncarnationID = domain.IncarnationID{99}
498 repository := &durableRecoveryRepository{generations: map[string]ports.SnapshotGeneration{record.Name: generation}, errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
499 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
500
501 d.restoreIncrementalSnapshots(context.Background())
502
503 require.Equal(t, protocol.SessionBroken, d.inactive[record.Name].state)
504 got, _, _ := catalogue.Record(record.Name)
505 require.Equal(t, "checkpoint validation failed", got.DegradedReason)
506}
507
508type cancellationRecoveryRepository struct {
509 ports.SnapshotRepository
510 started chan struct{}
511 once sync.Once
512}
513
514func (r *cancellationRecoveryRepository) LoadCheckpoint(ctx context.Context, _ domain.IncarnationID, _ string, _ ports.CheckpointRef) (ports.SnapshotGeneration, error) {
515 r.once.Do(func() { close(r.started) })
516 <-ctx.Done()
517 return ports.SnapshotGeneration{}, ctx.Err()
518}
519
520type recoveryCountingPTYFactory struct {
521 calls atomic.Int32
522}
523
524func (f *recoveryCountingPTYFactory) Open(context.Context, string, []string, []string, string, domain.Geometry) (ports.PTY, error) {
525 f.calls.Add(1)
526 return nil, errors.New("unexpected PTY open")
527}
528
529func TestSessionRecoveryUsesIncomingCommandContext(t *testing.T) {
530 type contextKey struct{}
531 key := contextKey{}
532 ctx := context.WithValue(t.Context(), key, "request")
533 wantErr := errors.New("stop after context capture")
534 record := durableRecoveryRecord(0)
535 record.DegradedReason = "checkpoint validation failed"
536 catalogue := portsmocks.NewMockCatalogue(t)
537 repository := portsmocks.NewMockSnapshotRepository(t)
538 catalogue.EXPECT().Record(record.Name).Return(record, true, nil).Once()
539 catalogue.EXPECT().Replace(record.Name, mock.Anything).Return(nil).Once()
540 seen := false
541 repository.EXPECT().DeleteIncarnation(mock.Anything, record.IncarnationID).
542 Run(func(got context.Context, _ domain.IncarnationID) { seen = got.Value(key) == "request" }).
543 Return(wantErr).Once()
544
545 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
546 d.catalogue = catalogue
547 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{9}, 16)))
548
549 _, err := (controlExec{ctx: ctx, d: d, recoveryName: record.Name}).SessionRecovery("discard")
550 require.ErrorIs(t, err, wantErr)
551 require.True(t, seen)
552}
553
554type temporaryRestoreError struct{}
555
556func (temporaryRestoreError) Error() string { return "temporary restore failure" }
557func (temporaryRestoreError) Temporary() bool { return true }
558
559type badVersionLookalikeError struct{}
560
561func (badVersionLookalikeError) Error() string { return "bad version lookalike" }
562func (badVersionLookalikeError) Is(target error) bool { return target == snapcodec.ErrBadVersion }
563
564type cyclicUnwrapError struct{}
565
566func (*cyclicUnwrapError) Error() string { return "cyclic unwrap" }
567func (e *cyclicUnwrapError) Unwrap() error {
568 return e
569}
570
571func TestUnambiguousBadVersionErrorClassification(t *testing.T) {
572 tests := []struct {
573 name string
574 err error
575 incompatible bool
576 }{
577 {name: "nil"},
578 {name: "exact sentinel", err: snapcodec.ErrBadVersion, incompatible: true},
579 {name: "single cause chain", err: fmt.Errorf("load checkpoint: %w", fmt.Errorf("decode manifest: %w", snapcodec.ErrBadVersion)), incompatible: true},
580 {name: "single entry multi cause", err: errors.Join(snapcodec.ErrBadVersion)},
581 {name: "joined retryable branch", err: errors.Join(snapcodec.ErrBadVersion, syscall.ENOSPC)},
582 {name: "custom Is lookalike", err: badVersionLookalikeError{}},
583 {name: "wrapped custom Is lookalike", err: fmt.Errorf("load checkpoint: %w", badVersionLookalikeError{})},
584 {name: "chain ending elsewhere", err: fmt.Errorf("load checkpoint: %w", errors.New("corrupt checkpoint"))},
585 {name: "cyclic custom chain", err: &cyclicUnwrapError{}},
586 }
587 for _, tt := range tests {
588 t.Run(tt.name, func(t *testing.T) {
589 require.Equal(t, tt.incompatible, isUnambiguousBadVersionError(tt.err))
590 })
591 }
592}
593
594func TestRetryableRestoreLoadErrorClassification(t *testing.T) {
595 tests := []struct {
596 name string
597 err error
598 retryable bool
599 }{
600 {name: "missing path", err: fmt.Errorf("wrapped: %w", &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.ENOENT})},
601 {name: "permission denied", err: fmt.Errorf("wrapped: %w", &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.EACCES})},
602 {name: "invalid path operation", err: fmt.Errorf("wrapped: %w", &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.EINVAL})},
603 {name: "temporary path failure", err: fmt.Errorf("wrapped: %w", &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.EAGAIN}), retryable: true},
604 {name: "resource exhausted path", err: fmt.Errorf("wrapped: %w", &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.ENOSPC}), retryable: true},
605 {name: "temporary interface", err: fmt.Errorf("wrapped: %w", temporaryRestoreError{}), retryable: true},
606 {name: "corrupt data", err: errors.New("corrupt checkpoint")},
607 {name: "context canceled", err: context.Canceled},
608 {name: "context deadline exceeded", err: context.DeadlineExceeded},
609 }
610 for _, tt := range tests {
611 t.Run(tt.name, func(t *testing.T) {
612 require.Equal(t, tt.retryable, isRetryableRestoreLoadError(tt.err))
613 })
614 }
615}
616
617func TestFinishRecordRestoreDoesNotDegradeStaleCatalogueAuthority(t *testing.T) {
618 for _, tt := range []struct {
619 name string
620 setup func(*Daemon, *durableRecoveryCatalogue, domain.CatalogueRecord)
621 }{
622 {
623 name: "missing record",
624 setup: func(_ *Daemon, c *durableRecoveryCatalogue, record domain.CatalogueRecord) {
625 c.mu.Lock()
626 delete(c.records, record.Name)
627 c.mu.Unlock()
628 },
629 },
630 {
631 name: "replacement incarnation",
632 setup: func(_ *Daemon, c *durableRecoveryCatalogue, record domain.CatalogueRecord) {
633 record.IncarnationID = domain.IncarnationID{99}
634 c.mu.Lock()
635 c.records[record.Name] = record
636 c.mu.Unlock()
637 },
638 },
639 {
640 name: "catalogue read error",
641 setup: func(d *Daemon, c *durableRecoveryCatalogue, _ domain.CatalogueRecord) {
642 d.catalogue = recordErrorCatalogue{durableRecoveryCatalogue: c, err: errors.New("read failed")}
643 d.recovery = recoveryusecase.NewCoordinator(d.catalogue, nil, nil)
644 },
645 },
646 } {
647 t.Run(tt.name, func(t *testing.T) {
648 record := durableRecoveryRecord(0)
649 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
650 tt.setup(d, catalogue, record)
651 d.mu.Lock()
652 before := d.inactive[record.Name]
653 d.mu.Unlock()
654
655 d.finishRecordRestore(record, errors.New("invalid checkpoint"), before.restoreDone)
656
657 d.mu.Lock()
658 after := d.inactive[record.Name]
659 d.mu.Unlock()
660 require.Equal(t, before.record, after.record)
661 require.Equal(t, before.state, after.state)
662 })
663 }
664}
665
666func TestPersistAndRegisterRestoredSessionRegistrationFailurePreservesStoppedEntry(t *testing.T) {
667 record := durableRecoveryRecord(0)
668 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
669 d.mu.Lock()
670 before := d.inactive[record.Name]
671 collisionID := domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
672 d.sessions[collisionID] = &session{sessionCore: sessionCore{id: collisionID, name: "collision"}}
673 d.mu.Unlock()
674 sess := newSnapshotTestSession(t, record.Name, false, record.Cwd)
675 sess.incarnation = record.IncarnationID
676
677 registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess, d.purgeEpochSnapshot())
678
679 require.ErrorContains(t, err, "registry rejected")
680 require.False(t, registered)
681 d.mu.Lock()
682 after := d.inactive[record.Name]
683 d.mu.Unlock()
684 require.Equal(t, before, after, "failed runtime publication must preserve the exact stopped authority and barrier")
685 require.Equal(t, before.restoreDone, after.restoreDone)
686}
687
688func TestCreateNamedSessionRegistrationFailureRollsBackDurableState(t *testing.T) {
689 for _, tc := range []struct {
690 name string
691 records []domain.CatalogueRecord
692 sessName string
693 cwd string
694 }{
695 {name: "fresh create", sessName: "fresh", cwd: "/fresh"},
696 {name: "stopped metadata update", records: []domain.CatalogueRecord{durableRecoveryRecord(0)}, sessName: durableRecoveryRecord(0).Name, cwd: "/resumed"},
697 } {
698 t.Run(tc.name, func(t *testing.T) {
699 pty, release := newBlockingPTY(t)
700 defer release()
701 catalogue := newDurableRecoveryCatalogue(tc.records)
702 coordinator := recoveryusecase.NewCoordinator(catalogue, nil, bytes.NewReader(bytes.Repeat([]byte{0x6a}, 16)))
703 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
704 WithCatalogue(catalogue, tc.records)(d)
705 WithRecoveryCoordinator(coordinator)(d)
706
707 d.mu.Lock()
708 for _, record := range tc.records {
709 d.inactive[record.Name] = inactiveSession{
710 name: record.Name, cwd: record.Cwd, createdAt: record.CreatedAt,
711 incarnation: record.IncarnationID, lastUsedSeq: record.LastUsedSeq,
712 tabNames: append([]string(nil), record.TabNames...), record: record,
713 state: protocol.SessionDown, restoreDone: make(chan struct{}),
714 }
715 }
716 beforeStopped, hadStopped := d.inactive[tc.sessName]
717 collisionID := domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
718 d.sessions[collisionID] = &session{sessionCore: sessionCore{id: collisionID, name: "collision"}}
719 d.mu.Unlock()
720 beforeRecords := mustDurableRecords(t, catalogue)
721
722 _, err := createSessionForTest(d, tc.sessName, false, tc.cwd, defaultSize, terminalEnv{}, d.baseEnv)
723
724 require.ErrorContains(t, err, "registry rejected")
725 require.ElementsMatch(t, beforeRecords, mustDurableRecords(t, catalogue), "failed runtime publication must leave no created record or metadata mutation")
726 d.mu.Lock()
727 afterStopped, stillStopped := d.inactive[tc.sessName]
728 d.mu.Unlock()
729 require.Equal(t, hadStopped, stillStopped)
730 if hadStopped {
731 require.Equal(t, beforeStopped, afterStopped, "resume publication failure must restore the exact stopped state")
732 require.Equal(t, beforeStopped.restoreDone, afterStopped.restoreDone)
733 }
734 })
735 }
736}
737
738func TestPersistAndRegisterRestoredSessionRejectsReplacementIncarnation(t *testing.T) {
739 record := durableRecoveryRecord(0)
740 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
741 replacement := record
742 replacement.IncarnationID = domain.IncarnationID{99}
743 require.NoError(t, catalogue.Replace(record.Name, replacement))
744 sess := newSnapshotTestSession(t, record.Name, false, record.Cwd)
745 sess.incarnation = record.IncarnationID
746
747 registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess, d.purgeEpochSnapshot())
748 require.ErrorContains(t, err, "incarnation")
749 require.False(t, registered)
750 require.Empty(t, d.sessions)
751}
752
753func TestSessionRecoveryCommand(t *testing.T) {
754 degraded := durableRecoveryRecord(2)
755 degraded.DegradedReason = "checkpoint validation failed"
756 fresh := degraded
757 fresh.IncarnationID = domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
758 fresh.Committed = nil
759 fresh.DegradedReason = ""
760 catalogue := portsmocks.NewMockCatalogue(t)
761 repository := portsmocks.NewMockSnapshotRepository(t)
762 catalogue.EXPECT().Record(degraded.Name).Return(degraded, true, nil).Once()
763 catalogue.EXPECT().Replace(degraded.Name, fresh).Return(nil).Once()
764 repository.EXPECT().DeleteIncarnation(mock.Anything, degraded.IncarnationID).Return(nil).Once()
765 catalogue.EXPECT().Record(degraded.Name).Return(fresh, true, nil).Once()
766 coordinator := recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{9}, 16)))
767 d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)),
768 WithCatalogue(catalogue, []domain.CatalogueRecord{degraded}),
769 WithSnapshotRepository(repository), WithRecoveryCoordinator(coordinator))
770
771 _, err := (controlExec{d: d, recoveryName: degraded.Name}).SessionRecovery("discard")
772 require.NoError(t, err)
773 require.Equal(t, protocol.SessionDown, d.inactive[degraded.Name].state)
774 require.Equal(t, fresh, d.inactive[degraded.Name].record)
775}
776
777func TestListShowsDegraded(t *testing.T) {
778 fresh := durableRecoveryRecord(0)
779 fresh.Committed = nil
780 restoring := durableRecoveryRecord(1)
781 degraded := durableRecoveryRecord(2)
782 degraded.DegradedReason = "checkpoint validation failed"
783 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{fresh, restoring, degraded}, &durableRecoveryRepository{})
784
785 listed := listSessions(t, d)
786 require.Equal(t, []protocol.SessionInfo{
787 {Name: fresh.Name, State: protocol.SessionDown},
788 {Name: restoring.Name, State: protocol.SessionDown},
789 {Name: degraded.Name, State: protocol.SessionBroken},
790 }, listed.Sessions)
791}
792
793func TestListHidesPurgingInactiveSession(t *testing.T) {
794 record := durableRecoveryRecord(0)
795 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
796 d.mu.Lock()
797 entry := d.inactive[record.Name]
798 entry.purging = true
799 entry.state = protocol.SessionDown
800 d.inactive[record.Name] = entry
801 d.mu.Unlock()
802
803 listed := listSessions(t, d)
804 require.Empty(t, listed.Sessions)
805}
806
807func TestAttachWaitsForRestore(t *testing.T) {
808 record := durableRecoveryRecord(0)
809 factory := &recoveryCountingPTYFactory{}
810 d := New(factory, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}), []domain.CatalogueRecord{record}))
811 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
812 defer d.serveCancel()
813
814 result := make(chan error, 1)
815 go func() {
816 _, _, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Name: record.Name, Size: domain.Size{Cols: 80, Rows: 24}}, nil)
817 result <- err
818 }()
819
820 select {
821 case err := <-result:
822 t.Fatalf("attach returned before target restoration completed: %v", err)
823 case <-time.After(25 * time.Millisecond):
824 }
825 require.Zero(t, factory.calls.Load())
826
827 d.setStoppedRecovery(record, protocol.SessionBroken)
828 d.mu.Lock()
829 closeRuntimeRestoreDoneLocked(d.inactive[record.Name].restoreDone)
830 d.mu.Unlock()
831 var protocolError *protoErr
832 require.ErrorAs(t, <-result, &protocolError)
833 require.Equal(t, protocol.ErrInternal, protocolError.code)
834 require.Zero(t, factory.calls.Load())
835}
836
837func TestAttachBarrierSurvivesLiveRegistryPublication(t *testing.T) {
838 record := durableRecoveryRecord(0)
839 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
840 d.mu.Lock()
841 done := d.inactive[record.Name].restoreDone
842 d.mu.Unlock()
843 sess := &session{sessionCore: sessionCore{name: record.Name, incarnation: record.IncarnationID}}
844
845 registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess, d.purgeEpochSnapshot())
846 require.NoError(t, err)
847 require.True(t, registered)
848
849 cancelled, cancel := context.WithCancel(context.Background())
850 cancel()
851 require.ErrorIs(t, d.waitForTargetRestore(cancelled, record.Name), context.Canceled,
852 "an attach racing after live publication must still observe the restore barrier")
853
854 d.finishRecordRestore(record, nil, done)
855 require.NoError(t, d.waitForTargetRestore(context.Background(), record.Name),
856 "attach must proceed after successful restoration completes")
857}
858
859func TestAttachRestoreWaitHonorsContextCancellation(t *testing.T) {
860 record := durableRecoveryRecord(0)
861 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
862 ctx, cancel := context.WithCancel(context.Background())
863 result := make(chan error, 1)
864 go func() { result <- d.waitForTargetRestore(ctx, record.Name) }()
865
866 select {
867 case err := <-result:
868 t.Fatalf("restore wait returned before cancellation: %v", err)
869 case <-time.After(25 * time.Millisecond):
870 }
871 cancel()
872 require.ErrorIs(t, <-result, context.Canceled)
873}
874
875func TestRestoreCancellationTransitionsBeforeAttachCompletion(t *testing.T) {
876 record := durableRecoveryRecord(0)
877 repository := &cancellationRecoveryRepository{started: make(chan struct{})}
878 d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
879 restoreCtx, cancelRestore := context.WithCancel(context.Background())
880 restored := make(chan struct{})
881 go func() {
882 d.restoreIncrementalSnapshots(restoreCtx)
883 close(restored)
884 }()
885 <-repository.started
886
887 attached := make(chan error, 1)
888 go func() {
889 _, _, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Name: record.Name, Size: domain.Size{Cols: 80, Rows: 24}}, nil)
890 attached <- err
891 }()
892 cancelRestore()
893
894 var protocolError *protoErr
895 require.ErrorAs(t, <-attached, &protocolError)
896 require.Equal(t, protocol.ErrInternal, protocolError.code)
897 <-restored
898 entry := d.inactive[record.Name]
899 require.Equal(t, protocol.SessionBroken, entry.state)
900 require.Equal(t, "restore interrupted", entry.record.DegradedReason)
901 select {
902 case <-entry.restoreDone:
903 default:
904 t.Fatal("restore completion was not signaled")
905 }
906 committed, _, _ := catalogue.Record(record.Name)
907 require.Equal(t, entry.record, committed)
908}
909
910func TestWaitForTargetRestoreRejectsClosedRestoringChannel(t *testing.T) {
911 record := durableRecoveryRecord(0)
912 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
913 d.mu.Lock()
914 closeRuntimeRestoreDoneLocked(d.inactive[record.Name].restoreDone)
915 d.mu.Unlock()
916
917 result := make(chan error, 1)
918 go func() { result <- d.waitForTargetRestore(context.Background(), record.Name) }()
919
920 select {
921 case err := <-result:
922 var protocolError *protoErr
923 require.ErrorAs(t, err, &protocolError)
924 require.Equal(t, protocol.ErrInternal, protocolError.code)
925 case <-time.After(time.Second):
926 t.Fatal("waiter spun on a closed restore channel")
927 }
928}
929
930// A healthy catalogue record whose runtime registration was skipped is not
931// degraded durable state, and must not send the operator to durable recovery.
932func TestWaitForTargetRestoreDistinguishesHealthyRuntime(t *testing.T) {
933 record := durableRecoveryRecord(0)
934 tests := []struct {
935 name string
936 state protocol.SessionState
937 message string
938 }{
939 {name: "healthy record without runtime", state: protocol.SessionDown, message: "session was not restored into this daemon: " + record.Name},
940 {name: "broken durable state", state: protocol.SessionBroken, message: "session durable state is broken: " + record.Name},
941 }
942 for _, tt := range tests {
943 t.Run(tt.name, func(t *testing.T) {
944 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
945 d.setStoppedRecovery(record, tt.state)
946 d.mu.Lock()
947 closeRuntimeRestoreDoneLocked(d.inactive[record.Name].restoreDone)
948 d.mu.Unlock()
949
950 var protocolError *protoErr
951 require.ErrorAs(t, d.waitForTargetRestore(context.Background(), record.Name), &protocolError)
952 require.Equal(t, protocol.ErrInternal, protocolError.code)
953 require.Equal(t, tt.message, protocolError.Error())
954 })
955 }
956}
957
958func TestAttachRejectsDegraded(t *testing.T) {
959 record := durableRecoveryRecord(0)
960 record.DegradedReason = "checkpoint validation failed"
961 factory := &recoveryCountingPTYFactory{}
962 d := New(factory, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}), []domain.CatalogueRecord{record}))
963 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
964 defer d.serveCancel()
965
966 _, _, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Name: record.Name, Size: domain.Size{Cols: 80, Rows: 24}}, nil)
967 var protocolError *protoErr
968 require.ErrorAs(t, err, &protocolError)
969 require.Equal(t, protocol.ErrInternal, protocolError.code)
970 require.Zero(t, factory.calls.Load())
971}
972
973func TestRestoreUnavailableReleasesPerRecordWaiters(t *testing.T) {
974 record := durableRecoveryRecord(0)
975 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, nil)
976 d.closeRestoreDone()
977
978 d.mu.Lock()
979 entry := d.inactive[record.Name]
980 d.mu.Unlock()
981 require.Equal(t, protocol.SessionBroken, entry.state)
982 select {
983 case <-entry.restoreDone:
984 default:
985 t.Fatal("per-record restore waiter remained blocked when repository was unavailable")
986 }
987}
988
989func TestCatalogueRestoreDegradedVisible(t *testing.T) {
990 record := durableRecoveryRecord(0)
991 record.DegradedReason = "checkpoint validation failed"
992 repository := &durableRecoveryRepository{generations: make(map[string]ports.SnapshotGeneration), errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
993 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
994
995 d.restoreIncrementalSnapshots(context.Background())
996
997 require.Equal(t, protocol.SessionBroken, d.inactive[record.Name].state)
998 require.Zero(t, repository.loads[record.Name])
999}