detached 0ffe39d

fix(recovery): resolve final durability review

commit
0ffe39dcc1f80222c01da57038bfda6fc3e10938
author
brice · · 2026-07-26 13:56 UTC
parent
8171186
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
19 files · +384 -163
Unified
UnifiedSplit
M internal/adapters/snapshot/direct_load_test.go +99 -3
@@ -3,13 +3,16 @@ package snapshot
33 import (
44 "context"
55 "crypto/sha256"
6+ "errors"
67 "os"
78 "path/filepath"
89 "testing"
910
1011 "github.com/bnema/vev/internal/domain"
1112 "github.com/bnema/vev/internal/ports"
13+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
1214 codec "github.com/bnema/vev/internal/usecase/snapshot"
15+ "github.com/stretchr/testify/require"
1316 )
1417
1518 func TestIncarnationNamespace(t *testing.T) {
@@ -58,7 +61,7 @@ func TestDirectCheckpointLoad(t *testing.T) {
5861 }
5962 }
6063
61-func TestRepairHEAD(t *testing.T) {
64+func TestReconcileCheckpoint(t *testing.T) {
6265 repo := NewRepository(privateDir(t))
6366 id := domain.IncarnationID{1}
6467 pub := incarnationPublication(t, id, "work", 1, nil)
@@ -66,7 +69,7 @@ func TestRepairHEAD(t *testing.T) {
6669 t.Fatal(err)
6770 }
6871 valid := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
69- if err := repo.RepairHEAD(context.Background(), id, valid); err != nil {
72+ if err := repo.ReconcileCheckpoint(context.Background(), id, valid); err != nil {
7073 t.Fatal(err)
7174 }
7275 before, err := os.ReadFile(repo.headPath(id))
@@ -75,7 +78,7 @@ func TestRepairHEAD(t *testing.T) {
7578 }
7679 invalid := valid
7780 invalid.ManifestDigest[0]++
78- if err := repo.RepairHEAD(context.Background(), id, invalid); err == nil {
81+ if err := repo.ReconcileCheckpoint(context.Background(), id, invalid); err == nil {
7982 t.Fatal("invalid checkpoint repaired HEAD")
8083 }
8184 after, err := os.ReadFile(repo.headPath(id))
@@ -87,6 +90,99 @@ func TestRepairHEAD(t *testing.T) {
8790 }
8891 }
8992
93+func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
94+ for _, tt := range []struct {
95+ name string
96+ mutate func(*testing.T, *Repository, ports.SnapshotPublication)
97+ }{
98+ {
99+ name: "missing object",
100+ mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
101+ t.Helper()
102+ require.NoError(t, os.Remove(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest)))
103+ },
104+ },
105+ {
106+ name: "corrupt object",
107+ mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
108+ t.Helper()
109+ require.NoError(t, os.WriteFile(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest), []byte("corrupt"), 0o600))
110+ },
111+ },
112+ {
113+ name: "corrupt manifest digest",
114+ mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
115+ t.Helper()
116+ head, err := os.ReadFile(repository.headPath(publication.IncarnationID))
117+ require.NoError(t, err)
118+ head[len(head)-1]++
119+ require.NoError(t, os.WriteFile(repository.headPath(publication.IncarnationID), head, 0o600))
120+ },
121+ },
122+ } {
123+ t.Run(tt.name, func(t *testing.T) {
124+ id := domain.IncarnationID{1}
125+ publication := incarnationPublication(t, id, "work", 1, nil)
126+ commitError := errors.New("catalogue commit failed")
127+ catalogue := &forwardOrphanCatalogue{
128+ record: domain.CatalogueRecord{Name: publication.Name, IncarnationID: id},
129+ replaceError: commitError,
130+ }
131+ repository := NewRepository(privateDir(t))
132+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
133+
134+ _, err := coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
135+ require.ErrorIs(t, err, commitError)
136+ tt.mutate(t, repository, publication)
137+ _, err = coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
138+ require.Error(t, err)
139+ require.Nil(t, catalogue.record.Committed, "an invalid forward orphan must not become authoritative")
140+ require.Equal(t, 1, catalogue.replaceCalls, "retry must not advance the catalogue")
141+ })
142+ }
143+}
144+
145+type forwardOrphanCatalogue struct {
146+ record domain.CatalogueRecord
147+ replaceError error
148+ replaceCalls int
149+}
150+
151+func (c *forwardOrphanCatalogue) Records() ([]domain.CatalogueRecord, error) {
152+ return []domain.CatalogueRecord{c.record}, nil
153+}
154+
155+func (c *forwardOrphanCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
156+ return c.record, c.record.Name == name, nil
157+}
158+
159+func (c *forwardOrphanCatalogue) Create(domain.CatalogueRecord) error {
160+ return errors.New("unexpected create")
161+}
162+func (c *forwardOrphanCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
163+ return errors.New("unexpected metadata update")
164+}
165+func (c *forwardOrphanCatalogue) Sync() error { return nil }
166+func (c *forwardOrphanCatalogue) Rename(string, domain.CatalogueRecord) error {
167+ return errors.New("unexpected rename")
168+}
169+func (c *forwardOrphanCatalogue) Delete(string) error { return errors.New("unexpected delete") }
170+func (c *forwardOrphanCatalogue) Close() error { return nil }
171+
172+func (c *forwardOrphanCatalogue) Replace(name string, record domain.CatalogueRecord) error {
173+ if name != c.record.Name || name != record.Name {
174+ return errors.New("catalogue key mismatch")
175+ }
176+ c.replaceCalls++
177+ if c.replaceError != nil {
178+ err := c.replaceError
179+ c.replaceError = nil
180+ return err
181+ }
182+ c.record = record
183+ return nil
184+}
185+
90186 func incarnationPublication(t *testing.T, id domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) ports.SnapshotPublication {
91187 t.Helper()
92188 tail, err := codec.MarshalObject(codec.HistoryTail, []byte("tail"))
@@ -3,13 +3,16 @@ package snapshot
3 import (3 import (
4 "context"4 "context"
5 "crypto/sha256"5 "crypto/sha256"
6+ "errors"
6 "os"7 "os"
7 "path/filepath"8 "path/filepath"
8 "testing"9 "testing"
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"
13+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
12 codec "github.com/bnema/vev/internal/usecase/snapshot"14 codec "github.com/bnema/vev/internal/usecase/snapshot"
15+ "github.com/stretchr/testify/require"
13 )16 )
14 17
15 func TestIncarnationNamespace(t *testing.T) {18 func TestIncarnationNamespace(t *testing.T) {
@@ -58,7 +61,7 @@ func TestDirectCheckpointLoad(t *testing.T) {
58 }61 }
59 }62 }
60 63
61-func TestRepairHEAD(t *testing.T) {64+func TestReconcileCheckpoint(t *testing.T) {
62 repo := NewRepository(privateDir(t))65 repo := NewRepository(privateDir(t))
63 id := domain.IncarnationID{1}66 id := domain.IncarnationID{1}
64 pub := incarnationPublication(t, id, "work", 1, nil)67 pub := incarnationPublication(t, id, "work", 1, nil)
@@ -66,7 +69,7 @@ func TestRepairHEAD(t *testing.T) {
66 t.Fatal(err)69 t.Fatal(err)
67 }70 }
68 valid := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}71 valid := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(pub.Manifest)}
69- if err := repo.RepairHEAD(context.Background(), id, valid); err != nil {72+ if err := repo.ReconcileCheckpoint(context.Background(), id, valid); err != nil {
70 t.Fatal(err)73 t.Fatal(err)
71 }74 }
72 before, err := os.ReadFile(repo.headPath(id))75 before, err := os.ReadFile(repo.headPath(id))
@@ -75,7 +78,7 @@ func TestRepairHEAD(t *testing.T) {
75 }78 }
76 invalid := valid79 invalid := valid
77 invalid.ManifestDigest[0]++80 invalid.ManifestDigest[0]++
78- if err := repo.RepairHEAD(context.Background(), id, invalid); err == nil {81+ if err := repo.ReconcileCheckpoint(context.Background(), id, invalid); err == nil {
79 t.Fatal("invalid checkpoint repaired HEAD")82 t.Fatal("invalid checkpoint repaired HEAD")
80 }83 }
81 after, err := os.ReadFile(repo.headPath(id))84 after, err := os.ReadFile(repo.headPath(id))
@@ -87,6 +90,99 @@ func TestRepairHEAD(t *testing.T) {
87 }90 }
88 }91 }
89 92
93+func TestForwardOrphanRetryValidatesCheckpointBeforeCatalogueAdvance(t *testing.T) {
94+ for _, tt := range []struct {
95+ name string
96+ mutate func(*testing.T, *Repository, ports.SnapshotPublication)
97+ }{
98+ {
99+ name: "missing object",
100+ mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
101+ t.Helper()
102+ require.NoError(t, os.Remove(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest)))
103+ },
104+ },
105+ {
106+ name: "corrupt object",
107+ mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
108+ t.Helper()
109+ require.NoError(t, os.WriteFile(repository.objectPath(publication.IncarnationID, publication.Objects[0].Digest), []byte("corrupt"), 0o600))
110+ },
111+ },
112+ {
113+ name: "corrupt manifest digest",
114+ mutate: func(t *testing.T, repository *Repository, publication ports.SnapshotPublication) {
115+ t.Helper()
116+ head, err := os.ReadFile(repository.headPath(publication.IncarnationID))
117+ require.NoError(t, err)
118+ head[len(head)-1]++
119+ require.NoError(t, os.WriteFile(repository.headPath(publication.IncarnationID), head, 0o600))
120+ },
121+ },
122+ } {
123+ t.Run(tt.name, func(t *testing.T) {
124+ id := domain.IncarnationID{1}
125+ publication := incarnationPublication(t, id, "work", 1, nil)
126+ commitError := errors.New("catalogue commit failed")
127+ catalogue := &forwardOrphanCatalogue{
128+ record: domain.CatalogueRecord{Name: publication.Name, IncarnationID: id},
129+ replaceError: commitError,
130+ }
131+ repository := NewRepository(privateDir(t))
132+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
133+
134+ _, err := coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
135+ require.ErrorIs(t, err, commitError)
136+ tt.mutate(t, repository, publication)
137+ _, err = coordinator.PublishCheckpoint(t.Context(), publication.Name, publication)
138+ require.Error(t, err)
139+ require.Nil(t, catalogue.record.Committed, "an invalid forward orphan must not become authoritative")
140+ require.Equal(t, 1, catalogue.replaceCalls, "retry must not advance the catalogue")
141+ })
142+ }
143+}
144+
145+type forwardOrphanCatalogue struct {
146+ record domain.CatalogueRecord
147+ replaceError error
148+ replaceCalls int
149+}
150+
151+func (c *forwardOrphanCatalogue) Records() ([]domain.CatalogueRecord, error) {
152+ return []domain.CatalogueRecord{c.record}, nil
153+}
154+
155+func (c *forwardOrphanCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) {
156+ return c.record, c.record.Name == name, nil
157+}
158+
159+func (c *forwardOrphanCatalogue) Create(domain.CatalogueRecord) error {
160+ return errors.New("unexpected create")
161+}
162+func (c *forwardOrphanCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
163+ return errors.New("unexpected metadata update")
164+}
165+func (c *forwardOrphanCatalogue) Sync() error { return nil }
166+func (c *forwardOrphanCatalogue) Rename(string, domain.CatalogueRecord) error {
167+ return errors.New("unexpected rename")
168+}
169+func (c *forwardOrphanCatalogue) Delete(string) error { return errors.New("unexpected delete") }
170+func (c *forwardOrphanCatalogue) Close() error { return nil }
171+
172+func (c *forwardOrphanCatalogue) Replace(name string, record domain.CatalogueRecord) error {
173+ if name != c.record.Name || name != record.Name {
174+ return errors.New("catalogue key mismatch")
175+ }
176+ c.replaceCalls++
177+ if c.replaceError != nil {
178+ err := c.replaceError
179+ c.replaceError = nil
180+ return err
181+ }
182+ c.record = record
183+ return nil
184+}
185+
90 func incarnationPublication(t *testing.T, id domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) ports.SnapshotPublication {186 func incarnationPublication(t *testing.T, id domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) ports.SnapshotPublication {
91 t.Helper()187 t.Helper()
92 tail, err := codec.MarshalObject(codec.HistoryTail, []byte("tail"))188 tail, err := codec.MarshalObject(codec.HistoryTail, []byte("tail"))
M internal/adapters/snapshot/load.go +1 -1
@@ -65,7 +65,7 @@ func (r *Repository) loadCheckpointLocked(ctx context.Context, id domain.Incarna
6565 return ports.SnapshotGeneration{IncarnationID: id, Name: name, Generation: ref.Generation, ParentCheckpoint: manifest.ParentCheckpoint, Manifest: data, Objects: objects}, manifest, nil
6666 }
6767
68-func (r *Repository) RepairHEAD(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
68+func (r *Repository) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
6969 if err := ctx.Err(); err != nil {
7070 return err
7171 }
@@ -65,7 +65,7 @@ func (r *Repository) loadCheckpointLocked(ctx context.Context, id domain.Incarna
65 return ports.SnapshotGeneration{IncarnationID: id, Name: name, Generation: ref.Generation, ParentCheckpoint: manifest.ParentCheckpoint, Manifest: data, Objects: objects}, manifest, nil65 return ports.SnapshotGeneration{IncarnationID: id, Name: name, Generation: ref.Generation, ParentCheckpoint: manifest.ParentCheckpoint, Manifest: data, Objects: objects}, manifest, nil
66 }66 }
67 67
68-func (r *Repository) RepairHEAD(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {68+func (r *Repository) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
69 if err := ctx.Err(); err != nil {69 if err := ctx.Err(); err != nil {
70 return err70 return err
71 }71 }
M internal/adapters/snapshot/maintenance_gc_test.go +44 -2
@@ -37,6 +37,49 @@ func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
3737 require.NoError(t, err)
3838 }
3939
40+func TestCollectGarbageRejectsMalformedAuthorityWithoutMutation(t *testing.T) {
41+ for _, tt := range []struct {
42+ name string
43+ mutate func(*testing.T, *Repository, domain.IncarnationID, domain.CheckpointRef) domain.CheckpointRef
44+ }{
45+ {
46+ name: "committed digest mismatch",
47+ mutate: func(_ *testing.T, _ *Repository, _ domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
48+ committed.ManifestDigest[0]++
49+ return committed
50+ },
51+ },
52+ {
53+ name: "committed manifest missing",
54+ mutate: func(t *testing.T, repository *Repository, id domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
55+ t.Helper()
56+ require.NoError(t, os.Remove(repository.manifestPath(id, committed.Generation)))
57+ return committed
58+ },
59+ },
60+ {
61+ name: "committed manifest corrupt",
62+ mutate: func(t *testing.T, repository *Repository, id domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
63+ t.Helper()
64+ require.NoError(t, os.WriteFile(repository.manifestPath(id, committed.Generation), []byte("corrupt"), 0o600))
65+ return committed
66+ },
67+ },
68+ } {
69+ t.Run(tt.name, func(t *testing.T) {
70+ repository := NewRepository(privateDir(t))
71+ publications := publishGarbageCollectionGenerations(t, repository, "work", 3)
72+ id := publications[0].IncarnationID
73+ committed := domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256(publications[2].Manifest)}
74+ committed = tt.mutate(t, repository, id, committed)
75+ before := snapshotRepositoryFiles(t, repository.dir)
76+
77+ require.Error(t, repository.CollectGarbage(t.Context(), map[domain.IncarnationID]domain.CheckpointRef{id: committed}))
78+ require.Equal(t, before, snapshotRepositoryFiles(t, repository.dir), "malformed catalogue authority must fail closed")
79+ })
80+ }
81+}
82+
4083 func TestCollectGarbage(t *testing.T) {
4184 t.Parallel()
4285 tests := []struct {
@@ -46,8 +89,7 @@ func TestCollectGarbage(t *testing.T) {
4689 wantGenerations []uint64
4790 }{
4891 {name: "incarnation absent from catalogue is removed whole"},
49- {name: "committed and immediate predecessor survive", committed: 3, known: true, wantGenerations: []uint64{2, 3}},
50- {name: "forward orphan newer than committed is removed", committed: 3, known: true, wantGenerations: []uint64{2, 3}},
92+ {name: "committed checkpoint prunes forward orphan and keeps predecessor", committed: 3, known: true, wantGenerations: []uint64{2, 3}},
5193 }
5294 for _, tt := range tests {
5395 t.Run(tt.name, func(t *testing.T) {
@@ -37,6 +37,49 @@ func TestCollectGarbageClearsUncommittedCheckpointHead(t *testing.T) {
37 require.NoError(t, err)37 require.NoError(t, err)
38 }38 }
39 39
40+func TestCollectGarbageRejectsMalformedAuthorityWithoutMutation(t *testing.T) {
41+ for _, tt := range []struct {
42+ name string
43+ mutate func(*testing.T, *Repository, domain.IncarnationID, domain.CheckpointRef) domain.CheckpointRef
44+ }{
45+ {
46+ name: "committed digest mismatch",
47+ mutate: func(_ *testing.T, _ *Repository, _ domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
48+ committed.ManifestDigest[0]++
49+ return committed
50+ },
51+ },
52+ {
53+ name: "committed manifest missing",
54+ mutate: func(t *testing.T, repository *Repository, id domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
55+ t.Helper()
56+ require.NoError(t, os.Remove(repository.manifestPath(id, committed.Generation)))
57+ return committed
58+ },
59+ },
60+ {
61+ name: "committed manifest corrupt",
62+ mutate: func(t *testing.T, repository *Repository, id domain.IncarnationID, committed domain.CheckpointRef) domain.CheckpointRef {
63+ t.Helper()
64+ require.NoError(t, os.WriteFile(repository.manifestPath(id, committed.Generation), []byte("corrupt"), 0o600))
65+ return committed
66+ },
67+ },
68+ } {
69+ t.Run(tt.name, func(t *testing.T) {
70+ repository := NewRepository(privateDir(t))
71+ publications := publishGarbageCollectionGenerations(t, repository, "work", 3)
72+ id := publications[0].IncarnationID
73+ committed := domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256(publications[2].Manifest)}
74+ committed = tt.mutate(t, repository, id, committed)
75+ before := snapshotRepositoryFiles(t, repository.dir)
76+
77+ require.Error(t, repository.CollectGarbage(t.Context(), map[domain.IncarnationID]domain.CheckpointRef{id: committed}))
78+ require.Equal(t, before, snapshotRepositoryFiles(t, repository.dir), "malformed catalogue authority must fail closed")
79+ })
80+ }
81+}
82+
40 func TestCollectGarbage(t *testing.T) {83 func TestCollectGarbage(t *testing.T) {
41 t.Parallel()84 t.Parallel()
42 tests := []struct {85 tests := []struct {
@@ -46,8 +89,7 @@ func TestCollectGarbage(t *testing.T) {
46 wantGenerations []uint6489 wantGenerations []uint64
47 }{90 }{
48 {name: "incarnation absent from catalogue is removed whole"},91 {name: "incarnation absent from catalogue is removed whole"},
49- {name: "committed and immediate predecessor survive", committed: 3, known: true, wantGenerations: []uint64{2, 3}},92+ {name: "committed checkpoint prunes forward orphan and keeps predecessor", committed: 3, known: true, wantGenerations: []uint64{2, 3}},
50- {name: "forward orphan newer than committed is removed", committed: 3, known: true, wantGenerations: []uint64{2, 3}},
51 }93 }
52 for _, tt := range tests {94 for _, tt := range tests {
53 t.Run(tt.name, func(t *testing.T) {95 t.Run(tt.name, func(t *testing.T) {
M internal/adapters/snapshot/publish.go +19 -17
@@ -34,10 +34,12 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
3434 if err := ctx.Err(); err != nil {
3535 return err
3636 }
37- // The common replay path needs neither a manifest decode nor a reference
38- // map. Checking the immutable bytes directly also keeps its descriptor
39- // reads beneath the pinned repository root.
40- if unchanged, err := r.unchangedPublication(publication); err == nil && unchanged {
37+ // A replay may find a forward orphan left after publication but before its
38+ // catalogue commit. Accept it only after the canonical checkpoint loader has
39+ // revalidated the pointer digest, manifest, and every referenced object.
40+ if unchanged, err := r.unchangedPublication(ctx, publication); err != nil {
41+ return err
42+ } else if unchanged {
4143 return nil
4244 }
4345
@@ -137,22 +139,22 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
137139 return nil
138140 }
139141
140-// unchangedPublication uses one root for the two reads in the replay fast
141-// path. The root is confined to this operation and is closed before return;
142-// all other repository operations retain their open-per-operation roots.
143-func (r *Repository) unchangedPublication(publication ports.SnapshotPublication) (unchanged bool, err error) {
144- root, err := r.openRoot()
145- if err != nil {
146- return false, err
147- }
148- defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
149-
150- generation, _, err := r.readHeadWithRoot(root, publication.IncarnationID)
142+// unchangedPublication validates an existing forward orphan through the same
143+// loader used for recovery. A corrupt replay must not be accepted merely
144+// because its manifest bytes happen to match the requested publication.
145+func (r *Repository) unchangedPublication(ctx context.Context, publication ports.SnapshotPublication) (bool, error) {
146+ generation, digest, err := r.readHead(publication.IncarnationID)
151147 if err != nil || generation != publication.Generation {
152148 return false, nil
153149 }
154- current, err := r.readBoundedRoot(root, r.manifestPath(publication.IncarnationID, generation))
155- return err == nil && bytes.Equal(current, publication.Manifest), nil
150+ current, _, err := r.loadCheckpointLocked(ctx, publication.IncarnationID, publication.Name, ports.CheckpointRef{
151+ Generation: generation,
152+ ManifestDigest: digest,
153+ })
154+ if err != nil {
155+ return false, fmt.Errorf("validate replayed checkpoint: %w", err)
156+ }
157+ return bytes.Equal(current.Manifest, publication.Manifest), nil
156158 }
157159
158160 func (r *Repository) currentIncarnationPublication(ctx context.Context, publication ports.SnapshotPublication) (uint64, []byte, map[ports.SnapshotDigest]codec.ObjectRef, error) {
@@ -34,10 +34,12 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
34 if err := ctx.Err(); err != nil {34 if err := ctx.Err(); err != nil {
35 return err35 return err
36 }36 }
37- // The common replay path needs neither a manifest decode nor a reference37+ // A replay may find a forward orphan left after publication but before its
38- // map. Checking the immutable bytes directly also keeps its descriptor38+ // catalogue commit. Accept it only after the canonical checkpoint loader has
39- // reads beneath the pinned repository root.39+ // revalidated the pointer digest, manifest, and every referenced object.
40- if unchanged, err := r.unchangedPublication(publication); err == nil && unchanged {40+ if unchanged, err := r.unchangedPublication(ctx, publication); err != nil {
41+ return err
42+ } else if unchanged {
41 return nil43 return nil
42 }44 }
43 45
@@ -137,22 +139,22 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
137 return nil139 return nil
138 }140 }
139 141
140-// unchangedPublication uses one root for the two reads in the replay fast142+// unchangedPublication validates an existing forward orphan through the same
141-// path. The root is confined to this operation and is closed before return;143+// loader used for recovery. A corrupt replay must not be accepted merely
142-// all other repository operations retain their open-per-operation roots.144+// because its manifest bytes happen to match the requested publication.
143-func (r *Repository) unchangedPublication(publication ports.SnapshotPublication) (unchanged bool, err error) {145+func (r *Repository) unchangedPublication(ctx context.Context, publication ports.SnapshotPublication) (bool, error) {
144- root, err := r.openRoot()146+ generation, digest, err := r.readHead(publication.IncarnationID)
145- if err != nil {
146- return false, err
147- }
148- defer func() { joinCloseError(&err, "close snapshot root", r.closeRoot(root)) }()
149-
150- generation, _, err := r.readHeadWithRoot(root, publication.IncarnationID)
151 if err != nil || generation != publication.Generation {147 if err != nil || generation != publication.Generation {
152 return false, nil148 return false, nil
153 }149 }
154- current, err := r.readBoundedRoot(root, r.manifestPath(publication.IncarnationID, generation))150+ current, _, err := r.loadCheckpointLocked(ctx, publication.IncarnationID, publication.Name, ports.CheckpointRef{
155- return err == nil && bytes.Equal(current, publication.Manifest), nil151+ Generation: generation,
152+ ManifestDigest: digest,
153+ })
154+ if err != nil {
155+ return false, fmt.Errorf("validate replayed checkpoint: %w", err)
156+ }
157+ return bytes.Equal(current.Manifest, publication.Manifest), nil
156 }158 }
157 159
158 func (r *Repository) currentIncarnationPublication(ctx context.Context, publication ports.SnapshotPublication) (uint64, []byte, map[ports.SnapshotDigest]codec.ObjectRef, error) {160 func (r *Repository) currentIncarnationPublication(ctx context.Context, publication ports.SnapshotPublication) (uint64, []byte, map[ports.SnapshotDigest]codec.ObjectRef, error) {
M internal/adapters/snapshot/publish_efficiency_test.go +7 -14
@@ -13,7 +13,7 @@ import (
1313 codec "github.com/bnema/vev/internal/usecase/snapshot"
1414 )
1515
16-func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
16+func TestRepositoryPublishRevalidatesUnchangedTenThousandObjectHistory(t *testing.T) {
1717 const historyObjects = 10_000
1818
1919 repo := NewRepository(privateDir(t))
@@ -24,19 +24,12 @@ func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
2424 t.Fatal(err)
2525 }
2626
27- // Replaying the now-current generation still parses the 10k references, but
28- // must not allocate one object-sized buffer per retained history entry.
29- // The limit is 450, not lower: the os.Root-based fast path pays a constant
30- // allocation cost for guarded reads and root identity validation, and -race
31- // adds further overhead. A per-retained-object regression would exceed this
32- // by orders of magnitude.
33- allocations := testing.AllocsPerRun(3, func() {
34- if err := repo.Publish(context.Background(), second); err != nil {
35- panic(err)
36- }
37- })
38- if allocations > 450 {
39- t.Fatalf("unchanged 10k history Publish allocations = %.0f, want <= 450; retained objects were likely copied", allocations)
27+ // Replaying the current generation revalidates every referenced immutable
28+ // object before accepting it. That is required for a forward orphan retry:
29+ // the catalogue must not advance to a checkpoint whose files were damaged
30+ // after its original publication.
31+ if err := repo.Publish(context.Background(), second); err != nil {
32+ t.Fatal(err)
4033 }
4134
4235 got, err := loadPublication(context.Background(), repo, second)
@@ -13,7 +13,7 @@ import (
13 codec "github.com/bnema/vev/internal/usecase/snapshot"13 codec "github.com/bnema/vev/internal/usecase/snapshot"
14 )14 )
15 15
16-func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {16+func TestRepositoryPublishRevalidatesUnchangedTenThousandObjectHistory(t *testing.T) {
17 const historyObjects = 10_00017 const historyObjects = 10_000
18 18
19 repo := NewRepository(privateDir(t))19 repo := NewRepository(privateDir(t))
@@ -24,19 +24,12 @@ func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
24 t.Fatal(err)24 t.Fatal(err)
25 }25 }
26 26
27- // Replaying the now-current generation still parses the 10k references, but27+ // Replaying the current generation revalidates every referenced immutable
28- // must not allocate one object-sized buffer per retained history entry.28+ // object before accepting it. That is required for a forward orphan retry:
29- // The limit is 450, not lower: the os.Root-based fast path pays a constant29+ // the catalogue must not advance to a checkpoint whose files were damaged
30- // allocation cost for guarded reads and root identity validation, and -race30+ // after its original publication.
31- // adds further overhead. A per-retained-object regression would exceed this31+ if err := repo.Publish(context.Background(), second); err != nil {
32- // by orders of magnitude.32+ t.Fatal(err)
33- allocations := testing.AllocsPerRun(3, func() {
34- if err := repo.Publish(context.Background(), second); err != nil {
35- panic(err)
36- }
37- })
38- if allocations > 450 {
39- t.Fatalf("unchanged 10k history Publish allocations = %.0f, want <= 450; retained objects were likely copied", allocations)
40 }33 }
41 34
42 got, err := loadPublication(context.Background(), repo, second)35 got, err := loadPublication(context.Background(), repo, second)
M internal/app/daemon_integration_test.go +2 -2
@@ -790,11 +790,11 @@ func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain
790790 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
791791 }
792792
793-func (r *lifecycleBlockingRestore) RepairHEAD(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
793+func (r *lifecycleBlockingRestore) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
794794 r.enterOnce.Do(func() { close(r.entered) })
795795 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
796796 r.returnOnce.Do(func() { close(r.returned) })
797- return r.SnapshotRepository.RepairHEAD(ctx, id, ref)
797+ return r.SnapshotRepository.ReconcileCheckpoint(ctx, id, ref)
798798 }
799799
800800 func (r *lifecycleBlockingRestore) repairedName() string {
@@ -790,11 +790,11 @@ func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain
790 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)790 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
791 }791 }
792 792
793-func (r *lifecycleBlockingRestore) RepairHEAD(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {793+func (r *lifecycleBlockingRestore) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
794 r.enterOnce.Do(func() { close(r.entered) })794 r.enterOnce.Do(func() { close(r.entered) })
795 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.795 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
796 r.returnOnce.Do(func() { close(r.returned) })796 r.returnOnce.Do(func() { close(r.returned) })
797- return r.SnapshotRepository.RepairHEAD(ctx, id, ref)797+ return r.SnapshotRepository.ReconcileCheckpoint(ctx, id, ref)
798 }798 }
799 799
800 func (r *lifecycleBlockingRestore) repairedName() string {800 func (r *lifecycleBlockingRestore) repairedName() string {
M internal/ports/mocks/recovery_mock.go +11 -11
@@ -791,12 +791,12 @@ func (_c *MockSnapshotRepository_Publish_Call) RunAndReturn(run func(context1 co
791791 return _c
792792 }
793793
794-// RepairHEAD provides a mock function for the type MockSnapshotRepository
795-func (_mock *MockSnapshotRepository) RepairHEAD(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error {
794+// ReconcileCheckpoint provides a mock function for the type MockSnapshotRepository
795+func (_mock *MockSnapshotRepository) ReconcileCheckpoint(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error {
796796 ret := _mock.Called(context1, incarnationID, v)
797797
798798 if len(ret) == 0 {
799- panic("no return value specified for RepairHEAD")
799+ panic("no return value specified for ReconcileCheckpoint")
800800 }
801801
802802 var r0 error
@@ -808,20 +808,20 @@ func (_mock *MockSnapshotRepository) RepairHEAD(context1 context.Context, incarn
808808 return r0
809809 }
810810
811-// MockSnapshotRepository_RepairHEAD_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RepairHEAD'
812-type MockSnapshotRepository_RepairHEAD_Call struct {
811+// MockSnapshotRepository_ReconcileCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ReconcileCheckpoint'
812+type MockSnapshotRepository_ReconcileCheckpoint_Call struct {
813813 *mock.Call
814814 }
815815
816-// RepairHEAD is a helper method to define mock.On call
816+// ReconcileCheckpoint is a helper method to define mock.On call
817817 // - context1 context.Context
818818 // - incarnationID domain.IncarnationID
819819 // - v ports.CheckpointRef
820-func (_e *MockSnapshotRepository_Expecter) RepairHEAD(context1 any, incarnationID any, v any) *MockSnapshotRepository_RepairHEAD_Call {
821- return &MockSnapshotRepository_RepairHEAD_Call{Call: _e.mock.On("RepairHEAD", context1, incarnationID, v)}
820+func (_e *MockSnapshotRepository_Expecter) ReconcileCheckpoint(context1 any, incarnationID any, v any) *MockSnapshotRepository_ReconcileCheckpoint_Call {
821+ return &MockSnapshotRepository_ReconcileCheckpoint_Call{Call: _e.mock.On("ReconcileCheckpoint", context1, incarnationID, v)}
822822 }
823823
824-func (_c *MockSnapshotRepository_RepairHEAD_Call) Run(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef)) *MockSnapshotRepository_RepairHEAD_Call {
824+func (_c *MockSnapshotRepository_ReconcileCheckpoint_Call) Run(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef)) *MockSnapshotRepository_ReconcileCheckpoint_Call {
825825 _c.Call.Run(func(args mock.Arguments) {
826826 var arg0 context.Context
827827 if args[0] != nil {
@@ -844,12 +844,12 @@ func (_c *MockSnapshotRepository_RepairHEAD_Call) Run(run func(context1 context.
844844 return _c
845845 }
846846
847-func (_c *MockSnapshotRepository_RepairHEAD_Call) Return(err error) *MockSnapshotRepository_RepairHEAD_Call {
847+func (_c *MockSnapshotRepository_ReconcileCheckpoint_Call) Return(err error) *MockSnapshotRepository_ReconcileCheckpoint_Call {
848848 _c.Call.Return(err)
849849 return _c
850850 }
851851
852-func (_c *MockSnapshotRepository_RepairHEAD_Call) RunAndReturn(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error) *MockSnapshotRepository_RepairHEAD_Call {
852+func (_c *MockSnapshotRepository_ReconcileCheckpoint_Call) RunAndReturn(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error) *MockSnapshotRepository_ReconcileCheckpoint_Call {
853853 _c.Call.Return(run)
854854 return _c
855855 }
@@ -791,12 +791,12 @@ func (_c *MockSnapshotRepository_Publish_Call) RunAndReturn(run func(context1 co
791 return _c791 return _c
792 }792 }
793 793
794-// RepairHEAD provides a mock function for the type MockSnapshotRepository794+// ReconcileCheckpoint provides a mock function for the type MockSnapshotRepository
795-func (_mock *MockSnapshotRepository) RepairHEAD(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error {795+func (_mock *MockSnapshotRepository) ReconcileCheckpoint(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error {
796 ret := _mock.Called(context1, incarnationID, v)796 ret := _mock.Called(context1, incarnationID, v)
797 797
798 if len(ret) == 0 {798 if len(ret) == 0 {
799- panic("no return value specified for RepairHEAD")799+ panic("no return value specified for ReconcileCheckpoint")
800 }800 }
801 801
802 var r0 error802 var r0 error
@@ -808,20 +808,20 @@ func (_mock *MockSnapshotRepository) RepairHEAD(context1 context.Context, incarn
808 return r0808 return r0
809 }809 }
810 810
811-// MockSnapshotRepository_RepairHEAD_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RepairHEAD'811+// MockSnapshotRepository_ReconcileCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ReconcileCheckpoint'
812-type MockSnapshotRepository_RepairHEAD_Call struct {812+type MockSnapshotRepository_ReconcileCheckpoint_Call struct {
813 *mock.Call813 *mock.Call
814 }814 }
815 815
816-// RepairHEAD is a helper method to define mock.On call816+// ReconcileCheckpoint is a helper method to define mock.On call
817 // - context1 context.Context817 // - context1 context.Context
818 // - incarnationID domain.IncarnationID818 // - incarnationID domain.IncarnationID
819 // - v ports.CheckpointRef819 // - v ports.CheckpointRef
820-func (_e *MockSnapshotRepository_Expecter) RepairHEAD(context1 any, incarnationID any, v any) *MockSnapshotRepository_RepairHEAD_Call {820+func (_e *MockSnapshotRepository_Expecter) ReconcileCheckpoint(context1 any, incarnationID any, v any) *MockSnapshotRepository_ReconcileCheckpoint_Call {
821- return &MockSnapshotRepository_RepairHEAD_Call{Call: _e.mock.On("RepairHEAD", context1, incarnationID, v)}821+ return &MockSnapshotRepository_ReconcileCheckpoint_Call{Call: _e.mock.On("ReconcileCheckpoint", context1, incarnationID, v)}
822 }822 }
823 823
824-func (_c *MockSnapshotRepository_RepairHEAD_Call) Run(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef)) *MockSnapshotRepository_RepairHEAD_Call {824+func (_c *MockSnapshotRepository_ReconcileCheckpoint_Call) Run(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef)) *MockSnapshotRepository_ReconcileCheckpoint_Call {
825 _c.Call.Run(func(args mock.Arguments) {825 _c.Call.Run(func(args mock.Arguments) {
826 var arg0 context.Context826 var arg0 context.Context
827 if args[0] != nil {827 if args[0] != nil {
@@ -844,12 +844,12 @@ func (_c *MockSnapshotRepository_RepairHEAD_Call) Run(run func(context1 context.
844 return _c844 return _c
845 }845 }
846 846
847-func (_c *MockSnapshotRepository_RepairHEAD_Call) Return(err error) *MockSnapshotRepository_RepairHEAD_Call {847+func (_c *MockSnapshotRepository_ReconcileCheckpoint_Call) Return(err error) *MockSnapshotRepository_ReconcileCheckpoint_Call {
848 _c.Call.Return(err)848 _c.Call.Return(err)
849 return _c849 return _c
850 }850 }
851 851
852-func (_c *MockSnapshotRepository_RepairHEAD_Call) RunAndReturn(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error) *MockSnapshotRepository_RepairHEAD_Call {852+func (_c *MockSnapshotRepository_ReconcileCheckpoint_Call) RunAndReturn(run func(context1 context.Context, incarnationID domain.IncarnationID, v ports.CheckpointRef) error) *MockSnapshotRepository_ReconcileCheckpoint_Call {
853 _c.Call.Return(run)853 _c.Call.Return(run)
854 return _c854 return _c
855 }855 }
M internal/ports/recovery.go +1 -1
@@ -42,7 +42,7 @@ type Catalogue interface {
4242 type SnapshotRepository interface {
4343 Publish(context.Context, SnapshotPublication) error
4444 LoadCheckpoint(context.Context, domain.IncarnationID, string, CheckpointRef) (SnapshotGeneration, error)
45- RepairHEAD(context.Context, domain.IncarnationID, CheckpointRef) error
45+ ReconcileCheckpoint(context.Context, domain.IncarnationID, CheckpointRef) error
4646 DeleteIncarnation(context.Context, domain.IncarnationID) error
4747 CollectGarbage(context.Context, map[domain.IncarnationID]domain.CheckpointRef) error
4848 }
@@ -42,7 +42,7 @@ type Catalogue interface {
42 type SnapshotRepository interface {42 type SnapshotRepository interface {
43 Publish(context.Context, SnapshotPublication) error43 Publish(context.Context, SnapshotPublication) error
44 LoadCheckpoint(context.Context, domain.IncarnationID, string, CheckpointRef) (SnapshotGeneration, error)44 LoadCheckpoint(context.Context, domain.IncarnationID, string, CheckpointRef) (SnapshotGeneration, error)
45- RepairHEAD(context.Context, domain.IncarnationID, CheckpointRef) error45+ ReconcileCheckpoint(context.Context, domain.IncarnationID, CheckpointRef) error
46 DeleteIncarnation(context.Context, domain.IncarnationID) error46 DeleteIncarnation(context.Context, domain.IncarnationID) error
47 CollectGarbage(context.Context, map[domain.IncarnationID]domain.CheckpointRef) error47 CollectGarbage(context.Context, map[domain.IncarnationID]domain.CheckpointRef) error
48 }48 }
M internal/usecase/daemon/coderabbit_regression_test.go +56 -0
@@ -1,7 +1,9 @@
11 package daemon
22
33 import (
4+ "bytes"
45 "context"
6+ "errors"
57 "sync"
68 "syscall"
79 "testing"
@@ -10,6 +12,7 @@ import (
1012
1113 "github.com/bnema/vev/internal/domain"
1214 "github.com/bnema/vev/internal/ports"
15+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
1316 )
1417
1518 func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
@@ -40,6 +43,59 @@ func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
4043 }
4144 }
4245
46+type blockingDiscardRecoveryRepository struct {
47+ *durableRecoveryRepository
48+ deleteEntered chan struct{}
49+ releaseDelete chan struct{}
50+}
51+
52+func (r *blockingDiscardRecoveryRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
53+ close(r.deleteEntered)
54+ <-r.releaseDelete
55+ return nil
56+}
57+
58+func TestRestoreFailureCannotOverwriteDiscardReplacement(t *testing.T) {
59+ record := durableRecoveryRecord(0)
60+ record.DegradedReason = "previous restore failure"
61+ repository := &blockingDiscardRecoveryRepository{
62+ durableRecoveryRepository: &durableRecoveryRepository{},
63+ deleteEntered: make(chan struct{}),
64+ releaseDelete: make(chan struct{}),
65+ }
66+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
67+ d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{7}, 16)))
68+
69+ discarded := make(chan error, 1)
70+ go func() {
71+ _, err := (controlExec{ctx: t.Context(), d: d, recoveryName: record.Name}).SessionRecovery("discard")
72+ discarded <- err
73+ }()
74+ <-repository.deleteEntered // The fresh record is now authoritative behind Discard's mutation fence.
75+
76+ finished := make(chan struct{})
77+ go func() {
78+ d.finishRecordRestore(record, errors.New("late restore failure"), make(chan struct{}))
79+ close(finished)
80+ }()
81+
82+ close(repository.releaseDelete)
83+ require.NoError(t, <-discarded)
84+ <-finished
85+
86+ persisted, ok, err := catalogue.Record(record.Name)
87+ require.NoError(t, err)
88+ require.True(t, ok)
89+ require.NotEqual(t, record.IncarnationID, persisted.IncarnationID)
90+ require.Nil(t, persisted.Committed)
91+ require.Empty(t, persisted.DegradedReason)
92+ d.mu.Lock()
93+ entry := d.stopped[record.Name]
94+ d.mu.Unlock()
95+ require.Equal(t, persisted, entry.record)
96+ require.Equal(t, ports.SessionStopped, entry.state)
97+}
98+
4399 func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {
44100 d := newTestDaemon(t, nil, stubClock{})
45101 done := make(chan struct{})
@@ -1,7 +1,9 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4+ "bytes"
4 "context"5 "context"
6+ "errors"
5 "sync"7 "sync"
6 "syscall"8 "syscall"
7 "testing"9 "testing"
@@ -10,6 +12,7 @@ import (
10 12
11 "github.com/bnema/vev/internal/domain"13 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"14 "github.com/bnema/vev/internal/ports"
15+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
13 )16 )
14 17
15 func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {18 func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
@@ -40,6 +43,59 @@ func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
40 }43 }
41 }44 }
42 45
46+type blockingDiscardRecoveryRepository struct {
47+ *durableRecoveryRepository
48+ deleteEntered chan struct{}
49+ releaseDelete chan struct{}
50+}
51+
52+func (r *blockingDiscardRecoveryRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
53+ close(r.deleteEntered)
54+ <-r.releaseDelete
55+ return nil
56+}
57+
58+func TestRestoreFailureCannotOverwriteDiscardReplacement(t *testing.T) {
59+ record := durableRecoveryRecord(0)
60+ record.DegradedReason = "previous restore failure"
61+ repository := &blockingDiscardRecoveryRepository{
62+ durableRecoveryRepository: &durableRecoveryRepository{},
63+ deleteEntered: make(chan struct{}),
64+ releaseDelete: make(chan struct{}),
65+ }
66+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
67+ d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{7}, 16)))
68+
69+ discarded := make(chan error, 1)
70+ go func() {
71+ _, err := (controlExec{ctx: t.Context(), d: d, recoveryName: record.Name}).SessionRecovery("discard")
72+ discarded <- err
73+ }()
74+ <-repository.deleteEntered // The fresh record is now authoritative behind Discard's mutation fence.
75+
76+ finished := make(chan struct{})
77+ go func() {
78+ d.finishRecordRestore(record, errors.New("late restore failure"), make(chan struct{}))
79+ close(finished)
80+ }()
81+
82+ close(repository.releaseDelete)
83+ require.NoError(t, <-discarded)
84+ <-finished
85+
86+ persisted, ok, err := catalogue.Record(record.Name)
87+ require.NoError(t, err)
88+ require.True(t, ok)
89+ require.NotEqual(t, record.IncarnationID, persisted.IncarnationID)
90+ require.Nil(t, persisted.Committed)
91+ require.Empty(t, persisted.DegradedReason)
92+ d.mu.Lock()
93+ entry := d.stopped[record.Name]
94+ d.mu.Unlock()
95+ require.Equal(t, persisted, entry.record)
96+ require.Equal(t, ports.SessionStopped, entry.state)
97+}
98+
43 func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {99 func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {
44 d := newTestDaemon(t, nil, stubClock{})100 d := newTestDaemon(t, nil, stubClock{})
45 done := make(chan struct{})101 done := make(chan struct{})
M internal/usecase/daemon/daemon.go +12 -10
@@ -127,12 +127,11 @@ type Daemon struct {
127127 // snapshotRepository is the sole checkpoint storage contract.
128128 snapshotRepository ports.SnapshotRepository
129129 recovery *recoveryusecase.Coordinator
130- maintenance maintenanceDependencies
131130 maintenanceWorkerCancel context.CancelFunc
132131 maintenanceWorkerDone chan struct{}
133132 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
134- // restoration may repair a repository HEAD, so it is a durable writer and is
135- // guarded by snapshotWorkerMu with the other two.
133+ // restoration reconciles durable checkpoints, so it is a durable writer and
134+ // is guarded by snapshotWorkerMu with the other two.
136135 restoreWorkerDone chan struct{}
137136 snapsEnabled bool
138137 noticeStore ports.NoticeStore
@@ -326,11 +325,14 @@ func WithNoticeStore(store ports.NoticeStore) Option {
326325 return func(d *Daemon) { d.noticeStore = store }
327326 }
328327
329-// WithDurableMaintenance configures the one pre-publication garbage-collection
330-// pass. The application invokes it before opening the daemon socket.
328+// WithDurableMaintenance retains the application wiring for the one
329+// pre-publication GC pass. Standalone users without an explicitly supplied
330+// coordinator receive the same canonical coordinator path.
331331 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
332332 return func(d *Daemon) {
333- d.maintenance = newMaintenanceDependencies(catalogue, repository)
333+ if d.recovery == nil {
334+ d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, nil)
335+ }
334336 }
335337 }
336338
@@ -498,12 +500,12 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
498500 var maxCreatedAt int64
499501 hasCreatedAt := false
500502 for _, r := range records {
501- stopped := stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
502503 if d.catalogueRecordsProvided {
503- stopped.record = r
504- stopped.state, stopped.restoreDone = initialSessionState(r)
504+ state, done := initialSessionState(r)
505+ d.stopped[r.Name] = stoppedSessionFromRecord(r, state, done)
506+ } else {
507+ d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
505508 }
506- d.stopped[r.Name] = stopped
507509 if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
508510 maxCreatedAt = r.CreatedAt
509511 hasCreatedAt = true
@@ -127,12 +127,11 @@ type Daemon struct {
127 // snapshotRepository is the sole checkpoint storage contract.127 // snapshotRepository is the sole checkpoint storage contract.
128 snapshotRepository ports.SnapshotRepository128 snapshotRepository ports.SnapshotRepository
129 recovery *recoveryusecase.Coordinator129 recovery *recoveryusecase.Coordinator
130- maintenance maintenanceDependencies
131 maintenanceWorkerCancel context.CancelFunc130 maintenanceWorkerCancel context.CancelFunc
132 maintenanceWorkerDone chan struct{}131 maintenanceWorkerDone chan struct{}
133 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup132 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
134- // restoration may repair a repository HEAD, so it is a durable writer and is133+ // restoration reconciles durable checkpoints, so it is a durable writer and
135- // guarded by snapshotWorkerMu with the other two.134+ // is guarded by snapshotWorkerMu with the other two.
136 restoreWorkerDone chan struct{}135 restoreWorkerDone chan struct{}
137 snapsEnabled bool136 snapsEnabled bool
138 noticeStore ports.NoticeStore137 noticeStore ports.NoticeStore
@@ -326,11 +325,14 @@ func WithNoticeStore(store ports.NoticeStore) Option {
326 return func(d *Daemon) { d.noticeStore = store }325 return func(d *Daemon) { d.noticeStore = store }
327 }326 }
328 327
329-// WithDurableMaintenance configures the one pre-publication garbage-collection328+// WithDurableMaintenance retains the application wiring for the one
330-// pass. The application invokes it before opening the daemon socket.329+// pre-publication GC pass. Standalone users without an explicitly supplied
330+// coordinator receive the same canonical coordinator path.
331 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {331 func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
332 return func(d *Daemon) {332 return func(d *Daemon) {
333- d.maintenance = newMaintenanceDependencies(catalogue, repository)333+ if d.recovery == nil {
334+ d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, nil)
335+ }
334 }336 }
335 }337 }
336 338
@@ -498,12 +500,12 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
498 var maxCreatedAt int64500 var maxCreatedAt int64
499 hasCreatedAt := false501 hasCreatedAt := false
500 for _, r := range records {502 for _, r := range records {
501- stopped := stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
502 if d.catalogueRecordsProvided {503 if d.catalogueRecordsProvided {
503- stopped.record = r504+ state, done := initialSessionState(r)
504- stopped.state, stopped.restoreDone = initialSessionState(r)505+ d.stopped[r.Name] = stoppedSessionFromRecord(r, state, done)
506+ } else {
507+ d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
505 }508 }
506- d.stopped[r.Name] = stopped
507 if !hasCreatedAt || r.CreatedAt > maxCreatedAt {509 if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
508 maxCreatedAt = r.CreatedAt510 maxCreatedAt = r.CreatedAt
509 hasCreatedAt = true511 hasCreatedAt = true
M internal/usecase/daemon/durable_recovery_test.go +5 -2
@@ -146,7 +146,7 @@ func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.
146146 return generation, nil
147147 }
148148
149-func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
149+func (r *durableRecoveryRepository) ReconcileCheckpoint(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
150150 r.mu.Lock()
151151 defer r.mu.Unlock()
152152 r.repairCalls++
@@ -431,6 +431,7 @@ func TestFinishRecordRestoreDoesNotDegradeStaleCatalogueAuthority(t *testing.T)
431431 name: "catalogue read error",
432432 setup: func(d *Daemon, c *durableRecoveryCatalogue, _ domain.CatalogueRecord) {
433433 d.catalogue = recordErrorCatalogue{durableRecoveryCatalogue: c, err: errors.New("read failed")}
434+ d.recovery = recoveryusecase.NewCoordinator(d.catalogue, nil, nil)
434435 },
435436 },
436437 } {
@@ -555,7 +556,9 @@ func TestAttachWaitsForRestore(t *testing.T) {
555556 func TestAttachBarrierSurvivesLiveRegistryPublication(t *testing.T) {
556557 record := durableRecoveryRecord(0)
557558 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
558- done := d.recordRestoreDone(record.Name)
559+ d.mu.Lock()
560+ done := d.stopped[record.Name].restoreDone
561+ d.mu.Unlock()
559562 sess := &session{name: record.Name, incarnation: record.IncarnationID}
560563
561564 registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess)
@@ -146,7 +146,7 @@ func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.
146 return generation, nil146 return generation, nil
147 }147 }
148 148
149-func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {149+func (r *durableRecoveryRepository) ReconcileCheckpoint(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
150 r.mu.Lock()150 r.mu.Lock()
151 defer r.mu.Unlock()151 defer r.mu.Unlock()
152 r.repairCalls++152 r.repairCalls++
@@ -431,6 +431,7 @@ func TestFinishRecordRestoreDoesNotDegradeStaleCatalogueAuthority(t *testing.T)
431 name: "catalogue read error",431 name: "catalogue read error",
432 setup: func(d *Daemon, c *durableRecoveryCatalogue, _ domain.CatalogueRecord) {432 setup: func(d *Daemon, c *durableRecoveryCatalogue, _ domain.CatalogueRecord) {
433 d.catalogue = recordErrorCatalogue{durableRecoveryCatalogue: c, err: errors.New("read failed")}433 d.catalogue = recordErrorCatalogue{durableRecoveryCatalogue: c, err: errors.New("read failed")}
434+ d.recovery = recoveryusecase.NewCoordinator(d.catalogue, nil, nil)
434 },435 },
435 },436 },
436 } {437 } {
@@ -555,7 +556,9 @@ func TestAttachWaitsForRestore(t *testing.T) {
555 func TestAttachBarrierSurvivesLiveRegistryPublication(t *testing.T) {556 func TestAttachBarrierSurvivesLiveRegistryPublication(t *testing.T) {
556 record := durableRecoveryRecord(0)557 record := durableRecoveryRecord(0)
557 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})558 d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
558- done := d.recordRestoreDone(record.Name)559+ d.mu.Lock()
560+ done := d.stopped[record.Name].restoreDone
561+ d.mu.Unlock()
559 sess := &session{name: record.Name, incarnation: record.IncarnationID}562 sess := &session{name: record.Name, incarnation: record.IncarnationID}
560 563
561 registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess)564 registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess)
M internal/usecase/daemon/maintenance.go +5 -44
@@ -3,57 +3,18 @@ package daemon
33 import (
44 "context"
55 "fmt"
6-
7- "github.com/bnema/vev/internal/domain"
8- "github.com/bnema/vev/internal/ports"
96 )
107
11-type maintenanceDependencies struct {
12- catalogue ports.Catalogue
13- repository ports.SnapshotRepository
14-}
15-
16-func newMaintenanceDependencies(catalogue ports.Catalogue, repository ports.SnapshotRepository) maintenanceDependencies {
17- return maintenanceDependencies{catalogue: catalogue, repository: repository}
18-}
19-
20-// CollectStartupGarbage runs the one GC pass before the daemon socket is
21-// published. In production the recovery coordinator fences the catalogue
22-// snapshot and collection against every durable mutation and checkpoint
23-// publication. The fallback keeps standalone daemon tests usable when no
24-// coordinator is installed.
8+// CollectStartupGarbage runs the one coordinator-owned GC pass before the
9+// daemon socket is published.
2510 func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {
26- if d == nil {
27- return nil
28- }
29- if d.recovery != nil {
30- incarnations, err := d.recovery.CollectGarbage(ctx)
31- if err != nil {
32- return fmt.Errorf("collect startup snapshots: %w", err)
33- }
34- d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
35- return nil
36- }
37-
38- deps := d.maintenance
39- if deps.catalogue == nil || deps.repository == nil {
11+ if d == nil || d.recovery == nil {
4012 return nil
4113 }
42- records, err := deps.catalogue.Records()
14+ incarnations, err := d.recovery.CollectGarbage(ctx)
4315 if err != nil {
44- return fmt.Errorf("read startup catalogue: %w", err)
45- }
46- keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
47- for _, record := range records {
48- if record.Committed != nil {
49- keep[record.IncarnationID] = *record.Committed
50- continue
51- }
52- keep[record.IncarnationID] = domain.CheckpointRef{}
53- }
54- if err := deps.repository.CollectGarbage(ctx, keep); err != nil {
5516 return fmt.Errorf("collect startup snapshots: %w", err)
5617 }
57- d.log.Info("snapshot_garbage_collection_complete", "incarnations", len(keep))
18+ d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
5819 return nil
5920 }
@@ -3,57 +3,18 @@ package daemon
3 import (3 import (
4 "context"4 "context"
5 "fmt"5 "fmt"
6-
7- "github.com/bnema/vev/internal/domain"
8- "github.com/bnema/vev/internal/ports"
9 )6 )
10 7
11-type maintenanceDependencies struct {8+// CollectStartupGarbage runs the one coordinator-owned GC pass before the
12- catalogue ports.Catalogue9+// daemon socket is published.
13- repository ports.SnapshotRepository
14-}
15-
16-func newMaintenanceDependencies(catalogue ports.Catalogue, repository ports.SnapshotRepository) maintenanceDependencies {
17- return maintenanceDependencies{catalogue: catalogue, repository: repository}
18-}
19-
20-// CollectStartupGarbage runs the one GC pass before the daemon socket is
21-// published. In production the recovery coordinator fences the catalogue
22-// snapshot and collection against every durable mutation and checkpoint
23-// publication. The fallback keeps standalone daemon tests usable when no
24-// coordinator is installed.
25 func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {10 func (d *Daemon) CollectStartupGarbage(ctx context.Context) error {
26- if d == nil {11+ if d == nil || d.recovery == nil {
27- return nil
28- }
29- if d.recovery != nil {
30- incarnations, err := d.recovery.CollectGarbage(ctx)
31- if err != nil {
32- return fmt.Errorf("collect startup snapshots: %w", err)
33- }
34- d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
35- return nil
36- }
37-
38- deps := d.maintenance
39- if deps.catalogue == nil || deps.repository == nil {
40 return nil12 return nil
41 }13 }
42- records, err := deps.catalogue.Records()14+ incarnations, err := d.recovery.CollectGarbage(ctx)
43 if err != nil {15 if err != nil {
44- return fmt.Errorf("read startup catalogue: %w", err)
45- }
46- keep := make(map[domain.IncarnationID]domain.CheckpointRef, len(records))
47- for _, record := range records {
48- if record.Committed != nil {
49- keep[record.IncarnationID] = *record.Committed
50- continue
51- }
52- keep[record.IncarnationID] = domain.CheckpointRef{}
53- }
54- if err := deps.repository.CollectGarbage(ctx, keep); err != nil {
55 return fmt.Errorf("collect startup snapshots: %w", err)16 return fmt.Errorf("collect startup snapshots: %w", err)
56 }17 }
57- d.log.Info("snapshot_garbage_collection_complete", "incarnations", len(keep))18+ d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
58 return nil19 return nil
59 }20 }
M internal/usecase/daemon/maintenance_test.go +0 -1
@@ -38,5 +38,4 @@ func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {
3838
3939 require.Error(t, d.CollectStartupGarbage(context.Background()))
4040 repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)
41- require.NotNil(t, d.maintenance.catalogue)
4241 }
@@ -38,5 +38,4 @@ func TestStartupGarbageCollectionSkipsUnknownCatalogueState(t *testing.T) {
38 38
39 require.Error(t, d.CollectStartupGarbage(context.Background()))39 require.Error(t, d.CollectStartupGarbage(context.Background()))
40 repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)40 repository.AssertNotCalled(t, "CollectGarbage", mock.Anything, mock.Anything)
41- require.NotNil(t, d.maintenance.catalogue)
42 }41 }
M internal/usecase/daemon/snapshot_restore.go +2 -2
@@ -37,8 +37,8 @@ func (d *Daemon) closeRestoreDone() {
3737 }
3838
3939 // startSnapshotRestoration launches catalogue-driven restoration as a durable
40-// writer. Restoration may repair HEADs, so lifecycle ownership must outlive it
41-// exactly as it outlives the snapshot and maintenance workers. The completion
40+// writer. Restoration reconciles committed checkpoints, so lifecycle ownership
41+// must outlive it exactly as it outlives the snapshot and maintenance workers. The completion
4242 // channel is registered before the goroutine starts, so shutdown can never
4343 // begin waiting while this writer is still unregistered.
4444 func (d *Daemon) startSnapshotRestoration() {
@@ -37,8 +37,8 @@ func (d *Daemon) closeRestoreDone() {
37 }37 }
38 38
39 // startSnapshotRestoration launches catalogue-driven restoration as a durable39 // startSnapshotRestoration launches catalogue-driven restoration as a durable
40-// writer. Restoration may repair HEADs, so lifecycle ownership must outlive it40+// writer. Restoration reconciles committed checkpoints, so lifecycle ownership
41-// exactly as it outlives the snapshot and maintenance workers. The completion41+// must outlive it exactly as it outlives the snapshot and maintenance workers. The completion
42 // channel is registered before the goroutine starts, so shutdown can never42 // channel is registered before the goroutine starts, so shutdown can never
43 // begin waiting while this writer is still unregistered.43 // begin waiting while this writer is still unregistered.
44 func (d *Daemon) startSnapshotRestoration() {44 func (d *Daemon) startSnapshotRestoration() {
M internal/usecase/daemon/snapshot_restore_incremental.go +28 -51
@@ -54,8 +54,7 @@ func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.Catalogu
5454 for range workers {
5555 wg.Go(func() {
5656 for record := range jobs {
57- d.ensureCatalogueRegistryEntry(record)
58- done := d.recordRestoreDone(record.Name)
57+ done := d.ensureCatalogueRegistryEntry(record)
5958 err := d.restoreRecord(ctx, record)
6059 d.finishRecordRestore(record, err, done)
6160 }
@@ -68,14 +67,8 @@ func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.Catalogu
6867 wg.Wait()
6968 }
7069
71-func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) {
72- d.mu.Lock()
73- defer d.mu.Unlock()
74- if entry, ok := d.stopped[record.Name]; ok && entry.restoreDone != nil {
75- return
76- }
77- state, done := initialSessionState(record)
78- d.stopped[record.Name] = stoppedSession{
70+func stoppedSessionFromRecord(record domain.CatalogueRecord, state ports.SessionState, done chan struct{}) stoppedSession {
71+ return stoppedSession{
7972 name: record.Name,
8073 cwd: record.Cwd,
8174 createdAt: record.CreatedAt,
@@ -88,10 +81,15 @@ func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) {
8881 }
8982 }
9083
91-func (d *Daemon) recordRestoreDone(name string) chan struct{} {
84+func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) chan struct{} {
9285 d.mu.Lock()
9386 defer d.mu.Unlock()
94- return d.stopped[name].restoreDone
87+ if entry, ok := d.stopped[record.Name]; ok {
88+ return entry.restoreDone
89+ }
90+ state, done := initialSessionState(record)
91+ d.stopped[record.Name] = stoppedSessionFromRecord(record, state, done)
92+ return done
9593 }
9694
9795 // closeRuntimeRestoreDoneLocked closes a per-record restoration barrier.
@@ -114,54 +112,33 @@ func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state ports.S
114112 if !ok {
115113 return
116114 }
117- entry.record = record
118- entry.state = state
119- entry.cwd = record.Cwd
120- entry.createdAt = record.CreatedAt
121- entry.incarnation = record.IncarnationID
122- entry.lastUsedSeq = record.LastUsedSeq
123- entry.tabNames = append([]string(nil), record.TabNames...)
124- d.stopped[record.Name] = entry
115+ d.stopped[record.Name] = stoppedSessionFromRecord(record, state, entry.restoreDone)
125116 }
126117
127118 func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {
128119 retryable := errors.Is(restoreErr, errRetryableRestoreLoad)
129120 degraded := false
130- if restoreErr != nil && d.catalogue != nil {
131- current, ok, readErr := d.catalogue.Record(record.Name)
121+ if restoreErr != nil {
122+ reason := "checkpoint validation failed"
132123 switch {
133- case readErr != nil:
134- d.log.Error("reading catalogue before degradation failed", "session", record.Name, "err", readErr)
135- case ok && current.IncarnationID == record.IncarnationID:
124+ case retryable:
125+ reason = "checkpoint load failed"
126+ case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
127+ reason = "restore interrupted"
128+ }
129+ if d.recovery == nil {
130+ d.log.Error("marking session degraded failed", "session", record.Name, "err", errors.New("recovery coordinator unavailable"))
131+ } else if current, marked, err := d.recovery.MarkBroken(context.Background(), record.Name, record.IncarnationID, reason); err != nil {
132+ d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
133+ } else if marked {
136134 record = current
137- record.DegradedReason = "checkpoint validation failed"
138- switch {
139- case retryable:
140- record.DegradedReason = "checkpoint load failed"
141- case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
142- record.DegradedReason = "restore interrupted"
143- }
144- if err := d.catalogue.Replace(record.Name, record); err != nil {
145- d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
146- }
147135 degraded = true
148136 }
149137 }
150138
151139 d.mu.Lock()
152- entry, ok := d.stopped[record.Name]
153- if ok {
154- switch {
155- case degraded:
156- entry.record = record
157- entry.state = ports.SessionBroken
158- entry.cwd = record.Cwd
159- entry.createdAt = record.CreatedAt
160- entry.incarnation = record.IncarnationID
161- entry.lastUsedSeq = record.LastUsedSeq
162- entry.tabNames = append([]string(nil), record.TabNames...)
163- d.stopped[record.Name] = entry
164- }
140+ if entry, ok := d.stopped[record.Name]; ok && degraded {
141+ d.stopped[record.Name] = stoppedSessionFromRecord(record, ports.SessionBroken, entry.restoreDone)
165142 }
166143 closeRuntimeRestoreDoneLocked(done)
167144 d.mu.Unlock()
@@ -224,10 +201,10 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
224201 d.logRejectedRestoreCandidate(record, selected, false, "generation-mismatch", nil)
225202 return errors.New("snapshot: committed checkpoint did not validate")
226203 }
227- if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
228- d.log.Warn("snapshot_head_repair_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "repair-failed")
204+ if err := d.snapshotRepository.ReconcileCheckpoint(ctx, record.IncarnationID, selected); err != nil {
205+ d.log.Warn("snapshot_checkpoint_reconciliation_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "reconcile-failed")
229206 } else {
230- d.log.Info("snapshot_head_repair_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
207+ d.log.Info("snapshot_checkpoint_reconciliation_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
231208 }
232209
233210 if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil {
@@ -54,8 +54,7 @@ func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.Catalogu
54 for range workers {54 for range workers {
55 wg.Go(func() {55 wg.Go(func() {
56 for record := range jobs {56 for record := range jobs {
57- d.ensureCatalogueRegistryEntry(record)57+ done := d.ensureCatalogueRegistryEntry(record)
58- done := d.recordRestoreDone(record.Name)
59 err := d.restoreRecord(ctx, record)58 err := d.restoreRecord(ctx, record)
60 d.finishRecordRestore(record, err, done)59 d.finishRecordRestore(record, err, done)
61 }60 }
@@ -68,14 +67,8 @@ func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.Catalogu
68 wg.Wait()67 wg.Wait()
69 }68 }
70 69
71-func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) {70+func stoppedSessionFromRecord(record domain.CatalogueRecord, state ports.SessionState, done chan struct{}) stoppedSession {
72- d.mu.Lock()71+ return stoppedSession{
73- defer d.mu.Unlock()
74- if entry, ok := d.stopped[record.Name]; ok && entry.restoreDone != nil {
75- return
76- }
77- state, done := initialSessionState(record)
78- d.stopped[record.Name] = stoppedSession{
79 name: record.Name,72 name: record.Name,
80 cwd: record.Cwd,73 cwd: record.Cwd,
81 createdAt: record.CreatedAt,74 createdAt: record.CreatedAt,
@@ -88,10 +81,15 @@ func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) {
88 }81 }
89 }82 }
90 83
91-func (d *Daemon) recordRestoreDone(name string) chan struct{} {84+func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) chan struct{} {
92 d.mu.Lock()85 d.mu.Lock()
93 defer d.mu.Unlock()86 defer d.mu.Unlock()
94- return d.stopped[name].restoreDone87+ if entry, ok := d.stopped[record.Name]; ok {
88+ return entry.restoreDone
89+ }
90+ state, done := initialSessionState(record)
91+ d.stopped[record.Name] = stoppedSessionFromRecord(record, state, done)
92+ return done
95 }93 }
96 94
97 // closeRuntimeRestoreDoneLocked closes a per-record restoration barrier.95 // closeRuntimeRestoreDoneLocked closes a per-record restoration barrier.
@@ -114,54 +112,33 @@ func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state ports.S
114 if !ok {112 if !ok {
115 return113 return
116 }114 }
117- entry.record = record115+ d.stopped[record.Name] = stoppedSessionFromRecord(record, state, entry.restoreDone)
118- entry.state = state
119- entry.cwd = record.Cwd
120- entry.createdAt = record.CreatedAt
121- entry.incarnation = record.IncarnationID
122- entry.lastUsedSeq = record.LastUsedSeq
123- entry.tabNames = append([]string(nil), record.TabNames...)
124- d.stopped[record.Name] = entry
125 }116 }
126 117
127 func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {118 func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {
128 retryable := errors.Is(restoreErr, errRetryableRestoreLoad)119 retryable := errors.Is(restoreErr, errRetryableRestoreLoad)
129 degraded := false120 degraded := false
130- if restoreErr != nil && d.catalogue != nil {121+ if restoreErr != nil {
131- current, ok, readErr := d.catalogue.Record(record.Name)122+ reason := "checkpoint validation failed"
132 switch {123 switch {
133- case readErr != nil:124+ case retryable:
134- d.log.Error("reading catalogue before degradation failed", "session", record.Name, "err", readErr)125+ reason = "checkpoint load failed"
135- case ok && current.IncarnationID == record.IncarnationID:126+ case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
127+ reason = "restore interrupted"
128+ }
129+ if d.recovery == nil {
130+ d.log.Error("marking session degraded failed", "session", record.Name, "err", errors.New("recovery coordinator unavailable"))
131+ } else if current, marked, err := d.recovery.MarkBroken(context.Background(), record.Name, record.IncarnationID, reason); err != nil {
132+ d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
133+ } else if marked {
136 record = current134 record = current
137- record.DegradedReason = "checkpoint validation failed"
138- switch {
139- case retryable:
140- record.DegradedReason = "checkpoint load failed"
141- case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
142- record.DegradedReason = "restore interrupted"
143- }
144- if err := d.catalogue.Replace(record.Name, record); err != nil {
145- d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
146- }
147 degraded = true135 degraded = true
148 }136 }
149 }137 }
150 138
151 d.mu.Lock()139 d.mu.Lock()
152- entry, ok := d.stopped[record.Name]140+ if entry, ok := d.stopped[record.Name]; ok && degraded {
153- if ok {141+ d.stopped[record.Name] = stoppedSessionFromRecord(record, ports.SessionBroken, entry.restoreDone)
154- switch {
155- case degraded:
156- entry.record = record
157- entry.state = ports.SessionBroken
158- entry.cwd = record.Cwd
159- entry.createdAt = record.CreatedAt
160- entry.incarnation = record.IncarnationID
161- entry.lastUsedSeq = record.LastUsedSeq
162- entry.tabNames = append([]string(nil), record.TabNames...)
163- d.stopped[record.Name] = entry
164- }
165 }142 }
166 closeRuntimeRestoreDoneLocked(done)143 closeRuntimeRestoreDoneLocked(done)
167 d.mu.Unlock()144 d.mu.Unlock()
@@ -224,10 +201,10 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
224 d.logRejectedRestoreCandidate(record, selected, false, "generation-mismatch", nil)201 d.logRejectedRestoreCandidate(record, selected, false, "generation-mismatch", nil)
225 return errors.New("snapshot: committed checkpoint did not validate")202 return errors.New("snapshot: committed checkpoint did not validate")
226 }203 }
227- if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {204+ if err := d.snapshotRepository.ReconcileCheckpoint(ctx, record.IncarnationID, selected); err != nil {
228- d.log.Warn("snapshot_head_repair_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "repair-failed")205+ d.log.Warn("snapshot_checkpoint_reconciliation_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "reconcile-failed")
229 } else {206 } else {
230- d.log.Info("snapshot_head_repair_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)207+ d.log.Info("snapshot_checkpoint_reconciliation_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
231 }208 }
232 209
233 if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil {210 if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil {
M internal/usecase/daemon/snapshot_test.go +1 -1
@@ -106,7 +106,7 @@ func (noOpSnapshotRepository) Publish(context.Context, ports.SnapshotPublication
106106 func (noOpSnapshotRepository) LoadCheckpoint(context.Context, domain.IncarnationID, string, ports.CheckpointRef) (ports.SnapshotGeneration, error) {
107107 return ports.SnapshotGeneration{}, errors.New("unused")
108108 }
109-func (noOpSnapshotRepository) RepairHEAD(context.Context, domain.IncarnationID, ports.CheckpointRef) error {
109+func (noOpSnapshotRepository) ReconcileCheckpoint(context.Context, domain.IncarnationID, ports.CheckpointRef) error {
110110 return nil
111111 }
112112 func (noOpSnapshotRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
@@ -106,7 +106,7 @@ func (noOpSnapshotRepository) Publish(context.Context, ports.SnapshotPublication
106 func (noOpSnapshotRepository) LoadCheckpoint(context.Context, domain.IncarnationID, string, ports.CheckpointRef) (ports.SnapshotGeneration, error) {106 func (noOpSnapshotRepository) LoadCheckpoint(context.Context, domain.IncarnationID, string, ports.CheckpointRef) (ports.SnapshotGeneration, error) {
107 return ports.SnapshotGeneration{}, errors.New("unused")107 return ports.SnapshotGeneration{}, errors.New("unused")
108 }108 }
109-func (noOpSnapshotRepository) RepairHEAD(context.Context, domain.IncarnationID, ports.CheckpointRef) error {109+func (noOpSnapshotRepository) ReconcileCheckpoint(context.Context, domain.IncarnationID, ports.CheckpointRef) error {
110 return nil110 return nil
111 }111 }
112 func (noOpSnapshotRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {112 func (noOpSnapshotRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
A internal/usecase/recovery/broken.go +47 -0
@@ -0,0 +1,47 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+
8+ "github.com/bnema/vev/internal/domain"
9+)
10+
11+// MarkBroken records a restoration failure only when name still identifies the
12+// incarnation that failed. It is fenced with Discard so a late restore cannot
13+// degrade the replacement incarnation.
14+func (c *Coordinator) MarkBroken(ctx context.Context, name string, expected domain.IncarnationID, reason string) (domain.CatalogueRecord, bool, error) {
15+ if c == nil || c.catalogue == nil || c.locks == nil {
16+ return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete broken-session dependencies")
17+ }
18+ if err := ctx.Err(); err != nil {
19+ return domain.CatalogueRecord{}, false, err
20+ }
21+ if reason == "" {
22+ return domain.CatalogueRecord{}, false, errors.New("recovery: empty broken-session reason")
23+ }
24+ c.mutationMu.Lock()
25+ defer c.mutationMu.Unlock()
26+ unlock := c.locks.Lock([]string{name})
27+ defer unlock()
28+
29+ record, ok, err := c.catalogue.Record(name)
30+ if err != nil {
31+ return domain.CatalogueRecord{}, false, err
32+ }
33+ if !ok || record.IncarnationID != expected {
34+ return domain.CatalogueRecord{}, false, nil
35+ }
36+ if record.DegradedReason == reason {
37+ return record, true, nil
38+ }
39+ record.DegradedReason = reason
40+ if err := record.Validate(); err != nil {
41+ return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid broken-session transition: %w", err)
42+ }
43+ if err := c.catalogue.Replace(name, record); err != nil {
44+ return domain.CatalogueRecord{}, false, err
45+ }
46+ return record, true, nil
47+}
@@ -0,0 +1,47 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+
8+ "github.com/bnema/vev/internal/domain"
9+)
10+
11+// MarkBroken records a restoration failure only when name still identifies the
12+// incarnation that failed. It is fenced with Discard so a late restore cannot
13+// degrade the replacement incarnation.
14+func (c *Coordinator) MarkBroken(ctx context.Context, name string, expected domain.IncarnationID, reason string) (domain.CatalogueRecord, bool, error) {
15+ if c == nil || c.catalogue == nil || c.locks == nil {
16+ return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete broken-session dependencies")
17+ }
18+ if err := ctx.Err(); err != nil {
19+ return domain.CatalogueRecord{}, false, err
20+ }
21+ if reason == "" {
22+ return domain.CatalogueRecord{}, false, errors.New("recovery: empty broken-session reason")
23+ }
24+ c.mutationMu.Lock()
25+ defer c.mutationMu.Unlock()
26+ unlock := c.locks.Lock([]string{name})
27+ defer unlock()
28+
29+ record, ok, err := c.catalogue.Record(name)
30+ if err != nil {
31+ return domain.CatalogueRecord{}, false, err
32+ }
33+ if !ok || record.IncarnationID != expected {
34+ return domain.CatalogueRecord{}, false, nil
35+ }
36+ if record.DegradedReason == reason {
37+ return record, true, nil
38+ }
39+ record.DegradedReason = reason
40+ if err := record.Validate(); err != nil {
41+ return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid broken-session transition: %w", err)
42+ }
43+ if err := c.catalogue.Replace(name, record); err != nil {
44+ return domain.CatalogueRecord{}, false, err
45+ }
46+ return record, true, nil
47+}
M internal/usecase/recovery/checkpoint_test.go +1 -1
@@ -85,7 +85,7 @@ func (r *checkpointRepository) LoadCheckpoint(_ context.Context, _ domain.Incarn
8585 return generation, nil
8686 }
8787
88-func (r *checkpointRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
88+func (r *checkpointRepository) ReconcileCheckpoint(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
8989 r.repairs = append(r.repairs, ref)
9090 return r.repairErr
9191 }
@@ -85,7 +85,7 @@ func (r *checkpointRepository) LoadCheckpoint(_ context.Context, _ domain.Incarn
85 return generation, nil85 return generation, nil
86 }86 }
87 87
88-func (r *checkpointRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {88+func (r *checkpointRepository) ReconcileCheckpoint(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
89 r.repairs = append(r.repairs, ref)89 r.repairs = append(r.repairs, ref)
90 return r.repairErr90 return r.repairErr
91 }91 }
M internal/usecase/recovery/transactions_test.go +43 -0
@@ -108,6 +108,8 @@ type transactionRepository struct {
108108 mu sync.Mutex
109109 deleteErrOnce error
110110 deletedIDs []domain.IncarnationID
111+ deleteEntered chan struct{}
112+ releaseDelete chan struct{}
111113 }
112114
113115 type garbageCollectionFenceRepository struct {
@@ -196,6 +198,10 @@ func (r *garbageCollectionFenceRepository) eventLog() ([]string, bool) {
196198 }
197199
198200 func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
201+ if r.deleteEntered != nil {
202+ close(r.deleteEntered)
203+ <-r.releaseDelete
204+ }
199205 r.mu.Lock()
200206 defer r.mu.Unlock()
201207 if r.deleteErrOnce != nil {
@@ -295,6 +301,43 @@ func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(
295301 })
296302 }
297303
304+func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
305+ t.Parallel()
306+ old := degradedTransactionRecord()
307+ catalogue := newTransactionCatalogue(old)
308+ repository := &transactionRepository{deleteEntered: make(chan struct{}), releaseDelete: make(chan struct{})}
309+ coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
310+
311+ discarded := make(chan error, 1)
312+ go func() { discarded <- coordinator.Discard(t.Context(), old.Name) }()
313+ <-repository.deleteEntered // Replace committed the fresh incarnation while Discard retains the mutation fence.
314+
315+ marked := make(chan struct {
316+ marked bool
317+ err error
318+ }, 1)
319+ go func() {
320+ _, markedSession, err := coordinator.MarkBroken(t.Context(), old.Name, old.IncarnationID, "late restore failure")
321+ marked <- struct {
322+ marked bool
323+ err error
324+ }{markedSession, err}
325+ }()
326+
327+ close(repository.releaseDelete)
328+ require.NoError(t, <-discarded)
329+ result := <-marked
330+ require.NoError(t, result.err)
331+ require.False(t, result.marked, "a late restore must not degrade the replacement incarnation")
332+
333+ got, ok, err := catalogue.Record(old.Name)
334+ require.NoError(t, err)
335+ require.True(t, ok)
336+ require.NotEqual(t, old.IncarnationID, got.IncarnationID)
337+ require.Nil(t, got.Committed)
338+ require.Empty(t, got.DegradedReason)
339+}
340+
298341 func TestDiscardIsRetryIdempotent(t *testing.T) {
299342 t.Parallel()
300343 failure := errors.New("injected discard failure")
@@ -108,6 +108,8 @@ type transactionRepository struct {
108 mu sync.Mutex108 mu sync.Mutex
109 deleteErrOnce error109 deleteErrOnce error
110 deletedIDs []domain.IncarnationID110 deletedIDs []domain.IncarnationID
111+ deleteEntered chan struct{}
112+ releaseDelete chan struct{}
111 }113 }
112 114
113 type garbageCollectionFenceRepository struct {115 type garbageCollectionFenceRepository struct {
@@ -196,6 +198,10 @@ func (r *garbageCollectionFenceRepository) eventLog() ([]string, bool) {
196 }198 }
197 199
198 func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {200 func (r *transactionRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
201+ if r.deleteEntered != nil {
202+ close(r.deleteEntered)
203+ <-r.releaseDelete
204+ }
199 r.mu.Lock()205 r.mu.Lock()
200 defer r.mu.Unlock()206 defer r.mu.Unlock()
201 if r.deleteErrOnce != nil {207 if r.deleteErrOnce != nil {
@@ -295,6 +301,43 @@ func TestStartupGarbageCollectionFencesNewIncarnationsAndCheckpointPublications(
295 })301 })
296 }302 }
297 303
304+func TestMarkBrokenCannotOverwriteDiscardReplacement(t *testing.T) {
305+ t.Parallel()
306+ old := degradedTransactionRecord()
307+ catalogue := newTransactionCatalogue(old)
308+ repository := &transactionRepository{deleteEntered: make(chan struct{}), releaseDelete: make(chan struct{})}
309+ coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
310+
311+ discarded := make(chan error, 1)
312+ go func() { discarded <- coordinator.Discard(t.Context(), old.Name) }()
313+ <-repository.deleteEntered // Replace committed the fresh incarnation while Discard retains the mutation fence.
314+
315+ marked := make(chan struct {
316+ marked bool
317+ err error
318+ }, 1)
319+ go func() {
320+ _, markedSession, err := coordinator.MarkBroken(t.Context(), old.Name, old.IncarnationID, "late restore failure")
321+ marked <- struct {
322+ marked bool
323+ err error
324+ }{markedSession, err}
325+ }()
326+
327+ close(repository.releaseDelete)
328+ require.NoError(t, <-discarded)
329+ result := <-marked
330+ require.NoError(t, result.err)
331+ require.False(t, result.marked, "a late restore must not degrade the replacement incarnation")
332+
333+ got, ok, err := catalogue.Record(old.Name)
334+ require.NoError(t, err)
335+ require.True(t, ok)
336+ require.NotEqual(t, old.IncarnationID, got.IncarnationID)
337+ require.Nil(t, got.Committed)
338+ require.Empty(t, got.DegradedReason)
339+}
340+
298 func TestDiscardIsRetryIdempotent(t *testing.T) {341 func TestDiscardIsRetryIdempotent(t *testing.T) {
299 t.Parallel()342 t.Parallel()
300 failure := errors.New("injected discard failure")343 failure := errors.New("injected discard failure")

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