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