detached 1b02243

feat(recovery): commit bounded checkpoint fallbacks

commit
1b022432fab47f60019b1834f83073012b87df11
author
brice · · 2026-07-25 23:33 UTC
parent
52e29a3
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
10 files · +666 -40
Split
UnifiedSplit
M internal/app/run.go +1 -0
@@ -530,6 +530,7 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
530530 if err := coordinator.Recover(ctx); err != nil {
531531 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
532532 }
533+ daemonOpts = append(daemonOpts, daemon.WithCheckpointCoordinator(coordinator))
533534 log.Info("session persistence enabled", "path", storePath)
534535 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
535536
@@ -530,6 +530,7 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
530 if err := coordinator.Recover(ctx); err != nil {530 if err := coordinator.Recover(ctx); err != nil {
531 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())531 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
532 }532 }
533+ daemonOpts = append(daemonOpts, daemon.WithCheckpointCoordinator(coordinator))
533 log.Info("session persistence enabled", "path", storePath)534 log.Info("session persistence enabled", "path", storePath)
534 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))535 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
535 536
M internal/ports/mocks/recovery_mock.go +171 -0
@@ -434,6 +434,177 @@ func (_c *MockRecoveryJournal_SaveDiscard_Call) RunAndReturn(run func(context1 c
434434 return _c
435435 }
436436
437+// NewMockCheckpointCoordinator creates a new instance of MockCheckpointCoordinator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
438+// The first argument is typically a *testing.T value.
439+func NewMockCheckpointCoordinator(t interface {
440+ mock.TestingT
441+ Cleanup(func())
442+}) *MockCheckpointCoordinator {
443+ mock := &MockCheckpointCoordinator{}
444+ mock.Mock.Test(t)
445+
446+ t.Cleanup(func() { mock.AssertExpectations(t) })
447+
448+ return mock
449+}
450+
451+// MockCheckpointCoordinator is an autogenerated mock type for the CheckpointCoordinator type
452+type MockCheckpointCoordinator struct {
453+ mock.Mock
454+}
455+
456+type MockCheckpointCoordinator_Expecter struct {
457+ mock *mock.Mock
458+}
459+
460+func (_m *MockCheckpointCoordinator) EXPECT() *MockCheckpointCoordinator_Expecter {
461+ return &MockCheckpointCoordinator_Expecter{mock: &_m.Mock}
462+}
463+
464+// PromoteFallback provides a mock function for the type MockCheckpointCoordinator
465+func (_mock *MockCheckpointCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error) {
466+ ret := _mock.Called(context1, s, checkpointRef)
467+
468+ if len(ret) == 0 {
469+ panic("no return value specified for PromoteFallback")
470+ }
471+
472+ var r0 domain.CatalogueRecord
473+ var r1 error
474+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)); ok {
475+ return returnFunc(context1, s, checkpointRef)
476+ }
477+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) domain.CatalogueRecord); ok {
478+ r0 = returnFunc(context1, s, checkpointRef)
479+ } else {
480+ r0 = ret.Get(0).(domain.CatalogueRecord)
481+ }
482+ if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {
483+ r1 = returnFunc(context1, s, checkpointRef)
484+ } else {
485+ r1 = ret.Error(1)
486+ }
487+ return r0, r1
488+}
489+
490+// MockCheckpointCoordinator_PromoteFallback_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PromoteFallback'
491+type MockCheckpointCoordinator_PromoteFallback_Call struct {
492+ *mock.Call
493+}
494+
495+// PromoteFallback is a helper method to define mock.On call
496+// - context1 context.Context
497+// - s string
498+// - checkpointRef domain.CheckpointRef
499+func (_e *MockCheckpointCoordinator_Expecter) PromoteFallback(context1 any, s any, checkpointRef any) *MockCheckpointCoordinator_PromoteFallback_Call {
500+ return &MockCheckpointCoordinator_PromoteFallback_Call{Call: _e.mock.On("PromoteFallback", context1, s, checkpointRef)}
501+}
502+
503+func (_c *MockCheckpointCoordinator_PromoteFallback_Call) Run(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef)) *MockCheckpointCoordinator_PromoteFallback_Call {
504+ _c.Call.Run(func(args mock.Arguments) {
505+ var arg0 context.Context
506+ if args[0] != nil {
507+ arg0 = args[0].(context.Context)
508+ }
509+ var arg1 string
510+ if args[1] != nil {
511+ arg1 = args[1].(string)
512+ }
513+ var arg2 domain.CheckpointRef
514+ if args[2] != nil {
515+ arg2 = args[2].(domain.CheckpointRef)
516+ }
517+ run(
518+ arg0,
519+ arg1,
520+ arg2,
521+ )
522+ })
523+ return _c
524+}
525+
526+func (_c *MockCheckpointCoordinator_PromoteFallback_Call) Return(catalogueRecord domain.CatalogueRecord, err error) *MockCheckpointCoordinator_PromoteFallback_Call {
527+ _c.Call.Return(catalogueRecord, err)
528+ return _c
529+}
530+
531+func (_c *MockCheckpointCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error)) *MockCheckpointCoordinator_PromoteFallback_Call {
532+ _c.Call.Return(run)
533+ return _c
534+}
535+
536+// PublishCheckpoint provides a mock function for the type MockCheckpointCoordinator
537+func (_mock *MockCheckpointCoordinator) PublishCheckpoint(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
538+ ret := _mock.Called(context1, s, snapshotPublication)
539+
540+ if len(ret) == 0 {
541+ panic("no return value specified for PublishCheckpoint")
542+ }
543+
544+ var r0 domain.CatalogueRecord
545+ var r1 error
546+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) (domain.CatalogueRecord, error)); ok {
547+ return returnFunc(context1, s, snapshotPublication)
548+ }
549+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) domain.CatalogueRecord); ok {
550+ r0 = returnFunc(context1, s, snapshotPublication)
551+ } else {
552+ r0 = ret.Get(0).(domain.CatalogueRecord)
553+ }
554+ if returnFunc, ok := ret.Get(1).(func(context.Context, string, ports.SnapshotPublication) error); ok {
555+ r1 = returnFunc(context1, s, snapshotPublication)
556+ } else {
557+ r1 = ret.Error(1)
558+ }
559+ return r0, r1
560+}
561+
562+// MockCheckpointCoordinator_PublishCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PublishCheckpoint'
563+type MockCheckpointCoordinator_PublishCheckpoint_Call struct {
564+ *mock.Call
565+}
566+
567+// PublishCheckpoint is a helper method to define mock.On call
568+// - context1 context.Context
569+// - s string
570+// - snapshotPublication ports.SnapshotPublication
571+func (_e *MockCheckpointCoordinator_Expecter) PublishCheckpoint(context1 any, s any, snapshotPublication any) *MockCheckpointCoordinator_PublishCheckpoint_Call {
572+ return &MockCheckpointCoordinator_PublishCheckpoint_Call{Call: _e.mock.On("PublishCheckpoint", context1, s, snapshotPublication)}
573+}
574+
575+func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) Run(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication)) *MockCheckpointCoordinator_PublishCheckpoint_Call {
576+ _c.Call.Run(func(args mock.Arguments) {
577+ var arg0 context.Context
578+ if args[0] != nil {
579+ arg0 = args[0].(context.Context)
580+ }
581+ var arg1 string
582+ if args[1] != nil {
583+ arg1 = args[1].(string)
584+ }
585+ var arg2 ports.SnapshotPublication
586+ if args[2] != nil {
587+ arg2 = args[2].(ports.SnapshotPublication)
588+ }
589+ run(
590+ arg0,
591+ arg1,
592+ arg2,
593+ )
594+ })
595+ return _c
596+}
597+
598+func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) Return(catalogueRecord domain.CatalogueRecord, err error) *MockCheckpointCoordinator_PublishCheckpoint_Call {
599+ _c.Call.Return(catalogueRecord, err)
600+ return _c
601+}
602+
603+func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) RunAndReturn(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error)) *MockCheckpointCoordinator_PublishCheckpoint_Call {
604+ _c.Call.Return(run)
605+ return _c
606+}
607+
437608 // NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
438609 // The first argument is typically a *testing.T value.
439610 func NewMockCatalogue(t interface {
@@ -434,6 +434,177 @@ func (_c *MockRecoveryJournal_SaveDiscard_Call) RunAndReturn(run func(context1 c
434 return _c434 return _c
435 }435 }
436 436
437+// NewMockCheckpointCoordinator creates a new instance of MockCheckpointCoordinator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
438+// The first argument is typically a *testing.T value.
439+func NewMockCheckpointCoordinator(t interface {
440+ mock.TestingT
441+ Cleanup(func())
442+}) *MockCheckpointCoordinator {
443+ mock := &MockCheckpointCoordinator{}
444+ mock.Mock.Test(t)
445+
446+ t.Cleanup(func() { mock.AssertExpectations(t) })
447+
448+ return mock
449+}
450+
451+// MockCheckpointCoordinator is an autogenerated mock type for the CheckpointCoordinator type
452+type MockCheckpointCoordinator struct {
453+ mock.Mock
454+}
455+
456+type MockCheckpointCoordinator_Expecter struct {
457+ mock *mock.Mock
458+}
459+
460+func (_m *MockCheckpointCoordinator) EXPECT() *MockCheckpointCoordinator_Expecter {
461+ return &MockCheckpointCoordinator_Expecter{mock: &_m.Mock}
462+}
463+
464+// PromoteFallback provides a mock function for the type MockCheckpointCoordinator
465+func (_mock *MockCheckpointCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error) {
466+ ret := _mock.Called(context1, s, checkpointRef)
467+
468+ if len(ret) == 0 {
469+ panic("no return value specified for PromoteFallback")
470+ }
471+
472+ var r0 domain.CatalogueRecord
473+ var r1 error
474+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)); ok {
475+ return returnFunc(context1, s, checkpointRef)
476+ }
477+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) domain.CatalogueRecord); ok {
478+ r0 = returnFunc(context1, s, checkpointRef)
479+ } else {
480+ r0 = ret.Get(0).(domain.CatalogueRecord)
481+ }
482+ if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {
483+ r1 = returnFunc(context1, s, checkpointRef)
484+ } else {
485+ r1 = ret.Error(1)
486+ }
487+ return r0, r1
488+}
489+
490+// MockCheckpointCoordinator_PromoteFallback_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PromoteFallback'
491+type MockCheckpointCoordinator_PromoteFallback_Call struct {
492+ *mock.Call
493+}
494+
495+// PromoteFallback is a helper method to define mock.On call
496+// - context1 context.Context
497+// - s string
498+// - checkpointRef domain.CheckpointRef
499+func (_e *MockCheckpointCoordinator_Expecter) PromoteFallback(context1 any, s any, checkpointRef any) *MockCheckpointCoordinator_PromoteFallback_Call {
500+ return &MockCheckpointCoordinator_PromoteFallback_Call{Call: _e.mock.On("PromoteFallback", context1, s, checkpointRef)}
501+}
502+
503+func (_c *MockCheckpointCoordinator_PromoteFallback_Call) Run(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef)) *MockCheckpointCoordinator_PromoteFallback_Call {
504+ _c.Call.Run(func(args mock.Arguments) {
505+ var arg0 context.Context
506+ if args[0] != nil {
507+ arg0 = args[0].(context.Context)
508+ }
509+ var arg1 string
510+ if args[1] != nil {
511+ arg1 = args[1].(string)
512+ }
513+ var arg2 domain.CheckpointRef
514+ if args[2] != nil {
515+ arg2 = args[2].(domain.CheckpointRef)
516+ }
517+ run(
518+ arg0,
519+ arg1,
520+ arg2,
521+ )
522+ })
523+ return _c
524+}
525+
526+func (_c *MockCheckpointCoordinator_PromoteFallback_Call) Return(catalogueRecord domain.CatalogueRecord, err error) *MockCheckpointCoordinator_PromoteFallback_Call {
527+ _c.Call.Return(catalogueRecord, err)
528+ return _c
529+}
530+
531+func (_c *MockCheckpointCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error)) *MockCheckpointCoordinator_PromoteFallback_Call {
532+ _c.Call.Return(run)
533+ return _c
534+}
535+
536+// PublishCheckpoint provides a mock function for the type MockCheckpointCoordinator
537+func (_mock *MockCheckpointCoordinator) PublishCheckpoint(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
538+ ret := _mock.Called(context1, s, snapshotPublication)
539+
540+ if len(ret) == 0 {
541+ panic("no return value specified for PublishCheckpoint")
542+ }
543+
544+ var r0 domain.CatalogueRecord
545+ var r1 error
546+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) (domain.CatalogueRecord, error)); ok {
547+ return returnFunc(context1, s, snapshotPublication)
548+ }
549+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) domain.CatalogueRecord); ok {
550+ r0 = returnFunc(context1, s, snapshotPublication)
551+ } else {
552+ r0 = ret.Get(0).(domain.CatalogueRecord)
553+ }
554+ if returnFunc, ok := ret.Get(1).(func(context.Context, string, ports.SnapshotPublication) error); ok {
555+ r1 = returnFunc(context1, s, snapshotPublication)
556+ } else {
557+ r1 = ret.Error(1)
558+ }
559+ return r0, r1
560+}
561+
562+// MockCheckpointCoordinator_PublishCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PublishCheckpoint'
563+type MockCheckpointCoordinator_PublishCheckpoint_Call struct {
564+ *mock.Call
565+}
566+
567+// PublishCheckpoint is a helper method to define mock.On call
568+// - context1 context.Context
569+// - s string
570+// - snapshotPublication ports.SnapshotPublication
571+func (_e *MockCheckpointCoordinator_Expecter) PublishCheckpoint(context1 any, s any, snapshotPublication any) *MockCheckpointCoordinator_PublishCheckpoint_Call {
572+ return &MockCheckpointCoordinator_PublishCheckpoint_Call{Call: _e.mock.On("PublishCheckpoint", context1, s, snapshotPublication)}
573+}
574+
575+func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) Run(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication)) *MockCheckpointCoordinator_PublishCheckpoint_Call {
576+ _c.Call.Run(func(args mock.Arguments) {
577+ var arg0 context.Context
578+ if args[0] != nil {
579+ arg0 = args[0].(context.Context)
580+ }
581+ var arg1 string
582+ if args[1] != nil {
583+ arg1 = args[1].(string)
584+ }
585+ var arg2 ports.SnapshotPublication
586+ if args[2] != nil {
587+ arg2 = args[2].(ports.SnapshotPublication)
588+ }
589+ run(
590+ arg0,
591+ arg1,
592+ arg2,
593+ )
594+ })
595+ return _c
596+}
597+
598+func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) Return(catalogueRecord domain.CatalogueRecord, err error) *MockCheckpointCoordinator_PublishCheckpoint_Call {
599+ _c.Call.Return(catalogueRecord, err)
600+ return _c
601+}
602+
603+func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) RunAndReturn(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error)) *MockCheckpointCoordinator_PublishCheckpoint_Call {
604+ _c.Call.Return(run)
605+ return _c
606+}
607+
437 // NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.608 // NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
438 // The first argument is typically a *testing.T value.609 // The first argument is typically a *testing.T value.
439 func NewMockCatalogue(t interface {610 func NewMockCatalogue(t interface {
M internal/ports/recovery.go +5 -0
@@ -123,6 +123,11 @@ type RecoveryJournal interface {
123123 DeleteDiscard(context.Context, domain.IncarnationID) error
124124 }
125125
126+type CheckpointCoordinator interface {
127+ PublishCheckpoint(context.Context, string, SnapshotPublication) (domain.CatalogueRecord, error)
128+ PromoteFallback(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)
129+}
130+
126131 type Catalogue interface {
127132 Records() []domain.CatalogueRecord
128133 Record(string) (domain.CatalogueRecord, bool)
@@ -123,6 +123,11 @@ type RecoveryJournal interface {
123 DeleteDiscard(context.Context, domain.IncarnationID) error123 DeleteDiscard(context.Context, domain.IncarnationID) error
124 }124 }
125 125
126+type CheckpointCoordinator interface {
127+ PublishCheckpoint(context.Context, string, SnapshotPublication) (domain.CatalogueRecord, error)
128+ PromoteFallback(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)
129+}
130+
126 type Catalogue interface {131 type Catalogue interface {
127 Records() []domain.CatalogueRecord132 Records() []domain.CatalogueRecord
128 Record(string) (domain.CatalogueRecord, bool)133 Record(string) (domain.CatalogueRecord, bool)
M internal/usecase/daemon/daemon.go +8 -0
@@ -125,6 +125,7 @@ type Daemon struct {
125125 // snapshotRepository is the sole checkpoint contract. legacySnapshots is
126126 // read-only migration input and is never used for new writes.
127127 snapshotRepository ports.SnapshotRepository
128+ checkpointRecovery ports.CheckpointCoordinator
128129 legacySnapshots ports.LegacySnapshotSource
129130 snapsEnabled bool
130131 // legacyImportPending holds verified legacy blobs whose source deletion
@@ -277,6 +278,13 @@ func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord)
277278 }
278279 }
279280
281+// WithCheckpointCoordinator installs the shared durable checkpoint transaction seam.
282+func WithCheckpointCoordinator(coordinator ports.CheckpointCoordinator) Option {
283+ return func(d *Daemon) {
284+ d.checkpointRecovery = coordinator
285+ }
286+}
287+
280288 // WithSnapshotRepository enables content-addressed incremental snapshots. The
281289 // legacy source is read-only migration input and is never used for new writes.
282290 func WithSnapshotRepository(repository any, legacy ports.LegacySnapshotSource) Option {
@@ -125,6 +125,7 @@ type Daemon struct {
125 // snapshotRepository is the sole checkpoint contract. legacySnapshots is125 // snapshotRepository is the sole checkpoint contract. legacySnapshots is
126 // read-only migration input and is never used for new writes.126 // read-only migration input and is never used for new writes.
127 snapshotRepository ports.SnapshotRepository127 snapshotRepository ports.SnapshotRepository
128+ checkpointRecovery ports.CheckpointCoordinator
128 legacySnapshots ports.LegacySnapshotSource129 legacySnapshots ports.LegacySnapshotSource
129 snapsEnabled bool130 snapsEnabled bool
130 // legacyImportPending holds verified legacy blobs whose source deletion131 // legacyImportPending holds verified legacy blobs whose source deletion
@@ -277,6 +278,13 @@ func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord)
277 }278 }
278 }279 }
279 280
281+// WithCheckpointCoordinator installs the shared durable checkpoint transaction seam.
282+func WithCheckpointCoordinator(coordinator ports.CheckpointCoordinator) Option {
283+ return func(d *Daemon) {
284+ d.checkpointRecovery = coordinator
285+ }
286+}
287+
280 // WithSnapshotRepository enables content-addressed incremental snapshots. The288 // WithSnapshotRepository enables content-addressed incremental snapshots. The
281 // legacy source is read-only migration input and is never used for new writes.289 // legacy source is read-only migration input and is never used for new writes.
282 func WithSnapshotRepository(repository any, legacy ports.LegacySnapshotSource) Option {290 func WithSnapshotRepository(repository any, legacy ports.LegacySnapshotSource) Option {
M internal/usecase/daemon/durable_recovery_test.go +45 -5
@@ -2,6 +2,7 @@ package daemon
22
33 import (
44 "context"
5+ "crypto/sha256"
56 "errors"
67 "fmt"
78 "io"
@@ -14,6 +15,7 @@ import (
1415
1516 "github.com/bnema/vev/internal/domain"
1617 "github.com/bnema/vev/internal/ports"
18+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
1719 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
1820 )
1921
@@ -125,12 +127,48 @@ func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.Snapshot
125127 func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {
126128 t.Helper()
127129 catalogue := newDurableRecoveryCatalogue(records)
128- d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository, nil))
130+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil)
131+ d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository, nil), WithCheckpointCoordinator(coordinator))
129132 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
130133 t.Cleanup(d.serveCancel)
131134 return d, catalogue
132135 }
133136
137+func TestFinalCheckpointFailurePreservesCatalogue(t *testing.T) {
138+ priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
139+ prior := domain.CatalogueRecord{
140+ Name: "work",
141+ IncarnationID: domain.IncarnationID{1},
142+ Cwd: "/work",
143+ CreatedAt: 42,
144+ RecoveryState: domain.RecoveryHealthy,
145+ Committed: &priorRef,
146+ }
147+ catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{prior})
148+ repository := &durableRecoveryRepository{}
149+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil)
150+ d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, []domain.CatalogueRecord{prior}), WithSnapshotRepository(repository, nil), WithCheckpointCoordinator(coordinator))
151+ startSnapshotEncodeWorker(t, d)
152+
153+ sess := newSnapshotTestSession(t, prior.Name, false, prior.Cwd)
154+ expired, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
155+ defer cancel()
156+ sess.snapshotMu.Lock()
157+ sess.snapshotPublishedGeneration = priorRef.Generation
158+ sess.snapshotPublishedCheckpoint = &priorRef
159+ sess.snapshotGeneration = priorRef.Generation + 1
160+ sess.snapshotPublicationContext = expired
161+ sess.snapDirty.Store(true)
162+ sess.snapshotMu.Unlock()
163+
164+ require.True(t, d.scheduleFinalSnapshot(sess))
165+ awaitSnapshotIdle(t, sess)
166+ got, ok := catalogue.Record(prior.Name)
167+ require.True(t, ok)
168+ require.Equal(t, prior, got, "a timed-out final checkpoint must not change authoritative bytes")
169+ require.Equal(t, priorRef, *sess.snapshotPublishedCheckpoint)
170+}
171+
134172 func TestCatalogueRestoreIndependent(t *testing.T) {
135173 emptyDaemon, emptyCatalogue := newDurableRecoveryDaemon(t, nil, &durableRecoveryRepository{})
136174 emptyDone := make(chan struct{})
@@ -173,12 +211,14 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
173211 require.Empty(t, d.sessions, "a corrupt record must never open a replacement runtime")
174212
175213 fallback := durableRecoveryRecord(2)
176- fallbackRef := domain.CheckpointRef{Generation: fallback.Committed.Generation - 1, ManifestDigest: [32]byte{77}}
177- fallback.Fallbacks[0] = &fallbackRef
214+ fallbackRef := domain.CheckpointRef{Generation: fallback.Committed.Generation - 1}
178215 fallbackSnapshotRecord := fallback
179216 fallbackSnapshotRecord.Committed = &fallbackRef
217+ fallbackGeneration := validGeneration(t, fallbackSnapshotRecord)
218+ fallbackRef.ManifestDigest = sha256.Sum256(fallbackGeneration.Manifest)
219+ fallback.Fallbacks[0] = &fallbackRef
180220 fallbackRepository := &durableRecoveryRepository{
181- generations: map[string]ports.SnapshotGeneration{fallback.Name: validGeneration(t, fallbackSnapshotRecord)},
221+ generations: map[string]ports.SnapshotGeneration{fallback.Name: fallbackGeneration},
182222 errors: make(map[string]error),
183223 loads: make(map[string]int),
184224 repairs: make(map[string]domain.CheckpointRef),
@@ -189,7 +229,7 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
189229 require.True(t, ok)
190230 require.Equal(t, fallbackRef, *promoted.Committed)
191231 require.Equal(t, runtimeHealthy, fallbackDaemon.stopped[fallback.Name].state)
192- require.Equal(t, 2, fallbackRepository.loads[fallback.Name])
232+ require.Equal(t, 3, fallbackRepository.loads[fallback.Name])
193233 require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])
194234 }
195235
@@ -2,6 +2,7 @@ package daemon
2 2
3 import (3 import (
4 "context"4 "context"
5+ "crypto/sha256"
5 "errors"6 "errors"
6 "fmt"7 "fmt"
7 "io"8 "io"
@@ -14,6 +15,7 @@ import (
14 15
15 "github.com/bnema/vev/internal/domain"16 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"17 "github.com/bnema/vev/internal/ports"
18+ recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
17 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"19 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
18 )20 )
19 21
@@ -125,12 +127,48 @@ func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.Snapshot
125 func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {127 func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (*Daemon, *durableRecoveryCatalogue) {
126 t.Helper()128 t.Helper()
127 catalogue := newDurableRecoveryCatalogue(records)129 catalogue := newDurableRecoveryCatalogue(records)
128- d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository, nil))130+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil)
131+ d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository, nil), WithCheckpointCoordinator(coordinator))
129 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())132 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
130 t.Cleanup(d.serveCancel)133 t.Cleanup(d.serveCancel)
131 return d, catalogue134 return d, catalogue
132 }135 }
133 136
137+func TestFinalCheckpointFailurePreservesCatalogue(t *testing.T) {
138+ priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
139+ prior := domain.CatalogueRecord{
140+ Name: "work",
141+ IncarnationID: domain.IncarnationID{1},
142+ Cwd: "/work",
143+ CreatedAt: 42,
144+ RecoveryState: domain.RecoveryHealthy,
145+ Committed: &priorRef,
146+ }
147+ catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{prior})
148+ repository := &durableRecoveryRepository{}
149+ coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil)
150+ d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, []domain.CatalogueRecord{prior}), WithSnapshotRepository(repository, nil), WithCheckpointCoordinator(coordinator))
151+ startSnapshotEncodeWorker(t, d)
152+
153+ sess := newSnapshotTestSession(t, prior.Name, false, prior.Cwd)
154+ expired, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
155+ defer cancel()
156+ sess.snapshotMu.Lock()
157+ sess.snapshotPublishedGeneration = priorRef.Generation
158+ sess.snapshotPublishedCheckpoint = &priorRef
159+ sess.snapshotGeneration = priorRef.Generation + 1
160+ sess.snapshotPublicationContext = expired
161+ sess.snapDirty.Store(true)
162+ sess.snapshotMu.Unlock()
163+
164+ require.True(t, d.scheduleFinalSnapshot(sess))
165+ awaitSnapshotIdle(t, sess)
166+ got, ok := catalogue.Record(prior.Name)
167+ require.True(t, ok)
168+ require.Equal(t, prior, got, "a timed-out final checkpoint must not change authoritative bytes")
169+ require.Equal(t, priorRef, *sess.snapshotPublishedCheckpoint)
170+}
171+
134 func TestCatalogueRestoreIndependent(t *testing.T) {172 func TestCatalogueRestoreIndependent(t *testing.T) {
135 emptyDaemon, emptyCatalogue := newDurableRecoveryDaemon(t, nil, &durableRecoveryRepository{})173 emptyDaemon, emptyCatalogue := newDurableRecoveryDaemon(t, nil, &durableRecoveryRepository{})
136 emptyDone := make(chan struct{})174 emptyDone := make(chan struct{})
@@ -173,12 +211,14 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
173 require.Empty(t, d.sessions, "a corrupt record must never open a replacement runtime")211 require.Empty(t, d.sessions, "a corrupt record must never open a replacement runtime")
174 212
175 fallback := durableRecoveryRecord(2)213 fallback := durableRecoveryRecord(2)
176- fallbackRef := domain.CheckpointRef{Generation: fallback.Committed.Generation - 1, ManifestDigest: [32]byte{77}}214+ fallbackRef := domain.CheckpointRef{Generation: fallback.Committed.Generation - 1}
177- fallback.Fallbacks[0] = &fallbackRef
178 fallbackSnapshotRecord := fallback215 fallbackSnapshotRecord := fallback
179 fallbackSnapshotRecord.Committed = &fallbackRef216 fallbackSnapshotRecord.Committed = &fallbackRef
217+ fallbackGeneration := validGeneration(t, fallbackSnapshotRecord)
218+ fallbackRef.ManifestDigest = sha256.Sum256(fallbackGeneration.Manifest)
219+ fallback.Fallbacks[0] = &fallbackRef
180 fallbackRepository := &durableRecoveryRepository{220 fallbackRepository := &durableRecoveryRepository{
181- generations: map[string]ports.SnapshotGeneration{fallback.Name: validGeneration(t, fallbackSnapshotRecord)},221+ generations: map[string]ports.SnapshotGeneration{fallback.Name: fallbackGeneration},
182 errors: make(map[string]error),222 errors: make(map[string]error),
183 loads: make(map[string]int),223 loads: make(map[string]int),
184 repairs: make(map[string]domain.CheckpointRef),224 repairs: make(map[string]domain.CheckpointRef),
@@ -189,7 +229,7 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
189 require.True(t, ok)229 require.True(t, ok)
190 require.Equal(t, fallbackRef, *promoted.Committed)230 require.Equal(t, fallbackRef, *promoted.Committed)
191 require.Equal(t, runtimeHealthy, fallbackDaemon.stopped[fallback.Name].state)231 require.Equal(t, runtimeHealthy, fallbackDaemon.stopped[fallback.Name].state)
192- require.Equal(t, 2, fallbackRepository.loads[fallback.Name])232+ require.Equal(t, 3, fallbackRepository.loads[fallback.Name])
193 require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])233 require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])
194 }234 }
195 235
M internal/usecase/daemon/snapshot_restore_incremental.go +7 -33
@@ -167,7 +167,6 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
167167 if len(candidates) == 0 {
168168 return errors.New("snapshot: healthy record has no checkpoint")
169169 }
170- valid := make([]domain.CheckpointRef, 0, len(candidates))
171170 var selected domain.CheckpointRef
172171 var selectedGeneration ports.SnapshotGeneration
173172 var selectedSnapshot snapcodec.Session
@@ -181,29 +180,22 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
181180 if err != nil || generation.IncarnationID != record.IncarnationID || generation.Name != record.Name || generation.Generation != candidate.Generation {
182181 continue
183182 }
184- valid = append(valid, candidate)
185- if selectedIndex < 0 {
186- selected, selectedGeneration, selectedSnapshot, selectedIndex = candidate, generation, snapshot, i
187- if i == 0 {
188- break
189- }
190- }
183+ selected, selectedGeneration, selectedSnapshot, selectedIndex = candidate, generation, snapshot, i
184+ break
191185 }
192186 if selectedIndex < 0 {
193187 return errors.New("snapshot: no catalogue checkpoint validated")
194188 }
195189
196190 if selectedIndex > 0 {
197- record = promoteFallback(record, selected, valid)
198- if d.catalogue == nil {
199- return errors.New("snapshot: catalogue unavailable for fallback promotion")
191+ if d.checkpointRecovery == nil {
192+ return errors.New("snapshot: checkpoint coordinator unavailable for fallback promotion")
200193 }
201- if err := d.catalogue.Replace(record.Name, record); err != nil {
194+ var err error
195+ record, err = d.checkpointRecovery.PromoteFallback(ctx, record.Name, selected)
196+ if err != nil {
202197 return fmt.Errorf("snapshot: promote fallback: %w", err)
203198 }
204- if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
205- return fmt.Errorf("snapshot: repair head: %w", err)
206- }
207199 d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
208200 }
209201
@@ -227,24 +219,6 @@ func checkpointCandidates(record domain.CatalogueRecord) []domain.CheckpointRef
227219 return candidates
228220 }
229221
230-func promoteFallback(record domain.CatalogueRecord, selected domain.CheckpointRef, valid []domain.CheckpointRef) domain.CatalogueRecord {
231- next := record
232- next.RecoveryState = domain.RecoveryHealthy
233- next.Committed = &selected
234- next.Fallbacks = [2]*domain.CheckpointRef{}
235- next.DegradedReason = ""
236- slot := 0
237- for _, candidate := range valid {
238- if candidate == selected || slot == len(next.Fallbacks) {
239- continue
240- }
241- copy := candidate
242- next.Fallbacks[slot] = &copy
243- slot++
244- }
245- return next
246-}
247-
248222 func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {
249223 manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)
250224 if err != nil {
@@ -167,7 +167,6 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
167 if len(candidates) == 0 {167 if len(candidates) == 0 {
168 return errors.New("snapshot: healthy record has no checkpoint")168 return errors.New("snapshot: healthy record has no checkpoint")
169 }169 }
170- valid := make([]domain.CheckpointRef, 0, len(candidates))
171 var selected domain.CheckpointRef170 var selected domain.CheckpointRef
172 var selectedGeneration ports.SnapshotGeneration171 var selectedGeneration ports.SnapshotGeneration
173 var selectedSnapshot snapcodec.Session172 var selectedSnapshot snapcodec.Session
@@ -181,29 +180,22 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
181 if err != nil || generation.IncarnationID != record.IncarnationID || generation.Name != record.Name || generation.Generation != candidate.Generation {180 if err != nil || generation.IncarnationID != record.IncarnationID || generation.Name != record.Name || generation.Generation != candidate.Generation {
182 continue181 continue
183 }182 }
184- valid = append(valid, candidate)183+ selected, selectedGeneration, selectedSnapshot, selectedIndex = candidate, generation, snapshot, i
185- if selectedIndex < 0 {184+ break
186- selected, selectedGeneration, selectedSnapshot, selectedIndex = candidate, generation, snapshot, i
187- if i == 0 {
188- break
189- }
190- }
191 }185 }
192 if selectedIndex < 0 {186 if selectedIndex < 0 {
193 return errors.New("snapshot: no catalogue checkpoint validated")187 return errors.New("snapshot: no catalogue checkpoint validated")
194 }188 }
195 189
196 if selectedIndex > 0 {190 if selectedIndex > 0 {
197- record = promoteFallback(record, selected, valid)191+ if d.checkpointRecovery == nil {
198- if d.catalogue == nil {192+ return errors.New("snapshot: checkpoint coordinator unavailable for fallback promotion")
199- return errors.New("snapshot: catalogue unavailable for fallback promotion")
200 }193 }
201- if err := d.catalogue.Replace(record.Name, record); err != nil {194+ var err error
195+ record, err = d.checkpointRecovery.PromoteFallback(ctx, record.Name, selected)
196+ if err != nil {
202 return fmt.Errorf("snapshot: promote fallback: %w", err)197 return fmt.Errorf("snapshot: promote fallback: %w", err)
203 }198 }
204- if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
205- return fmt.Errorf("snapshot: repair head: %w", err)
206- }
207 d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)199 d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
208 }200 }
209 201
@@ -227,24 +219,6 @@ func checkpointCandidates(record domain.CatalogueRecord) []domain.CheckpointRef
227 return candidates219 return candidates
228 }220 }
229 221
230-func promoteFallback(record domain.CatalogueRecord, selected domain.CheckpointRef, valid []domain.CheckpointRef) domain.CatalogueRecord {
231- next := record
232- next.RecoveryState = domain.RecoveryHealthy
233- next.Committed = &selected
234- next.Fallbacks = [2]*domain.CheckpointRef{}
235- next.DegradedReason = ""
236- slot := 0
237- for _, candidate := range valid {
238- if candidate == selected || slot == len(next.Fallbacks) {
239- continue
240- }
241- copy := candidate
242- next.Fallbacks[slot] = &copy
243- slot++
244- }
245- return next
246-}
247-
248 func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {222 func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {
249 manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)223 manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)
250 if err != nil {224 if err != nil {
M internal/usecase/daemon/snapshot_worker.go +9 -1
@@ -130,7 +130,15 @@ func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint
130130 }
131131 if err == nil && workerCtx.Err() == nil {
132132 if err = publicationContext.Err(); err == nil {
133- err = d.snapshotRepository.Publish(publicationContext, publication)
133+ if d.checkpointRecovery != nil {
134+ var record domain.CatalogueRecord
135+ record, err = d.checkpointRecovery.PublishCheckpoint(publicationContext, capture.name, publication)
136+ if err == nil && record.Committed != nil {
137+ capture.checkpoint = *record.Committed
138+ }
139+ } else {
140+ err = d.snapshotRepository.Publish(publicationContext, publication)
141+ }
134142 }
135143 }
136144 d.clearSnapshotWorkerInFlight(workerID, capture)
@@ -130,7 +130,15 @@ func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint
130 }130 }
131 if err == nil && workerCtx.Err() == nil {131 if err == nil && workerCtx.Err() == nil {
132 if err = publicationContext.Err(); err == nil {132 if err = publicationContext.Err(); err == nil {
133- err = d.snapshotRepository.Publish(publicationContext, publication)133+ if d.checkpointRecovery != nil {
134+ var record domain.CatalogueRecord
135+ record, err = d.checkpointRecovery.PublishCheckpoint(publicationContext, capture.name, publication)
136+ if err == nil && record.Committed != nil {
137+ capture.checkpoint = *record.Committed
138+ }
139+ } else {
140+ err = d.snapshotRepository.Publish(publicationContext, publication)
141+ }
134 }142 }
135 }143 }
136 d.clearSnapshotWorkerInFlight(workerID, capture)144 d.clearSnapshotWorkerInFlight(workerID, capture)
A internal/usecase/recovery/checkpoint.go +150 -0
@@ -0,0 +1,150 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "errors"
7+ "fmt"
8+
9+ "github.com/bnema/vev/internal/domain"
10+ "github.com/bnema/vev/internal/ports"
11+)
12+
13+var (
14+ ErrCheckpointRecordNotFound = errors.New("recovery: checkpoint catalogue record not found")
15+ ErrCheckpointConflict = errors.New("recovery: checkpoint publication conflicts with catalogue")
16+ ErrFallbackNotFound = errors.New("recovery: checkpoint is not a catalogue fallback")
17+)
18+
19+func shiftedCheckpoint(record domain.CatalogueRecord, next domain.CheckpointRef) domain.CatalogueRecord {
20+ prior := copyCheckpointRef(record.Committed)
21+ priorFallback := copyCheckpointRef(record.Fallbacks[0])
22+ nextRecord := record
23+ nextRecord.RecoveryState = domain.RecoveryHealthy
24+ nextRecord.Committed = copyCheckpointRef(&next)
25+ nextRecord.Fallbacks = [2]*domain.CheckpointRef{prior, priorFallback}
26+ nextRecord.DegradedReason = ""
27+ return nextRecord
28+}
29+
30+// PublishCheckpoint publishes repository data before atomically making its
31+// reference authoritative in the catalogue. A catalogue failure deliberately
32+// leaves the durable publication as a same-incarnation forward orphan.
33+func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, publication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
34+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
35+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
36+ }
37+ unlock := c.locks.Lock([]string{name})
38+ defer unlock()
39+
40+ record, ok := c.catalogue.Record(name)
41+ if !ok {
42+ return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
43+ }
44+ if publication.Name != name || publication.Name != record.Name || publication.IncarnationID != record.IncarnationID ||
45+ publication.Generation == 0 || len(publication.Manifest) == 0 ||
46+ !equalCheckpointRefs(publication.ParentCheckpoint, record.Committed) {
47+ return domain.CatalogueRecord{}, ErrCheckpointConflict
48+ }
49+ if err := c.repository.Publish(ctx, publication); err != nil {
50+ return domain.CatalogueRecord{}, err
51+ }
52+
53+ ref := domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
54+ next := shiftedCheckpoint(record, ref)
55+ if err := next.Validate(); err != nil {
56+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid checkpoint transition: %w", err)
57+ }
58+ if err := c.catalogue.Replace(name, next); err != nil {
59+ return domain.CatalogueRecord{}, err
60+ }
61+ return next, nil
62+}
63+
64+// PromoteFallback validates a directly indexed fallback and any alternatives,
65+// commits the bounded validated set, and repairs HEAD only after the catalogue
66+// commit is durable.
67+func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref domain.CheckpointRef) (domain.CatalogueRecord, error) {
68+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
69+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
70+ }
71+ unlock := c.locks.Lock([]string{name})
72+ defer unlock()
73+
74+ record, ok := c.catalogue.Record(name)
75+ if !ok {
76+ return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
77+ }
78+ found := false
79+ for _, fallback := range record.Fallbacks {
80+ if fallback != nil && *fallback == ref {
81+ found = true
82+ break
83+ }
84+ }
85+ if !found {
86+ return domain.CatalogueRecord{}, ErrFallbackNotFound
87+ }
88+ if err := c.validateCheckpoint(ctx, record, ref); err != nil {
89+ return domain.CatalogueRecord{}, err
90+ }
91+
92+ alternatives := make([]domain.CheckpointRef, 0, 2)
93+ candidates := make([]*domain.CheckpointRef, 0, 3)
94+ candidates = append(candidates, record.Committed)
95+ candidates = append(candidates, record.Fallbacks[:]...)
96+ for _, candidate := range candidates {
97+ if candidate == nil || *candidate == ref || candidate.Generation >= ref.Generation || len(alternatives) == 2 {
98+ continue
99+ }
100+ if err := c.validateCheckpoint(ctx, record, *candidate); err == nil {
101+ alternatives = append(alternatives, *candidate)
102+ }
103+ }
104+
105+ next := record
106+ next.RecoveryState = domain.RecoveryHealthy
107+ next.Committed = copyCheckpointRef(&ref)
108+ next.Fallbacks = [2]*domain.CheckpointRef{}
109+ next.DegradedReason = ""
110+ for i := range alternatives {
111+ next.Fallbacks[i] = copyCheckpointRef(&alternatives[i])
112+ }
113+ if err := next.Validate(); err != nil {
114+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fallback promotion: %w", err)
115+ }
116+ if err := c.catalogue.Replace(name, next); err != nil {
117+ return domain.CatalogueRecord{}, err
118+ }
119+ if err := c.repository.RepairHEAD(ctx, record.IncarnationID, ref); err != nil {
120+ return next, err
121+ }
122+ return next, nil
123+}
124+
125+func (c *Coordinator) validateCheckpoint(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef) error {
126+ generation, err := c.repository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
127+ if err != nil {
128+ return err
129+ }
130+ if generation.IncarnationID != record.IncarnationID || generation.Name != record.Name || generation.Generation != ref.Generation ||
131+ sha256.Sum256(generation.Manifest) != ref.ManifestDigest {
132+ return ErrCheckpointConflict
133+ }
134+ return nil
135+}
136+
137+func equalCheckpointRefs(a, b *domain.CheckpointRef) bool {
138+ if a == nil || b == nil {
139+ return a == nil && b == nil
140+ }
141+ return *a == *b
142+}
143+
144+func copyCheckpointRef(ref *domain.CheckpointRef) *domain.CheckpointRef {
145+ if ref == nil {
146+ return nil
147+ }
148+ copy := *ref
149+ return &copy
150+}
@@ -0,0 +1,150 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "errors"
7+ "fmt"
8+
9+ "github.com/bnema/vev/internal/domain"
10+ "github.com/bnema/vev/internal/ports"
11+)
12+
13+var (
14+ ErrCheckpointRecordNotFound = errors.New("recovery: checkpoint catalogue record not found")
15+ ErrCheckpointConflict = errors.New("recovery: checkpoint publication conflicts with catalogue")
16+ ErrFallbackNotFound = errors.New("recovery: checkpoint is not a catalogue fallback")
17+)
18+
19+func shiftedCheckpoint(record domain.CatalogueRecord, next domain.CheckpointRef) domain.CatalogueRecord {
20+ prior := copyCheckpointRef(record.Committed)
21+ priorFallback := copyCheckpointRef(record.Fallbacks[0])
22+ nextRecord := record
23+ nextRecord.RecoveryState = domain.RecoveryHealthy
24+ nextRecord.Committed = copyCheckpointRef(&next)
25+ nextRecord.Fallbacks = [2]*domain.CheckpointRef{prior, priorFallback}
26+ nextRecord.DegradedReason = ""
27+ return nextRecord
28+}
29+
30+// PublishCheckpoint publishes repository data before atomically making its
31+// reference authoritative in the catalogue. A catalogue failure deliberately
32+// leaves the durable publication as a same-incarnation forward orphan.
33+func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, publication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
34+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
35+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
36+ }
37+ unlock := c.locks.Lock([]string{name})
38+ defer unlock()
39+
40+ record, ok := c.catalogue.Record(name)
41+ if !ok {
42+ return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
43+ }
44+ if publication.Name != name || publication.Name != record.Name || publication.IncarnationID != record.IncarnationID ||
45+ publication.Generation == 0 || len(publication.Manifest) == 0 ||
46+ !equalCheckpointRefs(publication.ParentCheckpoint, record.Committed) {
47+ return domain.CatalogueRecord{}, ErrCheckpointConflict
48+ }
49+ if err := c.repository.Publish(ctx, publication); err != nil {
50+ return domain.CatalogueRecord{}, err
51+ }
52+
53+ ref := domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
54+ next := shiftedCheckpoint(record, ref)
55+ if err := next.Validate(); err != nil {
56+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid checkpoint transition: %w", err)
57+ }
58+ if err := c.catalogue.Replace(name, next); err != nil {
59+ return domain.CatalogueRecord{}, err
60+ }
61+ return next, nil
62+}
63+
64+// PromoteFallback validates a directly indexed fallback and any alternatives,
65+// commits the bounded validated set, and repairs HEAD only after the catalogue
66+// commit is durable.
67+func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref domain.CheckpointRef) (domain.CatalogueRecord, error) {
68+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
69+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
70+ }
71+ unlock := c.locks.Lock([]string{name})
72+ defer unlock()
73+
74+ record, ok := c.catalogue.Record(name)
75+ if !ok {
76+ return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
77+ }
78+ found := false
79+ for _, fallback := range record.Fallbacks {
80+ if fallback != nil && *fallback == ref {
81+ found = true
82+ break
83+ }
84+ }
85+ if !found {
86+ return domain.CatalogueRecord{}, ErrFallbackNotFound
87+ }
88+ if err := c.validateCheckpoint(ctx, record, ref); err != nil {
89+ return domain.CatalogueRecord{}, err
90+ }
91+
92+ alternatives := make([]domain.CheckpointRef, 0, 2)
93+ candidates := make([]*domain.CheckpointRef, 0, 3)
94+ candidates = append(candidates, record.Committed)
95+ candidates = append(candidates, record.Fallbacks[:]...)
96+ for _, candidate := range candidates {
97+ if candidate == nil || *candidate == ref || candidate.Generation >= ref.Generation || len(alternatives) == 2 {
98+ continue
99+ }
100+ if err := c.validateCheckpoint(ctx, record, *candidate); err == nil {
101+ alternatives = append(alternatives, *candidate)
102+ }
103+ }
104+
105+ next := record
106+ next.RecoveryState = domain.RecoveryHealthy
107+ next.Committed = copyCheckpointRef(&ref)
108+ next.Fallbacks = [2]*domain.CheckpointRef{}
109+ next.DegradedReason = ""
110+ for i := range alternatives {
111+ next.Fallbacks[i] = copyCheckpointRef(&alternatives[i])
112+ }
113+ if err := next.Validate(); err != nil {
114+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fallback promotion: %w", err)
115+ }
116+ if err := c.catalogue.Replace(name, next); err != nil {
117+ return domain.CatalogueRecord{}, err
118+ }
119+ if err := c.repository.RepairHEAD(ctx, record.IncarnationID, ref); err != nil {
120+ return next, err
121+ }
122+ return next, nil
123+}
124+
125+func (c *Coordinator) validateCheckpoint(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef) error {
126+ generation, err := c.repository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
127+ if err != nil {
128+ return err
129+ }
130+ if generation.IncarnationID != record.IncarnationID || generation.Name != record.Name || generation.Generation != ref.Generation ||
131+ sha256.Sum256(generation.Manifest) != ref.ManifestDigest {
132+ return ErrCheckpointConflict
133+ }
134+ return nil
135+}
136+
137+func equalCheckpointRefs(a, b *domain.CheckpointRef) bool {
138+ if a == nil || b == nil {
139+ return a == nil && b == nil
140+ }
141+ return *a == *b
142+}
143+
144+func copyCheckpointRef(ref *domain.CheckpointRef) *domain.CheckpointRef {
145+ if ref == nil {
146+ return nil
147+ }
148+ copy := *ref
149+ return &copy
150+}
A internal/usecase/recovery/checkpoint_test.go +199 -0
@@ -0,0 +1,199 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "errors"
7+ "sync"
8+ "testing"
9+
10+ "github.com/bnema/vev/internal/domain"
11+ "github.com/bnema/vev/internal/ports"
12+ "github.com/stretchr/testify/require"
13+)
14+
15+type checkpointCatalogue struct {
16+ mu sync.Mutex
17+ record domain.CatalogueRecord
18+ replaceErr error
19+ replaces int
20+ events *[]string
21+}
22+
23+func (c *checkpointCatalogue) Records() []domain.CatalogueRecord {
24+ return []domain.CatalogueRecord{c.record}
25+}
26+func (c *checkpointCatalogue) Record(name string) (domain.CatalogueRecord, bool) {
27+ c.mu.Lock()
28+ defer c.mu.Unlock()
29+ return c.record, c.record.Name == name
30+}
31+func (c *checkpointCatalogue) Create(domain.CatalogueRecord) error { return nil }
32+func (c *checkpointCatalogue) Replace(name string, record domain.CatalogueRecord) error {
33+ c.mu.Lock()
34+ defer c.mu.Unlock()
35+ c.replaces++
36+ if c.events != nil {
37+ *c.events = append(*c.events, "catalogue")
38+ }
39+ if c.replaceErr != nil {
40+ return c.replaceErr
41+ }
42+ if name != c.record.Name || record.Name != name {
43+ return errors.New("catalogue key mismatch")
44+ }
45+ c.record = record
46+ return nil
47+}
48+func (c *checkpointCatalogue) Rename(string, domain.CatalogueRecord) error { return nil }
49+func (c *checkpointCatalogue) Delete(string) error { return nil }
50+func (c *checkpointCatalogue) Close() error { return nil }
51+
52+type checkpointRepository struct {
53+ ports.SnapshotRepository
54+ publications []ports.SnapshotPublication
55+ generations map[uint64]ports.SnapshotGeneration
56+ loadErrors map[uint64]error
57+ loads []uint64
58+ repairs []domain.CheckpointRef
59+ events *[]string
60+}
61+
62+func (r *checkpointRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
63+ if r.events != nil {
64+ *r.events = append(*r.events, "repository")
65+ }
66+ r.publications = append(r.publications, publication)
67+ return nil
68+}
69+
70+func (r *checkpointRepository) LoadCheckpoint(_ context.Context, _ domain.IncarnationID, _ string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
71+ r.loads = append(r.loads, ref.Generation)
72+ if err := r.loadErrors[ref.Generation]; err != nil {
73+ return ports.SnapshotGeneration{}, err
74+ }
75+ generation, ok := r.generations[ref.Generation]
76+ if !ok {
77+ return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
78+ }
79+ return generation, nil
80+}
81+
82+func (r *checkpointRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
83+ r.repairs = append(r.repairs, ref)
84+ return nil
85+}
86+
87+func checkpointRecord() domain.CatalogueRecord {
88+ return domain.CatalogueRecord{
89+ Name: "work",
90+ IncarnationID: domain.IncarnationID{1},
91+ RecoveryState: domain.RecoveryFresh,
92+ }
93+}
94+
95+func checkpointPublication(record domain.CatalogueRecord, generation uint64) ports.SnapshotPublication {
96+ return ports.SnapshotPublication{
97+ IncarnationID: record.IncarnationID,
98+ Name: record.Name,
99+ Generation: generation,
100+ ParentCheckpoint: record.Committed,
101+ Manifest: []byte{byte(generation), 0xa5},
102+ }
103+}
104+
105+func populatedGenerations(refs [2]*domain.CheckpointRef) []uint64 {
106+ var generations []uint64
107+ for _, ref := range refs {
108+ if ref != nil {
109+ generations = append(generations, ref.Generation)
110+ }
111+ }
112+ return generations
113+}
114+
115+func runCheckpointCommit(t *testing.T, count int) (domain.CatalogueRecord, *checkpointCatalogue, *checkpointRepository) {
116+ t.Helper()
117+ events := make([]string, 0, count*2)
118+ catalogue := &checkpointCatalogue{record: checkpointRecord(), events: &events}
119+ repository := &checkpointRepository{events: &events}
120+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
121+ for generation := uint64(1); generation <= uint64(count); generation++ {
122+ var err error
123+ catalogue.record, err = coordinator.PublishCheckpoint(context.Background(), "work", checkpointPublication(catalogue.record, generation))
124+ require.NoError(t, err)
125+ }
126+ for i := 0; i < count; i++ {
127+ require.Equal(t, []string{"repository", "catalogue"}, events[i*2:i*2+2])
128+ }
129+ return catalogue.record, catalogue, repository
130+}
131+
132+func TestCheckpointCommitFirst(t *testing.T) {
133+ record, catalogue, _ := runCheckpointCommit(t, 1)
134+ require.Equal(t, uint64(1), record.Committed.Generation)
135+ require.Empty(t, populatedGenerations(record.Fallbacks))
136+ require.Equal(t, 1, catalogue.replaces)
137+}
138+
139+func TestCheckpointCommitSecond(t *testing.T) {
140+ record, catalogue, _ := runCheckpointCommit(t, 2)
141+ require.Equal(t, uint64(2), record.Committed.Generation)
142+ require.Equal(t, []uint64{1}, populatedGenerations(record.Fallbacks))
143+ require.Equal(t, 2, catalogue.replaces)
144+}
145+
146+func TestCheckpointCommitThird(t *testing.T) {
147+ record, catalogue, _ := runCheckpointCommit(t, 4)
148+ require.Equal(t, uint64(4), record.Committed.Generation)
149+ require.Equal(t, []uint64{3, 2}, populatedGenerations(record.Fallbacks))
150+ require.Equal(t, domain.RecoveryHealthy, record.RecoveryState)
151+ require.Empty(t, record.DegradedReason)
152+ require.Equal(t, 4, catalogue.replaces)
153+}
154+
155+func TestCheckpointCommitThirdPromotesOnlyValidatedDirectFallbacks(t *testing.T) {
156+ refs := []domain.CheckpointRef{
157+ {Generation: 3, ManifestDigest: sha256.Sum256([]byte("third"))},
158+ {Generation: 2, ManifestDigest: sha256.Sum256([]byte("second"))},
159+ {Generation: 1, ManifestDigest: sha256.Sum256([]byte("first"))},
160+ }
161+ record := checkpointRecord()
162+ record.RecoveryState = domain.RecoveryHealthy
163+ record.Committed = &refs[0]
164+ record.Fallbacks = [2]*domain.CheckpointRef{&refs[1], &refs[2]}
165+ catalogue := &checkpointCatalogue{record: record}
166+ repository := &checkpointRepository{
167+ generations: map[uint64]ports.SnapshotGeneration{
168+ 2: {IncarnationID: record.IncarnationID, Name: record.Name, Generation: 2, Manifest: []byte("second")},
169+ 1: {IncarnationID: record.IncarnationID, Name: record.Name, Generation: 1, Manifest: []byte("first")},
170+ },
171+ }
172+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
173+
174+ promoted, err := coordinator.PromoteFallback(context.Background(), record.Name, refs[1])
175+ require.NoError(t, err)
176+ require.Equal(t, refs[1], *promoted.Committed)
177+ require.Equal(t, []uint64{1}, populatedGenerations(promoted.Fallbacks))
178+ require.Equal(t, []uint64{2, 1}, repository.loads, "only the selected and older catalogue-indexed fallbacks are read directly")
179+ require.Equal(t, []domain.CheckpointRef{refs[1]}, repository.repairs)
180+ require.Equal(t, 1, catalogue.replaces)
181+}
182+
183+func TestCheckpointCommitPublishOrphan(t *testing.T) {
184+ prior := checkpointRecord()
185+ priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
186+ prior.RecoveryState = domain.RecoveryHealthy
187+ prior.Committed = &priorRef
188+ cause := errors.New("catalogue sync failed")
189+ catalogue := &checkpointCatalogue{record: prior, replaceErr: cause}
190+ repository := &checkpointRepository{}
191+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
192+ publication := checkpointPublication(prior, 8)
193+
194+ _, err := coordinator.PublishCheckpoint(context.Background(), prior.Name, publication)
195+ require.ErrorIs(t, err, cause)
196+ require.Equal(t, prior, catalogue.record, "failed catalogue commit must preserve the authoritative record")
197+ require.Len(t, repository.publications, 1, "the durable publication remains a forward orphan")
198+ require.Equal(t, publication, repository.publications[0])
199+}
@@ -0,0 +1,199 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "errors"
7+ "sync"
8+ "testing"
9+
10+ "github.com/bnema/vev/internal/domain"
11+ "github.com/bnema/vev/internal/ports"
12+ "github.com/stretchr/testify/require"
13+)
14+
15+type checkpointCatalogue struct {
16+ mu sync.Mutex
17+ record domain.CatalogueRecord
18+ replaceErr error
19+ replaces int
20+ events *[]string
21+}
22+
23+func (c *checkpointCatalogue) Records() []domain.CatalogueRecord {
24+ return []domain.CatalogueRecord{c.record}
25+}
26+func (c *checkpointCatalogue) Record(name string) (domain.CatalogueRecord, bool) {
27+ c.mu.Lock()
28+ defer c.mu.Unlock()
29+ return c.record, c.record.Name == name
30+}
31+func (c *checkpointCatalogue) Create(domain.CatalogueRecord) error { return nil }
32+func (c *checkpointCatalogue) Replace(name string, record domain.CatalogueRecord) error {
33+ c.mu.Lock()
34+ defer c.mu.Unlock()
35+ c.replaces++
36+ if c.events != nil {
37+ *c.events = append(*c.events, "catalogue")
38+ }
39+ if c.replaceErr != nil {
40+ return c.replaceErr
41+ }
42+ if name != c.record.Name || record.Name != name {
43+ return errors.New("catalogue key mismatch")
44+ }
45+ c.record = record
46+ return nil
47+}
48+func (c *checkpointCatalogue) Rename(string, domain.CatalogueRecord) error { return nil }
49+func (c *checkpointCatalogue) Delete(string) error { return nil }
50+func (c *checkpointCatalogue) Close() error { return nil }
51+
52+type checkpointRepository struct {
53+ ports.SnapshotRepository
54+ publications []ports.SnapshotPublication
55+ generations map[uint64]ports.SnapshotGeneration
56+ loadErrors map[uint64]error
57+ loads []uint64
58+ repairs []domain.CheckpointRef
59+ events *[]string
60+}
61+
62+func (r *checkpointRepository) Publish(_ context.Context, publication ports.SnapshotPublication) error {
63+ if r.events != nil {
64+ *r.events = append(*r.events, "repository")
65+ }
66+ r.publications = append(r.publications, publication)
67+ return nil
68+}
69+
70+func (r *checkpointRepository) LoadCheckpoint(_ context.Context, _ domain.IncarnationID, _ string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
71+ r.loads = append(r.loads, ref.Generation)
72+ if err := r.loadErrors[ref.Generation]; err != nil {
73+ return ports.SnapshotGeneration{}, err
74+ }
75+ generation, ok := r.generations[ref.Generation]
76+ if !ok {
77+ return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
78+ }
79+ return generation, nil
80+}
81+
82+func (r *checkpointRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
83+ r.repairs = append(r.repairs, ref)
84+ return nil
85+}
86+
87+func checkpointRecord() domain.CatalogueRecord {
88+ return domain.CatalogueRecord{
89+ Name: "work",
90+ IncarnationID: domain.IncarnationID{1},
91+ RecoveryState: domain.RecoveryFresh,
92+ }
93+}
94+
95+func checkpointPublication(record domain.CatalogueRecord, generation uint64) ports.SnapshotPublication {
96+ return ports.SnapshotPublication{
97+ IncarnationID: record.IncarnationID,
98+ Name: record.Name,
99+ Generation: generation,
100+ ParentCheckpoint: record.Committed,
101+ Manifest: []byte{byte(generation), 0xa5},
102+ }
103+}
104+
105+func populatedGenerations(refs [2]*domain.CheckpointRef) []uint64 {
106+ var generations []uint64
107+ for _, ref := range refs {
108+ if ref != nil {
109+ generations = append(generations, ref.Generation)
110+ }
111+ }
112+ return generations
113+}
114+
115+func runCheckpointCommit(t *testing.T, count int) (domain.CatalogueRecord, *checkpointCatalogue, *checkpointRepository) {
116+ t.Helper()
117+ events := make([]string, 0, count*2)
118+ catalogue := &checkpointCatalogue{record: checkpointRecord(), events: &events}
119+ repository := &checkpointRepository{events: &events}
120+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
121+ for generation := uint64(1); generation <= uint64(count); generation++ {
122+ var err error
123+ catalogue.record, err = coordinator.PublishCheckpoint(context.Background(), "work", checkpointPublication(catalogue.record, generation))
124+ require.NoError(t, err)
125+ }
126+ for i := 0; i < count; i++ {
127+ require.Equal(t, []string{"repository", "catalogue"}, events[i*2:i*2+2])
128+ }
129+ return catalogue.record, catalogue, repository
130+}
131+
132+func TestCheckpointCommitFirst(t *testing.T) {
133+ record, catalogue, _ := runCheckpointCommit(t, 1)
134+ require.Equal(t, uint64(1), record.Committed.Generation)
135+ require.Empty(t, populatedGenerations(record.Fallbacks))
136+ require.Equal(t, 1, catalogue.replaces)
137+}
138+
139+func TestCheckpointCommitSecond(t *testing.T) {
140+ record, catalogue, _ := runCheckpointCommit(t, 2)
141+ require.Equal(t, uint64(2), record.Committed.Generation)
142+ require.Equal(t, []uint64{1}, populatedGenerations(record.Fallbacks))
143+ require.Equal(t, 2, catalogue.replaces)
144+}
145+
146+func TestCheckpointCommitThird(t *testing.T) {
147+ record, catalogue, _ := runCheckpointCommit(t, 4)
148+ require.Equal(t, uint64(4), record.Committed.Generation)
149+ require.Equal(t, []uint64{3, 2}, populatedGenerations(record.Fallbacks))
150+ require.Equal(t, domain.RecoveryHealthy, record.RecoveryState)
151+ require.Empty(t, record.DegradedReason)
152+ require.Equal(t, 4, catalogue.replaces)
153+}
154+
155+func TestCheckpointCommitThirdPromotesOnlyValidatedDirectFallbacks(t *testing.T) {
156+ refs := []domain.CheckpointRef{
157+ {Generation: 3, ManifestDigest: sha256.Sum256([]byte("third"))},
158+ {Generation: 2, ManifestDigest: sha256.Sum256([]byte("second"))},
159+ {Generation: 1, ManifestDigest: sha256.Sum256([]byte("first"))},
160+ }
161+ record := checkpointRecord()
162+ record.RecoveryState = domain.RecoveryHealthy
163+ record.Committed = &refs[0]
164+ record.Fallbacks = [2]*domain.CheckpointRef{&refs[1], &refs[2]}
165+ catalogue := &checkpointCatalogue{record: record}
166+ repository := &checkpointRepository{
167+ generations: map[uint64]ports.SnapshotGeneration{
168+ 2: {IncarnationID: record.IncarnationID, Name: record.Name, Generation: 2, Manifest: []byte("second")},
169+ 1: {IncarnationID: record.IncarnationID, Name: record.Name, Generation: 1, Manifest: []byte("first")},
170+ },
171+ }
172+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
173+
174+ promoted, err := coordinator.PromoteFallback(context.Background(), record.Name, refs[1])
175+ require.NoError(t, err)
176+ require.Equal(t, refs[1], *promoted.Committed)
177+ require.Equal(t, []uint64{1}, populatedGenerations(promoted.Fallbacks))
178+ require.Equal(t, []uint64{2, 1}, repository.loads, "only the selected and older catalogue-indexed fallbacks are read directly")
179+ require.Equal(t, []domain.CheckpointRef{refs[1]}, repository.repairs)
180+ require.Equal(t, 1, catalogue.replaces)
181+}
182+
183+func TestCheckpointCommitPublishOrphan(t *testing.T) {
184+ prior := checkpointRecord()
185+ priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}
186+ prior.RecoveryState = domain.RecoveryHealthy
187+ prior.Committed = &priorRef
188+ cause := errors.New("catalogue sync failed")
189+ catalogue := &checkpointCatalogue{record: prior, replaceErr: cause}
190+ repository := &checkpointRepository{}
191+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
192+ publication := checkpointPublication(prior, 8)
193+
194+ _, err := coordinator.PublishCheckpoint(context.Background(), prior.Name, publication)
195+ require.ErrorIs(t, err, cause)
196+ require.Equal(t, prior, catalogue.record, "failed catalogue commit must preserve the authoritative record")
197+ require.Len(t, repository.publications, 1, "the durable publication remains a forward orphan")
198+ require.Equal(t, publication, repository.publications[0])
199+}
M internal/usecase/recovery/coordinator.go +71 -1
@@ -6,6 +6,8 @@ import (
66 "context"
77 "errors"
88 "io"
9+ "sort"
10+ "sync"
911
1012 "github.com/bnema/vev/internal/ports"
1113 )
@@ -14,15 +16,83 @@ var ErrPendingRecoveryUnsupported = errors.New("pending recovery intent requires
1416
1517 var recoveryListingBudget = ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10}
1618
19+type KeyLocks struct {
20+ mu sync.Mutex
21+ refs map[string]*keyLockRef
22+}
23+
24+type keyLockRef struct {
25+ mu sync.Mutex
26+ refs int
27+}
28+
29+func NewKeyLocks() *KeyLocks {
30+ return &KeyLocks{refs: make(map[string]*keyLockRef)}
31+}
32+
33+// Lock acquires each distinct stable key in lexical order. The returned
34+// function releases in reverse order and removes unused lock entries.
35+func (k *KeyLocks) Lock(names []string) func() {
36+ if k == nil {
37+ return func() {}
38+ }
39+ ordered := append([]string(nil), names...)
40+ sort.Strings(ordered)
41+ ordered = compactStrings(ordered)
42+
43+ refs := make([]*keyLockRef, len(ordered))
44+ k.mu.Lock()
45+ for i, name := range ordered {
46+ ref := k.refs[name]
47+ if ref == nil {
48+ ref = &keyLockRef{}
49+ k.refs[name] = ref
50+ }
51+ ref.refs++
52+ refs[i] = ref
53+ }
54+ k.mu.Unlock()
55+ for _, ref := range refs {
56+ ref.mu.Lock()
57+ }
58+ return func() {
59+ for i := len(refs) - 1; i >= 0; i-- {
60+ refs[i].mu.Unlock()
61+ }
62+ k.mu.Lock()
63+ for i, name := range ordered {
64+ refs[i].refs--
65+ if refs[i].refs == 0 {
66+ delete(k.refs, name)
67+ }
68+ }
69+ k.mu.Unlock()
70+ }
71+}
72+
73+func compactStrings(values []string) []string {
74+ if len(values) == 0 {
75+ return values
76+ }
77+ out := values[:1]
78+ for _, value := range values[1:] {
79+ if value != out[len(out)-1] {
80+ out = append(out, value)
81+ }
82+ }
83+ return out
84+}
85+
1786 type Coordinator struct {
1887 catalogue ports.Catalogue
1988 repository ports.SnapshotRepository
2089 journal ports.RecoveryJournal
90+ locks *KeyLocks
2191 random io.Reader
2292 }
2393
2494 func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, journal ports.RecoveryJournal, random io.Reader) *Coordinator {
25- return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, random: random}
95+ return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, locks: NewKeyLocks(), random: random}
2696 }
2797
2898 // Recover is deliberately fail-closed until the transaction roll-forward
@@ -6,6 +6,8 @@ import (
6 "context"6 "context"
7 "errors"7 "errors"
8 "io"8 "io"
9+ "sort"
10+ "sync"
9 11
10 "github.com/bnema/vev/internal/ports"12 "github.com/bnema/vev/internal/ports"
11 )13 )
@@ -14,15 +16,83 @@ var ErrPendingRecoveryUnsupported = errors.New("pending recovery intent requires
14 16
15 var recoveryListingBudget = ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10}17 var recoveryListingBudget = ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10}
16 18
19+type KeyLocks struct {
20+ mu sync.Mutex
21+ refs map[string]*keyLockRef
22+}
23+
24+type keyLockRef struct {
25+ mu sync.Mutex
26+ refs int
27+}
28+
29+func NewKeyLocks() *KeyLocks {
30+ return &KeyLocks{refs: make(map[string]*keyLockRef)}
31+}
32+
33+// Lock acquires each distinct stable key in lexical order. The returned
34+// function releases in reverse order and removes unused lock entries.
35+func (k *KeyLocks) Lock(names []string) func() {
36+ if k == nil {
37+ return func() {}
38+ }
39+ ordered := append([]string(nil), names...)
40+ sort.Strings(ordered)
41+ ordered = compactStrings(ordered)
42+
43+ refs := make([]*keyLockRef, len(ordered))
44+ k.mu.Lock()
45+ for i, name := range ordered {
46+ ref := k.refs[name]
47+ if ref == nil {
48+ ref = &keyLockRef{}
49+ k.refs[name] = ref
50+ }
51+ ref.refs++
52+ refs[i] = ref
53+ }
54+ k.mu.Unlock()
55+ for _, ref := range refs {
56+ ref.mu.Lock()
57+ }
58+ return func() {
59+ for i := len(refs) - 1; i >= 0; i-- {
60+ refs[i].mu.Unlock()
61+ }
62+ k.mu.Lock()
63+ for i, name := range ordered {
64+ refs[i].refs--
65+ if refs[i].refs == 0 {
66+ delete(k.refs, name)
67+ }
68+ }
69+ k.mu.Unlock()
70+ }
71+}
72+
73+func compactStrings(values []string) []string {
74+ if len(values) == 0 {
75+ return values
76+ }
77+ out := values[:1]
78+ for _, value := range values[1:] {
79+ if value != out[len(out)-1] {
80+ out = append(out, value)
81+ }
82+ }
83+ return out
84+}
85+
17 type Coordinator struct {86 type Coordinator struct {
18 catalogue ports.Catalogue87 catalogue ports.Catalogue
19 repository ports.SnapshotRepository88 repository ports.SnapshotRepository
20 journal ports.RecoveryJournal89 journal ports.RecoveryJournal
90+ locks *KeyLocks
21 random io.Reader91 random io.Reader
22 }92 }
23 93
24 func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, journal ports.RecoveryJournal, random io.Reader) *Coordinator {94 func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotRepository, journal ports.RecoveryJournal, random io.Reader) *Coordinator {
25- return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, random: random}95+ return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, locks: NewKeyLocks(), random: random}
26 }96 }
27 97
28 // Recover is deliberately fail-closed until the transaction roll-forward98 // Recover is deliberately fail-closed until the transaction roll-forward

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