detached 65de473

feat(recovery): reset incompatible session snapshots

commit
65de4733494831174b1e62365a470a03764df26b
author
brice · · 2026-07-27 13:19 UTC
parent
e362c80
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +265 -1
Split
UnifiedSplit
M internal/usecase/daemon/coderabbit_regression_test.go +70 -0
@@ -4,6 +4,7 @@ import (
44 "bytes"
55 "context"
66 "errors"
7+ "log/slog"
78 "sync"
89 "syscall"
910 "testing"
@@ -13,6 +14,7 @@ import (
1314 "github.com/bnema/vev/internal/domain"
1415 "github.com/bnema/vev/internal/ports"
1516 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
17+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
1618 )
1719
1820 func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
@@ -96,6 +98,74 @@ func TestRestoreFailureCannotOverwriteDiscardReplacement(t *testing.T) {
9698 require.Equal(t, ports.SessionStopped, entry.state)
9799 }
98100
101+func TestStaleIncompatibleRestoreCannotOverwriteNewerAuthority(t *testing.T) {
102+ tests := []struct {
103+ name string
104+ replacement func(domain.CatalogueRecord) domain.CatalogueRecord
105+ state ports.SessionState
106+ }{
107+ {
108+ name: "newer checkpoint",
109+ replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
110+ ref := *record.Committed
111+ ref.Generation++
112+ ref.ManifestDigest[0]++
113+ record.Committed = &ref
114+ return record
115+ },
116+ state: ports.SessionStopped,
117+ },
118+ {
119+ name: "newer incarnation",
120+ replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
121+ record.IncarnationID = domain.IncarnationID{9}
122+ return record
123+ },
124+ state: ports.SessionStopped,
125+ },
126+ {
127+ name: "degraded authority",
128+ replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
129+ record.DegradedReason = "newer restore decision"
130+ return record
131+ },
132+ state: ports.SessionBroken,
133+ },
134+ }
135+ for _, tt := range tests {
136+ t.Run(tt.name, func(t *testing.T) {
137+ record := durableRecoveryRecord(0)
138+ replacement := tt.replacement(record)
139+ repository := &durableRecoveryRepository{
140+ errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
141+ loads: make(map[string]int),
142+ repairs: make(map[string]domain.CheckpointRef),
143+ }
144+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
145+ var logs bytes.Buffer
146+ d.log = slog.New(slog.NewJSONHandler(&logs, nil))
147+ repository.onLoad = func() {
148+ require.NoError(t, catalogue.Replace(record.Name, replacement))
149+ d.setStoppedRecovery(replacement, tt.state)
150+ }
151+
152+ d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
153+
154+ persisted, ok, err := catalogue.Record(record.Name)
155+ require.NoError(t, err)
156+ require.True(t, ok)
157+ require.Equal(t, replacement, persisted)
158+ d.mu.Lock()
159+ entry := d.stopped[record.Name]
160+ d.mu.Unlock()
161+ require.Equal(t, tt.state, entry.state)
162+ require.Equal(t, replacement, entry.record)
163+ require.Empty(t, repository.deleted)
164+ daemonRequireEvent(t, daemonJSONLogs(t, logs.Bytes()), "snapshot_incompatible_reset_superseded")
165+ })
166+ }
167+}
168+
99169 func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {
100170 d := newTestDaemon(t, nil, stubClock{})
101171 done := make(chan struct{})
@@ -4,6 +4,7 @@ import (
4 "bytes"4 "bytes"
5 "context"5 "context"
6 "errors"6 "errors"
7+ "log/slog"
7 "sync"8 "sync"
8 "syscall"9 "syscall"
9 "testing"10 "testing"
@@ -13,6 +14,7 @@ import (
13 "github.com/bnema/vev/internal/domain"14 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"15 "github.com/bnema/vev/internal/ports"
15 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"16 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
17+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
16 )18 )
17 19
18 func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {20 func TestRestoreTransientLoadFailureBecomesDiscardable(t *testing.T) {
@@ -96,6 +98,74 @@ func TestRestoreFailureCannotOverwriteDiscardReplacement(t *testing.T) {
96 require.Equal(t, ports.SessionStopped, entry.state)98 require.Equal(t, ports.SessionStopped, entry.state)
97 }99 }
98 100
101+func TestStaleIncompatibleRestoreCannotOverwriteNewerAuthority(t *testing.T) {
102+ tests := []struct {
103+ name string
104+ replacement func(domain.CatalogueRecord) domain.CatalogueRecord
105+ state ports.SessionState
106+ }{
107+ {
108+ name: "newer checkpoint",
109+ replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
110+ ref := *record.Committed
111+ ref.Generation++
112+ ref.ManifestDigest[0]++
113+ record.Committed = &ref
114+ return record
115+ },
116+ state: ports.SessionStopped,
117+ },
118+ {
119+ name: "newer incarnation",
120+ replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
121+ record.IncarnationID = domain.IncarnationID{9}
122+ return record
123+ },
124+ state: ports.SessionStopped,
125+ },
126+ {
127+ name: "degraded authority",
128+ replacement: func(record domain.CatalogueRecord) domain.CatalogueRecord {
129+ record.DegradedReason = "newer restore decision"
130+ return record
131+ },
132+ state: ports.SessionBroken,
133+ },
134+ }
135+ for _, tt := range tests {
136+ t.Run(tt.name, func(t *testing.T) {
137+ record := durableRecoveryRecord(0)
138+ replacement := tt.replacement(record)
139+ repository := &durableRecoveryRepository{
140+ errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
141+ loads: make(map[string]int),
142+ repairs: make(map[string]domain.CheckpointRef),
143+ }
144+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
145+ var logs bytes.Buffer
146+ d.log = slog.New(slog.NewJSONHandler(&logs, nil))
147+ repository.onLoad = func() {
148+ require.NoError(t, catalogue.Replace(record.Name, replacement))
149+ d.setStoppedRecovery(replacement, tt.state)
150+ }
151+
152+ d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
153+
154+ persisted, ok, err := catalogue.Record(record.Name)
155+ require.NoError(t, err)
156+ require.True(t, ok)
157+ require.Equal(t, replacement, persisted)
158+ d.mu.Lock()
159+ entry := d.stopped[record.Name]
160+ d.mu.Unlock()
161+ require.Equal(t, tt.state, entry.state)
162+ require.Equal(t, replacement, entry.record)
163+ require.Empty(t, repository.deleted)
164+ daemonRequireEvent(t, daemonJSONLogs(t, logs.Bytes()), "snapshot_incompatible_reset_superseded")
165+ })
166+ }
167+}
168+
99 func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {169 func TestFinishRecordRestoreAlwaysClosesProvidedBarrier(t *testing.T) {
100 d := newTestDaemon(t, nil, stubClock{})170 d := newTestDaemon(t, nil, stubClock{})
101 done := make(chan struct{})171 done := make(chan struct{})
M internal/usecase/daemon/durable_recovery_test.go +153 -1
@@ -29,6 +29,7 @@ type durableRecoveryCatalogue struct {
2929 mu sync.Mutex
3030 records map[string]domain.CatalogueRecord
3131 metadataUpdates []domain.CatalogueMetadataUpdate
32+ replaceErr error
3233 }
3334
3435 type recordErrorCatalogue struct {
@@ -75,6 +76,9 @@ func mustDurableRecords(t *testing.T, c *durableRecoveryCatalogue) []domain.Cata
7576 func (c *durableRecoveryCatalogue) Replace(name string, record domain.CatalogueRecord) error {
7677 c.mu.Lock()
7778 defer c.mu.Unlock()
79+ if c.replaceErr != nil {
80+ return c.replaceErr
81+ }
7882 if name != record.Name {
7983 return errors.New("catalogue key mismatch")
8084 }
@@ -125,7 +129,10 @@ type durableRecoveryRepository struct {
125129 errors map[string]error
126130 loads map[string]int
127131 repairs map[string]domain.CheckpointRef
132+ deleted []domain.IncarnationID
133+ onLoad func()
128134 repairErr error
135+ deleteErr error
129136 repairCalls int
130137 }
131138
@@ -136,6 +143,9 @@ func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.
136143 r.loads = make(map[string]int)
137144 }
138145 r.loads[name]++
146+ if r.onLoad != nil {
147+ r.onLoad()
148+ }
139149 if err := r.errors[name]; err != nil {
140150 return ports.SnapshotGeneration{}, err
141151 }
@@ -160,6 +170,13 @@ func (r *durableRecoveryRepository) ReconcileCheckpoint(_ context.Context, id do
160170 return nil
161171 }
162172
173+func (r *durableRecoveryRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
174+ r.mu.Lock()
175+ defer r.mu.Unlock()
176+ r.deleted = append(r.deleted, id)
177+ return r.deleteErr
178+}
179+
163180 func durableRecoveryRecord(index int) domain.CatalogueRecord {
164181 id := domain.IncarnationID{0: byte(index>>8 + 1), 1: byte(index + 1)}
165182 ref := domain.CheckpointRef{Generation: uint64(index + 1), ManifestDigest: [32]byte{byte(index%255 + 1)}}
@@ -194,7 +211,7 @@ func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.Snapshot
194211 func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {
195212 t.Helper()
196213 catalogue := newDurableRecoveryCatalogue(records)
197- coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)
214+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{0xa5}, 16)))
198215 d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository), WithRecoveryCoordinator(coordinator))
199216 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
200217 t.Cleanup(d.serveCancel)
@@ -289,6 +306,141 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
289306
290307 }
291308
309+func TestCatalogueRestoreResetsOnlyIncompatibleCheckpoint(t *testing.T) {
310+ incompatible := durableRecoveryRecord(0)
311+ healthy := durableRecoveryRecord(1)
312+ repository := &durableRecoveryRepository{
313+ generations: map[string]ports.SnapshotGeneration{healthy.Name: validGeneration(t, healthy)},
314+ errors: map[string]error{incompatible.Name: fmt.Errorf("load envelope: %w", snapcodec.ErrBadVersion)},
315+ loads: make(map[string]int),
316+ repairs: make(map[string]domain.CheckpointRef),
317+ }
318+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{incompatible, healthy}, repository)
319+
320+ d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
321+
322+ fresh, ok, err := catalogue.Record(incompatible.Name)
323+ require.NoError(t, err)
324+ require.True(t, ok)
325+ require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{0xa5}, 16)), fresh.IncarnationID)
326+ require.Nil(t, fresh.Committed)
327+ require.Empty(t, fresh.DegradedReason)
328+ require.Empty(t, fresh.TabNames)
329+ require.Equal(t, []domain.IncarnationID{incompatible.IncarnationID}, repository.deleted)
330+ require.Equal(t, ports.SessionStopped, d.stopped[incompatible.Name].state)
331+ require.Equal(t, fresh, d.stopped[incompatible.Name].record)
332+
333+ restoredSibling, ok, err := catalogue.Record(healthy.Name)
334+ require.NoError(t, err)
335+ require.True(t, ok)
336+ require.Equal(t, healthy, restoredSibling)
337+ require.Equal(t, ports.SessionStopped, d.stopped[healthy.Name].state)
338+ require.Equal(t, 1, repository.loads[healthy.Name])
339+}
340+
341+func TestIncompatibleCheckpointDeleteFailurePublishesFreshStoppedAuthority(t *testing.T) {
342+ record := durableRecoveryRecord(0)
343+ deleteErr := errors.New("delete old incarnation")
344+ repository := &durableRecoveryRepository{
345+ errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
346+ loads: make(map[string]int),
347+ repairs: make(map[string]domain.CheckpointRef),
348+ deleteErr: deleteErr,
349+ }
350+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
351+ d.mu.Lock()
352+ restoreDone := d.stopped[record.Name].restoreDone
353+ d.mu.Unlock()
354+
355+ restoreErr := d.restoreRecord(context.Background(), record)
356+ require.NoError(t, restoreErr, "post-commit cleanup failure must not fail the authority transition")
357+
358+ fresh, ok, err := catalogue.Record(record.Name)
359+ require.NoError(t, err)
360+ require.True(t, ok)
361+ require.NotEqual(t, record.IncarnationID, fresh.IncarnationID)
362+ require.Nil(t, fresh.Committed)
363+ require.Empty(t, fresh.DegradedReason)
364+ d.mu.Lock()
365+ entry := d.stopped[record.Name]
366+ d.mu.Unlock()
367+ require.Equal(t, ports.SessionStopped, entry.state)
368+ require.Equal(t, fresh, entry.record)
369+ require.Equal(t, restoreDone, entry.restoreDone)
370+ select {
371+ case <-entry.restoreDone:
372+ t.Fatal("direct restore closed its coordinator-owned completion barrier")
373+ default:
374+ }
375+
376+ d.finishRecordRestore(record, restoreErr, restoreDone)
377+ d.mu.Lock()
378+ entry = d.stopped[record.Name]
379+ d.mu.Unlock()
380+ require.Equal(t, ports.SessionStopped, entry.state)
381+ require.Equal(t, fresh, entry.record)
382+ select {
383+ case <-entry.restoreDone:
384+ default:
385+ t.Fatal("restore completion was not signaled")
386+ }
387+ require.Equal(t, []domain.IncarnationID{record.IncarnationID}, repository.deleted)
388+}
389+
390+func TestIncompatibleCheckpointPreCommitFailureReturnsError(t *testing.T) {
391+ record := durableRecoveryRecord(0)
392+ cause := errors.New("catalogue replace failed")
393+ repository := &durableRecoveryRepository{
394+ errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
395+ loads: make(map[string]int),
396+ repairs: make(map[string]domain.CheckpointRef),
397+ }
398+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
399+ catalogue.replaceErr = cause
400+
401+ err := d.restoreRecord(context.Background(), record)
402+
403+ require.ErrorIs(t, err, cause)
404+ persisted, ok, recordErr := catalogue.Record(record.Name)
405+ require.NoError(t, recordErr)
406+ require.True(t, ok)
407+ require.Equal(t, record, persisted)
408+ require.Empty(t, repository.deleted)
409+}
410+
411+func TestRestoreLoadFailuresRetainCheckpointWithoutDeletion(t *testing.T) {
412+ tests := []struct {
413+ name string
414+ err error
415+ }{
416+ {name: "corruption", err: errors.New("corrupt checkpoint")},
417+ {name: "generic", err: errors.New("load failed")},
418+ {name: "io", err: &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.EACCES}},
419+ }
420+ for _, tt := range tests {
421+ t.Run(tt.name, func(t *testing.T) {
422+ record := durableRecoveryRecord(0)
423+ repository := &durableRecoveryRepository{
424+ errors: map[string]error{record.Name: tt.err},
425+ loads: make(map[string]int),
426+ repairs: make(map[string]domain.CheckpointRef),
427+ }
428+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
429+
430+ d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
431+
432+ persisted, ok, err := catalogue.Record(record.Name)
433+ require.NoError(t, err)
434+ require.True(t, ok)
435+ require.Equal(t, record.IncarnationID, persisted.IncarnationID)
436+ require.Equal(t, record.Committed, persisted.Committed)
437+ require.Equal(t, "checkpoint validation failed", persisted.DegradedReason)
438+ require.Equal(t, ports.SessionBroken, d.stopped[record.Name].state)
439+ require.Empty(t, repository.deleted)
440+ })
441+ }
442+}
443+
292444 func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
293445 const count = 4200
294446 records := make([]domain.CatalogueRecord, count)
@@ -29,6 +29,7 @@ type durableRecoveryCatalogue struct {
29 mu sync.Mutex29 mu sync.Mutex
30 records map[string]domain.CatalogueRecord30 records map[string]domain.CatalogueRecord
31 metadataUpdates []domain.CatalogueMetadataUpdate31 metadataUpdates []domain.CatalogueMetadataUpdate
32+ replaceErr error
32 }33 }
33 34
34 type recordErrorCatalogue struct {35 type recordErrorCatalogue struct {
@@ -75,6 +76,9 @@ func mustDurableRecords(t *testing.T, c *durableRecoveryCatalogue) []domain.Cata
75 func (c *durableRecoveryCatalogue) Replace(name string, record domain.CatalogueRecord) error {76 func (c *durableRecoveryCatalogue) Replace(name string, record domain.CatalogueRecord) error {
76 c.mu.Lock()77 c.mu.Lock()
77 defer c.mu.Unlock()78 defer c.mu.Unlock()
79+ if c.replaceErr != nil {
80+ return c.replaceErr
81+ }
78 if name != record.Name {82 if name != record.Name {
79 return errors.New("catalogue key mismatch")83 return errors.New("catalogue key mismatch")
80 }84 }
@@ -125,7 +129,10 @@ type durableRecoveryRepository struct {
125 errors map[string]error129 errors map[string]error
126 loads map[string]int130 loads map[string]int
127 repairs map[string]domain.CheckpointRef131 repairs map[string]domain.CheckpointRef
132+ deleted []domain.IncarnationID
133+ onLoad func()
128 repairErr error134 repairErr error
135+ deleteErr error
129 repairCalls int136 repairCalls int
130 }137 }
131 138
@@ -136,6 +143,9 @@ func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.
136 r.loads = make(map[string]int)143 r.loads = make(map[string]int)
137 }144 }
138 r.loads[name]++145 r.loads[name]++
146+ if r.onLoad != nil {
147+ r.onLoad()
148+ }
139 if err := r.errors[name]; err != nil {149 if err := r.errors[name]; err != nil {
140 return ports.SnapshotGeneration{}, err150 return ports.SnapshotGeneration{}, err
141 }151 }
@@ -160,6 +170,13 @@ func (r *durableRecoveryRepository) ReconcileCheckpoint(_ context.Context, id do
160 return nil170 return nil
161 }171 }
162 172
173+func (r *durableRecoveryRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
174+ r.mu.Lock()
175+ defer r.mu.Unlock()
176+ r.deleted = append(r.deleted, id)
177+ return r.deleteErr
178+}
179+
163 func durableRecoveryRecord(index int) domain.CatalogueRecord {180 func durableRecoveryRecord(index int) domain.CatalogueRecord {
164 id := domain.IncarnationID{0: byte(index>>8 + 1), 1: byte(index + 1)}181 id := domain.IncarnationID{0: byte(index>>8 + 1), 1: byte(index + 1)}
165 ref := domain.CheckpointRef{Generation: uint64(index + 1), ManifestDigest: [32]byte{byte(index%255 + 1)}}182 ref := domain.CheckpointRef{Generation: uint64(index + 1), ManifestDigest: [32]byte{byte(index%255 + 1)}}
@@ -194,7 +211,7 @@ func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.Snapshot
194 func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {211 func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {
195 t.Helper()212 t.Helper()
196 catalogue := newDurableRecoveryCatalogue(records)213 catalogue := newDurableRecoveryCatalogue(records)
197- coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil)214+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{0xa5}, 16)))
198 d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository), WithRecoveryCoordinator(coordinator))215 d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository), WithRecoveryCoordinator(coordinator))
199 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())216 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
200 t.Cleanup(d.serveCancel)217 t.Cleanup(d.serveCancel)
@@ -289,6 +306,141 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
289 306
290 }307 }
291 308
309+func TestCatalogueRestoreResetsOnlyIncompatibleCheckpoint(t *testing.T) {
310+ incompatible := durableRecoveryRecord(0)
311+ healthy := durableRecoveryRecord(1)
312+ repository := &durableRecoveryRepository{
313+ generations: map[string]ports.SnapshotGeneration{healthy.Name: validGeneration(t, healthy)},
314+ errors: map[string]error{incompatible.Name: fmt.Errorf("load envelope: %w", snapcodec.ErrBadVersion)},
315+ loads: make(map[string]int),
316+ repairs: make(map[string]domain.CheckpointRef),
317+ }
318+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{incompatible, healthy}, repository)
319+
320+ d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
321+
322+ fresh, ok, err := catalogue.Record(incompatible.Name)
323+ require.NoError(t, err)
324+ require.True(t, ok)
325+ require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{0xa5}, 16)), fresh.IncarnationID)
326+ require.Nil(t, fresh.Committed)
327+ require.Empty(t, fresh.DegradedReason)
328+ require.Empty(t, fresh.TabNames)
329+ require.Equal(t, []domain.IncarnationID{incompatible.IncarnationID}, repository.deleted)
330+ require.Equal(t, ports.SessionStopped, d.stopped[incompatible.Name].state)
331+ require.Equal(t, fresh, d.stopped[incompatible.Name].record)
332+
333+ restoredSibling, ok, err := catalogue.Record(healthy.Name)
334+ require.NoError(t, err)
335+ require.True(t, ok)
336+ require.Equal(t, healthy, restoredSibling)
337+ require.Equal(t, ports.SessionStopped, d.stopped[healthy.Name].state)
338+ require.Equal(t, 1, repository.loads[healthy.Name])
339+}
340+
341+func TestIncompatibleCheckpointDeleteFailurePublishesFreshStoppedAuthority(t *testing.T) {
342+ record := durableRecoveryRecord(0)
343+ deleteErr := errors.New("delete old incarnation")
344+ repository := &durableRecoveryRepository{
345+ errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
346+ loads: make(map[string]int),
347+ repairs: make(map[string]domain.CheckpointRef),
348+ deleteErr: deleteErr,
349+ }
350+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
351+ d.mu.Lock()
352+ restoreDone := d.stopped[record.Name].restoreDone
353+ d.mu.Unlock()
354+
355+ restoreErr := d.restoreRecord(context.Background(), record)
356+ require.NoError(t, restoreErr, "post-commit cleanup failure must not fail the authority transition")
357+
358+ fresh, ok, err := catalogue.Record(record.Name)
359+ require.NoError(t, err)
360+ require.True(t, ok)
361+ require.NotEqual(t, record.IncarnationID, fresh.IncarnationID)
362+ require.Nil(t, fresh.Committed)
363+ require.Empty(t, fresh.DegradedReason)
364+ d.mu.Lock()
365+ entry := d.stopped[record.Name]
366+ d.mu.Unlock()
367+ require.Equal(t, ports.SessionStopped, entry.state)
368+ require.Equal(t, fresh, entry.record)
369+ require.Equal(t, restoreDone, entry.restoreDone)
370+ select {
371+ case <-entry.restoreDone:
372+ t.Fatal("direct restore closed its coordinator-owned completion barrier")
373+ default:
374+ }
375+
376+ d.finishRecordRestore(record, restoreErr, restoreDone)
377+ d.mu.Lock()
378+ entry = d.stopped[record.Name]
379+ d.mu.Unlock()
380+ require.Equal(t, ports.SessionStopped, entry.state)
381+ require.Equal(t, fresh, entry.record)
382+ select {
383+ case <-entry.restoreDone:
384+ default:
385+ t.Fatal("restore completion was not signaled")
386+ }
387+ require.Equal(t, []domain.IncarnationID{record.IncarnationID}, repository.deleted)
388+}
389+
390+func TestIncompatibleCheckpointPreCommitFailureReturnsError(t *testing.T) {
391+ record := durableRecoveryRecord(0)
392+ cause := errors.New("catalogue replace failed")
393+ repository := &durableRecoveryRepository{
394+ errors: map[string]error{record.Name: snapcodec.ErrBadVersion},
395+ loads: make(map[string]int),
396+ repairs: make(map[string]domain.CheckpointRef),
397+ }
398+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
399+ catalogue.replaceErr = cause
400+
401+ err := d.restoreRecord(context.Background(), record)
402+
403+ require.ErrorIs(t, err, cause)
404+ persisted, ok, recordErr := catalogue.Record(record.Name)
405+ require.NoError(t, recordErr)
406+ require.True(t, ok)
407+ require.Equal(t, record, persisted)
408+ require.Empty(t, repository.deleted)
409+}
410+
411+func TestRestoreLoadFailuresRetainCheckpointWithoutDeletion(t *testing.T) {
412+ tests := []struct {
413+ name string
414+ err error
415+ }{
416+ {name: "corruption", err: errors.New("corrupt checkpoint")},
417+ {name: "generic", err: errors.New("load failed")},
418+ {name: "io", err: &os.PathError{Op: "open", Path: "/snapshot", Err: syscall.EACCES}},
419+ }
420+ for _, tt := range tests {
421+ t.Run(tt.name, func(t *testing.T) {
422+ record := durableRecoveryRecord(0)
423+ repository := &durableRecoveryRepository{
424+ errors: map[string]error{record.Name: tt.err},
425+ loads: make(map[string]int),
426+ repairs: make(map[string]domain.CheckpointRef),
427+ }
428+ d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
429+
430+ d.restoreCatalogue(context.Background(), mustDurableRecords(t, catalogue))
431+
432+ persisted, ok, err := catalogue.Record(record.Name)
433+ require.NoError(t, err)
434+ require.True(t, ok)
435+ require.Equal(t, record.IncarnationID, persisted.IncarnationID)
436+ require.Equal(t, record.Committed, persisted.Committed)
437+ require.Equal(t, "checkpoint validation failed", persisted.DegradedReason)
438+ require.Equal(t, ports.SessionBroken, d.stopped[record.Name].state)
439+ require.Empty(t, repository.deleted)
440+ })
441+ }
442+}
443+
292 func TestCatalogueRestoreAggregateOver4096(t *testing.T) {444 func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
293 const count = 4200445 const count = 4200
294 records := make([]domain.CatalogueRecord, count)446 records := make([]domain.CatalogueRecord, count)
M internal/usecase/daemon/snapshot_restore_incremental.go +42 -0
@@ -178,6 +178,9 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
178178 if ctxErr := ctx.Err(); ctxErr != nil {
179179 return ctxErr
180180 }
181+ if errors.Is(err, snapcodec.ErrBadVersion) {
182+ return d.resetIncompatibleCheckpoint(ctx, record, selected)
183+ }
181184 if isRetryableRestoreLoadError(err) {
182185 return fmt.Errorf("%w: %w", errRetryableRestoreLoad, err)
183186 }
@@ -215,6 +218,45 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
215218 return nil
216219 }
217220
221+func (d *Daemon) resetIncompatibleCheckpoint(ctx context.Context, record domain.CatalogueRecord, selected domain.CheckpointRef) error {
222+ fresh, committed, err := d.recovery.ResetIncompatible(ctx, record.Name, record.IncarnationID, selected)
223+ if !committed {
224+ if err != nil {
225+ return fmt.Errorf("snapshot: reset incompatible checkpoint: %w", err)
226+ }
227+ d.log.Info("snapshot_incompatible_reset_superseded",
228+ "session", record.Name,
229+ "incarnation", record.IncarnationID.String(),
230+ "generation", selected.Generation,
231+ )
232+ return nil
233+ }
234+
235+ d.mu.Lock()
236+ if entry, ok := d.stopped[record.Name]; ok {
237+ d.stopped[record.Name] = stoppedSessionFromRecord(fresh, ports.SessionStopped, entry.restoreDone)
238+ }
239+ d.mu.Unlock()
240+
241+ if err != nil {
242+ d.log.Warn("snapshot_incompatible_reset_cleanup_pending",
243+ "session", fresh.Name,
244+ "incarnation", fresh.IncarnationID.String(),
245+ "replaced_incarnation", record.IncarnationID.String(),
246+ "generation", selected.Generation,
247+ "err", err,
248+ )
249+ return nil
250+ }
251+ d.log.Info("snapshot_incompatible_reset_complete",
252+ "session", fresh.Name,
253+ "incarnation", fresh.IncarnationID.String(),
254+ "replaced_incarnation", record.IncarnationID.String(),
255+ "generation", selected.Generation,
256+ )
257+ return nil
258+}
259+
218260 func isRetryableRestoreLoadError(err error) bool {
219261 if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
220262 return false
@@ -178,6 +178,9 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
178 if ctxErr := ctx.Err(); ctxErr != nil {178 if ctxErr := ctx.Err(); ctxErr != nil {
179 return ctxErr179 return ctxErr
180 }180 }
181+ if errors.Is(err, snapcodec.ErrBadVersion) {
182+ return d.resetIncompatibleCheckpoint(ctx, record, selected)
183+ }
181 if isRetryableRestoreLoadError(err) {184 if isRetryableRestoreLoadError(err) {
182 return fmt.Errorf("%w: %w", errRetryableRestoreLoad, err)185 return fmt.Errorf("%w: %w", errRetryableRestoreLoad, err)
183 }186 }
@@ -215,6 +218,45 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
215 return nil218 return nil
216 }219 }
217 220
221+func (d *Daemon) resetIncompatibleCheckpoint(ctx context.Context, record domain.CatalogueRecord, selected domain.CheckpointRef) error {
222+ fresh, committed, err := d.recovery.ResetIncompatible(ctx, record.Name, record.IncarnationID, selected)
223+ if !committed {
224+ if err != nil {
225+ return fmt.Errorf("snapshot: reset incompatible checkpoint: %w", err)
226+ }
227+ d.log.Info("snapshot_incompatible_reset_superseded",
228+ "session", record.Name,
229+ "incarnation", record.IncarnationID.String(),
230+ "generation", selected.Generation,
231+ )
232+ return nil
233+ }
234+
235+ d.mu.Lock()
236+ if entry, ok := d.stopped[record.Name]; ok {
237+ d.stopped[record.Name] = stoppedSessionFromRecord(fresh, ports.SessionStopped, entry.restoreDone)
238+ }
239+ d.mu.Unlock()
240+
241+ if err != nil {
242+ d.log.Warn("snapshot_incompatible_reset_cleanup_pending",
243+ "session", fresh.Name,
244+ "incarnation", fresh.IncarnationID.String(),
245+ "replaced_incarnation", record.IncarnationID.String(),
246+ "generation", selected.Generation,
247+ "err", err,
248+ )
249+ return nil
250+ }
251+ d.log.Info("snapshot_incompatible_reset_complete",
252+ "session", fresh.Name,
253+ "incarnation", fresh.IncarnationID.String(),
254+ "replaced_incarnation", record.IncarnationID.String(),
255+ "generation", selected.Generation,
256+ )
257+ return nil
258+}
259+
218 func isRetryableRestoreLoadError(err error) bool {260 func isRetryableRestoreLoadError(err error) bool {
219 if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {261 if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
220 return false262 return false

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