detached 2fd50e3

feat(recovery): add explicit degraded session recovery

commit
2fd50e39e976f977e0c9b3d6d5cb690083302ac4
author
brice · · 2026-07-26 01:04 UTC
parent
66c3cd3
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
16 files · +1300 -25
Unified
UnifiedSplit
M internal/adapters/recoveryfs/journal_test.go +48 -0
@@ -11,6 +11,54 @@ import (
1111 "github.com/stretchr/testify/require"
1212 )
1313
14+func TestDiscardCrashMatrix(t *testing.T) {
15+ ctx := context.Background()
16+ state := t.TempDir()
17+ j := New(state)
18+ old := domain.IncarnationID{1}
19+ intent := domain.DiscardIntent{
20+ OldRecord: domain.CatalogueRecord{
21+ Name: "work", IncarnationID: old, RecoveryState: domain.RecoveryDegraded,
22+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "invalid",
23+ },
24+ OldIncarnation: old, NewIncarnation: domain.IncarnationID{2}, SessionName: "work", Reason: "discard",
25+ }
26+ require.NoError(t, j.SaveDiscard(ctx, intent))
27+
28+ t.Run("existing intent cannot be overwritten", func(t *testing.T) {
29+ changed := intent
30+ changed.Reason = "different"
31+ require.Error(t, j.SaveDiscard(ctx, changed))
32+ got, err := j.ListDiscards(ctx)
33+ require.NoError(t, err)
34+ require.Equal(t, []domain.DiscardIntent{intent}, got)
35+ })
36+
37+ t.Run("malformed intent remains visible", func(t *testing.T) {
38+ path := j.path(old)
39+ encoded, err := os.ReadFile(path)
40+ require.NoError(t, err)
41+ require.NoError(t, os.WriteFile(path, append(encoded, 0), 0o600))
42+ _, err = j.ListDiscards(ctx)
43+ require.Error(t, err)
44+ _, statErr := os.Stat(path)
45+ require.NoError(t, statErr)
46+ require.Error(t, j.SaveDiscard(ctx, intent))
47+ require.NoError(t, os.WriteFile(path, encoded, 0o600))
48+ })
49+
50+ t.Run("delete sync failure rolls forward", func(t *testing.T) {
51+ injected := errors.New("injected delete directory sync failure")
52+ j.hooks.syncDirectory = func(string) error { return injected }
53+ require.ErrorIs(t, j.DeleteDiscard(ctx, old), injected)
54+ j.hooks = journalHooks{}
55+ require.NoError(t, j.DeleteDiscard(ctx, old))
56+ got, err := j.ListDiscards(ctx)
57+ require.NoError(t, err)
58+ require.Empty(t, got)
59+ })
60+}
61+
1462 func TestRecoveryJournalRoundTripAndIdempotentDelete(t *testing.T) {
1563 j := New(t.TempDir())
1664 old := domain.IncarnationID{1}
@@ -11,6 +11,54 @@ import (
11 "github.com/stretchr/testify/require"11 "github.com/stretchr/testify/require"
12 )12 )
13 13
14+func TestDiscardCrashMatrix(t *testing.T) {
15+ ctx := context.Background()
16+ state := t.TempDir()
17+ j := New(state)
18+ old := domain.IncarnationID{1}
19+ intent := domain.DiscardIntent{
20+ OldRecord: domain.CatalogueRecord{
21+ Name: "work", IncarnationID: old, RecoveryState: domain.RecoveryDegraded,
22+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "invalid",
23+ },
24+ OldIncarnation: old, NewIncarnation: domain.IncarnationID{2}, SessionName: "work", Reason: "discard",
25+ }
26+ require.NoError(t, j.SaveDiscard(ctx, intent))
27+
28+ t.Run("existing intent cannot be overwritten", func(t *testing.T) {
29+ changed := intent
30+ changed.Reason = "different"
31+ require.Error(t, j.SaveDiscard(ctx, changed))
32+ got, err := j.ListDiscards(ctx)
33+ require.NoError(t, err)
34+ require.Equal(t, []domain.DiscardIntent{intent}, got)
35+ })
36+
37+ t.Run("malformed intent remains visible", func(t *testing.T) {
38+ path := j.path(old)
39+ encoded, err := os.ReadFile(path)
40+ require.NoError(t, err)
41+ require.NoError(t, os.WriteFile(path, append(encoded, 0), 0o600))
42+ _, err = j.ListDiscards(ctx)
43+ require.Error(t, err)
44+ _, statErr := os.Stat(path)
45+ require.NoError(t, statErr)
46+ require.Error(t, j.SaveDiscard(ctx, intent))
47+ require.NoError(t, os.WriteFile(path, encoded, 0o600))
48+ })
49+
50+ t.Run("delete sync failure rolls forward", func(t *testing.T) {
51+ injected := errors.New("injected delete directory sync failure")
52+ j.hooks.syncDirectory = func(string) error { return injected }
53+ require.ErrorIs(t, j.DeleteDiscard(ctx, old), injected)
54+ j.hooks = journalHooks{}
55+ require.NoError(t, j.DeleteDiscard(ctx, old))
56+ got, err := j.ListDiscards(ctx)
57+ require.NoError(t, err)
58+ require.Empty(t, got)
59+ })
60+}
61+
14 func TestRecoveryJournalRoundTripAndIdempotentDelete(t *testing.T) {62 func TestRecoveryJournalRoundTripAndIdempotentDelete(t *testing.T) {
15 j := New(t.TempDir())63 j := New(t.TempDir())
16 old := domain.IncarnationID{1}64 old := domain.IncarnationID{1}
M internal/adapters/snapshot/maintenance_delete.go +99 -5
@@ -18,9 +18,10 @@ import (
1818 )
1919
2020 const (
21- deletionTombstoneVersion = uint16(1)
22- deletionTombstoneMagic = "VEVD"
23- deletionTombstonesDir = "deletions"
21+ deletionTombstoneVersion = uint16(1)
22+ deletionTombstoneMagic = "VEVD"
23+ deletionTombstonesDir = "deletions"
24+ quarantineDescriptorVersion = 1
2425 )
2526
2627 var ErrMaintenanceBudgetTooSmall = errors.New("snapshot: maintenance budget too small")
@@ -537,6 +538,12 @@ func (r *Repository) QuarantineDeletionSources(ctx context.Context, tombstone do
537538 func (r *Repository) quarantineSource(source, target string) error {
538539 if _, err := os.Lstat(target); err == nil {
539540 if _, sourceErr := os.Lstat(source); errors.Is(sourceErr, os.ErrNotExist) {
541+ if err := r.syncDirectory(filepath.Dir(source)); err != nil {
542+ return err
543+ }
544+ if filepath.Clean(filepath.Dir(source)) != filepath.Clean(filepath.Dir(target)) {
545+ return r.syncDirectory(filepath.Dir(target))
546+ }
540547 return nil
541548 }
542549 return fmt.Errorf("snapshot: quarantine target already exists")
@@ -560,8 +567,95 @@ func (r *Repository) quarantineSource(source, target string) error {
560567 return nil
561568 }
562569
563-func (r *Repository) SaveQuarantineDescriptor(context.Context, domain.QuarantineDescriptor) error {
564- return errors.New("snapshot: quarantine descriptors not implemented")
570+type quarantineDescriptorFile struct {
571+ Version int `json:"version"`
572+ Descriptor domain.QuarantineDescriptor `json:"descriptor"`
573+}
574+
575+func validateQuarantineDescriptor(descriptor domain.QuarantineDescriptor) error {
576+ if err := descriptor.OldRecord.Validate(); err != nil {
577+ return fmt.Errorf("snapshot: quarantine old record: %w", err)
578+ }
579+ if descriptor.SessionName != descriptor.OldRecord.Name || descriptor.OldIncarnation != descriptor.OldRecord.IncarnationID {
580+ return errors.New("snapshot: inconsistent quarantine identity")
581+ }
582+ if descriptor.ReplacementIncarnation == (domain.IncarnationID{}) || descriptor.ReplacementIncarnation == descriptor.OldIncarnation || descriptor.Reason == "" {
583+ return errors.New("snapshot: incomplete quarantine descriptor")
584+ }
585+ return nil
586+}
587+
588+func encodeQuarantineDescriptor(descriptor domain.QuarantineDescriptor) ([]byte, error) {
589+ if err := validateQuarantineDescriptor(descriptor); err != nil {
590+ return nil, err
591+ }
592+ return json.Marshal(quarantineDescriptorFile{Version: quarantineDescriptorVersion, Descriptor: descriptor})
593+}
594+
595+func decodeQuarantineDescriptor(encoded []byte) (domain.QuarantineDescriptor, error) {
596+ decoder := json.NewDecoder(bytes.NewReader(encoded))
597+ decoder.DisallowUnknownFields()
598+ var file quarantineDescriptorFile
599+ if err := decoder.Decode(&file); err != nil {
600+ return domain.QuarantineDescriptor{}, err
601+ }
602+ if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
603+ return domain.QuarantineDescriptor{}, errors.New("snapshot: trailing quarantine descriptor data")
604+ }
605+ if file.Version != quarantineDescriptorVersion {
606+ return domain.QuarantineDescriptor{}, errors.New("snapshot: unsupported quarantine descriptor version")
607+ }
608+ return file.Descriptor, validateQuarantineDescriptor(file.Descriptor)
609+}
610+
611+func (r *Repository) SaveQuarantineDescriptor(ctx context.Context, descriptor domain.QuarantineDescriptor) error {
612+ if err := ctx.Err(); err != nil {
613+ return err
614+ }
615+ encoded, err := encodeQuarantineDescriptor(descriptor)
616+ if err != nil {
617+ return err
618+ }
619+ dir := filepath.Join(r.dir, "quarantine", descriptor.OldIncarnation.String())
620+ if err := r.ensurePrivateDirectory(dir); err != nil {
621+ return err
622+ }
623+ return r.writeImmutable(filepath.Join(dir, "record"), encoded, func(existing []byte) error {
624+ decoded, err := decodeQuarantineDescriptor(existing)
625+ if err != nil {
626+ return err
627+ }
628+ if !quarantineDescriptorsEqual(decoded, descriptor) || !bytes.Equal(existing, encoded) {
629+ return errors.New("snapshot: quarantine descriptor identity mismatch")
630+ }
631+ return nil
632+ })
633+}
634+
635+func quarantineDescriptorsEqual(a, b domain.QuarantineDescriptor) bool {
636+ return a.OldIncarnation == b.OldIncarnation && a.ReplacementIncarnation == b.ReplacementIncarnation &&
637+ a.SessionName == b.SessionName && a.Reason == b.Reason && catalogueRecordsEqual(a.OldRecord, b.OldRecord)
638+}
639+
640+func catalogueRecordsEqual(a, b domain.CatalogueRecord) bool {
641+ if a.Name != b.Name || a.IncarnationID != b.IncarnationID || a.Cwd != b.Cwd || a.CreatedAt != b.CreatedAt ||
642+ a.UpdatedAt != b.UpdatedAt || a.LastUsedSeq != b.LastUsedSeq || a.RecoveryState != b.RecoveryState ||
643+ a.DegradedReason != b.DegradedReason || len(a.TabNames) != len(b.TabNames) {
644+ return false
645+ }
646+ for i := range a.TabNames {
647+ if a.TabNames[i] != b.TabNames[i] {
648+ return false
649+ }
650+ }
651+ return checkpointRefsEqual(a.Committed, b.Committed) && checkpointRefsEqual(a.Fallbacks[0], b.Fallbacks[0]) && checkpointRefsEqual(a.Fallbacks[1], b.Fallbacks[1])
652+}
653+
654+func checkpointRefsEqual(a, b *domain.CheckpointRef) bool {
655+ if a == nil || b == nil {
656+ return a == nil && b == nil
657+ }
658+ return *a == *b
565659 }
566660
567661 func (r *Repository) QuarantineIncarnation(ctx context.Context, id domain.IncarnationID) error {
@@ -18,9 +18,10 @@ import (
18 )18 )
19 19
20 const (20 const (
21- deletionTombstoneVersion = uint16(1)21+ deletionTombstoneVersion = uint16(1)
22- deletionTombstoneMagic = "VEVD"22+ deletionTombstoneMagic = "VEVD"
23- deletionTombstonesDir = "deletions"23+ deletionTombstonesDir = "deletions"
24+ quarantineDescriptorVersion = 1
24 )25 )
25 26
26 var ErrMaintenanceBudgetTooSmall = errors.New("snapshot: maintenance budget too small")27 var ErrMaintenanceBudgetTooSmall = errors.New("snapshot: maintenance budget too small")
@@ -537,6 +538,12 @@ func (r *Repository) QuarantineDeletionSources(ctx context.Context, tombstone do
537 func (r *Repository) quarantineSource(source, target string) error {538 func (r *Repository) quarantineSource(source, target string) error {
538 if _, err := os.Lstat(target); err == nil {539 if _, err := os.Lstat(target); err == nil {
539 if _, sourceErr := os.Lstat(source); errors.Is(sourceErr, os.ErrNotExist) {540 if _, sourceErr := os.Lstat(source); errors.Is(sourceErr, os.ErrNotExist) {
541+ if err := r.syncDirectory(filepath.Dir(source)); err != nil {
542+ return err
543+ }
544+ if filepath.Clean(filepath.Dir(source)) != filepath.Clean(filepath.Dir(target)) {
545+ return r.syncDirectory(filepath.Dir(target))
546+ }
540 return nil547 return nil
541 }548 }
542 return fmt.Errorf("snapshot: quarantine target already exists")549 return fmt.Errorf("snapshot: quarantine target already exists")
@@ -560,8 +567,95 @@ func (r *Repository) quarantineSource(source, target string) error {
560 return nil567 return nil
561 }568 }
562 569
563-func (r *Repository) SaveQuarantineDescriptor(context.Context, domain.QuarantineDescriptor) error {570+type quarantineDescriptorFile struct {
564- return errors.New("snapshot: quarantine descriptors not implemented")571+ Version int `json:"version"`
572+ Descriptor domain.QuarantineDescriptor `json:"descriptor"`
573+}
574+
575+func validateQuarantineDescriptor(descriptor domain.QuarantineDescriptor) error {
576+ if err := descriptor.OldRecord.Validate(); err != nil {
577+ return fmt.Errorf("snapshot: quarantine old record: %w", err)
578+ }
579+ if descriptor.SessionName != descriptor.OldRecord.Name || descriptor.OldIncarnation != descriptor.OldRecord.IncarnationID {
580+ return errors.New("snapshot: inconsistent quarantine identity")
581+ }
582+ if descriptor.ReplacementIncarnation == (domain.IncarnationID{}) || descriptor.ReplacementIncarnation == descriptor.OldIncarnation || descriptor.Reason == "" {
583+ return errors.New("snapshot: incomplete quarantine descriptor")
584+ }
585+ return nil
586+}
587+
588+func encodeQuarantineDescriptor(descriptor domain.QuarantineDescriptor) ([]byte, error) {
589+ if err := validateQuarantineDescriptor(descriptor); err != nil {
590+ return nil, err
591+ }
592+ return json.Marshal(quarantineDescriptorFile{Version: quarantineDescriptorVersion, Descriptor: descriptor})
593+}
594+
595+func decodeQuarantineDescriptor(encoded []byte) (domain.QuarantineDescriptor, error) {
596+ decoder := json.NewDecoder(bytes.NewReader(encoded))
597+ decoder.DisallowUnknownFields()
598+ var file quarantineDescriptorFile
599+ if err := decoder.Decode(&file); err != nil {
600+ return domain.QuarantineDescriptor{}, err
601+ }
602+ if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
603+ return domain.QuarantineDescriptor{}, errors.New("snapshot: trailing quarantine descriptor data")
604+ }
605+ if file.Version != quarantineDescriptorVersion {
606+ return domain.QuarantineDescriptor{}, errors.New("snapshot: unsupported quarantine descriptor version")
607+ }
608+ return file.Descriptor, validateQuarantineDescriptor(file.Descriptor)
609+}
610+
611+func (r *Repository) SaveQuarantineDescriptor(ctx context.Context, descriptor domain.QuarantineDescriptor) error {
612+ if err := ctx.Err(); err != nil {
613+ return err
614+ }
615+ encoded, err := encodeQuarantineDescriptor(descriptor)
616+ if err != nil {
617+ return err
618+ }
619+ dir := filepath.Join(r.dir, "quarantine", descriptor.OldIncarnation.String())
620+ if err := r.ensurePrivateDirectory(dir); err != nil {
621+ return err
622+ }
623+ return r.writeImmutable(filepath.Join(dir, "record"), encoded, func(existing []byte) error {
624+ decoded, err := decodeQuarantineDescriptor(existing)
625+ if err != nil {
626+ return err
627+ }
628+ if !quarantineDescriptorsEqual(decoded, descriptor) || !bytes.Equal(existing, encoded) {
629+ return errors.New("snapshot: quarantine descriptor identity mismatch")
630+ }
631+ return nil
632+ })
633+}
634+
635+func quarantineDescriptorsEqual(a, b domain.QuarantineDescriptor) bool {
636+ return a.OldIncarnation == b.OldIncarnation && a.ReplacementIncarnation == b.ReplacementIncarnation &&
637+ a.SessionName == b.SessionName && a.Reason == b.Reason && catalogueRecordsEqual(a.OldRecord, b.OldRecord)
638+}
639+
640+func catalogueRecordsEqual(a, b domain.CatalogueRecord) bool {
641+ if a.Name != b.Name || a.IncarnationID != b.IncarnationID || a.Cwd != b.Cwd || a.CreatedAt != b.CreatedAt ||
642+ a.UpdatedAt != b.UpdatedAt || a.LastUsedSeq != b.LastUsedSeq || a.RecoveryState != b.RecoveryState ||
643+ a.DegradedReason != b.DegradedReason || len(a.TabNames) != len(b.TabNames) {
644+ return false
645+ }
646+ for i := range a.TabNames {
647+ if a.TabNames[i] != b.TabNames[i] {
648+ return false
649+ }
650+ }
651+ return checkpointRefsEqual(a.Committed, b.Committed) && checkpointRefsEqual(a.Fallbacks[0], b.Fallbacks[0]) && checkpointRefsEqual(a.Fallbacks[1], b.Fallbacks[1])
652+}
653+
654+func checkpointRefsEqual(a, b *domain.CheckpointRef) bool {
655+ if a == nil || b == nil {
656+ return a == nil && b == nil
657+ }
658+ return *a == *b
565 }659 }
566 660
567 func (r *Repository) QuarantineIncarnation(ctx context.Context, id domain.IncarnationID) error {661 func (r *Repository) QuarantineIncarnation(ctx context.Context, id domain.IncarnationID) error {
M internal/adapters/snapshot/maintenance_delete_test.go +149 -0
@@ -7,6 +7,7 @@ import (
77 "errors"
88 "os"
99 "path/filepath"
10+ "reflect"
1011 "slices"
1112 "testing"
1213
@@ -14,6 +15,154 @@ import (
1415 "github.com/bnema/vev/internal/ports"
1516 )
1617
18+func TestDiscardCrashMatrix(t *testing.T) {
19+ ctx := context.Background()
20+ old := domain.IncarnationID{1}
21+ replacement := domain.IncarnationID{2}
22+ record := domain.CatalogueRecord{
23+ Name: "broken", IncarnationID: old, RecoveryState: domain.RecoveryDegraded,
24+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "corrupt",
25+ }
26+ descriptor := domain.QuarantineDescriptor{
27+ OldRecord: record, OldIncarnation: old, ReplacementIncarnation: replacement,
28+ SessionName: record.Name, Reason: "explicit discard",
29+ }
30+
31+ t.Run("descriptor is immutable and idempotent", func(t *testing.T) {
32+ repo := NewRepository(privateDir(t))
33+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
34+ t.Fatal(err)
35+ }
36+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
37+ t.Fatalf("idempotent retry: %v", err)
38+ }
39+ path := filepath.Join(repo.dir, "quarantine", old.String(), "record")
40+ encoded, err := os.ReadFile(path)
41+ if err != nil {
42+ t.Fatal(err)
43+ }
44+ got, err := decodeQuarantineDescriptor(encoded)
45+ if err != nil || !reflect.DeepEqual(got, descriptor) {
46+ t.Fatalf("descriptor round trip = %#v, %v", got, err)
47+ }
48+ changed := descriptor
49+ changed.Reason = "different"
50+ if err := repo.SaveQuarantineDescriptor(ctx, changed); err == nil {
51+ t.Fatal("descriptor overwrite accepted")
52+ }
53+ if err := os.WriteFile(path, append(encoded, 0), 0o600); err != nil {
54+ t.Fatal(err)
55+ }
56+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err == nil {
57+ t.Fatal("malformed existing descriptor accepted")
58+ }
59+ })
60+
61+ for _, boundary := range []string{"descriptor-file-sync", "descriptor-install", "descriptor-directory-sync"} {
62+ t.Run(boundary, func(t *testing.T) {
63+ repo := NewRepository(privateDir(t))
64+ dir := filepath.Join(repo.dir, "quarantine", old.String())
65+ if err := repo.ensurePrivateDirectory(dir); err != nil {
66+ t.Fatal(err)
67+ }
68+ injected := errors.New("injected descriptor durability failure")
69+ switch boundary {
70+ case "descriptor-file-sync":
71+ repo.hooks.syncFile = func(string) error { return injected }
72+ case "descriptor-install":
73+ repo.hooks.installImmutable = func(string) error { return injected }
74+ case "descriptor-directory-sync":
75+ repo.hooks.syncDirectory = func(path string) error {
76+ if path == dir {
77+ return injected
78+ }
79+ return nil
80+ }
81+ }
82+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); !errors.Is(err, injected) {
83+ t.Fatalf("SaveQuarantineDescriptor() error = %v", err)
84+ }
85+ repo.hooks = repositoryHooks{}
86+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
87+ t.Fatalf("retry: %v", err)
88+ }
89+ })
90+ }
91+
92+ for _, boundary := range []string{"rename", "source-directory-sync", "destination-directory-sync"} {
93+ t.Run(boundary, func(t *testing.T) {
94+ repo := NewRepository(privateDir(t))
95+ source := repo.sessionPath(old)
96+ destinationDir := filepath.Join(repo.dir, "quarantine", old.String())
97+ if err := os.MkdirAll(source, 0o700); err != nil {
98+ t.Fatal(err)
99+ }
100+ if err := repo.ensurePrivateDirectory(destinationDir); err != nil {
101+ t.Fatal(err)
102+ }
103+ if err := os.WriteFile(filepath.Join(source, "payload"), []byte("uncertain"), 0o600); err != nil {
104+ t.Fatal(err)
105+ }
106+ injected := errors.New("injected quarantine durability failure")
107+ switch boundary {
108+ case "rename":
109+ repo.hooks.rename = func(string) error { return injected }
110+ case "source-directory-sync":
111+ repo.hooks.syncDirectory = func(path string) error {
112+ if path == filepath.Dir(source) {
113+ return injected
114+ }
115+ return nil
116+ }
117+ case "destination-directory-sync":
118+ repo.hooks.syncDirectory = func(path string) error {
119+ if path == destinationDir {
120+ return injected
121+ }
122+ return nil
123+ }
124+ }
125+ if err := repo.QuarantineIncarnation(ctx, old); !errors.Is(err, injected) {
126+ t.Fatalf("QuarantineIncarnation() error = %v", err)
127+ }
128+ repo.hooks = repositoryHooks{}
129+ if err := repo.QuarantineIncarnation(ctx, old); err != nil {
130+ t.Fatalf("retry: %v", err)
131+ }
132+ got, err := os.ReadFile(filepath.Join(destinationDir, "snapshot", "payload"))
133+ if err != nil || string(got) != "uncertain" {
134+ t.Fatalf("quarantined payload = %q, %v", got, err)
135+ }
136+ })
137+ }
138+
139+ t.Run("quarantine never removes source data", func(t *testing.T) {
140+ repo := NewRepository(privateDir(t))
141+ source := repo.sessionPath(old)
142+ if err := os.MkdirAll(source, 0o700); err != nil {
143+ t.Fatal(err)
144+ }
145+ payload := filepath.Join(source, "payload")
146+ if err := os.WriteFile(payload, []byte("uncertain"), 0o600); err != nil {
147+ t.Fatal(err)
148+ }
149+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
150+ t.Fatal(err)
151+ }
152+ if err := repo.QuarantineIncarnation(ctx, old); err != nil {
153+ t.Fatal(err)
154+ }
155+ target := filepath.Join(repo.dir, "quarantine", old.String(), "snapshot", "payload")
156+ got, err := os.ReadFile(target)
157+ if err != nil || string(got) != "uncertain" {
158+ t.Fatalf("quarantined payload = %q, %v", got, err)
159+ }
160+ if err := repo.QuarantineIncarnation(ctx, old); err != nil {
161+ t.Fatalf("idempotent retry: %v", err)
162+ }
163+ })
164+}
165+
17166 func TestDeletionTombstoneCodec(t *testing.T) {
18167 want := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{1}}
19168 encoded, err := encodeDeletionTombstone(want)
@@ -7,6 +7,7 @@ import (
7 "errors"7 "errors"
8 "os"8 "os"
9 "path/filepath"9 "path/filepath"
10+ "reflect"
10 "slices"11 "slices"
11 "testing"12 "testing"
12 13
@@ -14,6 +15,154 @@ import (
14 "github.com/bnema/vev/internal/ports"15 "github.com/bnema/vev/internal/ports"
15 )16 )
16 17
18+func TestDiscardCrashMatrix(t *testing.T) {
19+ ctx := context.Background()
20+ old := domain.IncarnationID{1}
21+ replacement := domain.IncarnationID{2}
22+ record := domain.CatalogueRecord{
23+ Name: "broken", IncarnationID: old, RecoveryState: domain.RecoveryDegraded,
24+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "corrupt",
25+ }
26+ descriptor := domain.QuarantineDescriptor{
27+ OldRecord: record, OldIncarnation: old, ReplacementIncarnation: replacement,
28+ SessionName: record.Name, Reason: "explicit discard",
29+ }
30+
31+ t.Run("descriptor is immutable and idempotent", func(t *testing.T) {
32+ repo := NewRepository(privateDir(t))
33+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
34+ t.Fatal(err)
35+ }
36+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
37+ t.Fatalf("idempotent retry: %v", err)
38+ }
39+ path := filepath.Join(repo.dir, "quarantine", old.String(), "record")
40+ encoded, err := os.ReadFile(path)
41+ if err != nil {
42+ t.Fatal(err)
43+ }
44+ got, err := decodeQuarantineDescriptor(encoded)
45+ if err != nil || !reflect.DeepEqual(got, descriptor) {
46+ t.Fatalf("descriptor round trip = %#v, %v", got, err)
47+ }
48+ changed := descriptor
49+ changed.Reason = "different"
50+ if err := repo.SaveQuarantineDescriptor(ctx, changed); err == nil {
51+ t.Fatal("descriptor overwrite accepted")
52+ }
53+ if err := os.WriteFile(path, append(encoded, 0), 0o600); err != nil {
54+ t.Fatal(err)
55+ }
56+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err == nil {
57+ t.Fatal("malformed existing descriptor accepted")
58+ }
59+ })
60+
61+ for _, boundary := range []string{"descriptor-file-sync", "descriptor-install", "descriptor-directory-sync"} {
62+ t.Run(boundary, func(t *testing.T) {
63+ repo := NewRepository(privateDir(t))
64+ dir := filepath.Join(repo.dir, "quarantine", old.String())
65+ if err := repo.ensurePrivateDirectory(dir); err != nil {
66+ t.Fatal(err)
67+ }
68+ injected := errors.New("injected descriptor durability failure")
69+ switch boundary {
70+ case "descriptor-file-sync":
71+ repo.hooks.syncFile = func(string) error { return injected }
72+ case "descriptor-install":
73+ repo.hooks.installImmutable = func(string) error { return injected }
74+ case "descriptor-directory-sync":
75+ repo.hooks.syncDirectory = func(path string) error {
76+ if path == dir {
77+ return injected
78+ }
79+ return nil
80+ }
81+ }
82+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); !errors.Is(err, injected) {
83+ t.Fatalf("SaveQuarantineDescriptor() error = %v", err)
84+ }
85+ repo.hooks = repositoryHooks{}
86+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
87+ t.Fatalf("retry: %v", err)
88+ }
89+ })
90+ }
91+
92+ for _, boundary := range []string{"rename", "source-directory-sync", "destination-directory-sync"} {
93+ t.Run(boundary, func(t *testing.T) {
94+ repo := NewRepository(privateDir(t))
95+ source := repo.sessionPath(old)
96+ destinationDir := filepath.Join(repo.dir, "quarantine", old.String())
97+ if err := os.MkdirAll(source, 0o700); err != nil {
98+ t.Fatal(err)
99+ }
100+ if err := repo.ensurePrivateDirectory(destinationDir); err != nil {
101+ t.Fatal(err)
102+ }
103+ if err := os.WriteFile(filepath.Join(source, "payload"), []byte("uncertain"), 0o600); err != nil {
104+ t.Fatal(err)
105+ }
106+ injected := errors.New("injected quarantine durability failure")
107+ switch boundary {
108+ case "rename":
109+ repo.hooks.rename = func(string) error { return injected }
110+ case "source-directory-sync":
111+ repo.hooks.syncDirectory = func(path string) error {
112+ if path == filepath.Dir(source) {
113+ return injected
114+ }
115+ return nil
116+ }
117+ case "destination-directory-sync":
118+ repo.hooks.syncDirectory = func(path string) error {
119+ if path == destinationDir {
120+ return injected
121+ }
122+ return nil
123+ }
124+ }
125+ if err := repo.QuarantineIncarnation(ctx, old); !errors.Is(err, injected) {
126+ t.Fatalf("QuarantineIncarnation() error = %v", err)
127+ }
128+ repo.hooks = repositoryHooks{}
129+ if err := repo.QuarantineIncarnation(ctx, old); err != nil {
130+ t.Fatalf("retry: %v", err)
131+ }
132+ got, err := os.ReadFile(filepath.Join(destinationDir, "snapshot", "payload"))
133+ if err != nil || string(got) != "uncertain" {
134+ t.Fatalf("quarantined payload = %q, %v", got, err)
135+ }
136+ })
137+ }
138+
139+ t.Run("quarantine never removes source data", func(t *testing.T) {
140+ repo := NewRepository(privateDir(t))
141+ source := repo.sessionPath(old)
142+ if err := os.MkdirAll(source, 0o700); err != nil {
143+ t.Fatal(err)
144+ }
145+ payload := filepath.Join(source, "payload")
146+ if err := os.WriteFile(payload, []byte("uncertain"), 0o600); err != nil {
147+ t.Fatal(err)
148+ }
149+ if err := repo.SaveQuarantineDescriptor(ctx, descriptor); err != nil {
150+ t.Fatal(err)
151+ }
152+ if err := repo.QuarantineIncarnation(ctx, old); err != nil {
153+ t.Fatal(err)
154+ }
155+ target := filepath.Join(repo.dir, "quarantine", old.String(), "snapshot", "payload")
156+ got, err := os.ReadFile(target)
157+ if err != nil || string(got) != "uncertain" {
158+ t.Fatalf("quarantined payload = %q, %v", got, err)
159+ }
160+ if err := repo.QuarantineIncarnation(ctx, old); err != nil {
161+ t.Fatalf("idempotent retry: %v", err)
162+ }
163+ })
164+}
165+
17 func TestDeletionTombstoneCodec(t *testing.T) {166 func TestDeletionTombstoneCodec(t *testing.T) {
18 want := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{1}}167 want := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{1}}
19 encoded, err := encodeDeletionTombstone(want)168 encoded, err := encodeDeletionTombstone(want)
M internal/ports/mocks/recovery_mock.go +283 -0
@@ -6,6 +6,7 @@ package portsmocks
66
77 import (
88 "context"
9+ "io"
910
1011 "github.com/bnema/vev/internal/domain"
1112 "github.com/bnema/vev/internal/ports"
@@ -971,6 +972,288 @@ func (_c *MockSessionLifecycleCoordinator_Rename_Call) RunAndReturn(run func(con
971972 return _c
972973 }
973974
975+// NewMockDegradedRecoveryCoordinator creates a new instance of MockDegradedRecoveryCoordinator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
976+// The first argument is typically a *testing.T value.
977+func NewMockDegradedRecoveryCoordinator(t interface {
978+ mock.TestingT
979+ Cleanup(func())
980+}) *MockDegradedRecoveryCoordinator {
981+ mock := &MockDegradedRecoveryCoordinator{}
982+ mock.Mock.Test(t)
983+
984+ t.Cleanup(func() { mock.AssertExpectations(t) })
985+
986+ return mock
987+}
988+
989+// MockDegradedRecoveryCoordinator is an autogenerated mock type for the DegradedRecoveryCoordinator type
990+type MockDegradedRecoveryCoordinator struct {
991+ mock.Mock
992+}
993+
994+type MockDegradedRecoveryCoordinator_Expecter struct {
995+ mock *mock.Mock
996+}
997+
998+func (_m *MockDegradedRecoveryCoordinator) EXPECT() *MockDegradedRecoveryCoordinator_Expecter {
999+ return &MockDegradedRecoveryCoordinator_Expecter{mock: &_m.Mock}
1000+}
1001+
1002+// Discard provides a mock function for the type MockDegradedRecoveryCoordinator
1003+func (_mock *MockDegradedRecoveryCoordinator) Discard(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error) {
1004+ ret := _mock.Called(context1, s, s1)
1005+
1006+ if len(ret) == 0 {
1007+ panic("no return value specified for Discard")
1008+ }
1009+
1010+ var r0 domain.CatalogueRecord
1011+ var r1 error
1012+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) (domain.CatalogueRecord, error)); ok {
1013+ return returnFunc(context1, s, s1)
1014+ }
1015+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) domain.CatalogueRecord); ok {
1016+ r0 = returnFunc(context1, s, s1)
1017+ } else {
1018+ r0 = ret.Get(0).(domain.CatalogueRecord)
1019+ }
1020+ if returnFunc, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
1021+ r1 = returnFunc(context1, s, s1)
1022+ } else {
1023+ r1 = ret.Error(1)
1024+ }
1025+ return r0, r1
1026+}
1027+
1028+// MockDegradedRecoveryCoordinator_Discard_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Discard'
1029+type MockDegradedRecoveryCoordinator_Discard_Call struct {
1030+ *mock.Call
1031+}
1032+
1033+// Discard is a helper method to define mock.On call
1034+// - context1 context.Context
1035+// - s string
1036+// - s1 string
1037+func (_e *MockDegradedRecoveryCoordinator_Expecter) Discard(context1 any, s any, s1 any) *MockDegradedRecoveryCoordinator_Discard_Call {
1038+ return &MockDegradedRecoveryCoordinator_Discard_Call{Call: _e.mock.On("Discard", context1, s, s1)}
1039+}
1040+
1041+func (_c *MockDegradedRecoveryCoordinator_Discard_Call) Run(run func(context1 context.Context, s string, s1 string)) *MockDegradedRecoveryCoordinator_Discard_Call {
1042+ _c.Call.Run(func(args mock.Arguments) {
1043+ var arg0 context.Context
1044+ if args[0] != nil {
1045+ arg0 = args[0].(context.Context)
1046+ }
1047+ var arg1 string
1048+ if args[1] != nil {
1049+ arg1 = args[1].(string)
1050+ }
1051+ var arg2 string
1052+ if args[2] != nil {
1053+ arg2 = args[2].(string)
1054+ }
1055+ run(
1056+ arg0,
1057+ arg1,
1058+ arg2,
1059+ )
1060+ })
1061+ return _c
1062+}
1063+
1064+func (_c *MockDegradedRecoveryCoordinator_Discard_Call) Return(catalogueRecord domain.CatalogueRecord, err error) *MockDegradedRecoveryCoordinator_Discard_Call {
1065+ _c.Call.Return(catalogueRecord, err)
1066+ return _c
1067+}
1068+
1069+func (_c *MockDegradedRecoveryCoordinator_Discard_Call) RunAndReturn(run func(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error)) *MockDegradedRecoveryCoordinator_Discard_Call {
1070+ _c.Call.Return(run)
1071+ return _c
1072+}
1073+
1074+// Export provides a mock function for the type MockDegradedRecoveryCoordinator
1075+func (_mock *MockDegradedRecoveryCoordinator) Export(context1 context.Context, s string, writer io.Writer) error {
1076+ ret := _mock.Called(context1, s, writer)
1077+
1078+ if len(ret) == 0 {
1079+ panic("no return value specified for Export")
1080+ }
1081+
1082+ var r0 error
1083+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, io.Writer) error); ok {
1084+ r0 = returnFunc(context1, s, writer)
1085+ } else {
1086+ r0 = ret.Error(0)
1087+ }
1088+ return r0
1089+}
1090+
1091+// MockDegradedRecoveryCoordinator_Export_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Export'
1092+type MockDegradedRecoveryCoordinator_Export_Call struct {
1093+ *mock.Call
1094+}
1095+
1096+// Export is a helper method to define mock.On call
1097+// - context1 context.Context
1098+// - s string
1099+// - writer io.Writer
1100+func (_e *MockDegradedRecoveryCoordinator_Expecter) Export(context1 any, s any, writer any) *MockDegradedRecoveryCoordinator_Export_Call {
1101+ return &MockDegradedRecoveryCoordinator_Export_Call{Call: _e.mock.On("Export", context1, s, writer)}
1102+}
1103+
1104+func (_c *MockDegradedRecoveryCoordinator_Export_Call) Run(run func(context1 context.Context, s string, writer io.Writer)) *MockDegradedRecoveryCoordinator_Export_Call {
1105+ _c.Call.Run(func(args mock.Arguments) {
1106+ var arg0 context.Context
1107+ if args[0] != nil {
1108+ arg0 = args[0].(context.Context)
1109+ }
1110+ var arg1 string
1111+ if args[1] != nil {
1112+ arg1 = args[1].(string)
1113+ }
1114+ var arg2 io.Writer
1115+ if args[2] != nil {
1116+ arg2 = args[2].(io.Writer)
1117+ }
1118+ run(
1119+ arg0,
1120+ arg1,
1121+ arg2,
1122+ )
1123+ })
1124+ return _c
1125+}
1126+
1127+func (_c *MockDegradedRecoveryCoordinator_Export_Call) Return(err error) *MockDegradedRecoveryCoordinator_Export_Call {
1128+ _c.Call.Return(err)
1129+ return _c
1130+}
1131+
1132+func (_c *MockDegradedRecoveryCoordinator_Export_Call) RunAndReturn(run func(context1 context.Context, s string, writer io.Writer) error) *MockDegradedRecoveryCoordinator_Export_Call {
1133+ _c.Call.Return(run)
1134+ return _c
1135+}
1136+
1137+// RestoreFallback provides a mock function for the type MockDegradedRecoveryCoordinator
1138+func (_mock *MockDegradedRecoveryCoordinator) RestoreFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) error {
1139+ ret := _mock.Called(context1, s, checkpointRef)
1140+
1141+ if len(ret) == 0 {
1142+ panic("no return value specified for RestoreFallback")
1143+ }
1144+
1145+ var r0 error
1146+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) error); ok {
1147+ r0 = returnFunc(context1, s, checkpointRef)
1148+ } else {
1149+ r0 = ret.Error(0)
1150+ }
1151+ return r0
1152+}
1153+
1154+// MockDegradedRecoveryCoordinator_RestoreFallback_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RestoreFallback'
1155+type MockDegradedRecoveryCoordinator_RestoreFallback_Call struct {
1156+ *mock.Call
1157+}
1158+
1159+// RestoreFallback is a helper method to define mock.On call
1160+// - context1 context.Context
1161+// - s string
1162+// - checkpointRef domain.CheckpointRef
1163+func (_e *MockDegradedRecoveryCoordinator_Expecter) RestoreFallback(context1 any, s any, checkpointRef any) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1164+ return &MockDegradedRecoveryCoordinator_RestoreFallback_Call{Call: _e.mock.On("RestoreFallback", context1, s, checkpointRef)}
1165+}
1166+
1167+func (_c *MockDegradedRecoveryCoordinator_RestoreFallback_Call) Run(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef)) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1168+ _c.Call.Run(func(args mock.Arguments) {
1169+ var arg0 context.Context
1170+ if args[0] != nil {
1171+ arg0 = args[0].(context.Context)
1172+ }
1173+ var arg1 string
1174+ if args[1] != nil {
1175+ arg1 = args[1].(string)
1176+ }
1177+ var arg2 domain.CheckpointRef
1178+ if args[2] != nil {
1179+ arg2 = args[2].(domain.CheckpointRef)
1180+ }
1181+ run(
1182+ arg0,
1183+ arg1,
1184+ arg2,
1185+ )
1186+ })
1187+ return _c
1188+}
1189+
1190+func (_c *MockDegradedRecoveryCoordinator_RestoreFallback_Call) Return(err error) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1191+ _c.Call.Return(err)
1192+ return _c
1193+}
1194+
1195+func (_c *MockDegradedRecoveryCoordinator_RestoreFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) error) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1196+ _c.Call.Return(run)
1197+ return _c
1198+}
1199+
1200+// Retry provides a mock function for the type MockDegradedRecoveryCoordinator
1201+func (_mock *MockDegradedRecoveryCoordinator) Retry(context1 context.Context, s string) error {
1202+ ret := _mock.Called(context1, s)
1203+
1204+ if len(ret) == 0 {
1205+ panic("no return value specified for Retry")
1206+ }
1207+
1208+ var r0 error
1209+ if returnFunc, ok := ret.Get(0).(func(context.Context, string) error); ok {
1210+ r0 = returnFunc(context1, s)
1211+ } else {
1212+ r0 = ret.Error(0)
1213+ }
1214+ return r0
1215+}
1216+
1217+// MockDegradedRecoveryCoordinator_Retry_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Retry'
1218+type MockDegradedRecoveryCoordinator_Retry_Call struct {
1219+ *mock.Call
1220+}
1221+
1222+// Retry is a helper method to define mock.On call
1223+// - context1 context.Context
1224+// - s string
1225+func (_e *MockDegradedRecoveryCoordinator_Expecter) Retry(context1 any, s any) *MockDegradedRecoveryCoordinator_Retry_Call {
1226+ return &MockDegradedRecoveryCoordinator_Retry_Call{Call: _e.mock.On("Retry", context1, s)}
1227+}
1228+
1229+func (_c *MockDegradedRecoveryCoordinator_Retry_Call) Run(run func(context1 context.Context, s string)) *MockDegradedRecoveryCoordinator_Retry_Call {
1230+ _c.Call.Run(func(args mock.Arguments) {
1231+ var arg0 context.Context
1232+ if args[0] != nil {
1233+ arg0 = args[0].(context.Context)
1234+ }
1235+ var arg1 string
1236+ if args[1] != nil {
1237+ arg1 = args[1].(string)
1238+ }
1239+ run(
1240+ arg0,
1241+ arg1,
1242+ )
1243+ })
1244+ return _c
1245+}
1246+
1247+func (_c *MockDegradedRecoveryCoordinator_Retry_Call) Return(err error) *MockDegradedRecoveryCoordinator_Retry_Call {
1248+ _c.Call.Return(err)
1249+ return _c
1250+}
1251+
1252+func (_c *MockDegradedRecoveryCoordinator_Retry_Call) RunAndReturn(run func(context1 context.Context, s string) error) *MockDegradedRecoveryCoordinator_Retry_Call {
1253+ _c.Call.Return(run)
1254+ return _c
1255+}
1256+
9741257 // NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
9751258 // The first argument is typically a *testing.T value.
9761259 func NewMockCatalogue(t interface {
@@ -6,6 +6,7 @@ package portsmocks
6 6
7 import (7 import (
8 "context"8 "context"
9+ "io"
9 10
10 "github.com/bnema/vev/internal/domain"11 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"12 "github.com/bnema/vev/internal/ports"
@@ -971,6 +972,288 @@ func (_c *MockSessionLifecycleCoordinator_Rename_Call) RunAndReturn(run func(con
971 return _c972 return _c
972 }973 }
973 974
975+// NewMockDegradedRecoveryCoordinator creates a new instance of MockDegradedRecoveryCoordinator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
976+// The first argument is typically a *testing.T value.
977+func NewMockDegradedRecoveryCoordinator(t interface {
978+ mock.TestingT
979+ Cleanup(func())
980+}) *MockDegradedRecoveryCoordinator {
981+ mock := &MockDegradedRecoveryCoordinator{}
982+ mock.Mock.Test(t)
983+
984+ t.Cleanup(func() { mock.AssertExpectations(t) })
985+
986+ return mock
987+}
988+
989+// MockDegradedRecoveryCoordinator is an autogenerated mock type for the DegradedRecoveryCoordinator type
990+type MockDegradedRecoveryCoordinator struct {
991+ mock.Mock
992+}
993+
994+type MockDegradedRecoveryCoordinator_Expecter struct {
995+ mock *mock.Mock
996+}
997+
998+func (_m *MockDegradedRecoveryCoordinator) EXPECT() *MockDegradedRecoveryCoordinator_Expecter {
999+ return &MockDegradedRecoveryCoordinator_Expecter{mock: &_m.Mock}
1000+}
1001+
1002+// Discard provides a mock function for the type MockDegradedRecoveryCoordinator
1003+func (_mock *MockDegradedRecoveryCoordinator) Discard(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error) {
1004+ ret := _mock.Called(context1, s, s1)
1005+
1006+ if len(ret) == 0 {
1007+ panic("no return value specified for Discard")
1008+ }
1009+
1010+ var r0 domain.CatalogueRecord
1011+ var r1 error
1012+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) (domain.CatalogueRecord, error)); ok {
1013+ return returnFunc(context1, s, s1)
1014+ }
1015+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) domain.CatalogueRecord); ok {
1016+ r0 = returnFunc(context1, s, s1)
1017+ } else {
1018+ r0 = ret.Get(0).(domain.CatalogueRecord)
1019+ }
1020+ if returnFunc, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
1021+ r1 = returnFunc(context1, s, s1)
1022+ } else {
1023+ r1 = ret.Error(1)
1024+ }
1025+ return r0, r1
1026+}
1027+
1028+// MockDegradedRecoveryCoordinator_Discard_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Discard'
1029+type MockDegradedRecoveryCoordinator_Discard_Call struct {
1030+ *mock.Call
1031+}
1032+
1033+// Discard is a helper method to define mock.On call
1034+// - context1 context.Context
1035+// - s string
1036+// - s1 string
1037+func (_e *MockDegradedRecoveryCoordinator_Expecter) Discard(context1 any, s any, s1 any) *MockDegradedRecoveryCoordinator_Discard_Call {
1038+ return &MockDegradedRecoveryCoordinator_Discard_Call{Call: _e.mock.On("Discard", context1, s, s1)}
1039+}
1040+
1041+func (_c *MockDegradedRecoveryCoordinator_Discard_Call) Run(run func(context1 context.Context, s string, s1 string)) *MockDegradedRecoveryCoordinator_Discard_Call {
1042+ _c.Call.Run(func(args mock.Arguments) {
1043+ var arg0 context.Context
1044+ if args[0] != nil {
1045+ arg0 = args[0].(context.Context)
1046+ }
1047+ var arg1 string
1048+ if args[1] != nil {
1049+ arg1 = args[1].(string)
1050+ }
1051+ var arg2 string
1052+ if args[2] != nil {
1053+ arg2 = args[2].(string)
1054+ }
1055+ run(
1056+ arg0,
1057+ arg1,
1058+ arg2,
1059+ )
1060+ })
1061+ return _c
1062+}
1063+
1064+func (_c *MockDegradedRecoveryCoordinator_Discard_Call) Return(catalogueRecord domain.CatalogueRecord, err error) *MockDegradedRecoveryCoordinator_Discard_Call {
1065+ _c.Call.Return(catalogueRecord, err)
1066+ return _c
1067+}
1068+
1069+func (_c *MockDegradedRecoveryCoordinator_Discard_Call) RunAndReturn(run func(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error)) *MockDegradedRecoveryCoordinator_Discard_Call {
1070+ _c.Call.Return(run)
1071+ return _c
1072+}
1073+
1074+// Export provides a mock function for the type MockDegradedRecoveryCoordinator
1075+func (_mock *MockDegradedRecoveryCoordinator) Export(context1 context.Context, s string, writer io.Writer) error {
1076+ ret := _mock.Called(context1, s, writer)
1077+
1078+ if len(ret) == 0 {
1079+ panic("no return value specified for Export")
1080+ }
1081+
1082+ var r0 error
1083+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, io.Writer) error); ok {
1084+ r0 = returnFunc(context1, s, writer)
1085+ } else {
1086+ r0 = ret.Error(0)
1087+ }
1088+ return r0
1089+}
1090+
1091+// MockDegradedRecoveryCoordinator_Export_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Export'
1092+type MockDegradedRecoveryCoordinator_Export_Call struct {
1093+ *mock.Call
1094+}
1095+
1096+// Export is a helper method to define mock.On call
1097+// - context1 context.Context
1098+// - s string
1099+// - writer io.Writer
1100+func (_e *MockDegradedRecoveryCoordinator_Expecter) Export(context1 any, s any, writer any) *MockDegradedRecoveryCoordinator_Export_Call {
1101+ return &MockDegradedRecoveryCoordinator_Export_Call{Call: _e.mock.On("Export", context1, s, writer)}
1102+}
1103+
1104+func (_c *MockDegradedRecoveryCoordinator_Export_Call) Run(run func(context1 context.Context, s string, writer io.Writer)) *MockDegradedRecoveryCoordinator_Export_Call {
1105+ _c.Call.Run(func(args mock.Arguments) {
1106+ var arg0 context.Context
1107+ if args[0] != nil {
1108+ arg0 = args[0].(context.Context)
1109+ }
1110+ var arg1 string
1111+ if args[1] != nil {
1112+ arg1 = args[1].(string)
1113+ }
1114+ var arg2 io.Writer
1115+ if args[2] != nil {
1116+ arg2 = args[2].(io.Writer)
1117+ }
1118+ run(
1119+ arg0,
1120+ arg1,
1121+ arg2,
1122+ )
1123+ })
1124+ return _c
1125+}
1126+
1127+func (_c *MockDegradedRecoveryCoordinator_Export_Call) Return(err error) *MockDegradedRecoveryCoordinator_Export_Call {
1128+ _c.Call.Return(err)
1129+ return _c
1130+}
1131+
1132+func (_c *MockDegradedRecoveryCoordinator_Export_Call) RunAndReturn(run func(context1 context.Context, s string, writer io.Writer) error) *MockDegradedRecoveryCoordinator_Export_Call {
1133+ _c.Call.Return(run)
1134+ return _c
1135+}
1136+
1137+// RestoreFallback provides a mock function for the type MockDegradedRecoveryCoordinator
1138+func (_mock *MockDegradedRecoveryCoordinator) RestoreFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) error {
1139+ ret := _mock.Called(context1, s, checkpointRef)
1140+
1141+ if len(ret) == 0 {
1142+ panic("no return value specified for RestoreFallback")
1143+ }
1144+
1145+ var r0 error
1146+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) error); ok {
1147+ r0 = returnFunc(context1, s, checkpointRef)
1148+ } else {
1149+ r0 = ret.Error(0)
1150+ }
1151+ return r0
1152+}
1153+
1154+// MockDegradedRecoveryCoordinator_RestoreFallback_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RestoreFallback'
1155+type MockDegradedRecoveryCoordinator_RestoreFallback_Call struct {
1156+ *mock.Call
1157+}
1158+
1159+// RestoreFallback is a helper method to define mock.On call
1160+// - context1 context.Context
1161+// - s string
1162+// - checkpointRef domain.CheckpointRef
1163+func (_e *MockDegradedRecoveryCoordinator_Expecter) RestoreFallback(context1 any, s any, checkpointRef any) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1164+ return &MockDegradedRecoveryCoordinator_RestoreFallback_Call{Call: _e.mock.On("RestoreFallback", context1, s, checkpointRef)}
1165+}
1166+
1167+func (_c *MockDegradedRecoveryCoordinator_RestoreFallback_Call) Run(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef)) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1168+ _c.Call.Run(func(args mock.Arguments) {
1169+ var arg0 context.Context
1170+ if args[0] != nil {
1171+ arg0 = args[0].(context.Context)
1172+ }
1173+ var arg1 string
1174+ if args[1] != nil {
1175+ arg1 = args[1].(string)
1176+ }
1177+ var arg2 domain.CheckpointRef
1178+ if args[2] != nil {
1179+ arg2 = args[2].(domain.CheckpointRef)
1180+ }
1181+ run(
1182+ arg0,
1183+ arg1,
1184+ arg2,
1185+ )
1186+ })
1187+ return _c
1188+}
1189+
1190+func (_c *MockDegradedRecoveryCoordinator_RestoreFallback_Call) Return(err error) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1191+ _c.Call.Return(err)
1192+ return _c
1193+}
1194+
1195+func (_c *MockDegradedRecoveryCoordinator_RestoreFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) error) *MockDegradedRecoveryCoordinator_RestoreFallback_Call {
1196+ _c.Call.Return(run)
1197+ return _c
1198+}
1199+
1200+// Retry provides a mock function for the type MockDegradedRecoveryCoordinator
1201+func (_mock *MockDegradedRecoveryCoordinator) Retry(context1 context.Context, s string) error {
1202+ ret := _mock.Called(context1, s)
1203+
1204+ if len(ret) == 0 {
1205+ panic("no return value specified for Retry")
1206+ }
1207+
1208+ var r0 error
1209+ if returnFunc, ok := ret.Get(0).(func(context.Context, string) error); ok {
1210+ r0 = returnFunc(context1, s)
1211+ } else {
1212+ r0 = ret.Error(0)
1213+ }
1214+ return r0
1215+}
1216+
1217+// MockDegradedRecoveryCoordinator_Retry_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Retry'
1218+type MockDegradedRecoveryCoordinator_Retry_Call struct {
1219+ *mock.Call
1220+}
1221+
1222+// Retry is a helper method to define mock.On call
1223+// - context1 context.Context
1224+// - s string
1225+func (_e *MockDegradedRecoveryCoordinator_Expecter) Retry(context1 any, s any) *MockDegradedRecoveryCoordinator_Retry_Call {
1226+ return &MockDegradedRecoveryCoordinator_Retry_Call{Call: _e.mock.On("Retry", context1, s)}
1227+}
1228+
1229+func (_c *MockDegradedRecoveryCoordinator_Retry_Call) Run(run func(context1 context.Context, s string)) *MockDegradedRecoveryCoordinator_Retry_Call {
1230+ _c.Call.Run(func(args mock.Arguments) {
1231+ var arg0 context.Context
1232+ if args[0] != nil {
1233+ arg0 = args[0].(context.Context)
1234+ }
1235+ var arg1 string
1236+ if args[1] != nil {
1237+ arg1 = args[1].(string)
1238+ }
1239+ run(
1240+ arg0,
1241+ arg1,
1242+ )
1243+ })
1244+ return _c
1245+}
1246+
1247+func (_c *MockDegradedRecoveryCoordinator_Retry_Call) Return(err error) *MockDegradedRecoveryCoordinator_Retry_Call {
1248+ _c.Call.Return(err)
1249+ return _c
1250+}
1251+
1252+func (_c *MockDegradedRecoveryCoordinator_Retry_Call) RunAndReturn(run func(context1 context.Context, s string) error) *MockDegradedRecoveryCoordinator_Retry_Call {
1253+ _c.Call.Return(run)
1254+ return _c
1255+}
1256+
974 // NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.1257 // NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
975 // The first argument is typically a *testing.T value.1258 // The first argument is typically a *testing.T value.
976 func NewMockCatalogue(t interface {1259 func NewMockCatalogue(t interface {
M internal/ports/recovery.go +9 -0
@@ -3,6 +3,7 @@ package ports
33 import (
44 "context"
55 "errors"
6+ "io"
67
78 "github.com/bnema/vev/internal/domain"
89 )
@@ -144,6 +145,14 @@ type SessionLifecycleCoordinator interface {
144145 Delete(context.Context, string) error
145146 }
146147
148+// DegradedRecoveryCoordinator exposes only explicit operator recovery actions.
149+type DegradedRecoveryCoordinator interface {
150+ Retry(context.Context, string) error
151+ RestoreFallback(context.Context, string, domain.CheckpointRef) error
152+ Export(context.Context, string, io.Writer) error
153+ Discard(context.Context, string, string) (domain.CatalogueRecord, error)
154+}
155+
147156 type Catalogue interface {
148157 Records() []domain.CatalogueRecord
149158 Record(string) (domain.CatalogueRecord, bool)
@@ -3,6 +3,7 @@ package ports
3 import (3 import (
4 "context"4 "context"
5 "errors"5 "errors"
6+ "io"
6 7
7 "github.com/bnema/vev/internal/domain"8 "github.com/bnema/vev/internal/domain"
8 )9 )
@@ -144,6 +145,14 @@ type SessionLifecycleCoordinator interface {
144 Delete(context.Context, string) error145 Delete(context.Context, string) error
145 }146 }
146 147
148+// DegradedRecoveryCoordinator exposes only explicit operator recovery actions.
149+type DegradedRecoveryCoordinator interface {
150+ Retry(context.Context, string) error
151+ RestoreFallback(context.Context, string, domain.CheckpointRef) error
152+ Export(context.Context, string, io.Writer) error
153+ Discard(context.Context, string, string) (domain.CatalogueRecord, error)
154+}
155+
147 type Catalogue interface {156 type Catalogue interface {
148 Records() []domain.CatalogueRecord157 Records() []domain.CatalogueRecord
149 Record(string) (domain.CatalogueRecord, bool)158 Record(string) (domain.CatalogueRecord, bool)
M internal/usecase/command/command.go +1 -0
@@ -124,6 +124,7 @@ type ControlContext interface {
124124 RenameSessionTo(name string) error
125125 RenameTabTo(name string) error
126126 Toast(severity, message string) error
127+ SessionRecovery(action, argument string) (string, error)
127128 ListSessions(json bool) (string, error)
128129 ListTabs(json bool) (string, error)
129130 ListPanes(json bool) (string, error)
@@ -124,6 +124,7 @@ type ControlContext interface {
124 RenameSessionTo(name string) error124 RenameSessionTo(name string) error
125 RenameTabTo(name string) error125 RenameTabTo(name string) error
126 Toast(severity, message string) error126 Toast(severity, message string) error
127+ SessionRecovery(action, argument string) (string, error)
127 ListSessions(json bool) (string, error)128 ListSessions(json bool) (string, error)
128 ListTabs(json bool) (string, error)129 ListTabs(json bool) (string, error)
129 ListPanes(json bool) (string, error)130 ListPanes(json bool) (string, error)
M internal/usecase/command/registry.go +27 -0
@@ -47,6 +47,7 @@ func Registry() []Command {
4747 listCommand("list-sessions", "List sessions", "List sessions with active markers", TargetNone, func(ctx ControlContext, json bool) (string, error) { return ctx.ListSessions(json) }),
4848 listCommand("list-tabs", "List tabs", "List tabs in the target session", TargetSession, func(ctx ControlContext, json bool) (string, error) { return ctx.ListTabs(json) }),
4949 listCommand("list-panes", "List panes", "List panes in the target tab", TargetTab, func(ctx ControlContext, json bool) (string, error) { return ctx.ListPanes(json) }),
50+ sessionRecoveryCommand(),
5051 }
5152 return commands
5253 }
@@ -89,6 +90,32 @@ func toastCommand() Command {
8990 }}
9091 }
9192
93+func sessionRecoveryCommand() Command {
94+ return Command{
95+ Slug: "session-recovery", Name: "Session recovery", Desc: "Explicitly recover a degraded durable session",
96+ Usage: "session-recovery retry\nsession-recovery restore <generation>\nsession-recovery export <path>\nsession-recovery discard",
97+ Scriptable: true, Target: TargetNone,
98+ Control: func(ctx ControlContext, args []string, _ ControlOptions) (ControlResult, error) {
99+ var action, argument string
100+ switch {
101+ case len(args) == 1 && (args[0] == "retry" || args[0] == "discard"):
102+ action = args[0]
103+ case len(args) == 2 && args[0] == "restore":
104+ if _, err := ParsePositiveDecimal(args[1:]); err != nil {
105+ return ControlResult{}, ErrInvalidArguments
106+ }
107+ action, argument = args[0], args[1]
108+ case len(args) == 2 && args[0] == "export" && args[1] != "":
109+ action, argument = args[0], args[1]
110+ default:
111+ return ControlResult{}, ErrInvalidArguments
112+ }
113+ output, err := ctx.SessionRecovery(action, argument)
114+ return ControlResult{Output: output}, err
115+ },
116+ }
117+}
118+
92119 func listCommand(slug, name, desc string, target TargetKind, list func(ControlContext, bool) (string, error)) Command {
93120 return Command{Slug: slug, Name: name, Desc: desc, Usage: slug + " [--json]", Scriptable: true, Target: target, Control: func(ctx ControlContext, args []string, opts ControlOptions) (ControlResult, error) {
94121 if len(args) != 0 {
@@ -47,6 +47,7 @@ func Registry() []Command {
47 listCommand("list-sessions", "List sessions", "List sessions with active markers", TargetNone, func(ctx ControlContext, json bool) (string, error) { return ctx.ListSessions(json) }),47 listCommand("list-sessions", "List sessions", "List sessions with active markers", TargetNone, func(ctx ControlContext, json bool) (string, error) { return ctx.ListSessions(json) }),
48 listCommand("list-tabs", "List tabs", "List tabs in the target session", TargetSession, func(ctx ControlContext, json bool) (string, error) { return ctx.ListTabs(json) }),48 listCommand("list-tabs", "List tabs", "List tabs in the target session", TargetSession, func(ctx ControlContext, json bool) (string, error) { return ctx.ListTabs(json) }),
49 listCommand("list-panes", "List panes", "List panes in the target tab", TargetTab, func(ctx ControlContext, json bool) (string, error) { return ctx.ListPanes(json) }),49 listCommand("list-panes", "List panes", "List panes in the target tab", TargetTab, func(ctx ControlContext, json bool) (string, error) { return ctx.ListPanes(json) }),
50+ sessionRecoveryCommand(),
50 }51 }
51 return commands52 return commands
52 }53 }
@@ -89,6 +90,32 @@ func toastCommand() Command {
89 }}90 }}
90 }91 }
91 92
93+func sessionRecoveryCommand() Command {
94+ return Command{
95+ Slug: "session-recovery", Name: "Session recovery", Desc: "Explicitly recover a degraded durable session",
96+ Usage: "session-recovery retry\nsession-recovery restore <generation>\nsession-recovery export <path>\nsession-recovery discard",
97+ Scriptable: true, Target: TargetNone,
98+ Control: func(ctx ControlContext, args []string, _ ControlOptions) (ControlResult, error) {
99+ var action, argument string
100+ switch {
101+ case len(args) == 1 && (args[0] == "retry" || args[0] == "discard"):
102+ action = args[0]
103+ case len(args) == 2 && args[0] == "restore":
104+ if _, err := ParsePositiveDecimal(args[1:]); err != nil {
105+ return ControlResult{}, ErrInvalidArguments
106+ }
107+ action, argument = args[0], args[1]
108+ case len(args) == 2 && args[0] == "export" && args[1] != "":
109+ action, argument = args[0], args[1]
110+ default:
111+ return ControlResult{}, ErrInvalidArguments
112+ }
113+ output, err := ctx.SessionRecovery(action, argument)
114+ return ControlResult{Output: output}, err
115+ },
116+ }
117+}
118+
92 func listCommand(slug, name, desc string, target TargetKind, list func(ControlContext, bool) (string, error)) Command {119 func listCommand(slug, name, desc string, target TargetKind, list func(ControlContext, bool) (string, error)) Command {
93 return Command{Slug: slug, Name: name, Desc: desc, Usage: slug + " [--json]", Scriptable: true, Target: target, Control: func(ctx ControlContext, args []string, opts ControlOptions) (ControlResult, error) {120 return Command{Slug: slug, Name: name, Desc: desc, Usage: slug + " [--json]", Scriptable: true, Target: target, Control: func(ctx ControlContext, args []string, opts ControlOptions) (ControlResult, error) {
94 if len(args) != 0 {121 if len(args) != 0 {
M internal/usecase/command/registry_test.go +29 -0
@@ -105,6 +105,32 @@ func TestPaletteRegistryHidesAPIOnly(t *testing.T) {
105105 }
106106 }
107107
108+func TestSessionRecoveryCommand(t *testing.T) {
109+ cmd, ok := BySlug("session-recovery")
110+ if !ok || !cmd.Scriptable || cmd.Target != TargetNone {
111+ t.Fatalf("session-recovery command = %#v, ok=%v", cmd, ok)
112+ }
113+ tests := []struct {
114+ args []string
115+ wantCall string
116+ wantErr error
117+ }{
118+ {args: []string{"retry"}, wantCall: "session-recovery:retry:"},
119+ {args: []string{"restore", "7"}, wantCall: "session-recovery:restore:7"},
120+ {args: []string{"export", "/tmp/export"}, wantCall: "session-recovery:export:/tmp/export"},
121+ {args: []string{"discard"}, wantCall: "session-recovery:discard:"},
122+ {args: []string{"restore", "0"}, wantErr: ErrInvalidArguments},
123+ {args: []string{"unknown"}, wantErr: ErrInvalidArguments},
124+ }
125+ for _, test := range tests {
126+ ctx := &controlSpy{}
127+ _, err := cmd.Control(ctx, test.args, ControlOptions{})
128+ if !errors.Is(err, test.wantErr) || ctx.call != test.wantCall {
129+ t.Errorf("args %v: call/error = %q/%v, want %q/%v", test.args, ctx.call, err, test.wantCall, test.wantErr)
130+ }
131+ }
132+}
133+
108134 func TestControlHandlersValidateAndDelegate(t *testing.T) {
109135 tests := []struct {
110136 name string
@@ -178,6 +204,9 @@ func (s *controlSpy) RenameTabTo(v string) error { return s.record("renam
178204 func (s *controlSpy) Toast(level, message string) error {
179205 return s.record("toast:" + level + ":" + message)
180206 }
207+func (s *controlSpy) SessionRecovery(action, argument string) (string, error) {
208+ return "", s.record("session-recovery:" + action + ":" + argument)
209+}
181210 func (s *controlSpy) ListSessions(json bool) (string, error) {
182211 _ = s.record("list-sessions:" + strconv.FormatBool(json))
183212 return "sessions", nil
@@ -105,6 +105,32 @@ func TestPaletteRegistryHidesAPIOnly(t *testing.T) {
105 }105 }
106 }106 }
107 107
108+func TestSessionRecoveryCommand(t *testing.T) {
109+ cmd, ok := BySlug("session-recovery")
110+ if !ok || !cmd.Scriptable || cmd.Target != TargetNone {
111+ t.Fatalf("session-recovery command = %#v, ok=%v", cmd, ok)
112+ }
113+ tests := []struct {
114+ args []string
115+ wantCall string
116+ wantErr error
117+ }{
118+ {args: []string{"retry"}, wantCall: "session-recovery:retry:"},
119+ {args: []string{"restore", "7"}, wantCall: "session-recovery:restore:7"},
120+ {args: []string{"export", "/tmp/export"}, wantCall: "session-recovery:export:/tmp/export"},
121+ {args: []string{"discard"}, wantCall: "session-recovery:discard:"},
122+ {args: []string{"restore", "0"}, wantErr: ErrInvalidArguments},
123+ {args: []string{"unknown"}, wantErr: ErrInvalidArguments},
124+ }
125+ for _, test := range tests {
126+ ctx := &controlSpy{}
127+ _, err := cmd.Control(ctx, test.args, ControlOptions{})
128+ if !errors.Is(err, test.wantErr) || ctx.call != test.wantCall {
129+ t.Errorf("args %v: call/error = %q/%v, want %q/%v", test.args, ctx.call, err, test.wantCall, test.wantErr)
130+ }
131+ }
132+}
133+
108 func TestControlHandlersValidateAndDelegate(t *testing.T) {134 func TestControlHandlersValidateAndDelegate(t *testing.T) {
109 tests := []struct {135 tests := []struct {
110 name string136 name string
@@ -178,6 +204,9 @@ func (s *controlSpy) RenameTabTo(v string) error { return s.record("renam
178 func (s *controlSpy) Toast(level, message string) error {204 func (s *controlSpy) Toast(level, message string) error {
179 return s.record("toast:" + level + ":" + message)205 return s.record("toast:" + level + ":" + message)
180 }206 }
207+func (s *controlSpy) SessionRecovery(action, argument string) (string, error) {
208+ return "", s.record("session-recovery:" + action + ":" + argument)
209+}
181 func (s *controlSpy) ListSessions(json bool) (string, error) {210 func (s *controlSpy) ListSessions(json bool) (string, error) {
182 _ = s.record("list-sessions:" + strconv.FormatBool(json))211 _ = s.record("list-sessions:" + strconv.FormatBool(json))
183 return "sessions", nil212 return "sessions", nil
M internal/usecase/daemon/control.go +87 -6
@@ -1,10 +1,13 @@
11 package daemon
22
33 import (
4+ "context"
45 "encoding/json"
56 "errors"
67 "fmt"
8+ "os"
79 "sort"
10+ "strconv"
811 "strings"
912
1013 "github.com/bnema/vev/internal/domain"
@@ -43,7 +46,7 @@ func (d *Daemon) dispatchCommand(request ports.CommandRequest) ports.CommandResu
4346 return commandFailure(ports.ErrNotScriptable, request.Slug+" requires an attached client")
4447 }
4548 if cmd.Target == command.TargetNone && !request.Self {
46- return d.runControl(cmd, controlExec{d: d}, request)
49+ return d.runControl(cmd, controlExec{d: d, recoveryName: request.TargetSession}, request)
4750 }
4851
4952 sess, code, text := d.resolveTargetSession(request)
@@ -332,11 +335,12 @@ func (a daemonActions) switchRelative(sess *session, delta int) error {
332335 // controlExec implements command.ControlContext against resolved daemon-owned
333336 // targets. It deliberately contains no attached-client state.
334337 type controlExec struct {
335- d *Daemon
336- sess *session
337- tab *tab
338- target daemonActionTarget
339- actions daemonActionRunner
338+ d *Daemon
339+ sess *session
340+ tab *tab
341+ recoveryName string
342+ target daemonActionTarget
343+ actions daemonActionRunner
340344 }
341345
342346 func (e controlExec) runAction(request daemonActionRequest) error {
@@ -457,6 +461,83 @@ func (e controlExec) Toast(severity, message string) error {
457461 return nil
458462 }
459463
464+func (e controlExec) SessionRecovery(action, argument string) (string, error) {
465+ if e.d == nil || e.d.degradedRecovery == nil || e.recoveryName == "" {
466+ return "", command.ErrInvalidArguments
467+ }
468+ ctx := context.Background()
469+ switch action {
470+ case "retry":
471+ if err := e.d.degradedRecovery.Retry(ctx, e.recoveryName); err != nil {
472+ return "", err
473+ }
474+ return "", e.restoreExplicitRecovery(ctx)
475+ case "restore":
476+ if e.d.catalogue == nil {
477+ return "", errors.New("durable session catalogue unavailable")
478+ }
479+ generation, err := strconv.ParseUint(argument, 10, 64)
480+ if err != nil || generation == 0 {
481+ return "", command.ErrInvalidArguments
482+ }
483+ record, ok := e.d.catalogue.Record(e.recoveryName)
484+ if !ok {
485+ return "", errors.New("session recovery record not found")
486+ }
487+ for _, fallback := range record.Fallbacks {
488+ if fallback != nil && fallback.Generation == generation {
489+ if err := e.d.degradedRecovery.RestoreFallback(ctx, e.recoveryName, *fallback); err != nil {
490+ return "", err
491+ }
492+ return "", e.restoreExplicitRecovery(ctx)
493+ }
494+ }
495+ return "", errors.New("requested recovery fallback is not available")
496+ case "export":
497+ file, err := os.OpenFile(argument, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
498+ if err != nil {
499+ return "", err
500+ }
501+ exportErr := e.d.degradedRecovery.Export(ctx, e.recoveryName, file)
502+ closeErr := file.Close()
503+ if exportErr != nil || closeErr != nil {
504+ removeErr := os.Remove(argument)
505+ return "", errors.Join(exportErr, closeErr, removeErr)
506+ }
507+ return argument, nil
508+ case "discard":
509+ record, err := e.d.degradedRecovery.Discard(ctx, e.recoveryName, "explicit operator discard")
510+ if err == nil {
511+ e.d.setStoppedRecovery(record, runtimeFresh)
512+ }
513+ return "", err
514+ default:
515+ return "", command.ErrInvalidArguments
516+ }
517+}
518+
519+func (e controlExec) restoreExplicitRecovery(ctx context.Context) error {
520+ if e.d.catalogue == nil {
521+ return errors.New("durable session catalogue unavailable")
522+ }
523+ record, ok := e.d.catalogue.Record(e.recoveryName)
524+ if !ok {
525+ return errors.New("session recovery record not found")
526+ }
527+ done := make(chan struct{})
528+ e.d.mu.Lock()
529+ if entry, exists := e.d.stopped[record.Name]; exists {
530+ entry.record = record
531+ entry.state = runtimeRestoring
532+ entry.restoreDone = done
533+ e.d.stopped[record.Name] = entry
534+ }
535+ e.d.mu.Unlock()
536+ err := e.d.restoreRecord(ctx, record)
537+ e.d.finishRecordRestore(record, err, done)
538+ return err
539+}
540+
460541 func (e controlExec) ListSessions(asJSON bool) (string, error) {
461542 type row struct {
462543 Name string `json:"name"`
@@ -1,10 +1,13 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4+ "context"
4 "encoding/json"5 "encoding/json"
5 "errors"6 "errors"
6 "fmt"7 "fmt"
8+ "os"
7 "sort"9 "sort"
10+ "strconv"
8 "strings"11 "strings"
9 12
10 "github.com/bnema/vev/internal/domain"13 "github.com/bnema/vev/internal/domain"
@@ -43,7 +46,7 @@ func (d *Daemon) dispatchCommand(request ports.CommandRequest) ports.CommandResu
43 return commandFailure(ports.ErrNotScriptable, request.Slug+" requires an attached client")46 return commandFailure(ports.ErrNotScriptable, request.Slug+" requires an attached client")
44 }47 }
45 if cmd.Target == command.TargetNone && !request.Self {48 if cmd.Target == command.TargetNone && !request.Self {
46- return d.runControl(cmd, controlExec{d: d}, request)49+ return d.runControl(cmd, controlExec{d: d, recoveryName: request.TargetSession}, request)
47 }50 }
48 51
49 sess, code, text := d.resolveTargetSession(request)52 sess, code, text := d.resolveTargetSession(request)
@@ -332,11 +335,12 @@ func (a daemonActions) switchRelative(sess *session, delta int) error {
332 // controlExec implements command.ControlContext against resolved daemon-owned335 // controlExec implements command.ControlContext against resolved daemon-owned
333 // targets. It deliberately contains no attached-client state.336 // targets. It deliberately contains no attached-client state.
334 type controlExec struct {337 type controlExec struct {
335- d *Daemon338+ d *Daemon
336- sess *session339+ sess *session
337- tab *tab340+ tab *tab
338- target daemonActionTarget341+ recoveryName string
339- actions daemonActionRunner342+ target daemonActionTarget
343+ actions daemonActionRunner
340 }344 }
341 345
342 func (e controlExec) runAction(request daemonActionRequest) error {346 func (e controlExec) runAction(request daemonActionRequest) error {
@@ -457,6 +461,83 @@ func (e controlExec) Toast(severity, message string) error {
457 return nil461 return nil
458 }462 }
459 463
464+func (e controlExec) SessionRecovery(action, argument string) (string, error) {
465+ if e.d == nil || e.d.degradedRecovery == nil || e.recoveryName == "" {
466+ return "", command.ErrInvalidArguments
467+ }
468+ ctx := context.Background()
469+ switch action {
470+ case "retry":
471+ if err := e.d.degradedRecovery.Retry(ctx, e.recoveryName); err != nil {
472+ return "", err
473+ }
474+ return "", e.restoreExplicitRecovery(ctx)
475+ case "restore":
476+ if e.d.catalogue == nil {
477+ return "", errors.New("durable session catalogue unavailable")
478+ }
479+ generation, err := strconv.ParseUint(argument, 10, 64)
480+ if err != nil || generation == 0 {
481+ return "", command.ErrInvalidArguments
482+ }
483+ record, ok := e.d.catalogue.Record(e.recoveryName)
484+ if !ok {
485+ return "", errors.New("session recovery record not found")
486+ }
487+ for _, fallback := range record.Fallbacks {
488+ if fallback != nil && fallback.Generation == generation {
489+ if err := e.d.degradedRecovery.RestoreFallback(ctx, e.recoveryName, *fallback); err != nil {
490+ return "", err
491+ }
492+ return "", e.restoreExplicitRecovery(ctx)
493+ }
494+ }
495+ return "", errors.New("requested recovery fallback is not available")
496+ case "export":
497+ file, err := os.OpenFile(argument, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
498+ if err != nil {
499+ return "", err
500+ }
501+ exportErr := e.d.degradedRecovery.Export(ctx, e.recoveryName, file)
502+ closeErr := file.Close()
503+ if exportErr != nil || closeErr != nil {
504+ removeErr := os.Remove(argument)
505+ return "", errors.Join(exportErr, closeErr, removeErr)
506+ }
507+ return argument, nil
508+ case "discard":
509+ record, err := e.d.degradedRecovery.Discard(ctx, e.recoveryName, "explicit operator discard")
510+ if err == nil {
511+ e.d.setStoppedRecovery(record, runtimeFresh)
512+ }
513+ return "", err
514+ default:
515+ return "", command.ErrInvalidArguments
516+ }
517+}
518+
519+func (e controlExec) restoreExplicitRecovery(ctx context.Context) error {
520+ if e.d.catalogue == nil {
521+ return errors.New("durable session catalogue unavailable")
522+ }
523+ record, ok := e.d.catalogue.Record(e.recoveryName)
524+ if !ok {
525+ return errors.New("session recovery record not found")
526+ }
527+ done := make(chan struct{})
528+ e.d.mu.Lock()
529+ if entry, exists := e.d.stopped[record.Name]; exists {
530+ entry.record = record
531+ entry.state = runtimeRestoring
532+ entry.restoreDone = done
533+ e.d.stopped[record.Name] = entry
534+ }
535+ e.d.mu.Unlock()
536+ err := e.d.restoreRecord(ctx, record)
537+ e.d.finishRecordRestore(record, err, done)
538+ return err
539+}
540+
460 func (e controlExec) ListSessions(asJSON bool) (string, error) {541 func (e controlExec) ListSessions(asJSON bool) (string, error) {
461 type row struct {542 type row struct {
462 Name string `json:"name"`543 Name string `json:"name"`
M internal/usecase/daemon/daemon.go +4 -0
@@ -127,6 +127,7 @@ type Daemon struct {
127127 snapshotRepository ports.SnapshotRepository
128128 checkpointRecovery ports.CheckpointCoordinator
129129 lifecycleRecovery ports.SessionLifecycleCoordinator
130+ degradedRecovery ports.DegradedRecoveryCoordinator
130131 legacySnapshots ports.LegacySnapshotSource
131132 snapsEnabled bool
132133 noticeStore ports.NoticeStore
@@ -286,6 +287,9 @@ func WithRecoveryCoordinator(coordinator ports.SessionLifecycleCoordinator) Opti
286287 return func(d *Daemon) {
287288 d.checkpointRecovery = coordinator
288289 d.lifecycleRecovery = coordinator
290+ if degraded, ok := coordinator.(ports.DegradedRecoveryCoordinator); ok {
291+ d.degradedRecovery = degraded
292+ }
289293 }
290294 }
291295
@@ -127,6 +127,7 @@ type Daemon struct {
127 snapshotRepository ports.SnapshotRepository127 snapshotRepository ports.SnapshotRepository
128 checkpointRecovery ports.CheckpointCoordinator128 checkpointRecovery ports.CheckpointCoordinator
129 lifecycleRecovery ports.SessionLifecycleCoordinator129 lifecycleRecovery ports.SessionLifecycleCoordinator
130+ degradedRecovery ports.DegradedRecoveryCoordinator
130 legacySnapshots ports.LegacySnapshotSource131 legacySnapshots ports.LegacySnapshotSource
131 snapsEnabled bool132 snapsEnabled bool
132 noticeStore ports.NoticeStore133 noticeStore ports.NoticeStore
@@ -286,6 +287,9 @@ func WithRecoveryCoordinator(coordinator ports.SessionLifecycleCoordinator) Opti
286 return func(d *Daemon) {287 return func(d *Daemon) {
287 d.checkpointRecovery = coordinator288 d.checkpointRecovery = coordinator
288 d.lifecycleRecovery = coordinator289 d.lifecycleRecovery = coordinator
290+ if degraded, ok := coordinator.(ports.DegradedRecoveryCoordinator); ok {
291+ d.degradedRecovery = degraded
292+ }
289 }293 }
290 }294 }
291 295
M internal/usecase/daemon/durable_recovery_test.go +35 -0
@@ -375,6 +375,41 @@ func (f *recoveryCountingPTYFactory) Open(context.Context, string, []string, []s
375375 return nil, errors.New("unexpected PTY open")
376376 }
377377
378+type explicitRecoveryStub struct {
379+ record domain.CatalogueRecord
380+}
381+
382+func (s explicitRecoveryStub) Retry(context.Context, string) error { return nil }
383+func (s explicitRecoveryStub) RestoreFallback(context.Context, string, domain.CheckpointRef) error {
384+ return nil
385+}
386+func (s explicitRecoveryStub) Export(_ context.Context, _ string, w io.Writer) error {
387+ _, err := w.Write([]byte("export"))
388+ return err
389+}
390+func (s explicitRecoveryStub) Discard(context.Context, string, string) (domain.CatalogueRecord, error) {
391+ return s.record, nil
392+}
393+
394+func TestSessionRecoveryCommand(t *testing.T) {
395+ degraded := durableRecoveryRecord(2)
396+ degraded.RecoveryState = domain.RecoveryDegraded
397+ degraded.DegradedReason = "checkpoint validation failed"
398+ d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{degraded}, &durableRecoveryRepository{})
399+ fresh := degraded
400+ fresh.IncarnationID = domain.IncarnationID{9}
401+ fresh.RecoveryState = domain.RecoveryFresh
402+ fresh.Committed = nil
403+ fresh.Fallbacks = [2]*domain.CheckpointRef{}
404+ fresh.DegradedReason = ""
405+ d.degradedRecovery = explicitRecoveryStub{record: fresh}
406+
407+ _, err := (controlExec{d: d, recoveryName: degraded.Name}).SessionRecovery("discard", "")
408+ require.NoError(t, err)
409+ require.Equal(t, runtimeFresh, d.stopped[degraded.Name].state)
410+ require.Equal(t, fresh, d.stopped[degraded.Name].record)
411+}
412+
378413 func TestListShowsDegraded(t *testing.T) {
379414 fresh := durableRecoveryRecord(0)
380415 fresh.RecoveryState = domain.RecoveryFresh
@@ -375,6 +375,41 @@ func (f *recoveryCountingPTYFactory) Open(context.Context, string, []string, []s
375 return nil, errors.New("unexpected PTY open")375 return nil, errors.New("unexpected PTY open")
376 }376 }
377 377
378+type explicitRecoveryStub struct {
379+ record domain.CatalogueRecord
380+}
381+
382+func (s explicitRecoveryStub) Retry(context.Context, string) error { return nil }
383+func (s explicitRecoveryStub) RestoreFallback(context.Context, string, domain.CheckpointRef) error {
384+ return nil
385+}
386+func (s explicitRecoveryStub) Export(_ context.Context, _ string, w io.Writer) error {
387+ _, err := w.Write([]byte("export"))
388+ return err
389+}
390+func (s explicitRecoveryStub) Discard(context.Context, string, string) (domain.CatalogueRecord, error) {
391+ return s.record, nil
392+}
393+
394+func TestSessionRecoveryCommand(t *testing.T) {
395+ degraded := durableRecoveryRecord(2)
396+ degraded.RecoveryState = domain.RecoveryDegraded
397+ degraded.DegradedReason = "checkpoint validation failed"
398+ d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{degraded}, &durableRecoveryRepository{})
399+ fresh := degraded
400+ fresh.IncarnationID = domain.IncarnationID{9}
401+ fresh.RecoveryState = domain.RecoveryFresh
402+ fresh.Committed = nil
403+ fresh.Fallbacks = [2]*domain.CheckpointRef{}
404+ fresh.DegradedReason = ""
405+ d.degradedRecovery = explicitRecoveryStub{record: fresh}
406+
407+ _, err := (controlExec{d: d, recoveryName: degraded.Name}).SessionRecovery("discard", "")
408+ require.NoError(t, err)
409+ require.Equal(t, runtimeFresh, d.stopped[degraded.Name].state)
410+ require.Equal(t, fresh, d.stopped[degraded.Name].record)
411+}
412+
378 func TestListShowsDegraded(t *testing.T) {413 func TestListShowsDegraded(t *testing.T) {
379 fresh := durableRecoveryRecord(0)414 fresh := durableRecoveryRecord(0)
380 fresh.RecoveryState = domain.RecoveryFresh415 fresh.RecoveryState = domain.RecoveryFresh
M internal/usecase/recovery/checkpoint.go +3 -0
@@ -70,7 +70,10 @@ func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref doma
7070 }
7171 unlock := c.locks.Lock([]string{name})
7272 defer unlock()
73+ return c.promoteFallbackLocked(ctx, name, ref)
74+}
7375
76+func (c *Coordinator) promoteFallbackLocked(ctx context.Context, name string, ref domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
7477 record, ok := c.catalogue.Record(name)
7578 if !ok {
7679 return ports.FallbackPromotionOutcome{}, ErrCheckpointRecordNotFound
@@ -70,7 +70,10 @@ func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref doma
70 }70 }
71 unlock := c.locks.Lock([]string{name})71 unlock := c.locks.Lock([]string{name})
72 defer unlock()72 defer unlock()
73+ return c.promoteFallbackLocked(ctx, name, ref)
74+}
73 75
76+func (c *Coordinator) promoteFallbackLocked(ctx context.Context, name string, ref domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
74 record, ok := c.catalogue.Record(name)77 record, ok := c.catalogue.Record(name)
75 if !ok {78 if !ok {
76 return ports.FallbackPromotionOutcome{}, ErrCheckpointRecordNotFound79 return ports.FallbackPromotionOutcome{}, ErrCheckpointRecordNotFound
M internal/usecase/recovery/coordinator.go +5 -5
@@ -14,8 +14,6 @@ import (
1414 "github.com/bnema/vev/internal/ports"
1515 )
1616
17-var ErrPendingRecoveryUnsupported = errors.New("pending recovery intent requires transaction recovery")
18-
1917 var recoveryListingBudget = ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10}
2018
2119 type KeyLocks struct {
@@ -212,6 +210,11 @@ func (c *Coordinator) Recover(ctx context.Context) error {
212210 if err != nil {
213211 return err
214212 }
213+ for _, intent := range intents {
214+ if err := c.recoverDiscard(ctx, intent); err != nil {
215+ return err
216+ }
217+ }
215218 for _, candidate := range c.catalogue.Records() {
216219 if candidate.RecoveryState != domain.RecoveryDeleting {
217220 continue
@@ -239,9 +242,6 @@ func (c *Coordinator) Recover(ctx context.Context) error {
239242 }
240243 }
241244 if page.Done {
242- if len(intents) != 0 {
243- return ErrPendingRecoveryUnsupported
244- }
245245 return nil
246246 }
247247 if page.Next.After == cursor.After {
@@ -14,8 +14,6 @@ import (
14 "github.com/bnema/vev/internal/ports"14 "github.com/bnema/vev/internal/ports"
15 )15 )
16 16
17-var ErrPendingRecoveryUnsupported = errors.New("pending recovery intent requires transaction recovery")
18-
19 var recoveryListingBudget = ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10}17 var recoveryListingBudget = ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10}
20 18
21 type KeyLocks struct {19 type KeyLocks struct {
@@ -212,6 +210,11 @@ func (c *Coordinator) Recover(ctx context.Context) error {
212 if err != nil {210 if err != nil {
213 return err211 return err
214 }212 }
213+ for _, intent := range intents {
214+ if err := c.recoverDiscard(ctx, intent); err != nil {
215+ return err
216+ }
217+ }
215 for _, candidate := range c.catalogue.Records() {218 for _, candidate := range c.catalogue.Records() {
216 if candidate.RecoveryState != domain.RecoveryDeleting {219 if candidate.RecoveryState != domain.RecoveryDeleting {
217 continue220 continue
@@ -239,9 +242,6 @@ func (c *Coordinator) Recover(ctx context.Context) error {
239 }242 }
240 }243 }
241 if page.Done {244 if page.Done {
242- if len(intents) != 0 {
243- return ErrPendingRecoveryUnsupported
244- }
245 return nil245 return nil
246 }246 }
247 if page.Next.After == cursor.After {247 if page.Next.After == cursor.After {
M internal/usecase/recovery/coordinator_test.go +17 -4
@@ -22,12 +22,25 @@ func (j journalStub) ListDiscards(context.Context) ([]domain.DiscardIntent, erro
2222 }
2323 func (j journalStub) DeleteDiscard(context.Context, domain.IncarnationID) error { return nil }
2424
25-func TestCoordinatorRecoverFailsClosedForPendingWork(t *testing.T) {
25+func TestCoordinatorRecoverRollsPendingDiscardForward(t *testing.T) {
26+ ctx := context.Background()
27+ old := domain.CatalogueRecord{
28+ Name: "broken", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryDegraded,
29+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "corrupt",
30+ }
31+ intent := domain.DiscardIntent{
32+ OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2},
33+ SessionName: old.Name, Reason: "discard",
34+ }
2635 repo := portsmocks.NewMockSnapshotRepository(t)
27- repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).
36+ repo.EXPECT().SaveQuarantineDescriptor(ctx, quarantineDescriptor(intent)).Return(nil)
37+ repo.EXPECT().QuarantineIncarnation(ctx, old.IncarnationID).Return(nil)
38+ repo.EXPECT().ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, recoveryListingBudget).
2839 Return(ports.DeletionTombstonePage{Done: true}, nil)
29- c := NewCoordinator(newTransactionCatalogue(), repo, journalStub{intents: []domain.DiscardIntent{{}}}, nil)
30- require.ErrorIs(t, c.Recover(context.Background()), ErrPendingRecoveryUnsupported)
40+ catalogue := newTransactionCatalogue(old)
41+ c := NewCoordinator(catalogue, repo, journalStub{intents: []domain.DiscardIntent{intent}}, nil)
42+ require.NoError(t, c.Recover(ctx))
43+ require.Equal(t, intent.NewIncarnation, catalogue.records[old.Name].IncarnationID)
3144 }
3245
3346 func TestCoordinatorRecoverPagesTombstonesAndPropagatesErrors(t *testing.T) {
@@ -22,12 +22,25 @@ func (j journalStub) ListDiscards(context.Context) ([]domain.DiscardIntent, erro
22 }22 }
23 func (j journalStub) DeleteDiscard(context.Context, domain.IncarnationID) error { return nil }23 func (j journalStub) DeleteDiscard(context.Context, domain.IncarnationID) error { return nil }
24 24
25-func TestCoordinatorRecoverFailsClosedForPendingWork(t *testing.T) {25+func TestCoordinatorRecoverRollsPendingDiscardForward(t *testing.T) {
26+ ctx := context.Background()
27+ old := domain.CatalogueRecord{
28+ Name: "broken", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryDegraded,
29+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}, DegradedReason: "corrupt",
30+ }
31+ intent := domain.DiscardIntent{
32+ OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2},
33+ SessionName: old.Name, Reason: "discard",
34+ }
26 repo := portsmocks.NewMockSnapshotRepository(t)35 repo := portsmocks.NewMockSnapshotRepository(t)
27- repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).36+ repo.EXPECT().SaveQuarantineDescriptor(ctx, quarantineDescriptor(intent)).Return(nil)
37+ repo.EXPECT().QuarantineIncarnation(ctx, old.IncarnationID).Return(nil)
38+ repo.EXPECT().ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, recoveryListingBudget).
28 Return(ports.DeletionTombstonePage{Done: true}, nil)39 Return(ports.DeletionTombstonePage{Done: true}, nil)
29- c := NewCoordinator(newTransactionCatalogue(), repo, journalStub{intents: []domain.DiscardIntent{{}}}, nil)40+ catalogue := newTransactionCatalogue(old)
30- require.ErrorIs(t, c.Recover(context.Background()), ErrPendingRecoveryUnsupported)41+ c := NewCoordinator(catalogue, repo, journalStub{intents: []domain.DiscardIntent{intent}}, nil)
42+ require.NoError(t, c.Recover(ctx))
43+ require.Equal(t, intent.NewIncarnation, catalogue.records[old.Name].IncarnationID)
31 }44 }
32 45
33 func TestCoordinatorRecoverPagesTombstonesAndPropagatesErrors(t *testing.T) {46 func TestCoordinatorRecoverPagesTombstonesAndPropagatesErrors(t *testing.T) {
A internal/usecase/recovery/discard.go +278 -0
@@ -0,0 +1,278 @@
1+package recovery
2+
3+import (
4+ "bytes"
5+ "context"
6+ "crypto/sha256"
7+ "encoding/binary"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "slices"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+)
16+
17+// RecoveryAction identifies an explicit operator choice for degraded state.
18+type RecoveryAction uint8
19+
20+const (
21+ RecoveryRetry RecoveryAction = iota + 1
22+ RecoveryRestoreFallback
23+ RecoveryExport
24+ RecoveryDiscard
25+)
26+
27+var (
28+ ErrRecoveryRecordNotFound = errors.New("recovery: session not found")
29+ ErrSessionNotDegraded = errors.New("recovery: session is not degraded")
30+ ErrDiscardConflict = errors.New("recovery: discard intent conflicts with catalogue")
31+)
32+
33+// Retry directly validates the committed checkpoint. A failed validation is
34+// read-only; only a successful validation is allowed to clear degraded state.
35+func (c *Coordinator) Retry(ctx context.Context, name string) error {
36+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
37+ return errors.New("recovery: incomplete retry dependencies")
38+ }
39+ unlock := c.locks.Lock([]string{name})
40+ defer unlock()
41+ record, ok := c.catalogue.Record(name)
42+ if !ok {
43+ return ErrRecoveryRecordNotFound
44+ }
45+ if record.RecoveryState != domain.RecoveryDegraded || record.Committed == nil {
46+ return ErrSessionNotDegraded
47+ }
48+ if err := c.validateCheckpoint(ctx, record, *record.Committed); err != nil {
49+ return err
50+ }
51+ next := record
52+ next.RecoveryState = domain.RecoveryHealthy
53+ next.DegradedReason = ""
54+ if err := next.Validate(); err != nil {
55+ return fmt.Errorf("recovery: invalid retry transition: %w", err)
56+ }
57+ return c.catalogue.Replace(name, next)
58+}
59+
60+// RestoreFallback promotes only a fallback already displayed from the
61+// catalogue. PromoteFallback performs direct validation before replacement.
62+func (c *Coordinator) RestoreFallback(ctx context.Context, name string, ref domain.CheckpointRef) error {
63+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
64+ return errors.New("recovery: incomplete fallback dependencies")
65+ }
66+ unlock := c.locks.Lock([]string{name})
67+ defer unlock()
68+ record, ok := c.catalogue.Record(name)
69+ if !ok {
70+ return ErrRecoveryRecordNotFound
71+ }
72+ if record.RecoveryState != domain.RecoveryDegraded {
73+ return ErrSessionNotDegraded
74+ }
75+ outcome, err := c.promoteFallbackLocked(ctx, name, ref)
76+ if err != nil {
77+ return err
78+ }
79+ return outcome.HEADRepairError
80+}
81+
82+// Export writes a deterministic, self-contained copy of the committed
83+// generation without changing catalogue or repository state.
84+func (c *Coordinator) Export(ctx context.Context, name string, w io.Writer) error {
85+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || w == nil {
86+ return errors.New("recovery: incomplete export dependencies")
87+ }
88+ unlock := c.locks.Lock([]string{name})
89+ defer unlock()
90+ record, ok := c.catalogue.Record(name)
91+ if !ok {
92+ return ErrRecoveryRecordNotFound
93+ }
94+ if record.RecoveryState != domain.RecoveryDegraded || record.Committed == nil {
95+ return ErrSessionNotDegraded
96+ }
97+ generation, err := c.repository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, *record.Committed)
98+ if err != nil {
99+ return err
100+ }
101+ if err := validateLoadedGeneration(record, *record.Committed, generation); err != nil {
102+ return err
103+ }
104+ encoded, err := encodeExport(generation)
105+ if err != nil {
106+ return err
107+ }
108+ _, err = w.Write(encoded)
109+ return err
110+}
111+
112+func validateLoadedGeneration(record domain.CatalogueRecord, ref domain.CheckpointRef, generation ports.SnapshotGeneration) error {
113+ if generation.IncarnationID != record.IncarnationID || generation.Name != record.Name ||
114+ generation.Generation != ref.Generation || sha256.Sum256(generation.Manifest) != ref.ManifestDigest {
115+ return ErrCheckpointConflict
116+ }
117+ return nil
118+}
119+
120+func encodeExport(generation ports.SnapshotGeneration) ([]byte, error) {
121+ var out bytes.Buffer
122+ out.WriteString("VEVX")
123+ _ = binary.Write(&out, binary.BigEndian, uint16(1))
124+ out.Write(generation.IncarnationID[:])
125+ _ = binary.Write(&out, binary.BigEndian, generation.Generation)
126+ if err := writeExportBytes(&out, []byte(generation.Name)); err != nil {
127+ return nil, err
128+ }
129+ if err := writeExportBytes(&out, generation.Manifest); err != nil {
130+ return nil, err
131+ }
132+ digests := make([]ports.SnapshotDigest, 0, len(generation.Objects))
133+ for digest := range generation.Objects {
134+ digests = append(digests, digest)
135+ }
136+ slices.SortFunc(digests, func(a, b ports.SnapshotDigest) int { return bytes.Compare(a[:], b[:]) })
137+ if uint64(len(digests)) > uint64(^uint32(0)) {
138+ return nil, errors.New("recovery: too many export objects")
139+ }
140+ _ = binary.Write(&out, binary.BigEndian, uint32(len(digests)))
141+ for _, digest := range digests {
142+ out.Write(digest[:])
143+ if err := writeExportBytes(&out, generation.Objects[digest]); err != nil {
144+ return nil, err
145+ }
146+ }
147+ return out.Bytes(), nil
148+}
149+
150+func writeExportBytes(w io.Writer, data []byte) error {
151+ if uint64(len(data)) > uint64(^uint32(0)) {
152+ return errors.New("recovery: export field too large")
153+ }
154+ if err := binary.Write(w, binary.BigEndian, uint32(len(data))); err != nil {
155+ return err
156+ }
157+ _, err := w.Write(data)
158+ return err
159+}
160+
161+// Discard commits to a new incarnation by saving the complete intent before
162+// any quarantine or catalogue mutation. There is no rollback after that point.
163+func (c *Coordinator) Discard(ctx context.Context, name, reason string) (domain.CatalogueRecord, error) {
164+ if c == nil || c.catalogue == nil || c.repository == nil || c.journal == nil || c.locks == nil || c.random == nil {
165+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete discard dependencies")
166+ }
167+ unlock := c.locks.Lock([]string{name})
168+ defer unlock()
169+ record, ok := c.catalogue.Record(name)
170+ if !ok {
171+ return domain.CatalogueRecord{}, ErrRecoveryRecordNotFound
172+ }
173+ if record.RecoveryState != domain.RecoveryDegraded {
174+ return domain.CatalogueRecord{}, ErrSessionNotDegraded
175+ }
176+ if reason == "" {
177+ return domain.CatalogueRecord{}, errors.New("recovery: discard reason is required")
178+ }
179+ newID, err := domain.NewIncarnationID(c.random)
180+ if err != nil {
181+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: generate replacement incarnation: %w", err)
182+ }
183+ intent := domain.DiscardIntent{
184+ OldRecord: record, OldIncarnation: record.IncarnationID, NewIncarnation: newID,
185+ SessionName: record.Name, Reason: reason,
186+ }
187+ if err := c.journal.SaveDiscard(ctx, intent); err != nil {
188+ return domain.CatalogueRecord{}, err
189+ }
190+ if err := c.recoverDiscardLocked(ctx, intent); err != nil {
191+ return domain.CatalogueRecord{}, err
192+ }
193+ return freshReplacement(intent), nil
194+}
195+
196+func (c *Coordinator) recoverDiscard(ctx context.Context, intent domain.DiscardIntent) error {
197+ if c == nil || c.catalogue == nil || c.repository == nil || c.journal == nil || c.locks == nil {
198+ return errors.New("recovery: incomplete discard recovery dependencies")
199+ }
200+ unlock := c.locks.Lock([]string{intent.SessionName})
201+ defer unlock()
202+ return c.recoverDiscardLocked(ctx, intent)
203+}
204+
205+func (c *Coordinator) recoverDiscardLocked(ctx context.Context, intent domain.DiscardIntent) error {
206+ if err := validateDiscardIntent(intent); err != nil {
207+ return err
208+ }
209+ current, ok := c.catalogue.Record(intent.SessionName)
210+ if !ok {
211+ return fmt.Errorf("%w: missing session %q", ErrDiscardConflict, intent.SessionName)
212+ }
213+ next := freshReplacement(intent)
214+ if err := next.Validate(); err != nil {
215+ return fmt.Errorf("recovery: invalid discard replacement: %w", err)
216+ }
217+ switch current.IncarnationID {
218+ case intent.OldIncarnation:
219+ if !catalogueRecordsEqual(current, intent.OldRecord) {
220+ return fmt.Errorf("%w: old record changed", ErrDiscardConflict)
221+ }
222+ case intent.NewIncarnation:
223+ if !catalogueRecordsEqual(current, next) {
224+ return fmt.Errorf("%w: replacement record changed", ErrDiscardConflict)
225+ }
226+ default:
227+ return fmt.Errorf("%w: unexpected incarnation", ErrDiscardConflict)
228+ }
229+
230+ if err := c.repository.SaveQuarantineDescriptor(ctx, quarantineDescriptor(intent)); err != nil {
231+ return err
232+ }
233+ if err := c.repository.QuarantineIncarnation(ctx, intent.OldIncarnation); err != nil {
234+ return err
235+ }
236+ if current.IncarnationID == intent.OldIncarnation {
237+ if err := c.catalogue.Replace(intent.SessionName, next); err != nil {
238+ return err
239+ }
240+ }
241+ return c.journal.DeleteDiscard(ctx, intent.OldIncarnation)
242+}
243+
244+func validateDiscardIntent(intent domain.DiscardIntent) error {
245+ if err := intent.OldRecord.Validate(); err != nil {
246+ return fmt.Errorf("recovery: invalid discard old record: %w", err)
247+ }
248+ if intent.SessionName != intent.OldRecord.Name || intent.OldIncarnation != intent.OldRecord.IncarnationID ||
249+ intent.NewIncarnation == (domain.IncarnationID{}) || intent.NewIncarnation == intent.OldIncarnation || intent.Reason == "" {
250+ return errors.New("recovery: invalid discard intent")
251+ }
252+ return nil
253+}
254+
255+func quarantineDescriptor(intent domain.DiscardIntent) domain.QuarantineDescriptor {
256+ return domain.QuarantineDescriptor{
257+ OldRecord: intent.OldRecord, OldIncarnation: intent.OldIncarnation,
258+ ReplacementIncarnation: intent.NewIncarnation, SessionName: intent.SessionName, Reason: intent.Reason,
259+ }
260+}
261+
262+func freshReplacement(intent domain.DiscardIntent) domain.CatalogueRecord {
263+ next := intent.OldRecord
264+ next.IncarnationID = intent.NewIncarnation
265+ next.RecoveryState = domain.RecoveryFresh
266+ next.Committed = nil
267+ next.Fallbacks = [2]*domain.CheckpointRef{}
268+ next.DegradedReason = ""
269+ return next
270+}
271+
272+func catalogueRecordsEqual(a, b domain.CatalogueRecord) bool {
273+ return a.Name == b.Name && a.IncarnationID == b.IncarnationID && a.Cwd == b.Cwd &&
274+ a.CreatedAt == b.CreatedAt && a.UpdatedAt == b.UpdatedAt && a.LastUsedSeq == b.LastUsedSeq &&
275+ slices.Equal(a.TabNames, b.TabNames) && a.RecoveryState == b.RecoveryState &&
276+ equalCheckpointRefs(a.Committed, b.Committed) && equalCheckpointRefs(a.Fallbacks[0], b.Fallbacks[0]) &&
277+ equalCheckpointRefs(a.Fallbacks[1], b.Fallbacks[1]) && a.DegradedReason == b.DegradedReason
278+}
@@ -0,0 +1,278 @@
1+package recovery
2+
3+import (
4+ "bytes"
5+ "context"
6+ "crypto/sha256"
7+ "encoding/binary"
8+ "errors"
9+ "fmt"
10+ "io"
11+ "slices"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+)
16+
17+// RecoveryAction identifies an explicit operator choice for degraded state.
18+type RecoveryAction uint8
19+
20+const (
21+ RecoveryRetry RecoveryAction = iota + 1
22+ RecoveryRestoreFallback
23+ RecoveryExport
24+ RecoveryDiscard
25+)
26+
27+var (
28+ ErrRecoveryRecordNotFound = errors.New("recovery: session not found")
29+ ErrSessionNotDegraded = errors.New("recovery: session is not degraded")
30+ ErrDiscardConflict = errors.New("recovery: discard intent conflicts with catalogue")
31+)
32+
33+// Retry directly validates the committed checkpoint. A failed validation is
34+// read-only; only a successful validation is allowed to clear degraded state.
35+func (c *Coordinator) Retry(ctx context.Context, name string) error {
36+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
37+ return errors.New("recovery: incomplete retry dependencies")
38+ }
39+ unlock := c.locks.Lock([]string{name})
40+ defer unlock()
41+ record, ok := c.catalogue.Record(name)
42+ if !ok {
43+ return ErrRecoveryRecordNotFound
44+ }
45+ if record.RecoveryState != domain.RecoveryDegraded || record.Committed == nil {
46+ return ErrSessionNotDegraded
47+ }
48+ if err := c.validateCheckpoint(ctx, record, *record.Committed); err != nil {
49+ return err
50+ }
51+ next := record
52+ next.RecoveryState = domain.RecoveryHealthy
53+ next.DegradedReason = ""
54+ if err := next.Validate(); err != nil {
55+ return fmt.Errorf("recovery: invalid retry transition: %w", err)
56+ }
57+ return c.catalogue.Replace(name, next)
58+}
59+
60+// RestoreFallback promotes only a fallback already displayed from the
61+// catalogue. PromoteFallback performs direct validation before replacement.
62+func (c *Coordinator) RestoreFallback(ctx context.Context, name string, ref domain.CheckpointRef) error {
63+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
64+ return errors.New("recovery: incomplete fallback dependencies")
65+ }
66+ unlock := c.locks.Lock([]string{name})
67+ defer unlock()
68+ record, ok := c.catalogue.Record(name)
69+ if !ok {
70+ return ErrRecoveryRecordNotFound
71+ }
72+ if record.RecoveryState != domain.RecoveryDegraded {
73+ return ErrSessionNotDegraded
74+ }
75+ outcome, err := c.promoteFallbackLocked(ctx, name, ref)
76+ if err != nil {
77+ return err
78+ }
79+ return outcome.HEADRepairError
80+}
81+
82+// Export writes a deterministic, self-contained copy of the committed
83+// generation without changing catalogue or repository state.
84+func (c *Coordinator) Export(ctx context.Context, name string, w io.Writer) error {
85+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || w == nil {
86+ return errors.New("recovery: incomplete export dependencies")
87+ }
88+ unlock := c.locks.Lock([]string{name})
89+ defer unlock()
90+ record, ok := c.catalogue.Record(name)
91+ if !ok {
92+ return ErrRecoveryRecordNotFound
93+ }
94+ if record.RecoveryState != domain.RecoveryDegraded || record.Committed == nil {
95+ return ErrSessionNotDegraded
96+ }
97+ generation, err := c.repository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, *record.Committed)
98+ if err != nil {
99+ return err
100+ }
101+ if err := validateLoadedGeneration(record, *record.Committed, generation); err != nil {
102+ return err
103+ }
104+ encoded, err := encodeExport(generation)
105+ if err != nil {
106+ return err
107+ }
108+ _, err = w.Write(encoded)
109+ return err
110+}
111+
112+func validateLoadedGeneration(record domain.CatalogueRecord, ref domain.CheckpointRef, generation ports.SnapshotGeneration) error {
113+ if generation.IncarnationID != record.IncarnationID || generation.Name != record.Name ||
114+ generation.Generation != ref.Generation || sha256.Sum256(generation.Manifest) != ref.ManifestDigest {
115+ return ErrCheckpointConflict
116+ }
117+ return nil
118+}
119+
120+func encodeExport(generation ports.SnapshotGeneration) ([]byte, error) {
121+ var out bytes.Buffer
122+ out.WriteString("VEVX")
123+ _ = binary.Write(&out, binary.BigEndian, uint16(1))
124+ out.Write(generation.IncarnationID[:])
125+ _ = binary.Write(&out, binary.BigEndian, generation.Generation)
126+ if err := writeExportBytes(&out, []byte(generation.Name)); err != nil {
127+ return nil, err
128+ }
129+ if err := writeExportBytes(&out, generation.Manifest); err != nil {
130+ return nil, err
131+ }
132+ digests := make([]ports.SnapshotDigest, 0, len(generation.Objects))
133+ for digest := range generation.Objects {
134+ digests = append(digests, digest)
135+ }
136+ slices.SortFunc(digests, func(a, b ports.SnapshotDigest) int { return bytes.Compare(a[:], b[:]) })
137+ if uint64(len(digests)) > uint64(^uint32(0)) {
138+ return nil, errors.New("recovery: too many export objects")
139+ }
140+ _ = binary.Write(&out, binary.BigEndian, uint32(len(digests)))
141+ for _, digest := range digests {
142+ out.Write(digest[:])
143+ if err := writeExportBytes(&out, generation.Objects[digest]); err != nil {
144+ return nil, err
145+ }
146+ }
147+ return out.Bytes(), nil
148+}
149+
150+func writeExportBytes(w io.Writer, data []byte) error {
151+ if uint64(len(data)) > uint64(^uint32(0)) {
152+ return errors.New("recovery: export field too large")
153+ }
154+ if err := binary.Write(w, binary.BigEndian, uint32(len(data))); err != nil {
155+ return err
156+ }
157+ _, err := w.Write(data)
158+ return err
159+}
160+
161+// Discard commits to a new incarnation by saving the complete intent before
162+// any quarantine or catalogue mutation. There is no rollback after that point.
163+func (c *Coordinator) Discard(ctx context.Context, name, reason string) (domain.CatalogueRecord, error) {
164+ if c == nil || c.catalogue == nil || c.repository == nil || c.journal == nil || c.locks == nil || c.random == nil {
165+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete discard dependencies")
166+ }
167+ unlock := c.locks.Lock([]string{name})
168+ defer unlock()
169+ record, ok := c.catalogue.Record(name)
170+ if !ok {
171+ return domain.CatalogueRecord{}, ErrRecoveryRecordNotFound
172+ }
173+ if record.RecoveryState != domain.RecoveryDegraded {
174+ return domain.CatalogueRecord{}, ErrSessionNotDegraded
175+ }
176+ if reason == "" {
177+ return domain.CatalogueRecord{}, errors.New("recovery: discard reason is required")
178+ }
179+ newID, err := domain.NewIncarnationID(c.random)
180+ if err != nil {
181+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: generate replacement incarnation: %w", err)
182+ }
183+ intent := domain.DiscardIntent{
184+ OldRecord: record, OldIncarnation: record.IncarnationID, NewIncarnation: newID,
185+ SessionName: record.Name, Reason: reason,
186+ }
187+ if err := c.journal.SaveDiscard(ctx, intent); err != nil {
188+ return domain.CatalogueRecord{}, err
189+ }
190+ if err := c.recoverDiscardLocked(ctx, intent); err != nil {
191+ return domain.CatalogueRecord{}, err
192+ }
193+ return freshReplacement(intent), nil
194+}
195+
196+func (c *Coordinator) recoverDiscard(ctx context.Context, intent domain.DiscardIntent) error {
197+ if c == nil || c.catalogue == nil || c.repository == nil || c.journal == nil || c.locks == nil {
198+ return errors.New("recovery: incomplete discard recovery dependencies")
199+ }
200+ unlock := c.locks.Lock([]string{intent.SessionName})
201+ defer unlock()
202+ return c.recoverDiscardLocked(ctx, intent)
203+}
204+
205+func (c *Coordinator) recoverDiscardLocked(ctx context.Context, intent domain.DiscardIntent) error {
206+ if err := validateDiscardIntent(intent); err != nil {
207+ return err
208+ }
209+ current, ok := c.catalogue.Record(intent.SessionName)
210+ if !ok {
211+ return fmt.Errorf("%w: missing session %q", ErrDiscardConflict, intent.SessionName)
212+ }
213+ next := freshReplacement(intent)
214+ if err := next.Validate(); err != nil {
215+ return fmt.Errorf("recovery: invalid discard replacement: %w", err)
216+ }
217+ switch current.IncarnationID {
218+ case intent.OldIncarnation:
219+ if !catalogueRecordsEqual(current, intent.OldRecord) {
220+ return fmt.Errorf("%w: old record changed", ErrDiscardConflict)
221+ }
222+ case intent.NewIncarnation:
223+ if !catalogueRecordsEqual(current, next) {
224+ return fmt.Errorf("%w: replacement record changed", ErrDiscardConflict)
225+ }
226+ default:
227+ return fmt.Errorf("%w: unexpected incarnation", ErrDiscardConflict)
228+ }
229+
230+ if err := c.repository.SaveQuarantineDescriptor(ctx, quarantineDescriptor(intent)); err != nil {
231+ return err
232+ }
233+ if err := c.repository.QuarantineIncarnation(ctx, intent.OldIncarnation); err != nil {
234+ return err
235+ }
236+ if current.IncarnationID == intent.OldIncarnation {
237+ if err := c.catalogue.Replace(intent.SessionName, next); err != nil {
238+ return err
239+ }
240+ }
241+ return c.journal.DeleteDiscard(ctx, intent.OldIncarnation)
242+}
243+
244+func validateDiscardIntent(intent domain.DiscardIntent) error {
245+ if err := intent.OldRecord.Validate(); err != nil {
246+ return fmt.Errorf("recovery: invalid discard old record: %w", err)
247+ }
248+ if intent.SessionName != intent.OldRecord.Name || intent.OldIncarnation != intent.OldRecord.IncarnationID ||
249+ intent.NewIncarnation == (domain.IncarnationID{}) || intent.NewIncarnation == intent.OldIncarnation || intent.Reason == "" {
250+ return errors.New("recovery: invalid discard intent")
251+ }
252+ return nil
253+}
254+
255+func quarantineDescriptor(intent domain.DiscardIntent) domain.QuarantineDescriptor {
256+ return domain.QuarantineDescriptor{
257+ OldRecord: intent.OldRecord, OldIncarnation: intent.OldIncarnation,
258+ ReplacementIncarnation: intent.NewIncarnation, SessionName: intent.SessionName, Reason: intent.Reason,
259+ }
260+}
261+
262+func freshReplacement(intent domain.DiscardIntent) domain.CatalogueRecord {
263+ next := intent.OldRecord
264+ next.IncarnationID = intent.NewIncarnation
265+ next.RecoveryState = domain.RecoveryFresh
266+ next.Committed = nil
267+ next.Fallbacks = [2]*domain.CheckpointRef{}
268+ next.DegradedReason = ""
269+ return next
270+}
271+
272+func catalogueRecordsEqual(a, b domain.CatalogueRecord) bool {
273+ return a.Name == b.Name && a.IncarnationID == b.IncarnationID && a.Cwd == b.Cwd &&
274+ a.CreatedAt == b.CreatedAt && a.UpdatedAt == b.UpdatedAt && a.LastUsedSeq == b.LastUsedSeq &&
275+ slices.Equal(a.TabNames, b.TabNames) && a.RecoveryState == b.RecoveryState &&
276+ equalCheckpointRefs(a.Committed, b.Committed) && equalCheckpointRefs(a.Fallbacks[0], b.Fallbacks[0]) &&
277+ equalCheckpointRefs(a.Fallbacks[1], b.Fallbacks[1]) && a.DegradedReason == b.DegradedReason
278+}
M internal/usecase/recovery/transactions_test.go +226 -5
@@ -3,7 +3,9 @@ package recovery
33 import (
44 "bytes"
55 "context"
6+ "crypto/sha256"
67 "errors"
8+ "strconv"
79 "sync"
810 "testing"
911
@@ -87,14 +89,56 @@ func (c *transactionCatalogue) Close() error { return nil }
8789
8890 type transactionRepository struct {
8991 ports.SnapshotRepository
90- mu sync.Mutex
91- tombstones map[domain.IncarnationID]domain.DeletionTombstone
92- events []string
93- quarantine []bool
92+ mu sync.Mutex
93+ tombstones map[domain.IncarnationID]domain.DeletionTombstone
94+ generations map[domain.CheckpointRef]ports.SnapshotGeneration
95+ loadErr error
96+ events []string
97+ quarantine []bool
98+ descriptors []domain.QuarantineDescriptor
99+ quarantinedIDs []domain.IncarnationID
100+ repaired []domain.CheckpointRef
94101 }
95102
96103 func newTransactionRepository() *transactionRepository {
97- return &transactionRepository{tombstones: make(map[domain.IncarnationID]domain.DeletionTombstone)}
104+ return &transactionRepository{
105+ tombstones: make(map[domain.IncarnationID]domain.DeletionTombstone),
106+ generations: make(map[domain.CheckpointRef]ports.SnapshotGeneration),
107+ }
108+}
109+func (r *transactionRepository) LoadCheckpoint(_ context.Context, _ domain.IncarnationID, _ string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
110+ r.mu.Lock()
111+ defer r.mu.Unlock()
112+ r.events = append(r.events, "load")
113+ if r.loadErr != nil {
114+ return ports.SnapshotGeneration{}, r.loadErr
115+ }
116+ generation, ok := r.generations[ref]
117+ if !ok {
118+ return ports.SnapshotGeneration{}, errors.New("checkpoint missing")
119+ }
120+ return generation, nil
121+}
122+func (r *transactionRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref domain.CheckpointRef) error {
123+ r.mu.Lock()
124+ defer r.mu.Unlock()
125+ r.repaired = append(r.repaired, ref)
126+ r.events = append(r.events, "repair-head")
127+ return nil
128+}
129+func (r *transactionRepository) SaveQuarantineDescriptor(_ context.Context, descriptor domain.QuarantineDescriptor) error {
130+ r.mu.Lock()
131+ defer r.mu.Unlock()
132+ r.descriptors = append(r.descriptors, descriptor)
133+ r.events = append(r.events, "descriptor")
134+ return nil
135+}
136+func (r *transactionRepository) QuarantineIncarnation(_ context.Context, id domain.IncarnationID) error {
137+ r.mu.Lock()
138+ defer r.mu.Unlock()
139+ r.quarantinedIDs = append(r.quarantinedIDs, id)
140+ r.events = append(r.events, "quarantine-incarnation")
141+ return nil
98142 }
99143 func (r *transactionRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {
100144 r.mu.Lock()
@@ -202,6 +246,183 @@ func TestTransactionCrashMatrix(t *testing.T) {
202246 })
203247 }
204248
249+type transactionJournal struct {
250+ mu sync.Mutex
251+ intents map[domain.IncarnationID]domain.DiscardIntent
252+ events []string
253+}
254+
255+func newTransactionJournal() *transactionJournal {
256+ return &transactionJournal{intents: make(map[domain.IncarnationID]domain.DiscardIntent)}
257+}
258+func (j *transactionJournal) SaveDiscard(_ context.Context, intent domain.DiscardIntent) error {
259+ j.mu.Lock()
260+ defer j.mu.Unlock()
261+ j.intents[intent.OldIncarnation] = intent
262+ j.events = append(j.events, "intent")
263+ return nil
264+}
265+func (j *transactionJournal) ListDiscards(context.Context) ([]domain.DiscardIntent, error) {
266+ j.mu.Lock()
267+ defer j.mu.Unlock()
268+ out := make([]domain.DiscardIntent, 0, len(j.intents))
269+ for _, intent := range j.intents {
270+ out = append(out, intent)
271+ }
272+ return out, nil
273+}
274+func (j *transactionJournal) DeleteDiscard(_ context.Context, id domain.IncarnationID) error {
275+ j.mu.Lock()
276+ defer j.mu.Unlock()
277+ delete(j.intents, id)
278+ j.events = append(j.events, "delete-intent")
279+ return nil
280+}
281+
282+func degradedTransactionRecord() domain.CatalogueRecord {
283+ return domain.CatalogueRecord{
284+ Name: "broken", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryDegraded,
285+ Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
286+ Fallbacks: [2]*domain.CheckpointRef{{Generation: 2, ManifestDigest: [32]byte{2}}, {Generation: 1, ManifestDigest: [32]byte{1}}},
287+ DegradedReason: "checkpoint unreadable",
288+ }
289+}
290+
291+func TestDegradedRetryIsReadOnly(t *testing.T) {
292+ record := degradedTransactionRecord()
293+ catalogue := newTransactionCatalogue(record)
294+ repository := newTransactionRepository()
295+ repository.loadErr = errors.New("corrupt checkpoint")
296+ coordinator := NewCoordinator(catalogue, repository, newTransactionJournal(), bytes.NewReader(bytes.Repeat([]byte{9}, 16)))
297+
298+ require.Error(t, coordinator.Retry(context.Background(), record.Name))
299+ require.Equal(t, record, catalogue.records[record.Name])
300+ require.Empty(t, catalogue.events)
301+}
302+
303+func TestFallbackPromotion(t *testing.T) {
304+ record := degradedTransactionRecord()
305+ fallback := *record.Fallbacks[0]
306+ catalogue := newTransactionCatalogue(record)
307+ repository := newTransactionRepository()
308+ repository.generations[fallback] = ports.SnapshotGeneration{
309+ IncarnationID: record.IncarnationID, Name: record.Name, Generation: fallback.Generation,
310+ Manifest: []byte("fallback"),
311+ }
312+ fallback.ManifestDigest = checkpointDigest(repository.generations[*record.Fallbacks[0]].Manifest)
313+ record.Fallbacks[0] = &fallback
314+ catalogue.records[record.Name] = record
315+ repository.generations[fallback] = ports.SnapshotGeneration{
316+ IncarnationID: record.IncarnationID, Name: record.Name, Generation: fallback.Generation,
317+ Manifest: []byte("fallback"),
318+ }
319+ coordinator := NewCoordinator(catalogue, repository, newTransactionJournal(), nil)
320+
321+ require.NoError(t, coordinator.RestoreFallback(context.Background(), record.Name, fallback))
322+ got := catalogue.records[record.Name]
323+ require.Equal(t, domain.RecoveryHealthy, got.RecoveryState)
324+ require.Equal(t, fallback, *got.Committed)
325+ require.Equal(t, []domain.CheckpointRef{fallback}, repository.repaired)
326+ require.Equal(t, []string{"load", "load", "repair-head"}, repository.events)
327+}
328+
329+func checkpointDigest(data []byte) [32]byte {
330+ return sha256.Sum256(data)
331+}
332+
333+func TestDegradedExportIsReadOnly(t *testing.T) {
334+ record := degradedTransactionRecord()
335+ manifest := []byte("manifest")
336+ ref := *record.Committed
337+ ref.ManifestDigest = checkpointDigest(manifest)
338+ record.Committed = &ref
339+ catalogue := newTransactionCatalogue(record)
340+ repository := newTransactionRepository()
341+ repository.generations[ref] = ports.SnapshotGeneration{
342+ IncarnationID: record.IncarnationID, Name: record.Name, Generation: ref.Generation,
343+ Manifest: manifest, Objects: map[ports.SnapshotDigest][]byte{{2}: []byte("second"), {1}: []byte("first")},
344+ }
345+ coordinator := NewCoordinator(catalogue, repository, newTransactionJournal(), nil)
346+ var exported, repeated bytes.Buffer
347+
348+ require.NoError(t, coordinator.Export(context.Background(), record.Name, &exported))
349+ require.NoError(t, coordinator.Export(context.Background(), record.Name, &repeated))
350+ require.NotEmpty(t, exported.Bytes())
351+ require.Equal(t, exported.Bytes(), repeated.Bytes())
352+ require.Equal(t, record, catalogue.records[record.Name])
353+ require.Empty(t, catalogue.events)
354+}
355+
356+func TestDiscardCrashMatrix(t *testing.T) {
357+ ctx := context.Background()
358+ old := degradedTransactionRecord()
359+ newID := domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
360+ intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: newID, SessionName: old.Name, Reason: "explicit discard"}
361+
362+ t.Run("commit order", func(t *testing.T) {
363+ catalogue := newTransactionCatalogue(old)
364+ repository := newTransactionRepository()
365+ journal := newTransactionJournal()
366+ coordinator := NewCoordinator(catalogue, repository, journal, bytes.NewReader(newID[:]))
367+ got, err := coordinator.Discard(ctx, old.Name, intent.Reason)
368+ require.NoError(t, err)
369+ require.Equal(t, freshReplacement(intent), got)
370+ require.Equal(t, []string{"intent", "delete-intent"}, journal.events)
371+ require.Equal(t, []string{"descriptor", "quarantine-incarnation"}, repository.events)
372+ require.Equal(t, []string{"replace:" + old.Name}, catalogue.events)
373+ require.Empty(t, journal.intents)
374+ })
375+
376+ for step := 1; step <= 6; step++ {
377+ t.Run(strconv.Itoa(step), func(t *testing.T) {
378+ catalogue := newTransactionCatalogue(old)
379+ repository := newTransactionRepository()
380+ journal := newTransactionJournal()
381+ if step >= 1 {
382+ journal.intents[old.IncarnationID] = intent
383+ }
384+ if step >= 2 {
385+ repository.descriptors = append(repository.descriptors, quarantineDescriptor(intent))
386+ }
387+ if step >= 3 {
388+ repository.quarantinedIDs = append(repository.quarantinedIDs, old.IncarnationID)
389+ }
390+ if step >= 4 {
391+ catalogue.records[old.Name] = freshReplacement(intent)
392+ }
393+ if step >= 5 {
394+ delete(journal.intents, old.IncarnationID)
395+ }
396+ if step == 6 {
397+ // Runtime exposure is deliberately outside durable recovery.
398+ }
399+ coordinator := NewCoordinator(catalogue, repository, journal, nil)
400+ require.NoError(t, coordinator.Recover(ctx))
401+ got := catalogue.records[old.Name]
402+ require.Equal(t, newID, got.IncarnationID)
403+ require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
404+ require.NotEmpty(t, repository.descriptors)
405+ require.NotEmpty(t, repository.quarantinedIDs)
406+ require.Empty(t, journal.intents)
407+ })
408+ }
409+}
410+
411+func TestDiscardIncarnationConflict(t *testing.T) {
412+ old := degradedTransactionRecord()
413+ intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}
414+ conflict := old
415+ conflict.IncarnationID = domain.IncarnationID{3}
416+ catalogue := newTransactionCatalogue(conflict)
417+ journal := newTransactionJournal()
418+ journal.intents[old.IncarnationID] = intent
419+ coordinator := NewCoordinator(catalogue, newTransactionRepository(), journal, nil)
420+
421+ require.Error(t, coordinator.Recover(context.Background()))
422+ require.Equal(t, conflict, catalogue.records[old.Name])
423+ require.Contains(t, journal.intents, old.IncarnationID)
424+}
425+
205426 func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {
206427 ctx := context.Background()
207428 old := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}
@@ -3,7 +3,9 @@ package recovery
3 import (3 import (
4 "bytes"4 "bytes"
5 "context"5 "context"
6+ "crypto/sha256"
6 "errors"7 "errors"
8+ "strconv"
7 "sync"9 "sync"
8 "testing"10 "testing"
9 11
@@ -87,14 +89,56 @@ func (c *transactionCatalogue) Close() error { return nil }
87 89
88 type transactionRepository struct {90 type transactionRepository struct {
89 ports.SnapshotRepository91 ports.SnapshotRepository
90- mu sync.Mutex92+ mu sync.Mutex
91- tombstones map[domain.IncarnationID]domain.DeletionTombstone93+ tombstones map[domain.IncarnationID]domain.DeletionTombstone
92- events []string94+ generations map[domain.CheckpointRef]ports.SnapshotGeneration
93- quarantine []bool95+ loadErr error
96+ events []string
97+ quarantine []bool
98+ descriptors []domain.QuarantineDescriptor
99+ quarantinedIDs []domain.IncarnationID
100+ repaired []domain.CheckpointRef
94 }101 }
95 102
96 func newTransactionRepository() *transactionRepository {103 func newTransactionRepository() *transactionRepository {
97- return &transactionRepository{tombstones: make(map[domain.IncarnationID]domain.DeletionTombstone)}104+ return &transactionRepository{
105+ tombstones: make(map[domain.IncarnationID]domain.DeletionTombstone),
106+ generations: make(map[domain.CheckpointRef]ports.SnapshotGeneration),
107+ }
108+}
109+func (r *transactionRepository) LoadCheckpoint(_ context.Context, _ domain.IncarnationID, _ string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
110+ r.mu.Lock()
111+ defer r.mu.Unlock()
112+ r.events = append(r.events, "load")
113+ if r.loadErr != nil {
114+ return ports.SnapshotGeneration{}, r.loadErr
115+ }
116+ generation, ok := r.generations[ref]
117+ if !ok {
118+ return ports.SnapshotGeneration{}, errors.New("checkpoint missing")
119+ }
120+ return generation, nil
121+}
122+func (r *transactionRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref domain.CheckpointRef) error {
123+ r.mu.Lock()
124+ defer r.mu.Unlock()
125+ r.repaired = append(r.repaired, ref)
126+ r.events = append(r.events, "repair-head")
127+ return nil
128+}
129+func (r *transactionRepository) SaveQuarantineDescriptor(_ context.Context, descriptor domain.QuarantineDescriptor) error {
130+ r.mu.Lock()
131+ defer r.mu.Unlock()
132+ r.descriptors = append(r.descriptors, descriptor)
133+ r.events = append(r.events, "descriptor")
134+ return nil
135+}
136+func (r *transactionRepository) QuarantineIncarnation(_ context.Context, id domain.IncarnationID) error {
137+ r.mu.Lock()
138+ defer r.mu.Unlock()
139+ r.quarantinedIDs = append(r.quarantinedIDs, id)
140+ r.events = append(r.events, "quarantine-incarnation")
141+ return nil
98 }142 }
99 func (r *transactionRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {143 func (r *transactionRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {
100 r.mu.Lock()144 r.mu.Lock()
@@ -202,6 +246,183 @@ func TestTransactionCrashMatrix(t *testing.T) {
202 })246 })
203 }247 }
204 248
249+type transactionJournal struct {
250+ mu sync.Mutex
251+ intents map[domain.IncarnationID]domain.DiscardIntent
252+ events []string
253+}
254+
255+func newTransactionJournal() *transactionJournal {
256+ return &transactionJournal{intents: make(map[domain.IncarnationID]domain.DiscardIntent)}
257+}
258+func (j *transactionJournal) SaveDiscard(_ context.Context, intent domain.DiscardIntent) error {
259+ j.mu.Lock()
260+ defer j.mu.Unlock()
261+ j.intents[intent.OldIncarnation] = intent
262+ j.events = append(j.events, "intent")
263+ return nil
264+}
265+func (j *transactionJournal) ListDiscards(context.Context) ([]domain.DiscardIntent, error) {
266+ j.mu.Lock()
267+ defer j.mu.Unlock()
268+ out := make([]domain.DiscardIntent, 0, len(j.intents))
269+ for _, intent := range j.intents {
270+ out = append(out, intent)
271+ }
272+ return out, nil
273+}
274+func (j *transactionJournal) DeleteDiscard(_ context.Context, id domain.IncarnationID) error {
275+ j.mu.Lock()
276+ defer j.mu.Unlock()
277+ delete(j.intents, id)
278+ j.events = append(j.events, "delete-intent")
279+ return nil
280+}
281+
282+func degradedTransactionRecord() domain.CatalogueRecord {
283+ return domain.CatalogueRecord{
284+ Name: "broken", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryDegraded,
285+ Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
286+ Fallbacks: [2]*domain.CheckpointRef{{Generation: 2, ManifestDigest: [32]byte{2}}, {Generation: 1, ManifestDigest: [32]byte{1}}},
287+ DegradedReason: "checkpoint unreadable",
288+ }
289+}
290+
291+func TestDegradedRetryIsReadOnly(t *testing.T) {
292+ record := degradedTransactionRecord()
293+ catalogue := newTransactionCatalogue(record)
294+ repository := newTransactionRepository()
295+ repository.loadErr = errors.New("corrupt checkpoint")
296+ coordinator := NewCoordinator(catalogue, repository, newTransactionJournal(), bytes.NewReader(bytes.Repeat([]byte{9}, 16)))
297+
298+ require.Error(t, coordinator.Retry(context.Background(), record.Name))
299+ require.Equal(t, record, catalogue.records[record.Name])
300+ require.Empty(t, catalogue.events)
301+}
302+
303+func TestFallbackPromotion(t *testing.T) {
304+ record := degradedTransactionRecord()
305+ fallback := *record.Fallbacks[0]
306+ catalogue := newTransactionCatalogue(record)
307+ repository := newTransactionRepository()
308+ repository.generations[fallback] = ports.SnapshotGeneration{
309+ IncarnationID: record.IncarnationID, Name: record.Name, Generation: fallback.Generation,
310+ Manifest: []byte("fallback"),
311+ }
312+ fallback.ManifestDigest = checkpointDigest(repository.generations[*record.Fallbacks[0]].Manifest)
313+ record.Fallbacks[0] = &fallback
314+ catalogue.records[record.Name] = record
315+ repository.generations[fallback] = ports.SnapshotGeneration{
316+ IncarnationID: record.IncarnationID, Name: record.Name, Generation: fallback.Generation,
317+ Manifest: []byte("fallback"),
318+ }
319+ coordinator := NewCoordinator(catalogue, repository, newTransactionJournal(), nil)
320+
321+ require.NoError(t, coordinator.RestoreFallback(context.Background(), record.Name, fallback))
322+ got := catalogue.records[record.Name]
323+ require.Equal(t, domain.RecoveryHealthy, got.RecoveryState)
324+ require.Equal(t, fallback, *got.Committed)
325+ require.Equal(t, []domain.CheckpointRef{fallback}, repository.repaired)
326+ require.Equal(t, []string{"load", "load", "repair-head"}, repository.events)
327+}
328+
329+func checkpointDigest(data []byte) [32]byte {
330+ return sha256.Sum256(data)
331+}
332+
333+func TestDegradedExportIsReadOnly(t *testing.T) {
334+ record := degradedTransactionRecord()
335+ manifest := []byte("manifest")
336+ ref := *record.Committed
337+ ref.ManifestDigest = checkpointDigest(manifest)
338+ record.Committed = &ref
339+ catalogue := newTransactionCatalogue(record)
340+ repository := newTransactionRepository()
341+ repository.generations[ref] = ports.SnapshotGeneration{
342+ IncarnationID: record.IncarnationID, Name: record.Name, Generation: ref.Generation,
343+ Manifest: manifest, Objects: map[ports.SnapshotDigest][]byte{{2}: []byte("second"), {1}: []byte("first")},
344+ }
345+ coordinator := NewCoordinator(catalogue, repository, newTransactionJournal(), nil)
346+ var exported, repeated bytes.Buffer
347+
348+ require.NoError(t, coordinator.Export(context.Background(), record.Name, &exported))
349+ require.NoError(t, coordinator.Export(context.Background(), record.Name, &repeated))
350+ require.NotEmpty(t, exported.Bytes())
351+ require.Equal(t, exported.Bytes(), repeated.Bytes())
352+ require.Equal(t, record, catalogue.records[record.Name])
353+ require.Empty(t, catalogue.events)
354+}
355+
356+func TestDiscardCrashMatrix(t *testing.T) {
357+ ctx := context.Background()
358+ old := degradedTransactionRecord()
359+ newID := domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
360+ intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: newID, SessionName: old.Name, Reason: "explicit discard"}
361+
362+ t.Run("commit order", func(t *testing.T) {
363+ catalogue := newTransactionCatalogue(old)
364+ repository := newTransactionRepository()
365+ journal := newTransactionJournal()
366+ coordinator := NewCoordinator(catalogue, repository, journal, bytes.NewReader(newID[:]))
367+ got, err := coordinator.Discard(ctx, old.Name, intent.Reason)
368+ require.NoError(t, err)
369+ require.Equal(t, freshReplacement(intent), got)
370+ require.Equal(t, []string{"intent", "delete-intent"}, journal.events)
371+ require.Equal(t, []string{"descriptor", "quarantine-incarnation"}, repository.events)
372+ require.Equal(t, []string{"replace:" + old.Name}, catalogue.events)
373+ require.Empty(t, journal.intents)
374+ })
375+
376+ for step := 1; step <= 6; step++ {
377+ t.Run(strconv.Itoa(step), func(t *testing.T) {
378+ catalogue := newTransactionCatalogue(old)
379+ repository := newTransactionRepository()
380+ journal := newTransactionJournal()
381+ if step >= 1 {
382+ journal.intents[old.IncarnationID] = intent
383+ }
384+ if step >= 2 {
385+ repository.descriptors = append(repository.descriptors, quarantineDescriptor(intent))
386+ }
387+ if step >= 3 {
388+ repository.quarantinedIDs = append(repository.quarantinedIDs, old.IncarnationID)
389+ }
390+ if step >= 4 {
391+ catalogue.records[old.Name] = freshReplacement(intent)
392+ }
393+ if step >= 5 {
394+ delete(journal.intents, old.IncarnationID)
395+ }
396+ if step == 6 {
397+ // Runtime exposure is deliberately outside durable recovery.
398+ }
399+ coordinator := NewCoordinator(catalogue, repository, journal, nil)
400+ require.NoError(t, coordinator.Recover(ctx))
401+ got := catalogue.records[old.Name]
402+ require.Equal(t, newID, got.IncarnationID)
403+ require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
404+ require.NotEmpty(t, repository.descriptors)
405+ require.NotEmpty(t, repository.quarantinedIDs)
406+ require.Empty(t, journal.intents)
407+ })
408+ }
409+}
410+
411+func TestDiscardIncarnationConflict(t *testing.T) {
412+ old := degradedTransactionRecord()
413+ intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}
414+ conflict := old
415+ conflict.IncarnationID = domain.IncarnationID{3}
416+ catalogue := newTransactionCatalogue(conflict)
417+ journal := newTransactionJournal()
418+ journal.intents[old.IncarnationID] = intent
419+ coordinator := NewCoordinator(catalogue, newTransactionRepository(), journal, nil)
420+
421+ require.Error(t, coordinator.Recover(context.Background()))
422+ require.Equal(t, conflict, catalogue.records[old.Name])
423+ require.Contains(t, journal.intents, old.IncarnationID)
424+}
425+
205 func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {426 func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {
206 ctx := context.Background()427 ctx := context.Background()
207 old := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}428 old := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close