detached a6f1d55

fix(app): report fenced recoveries and serve offline kill from recoverable state

fenced recovery conflicts are now logged and persisted as notices so a fenced startup cannot read as clean. runOfflineNamedKill's os.Stat precheck denied a session whenever sessions.kv was absent even with a recoverable .prev/.next.
commit
a6f1d55b9d5101057c3fc72d03a5f4fd4e4e537e
author
brice · · 2026-07-26 04:55 UTC
parent
808c829
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +359 -9
Split
UnifiedSplit
M internal/app/daemon_integration_test.go +155 -0
@@ -659,6 +659,161 @@ func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
659659 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
660660 }
661661
662+// Startup restoration repairs HEADs, promotes fallbacks, and replaces catalogue
663+// records. Lifecycle ownership must therefore outlive it: releasing the flock
664+// while a restoration repository call is still running would let a second
665+// daemon mutate the same durable state.
666+func TestLifecycleOwnershipOutlivesRestorationWriter(t *testing.T) {
667+ stateDir := filepath.Join(t.TempDir(), "state")
668+ repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
669+ checkpointed := publishRestorableCheckpoint(t, stateDir, repository)
670+
671+ runtimeDir := filepath.Join(t.TempDir(), "runtime")
672+ blocking := &lifecycleBlockingRestore{
673+ SnapshotRepository: repository,
674+ entered: make(chan struct{}),
675+ release: make(chan struct{}),
676+ returned: make(chan struct{}),
677+ }
678+ var releaseOnce sync.Once
679+ releaseRestore := func() { releaseOnce.Do(func() { close(blocking.release) }) }
680+ t.Cleanup(releaseRestore)
681+
682+ shutdownClock := newLifecycleShutdownClock()
683+ ctx, cancel := context.WithCancel(context.Background())
684+ t.Cleanup(cancel)
685+ listenerReady := make(chan struct{})
686+ callbackReturned := make(chan struct{})
687+ wrapperReturned := make(chan error, 1)
688+ go func() {
689+ wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
690+ defer close(callbackReturned)
691+ opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
692+ StateDir: stateDir,
693+ Random: rand.Reader,
694+ SnapshotMigration: repository,
695+ })
696+ if err != nil {
697+ return err
698+ }
699+ listener, err := ipc.Listen(runtimeDir)
700+ if err != nil {
701+ return errors.Join(err, opened.Catalogue.Close())
702+ }
703+ close(listenerReady)
704+ d := daemon.New(
705+ pty.NewFactory(),
706+ shutdownClock,
707+ discardLog(),
708+ daemon.WithShell("/bin/cat", nil),
709+ daemon.WithCatalogue(opened.Catalogue, opened.Records),
710+ daemon.WithSnapshotRepository(blocking, nil),
711+ )
712+ return d.Serve(ctx, listener)
713+ })
714+ }()
715+
716+ awaitLifecycleStage(t, listenerReady, "daemon listener")
717+ awaitLifecycleStage(t, blocking.entered, "restoration repository call")
718+ require.Equal(t, checkpointed, blocking.repairedName(), "restoration must target the checkpointed session")
719+
720+ cancel()
721+ shutdownClock.nextTimer(t).fire()
722+ select {
723+ case <-callbackReturned:
724+ t.Fatal("Serve callback returned after shutdown deadline while restoration was blocked")
725+ case <-time.After(100 * time.Millisecond):
726+ }
727+ assertLifecycleStagePending(t, callbackReturned, "Serve callback return while restoration is blocked")
728+ assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while restoration is blocked")
729+ _, err := lifecycle.TryAcquire(runtimeDir)
730+ require.ErrorIs(t, err, lifecycle.ErrBusy)
731+
732+ releaseRestore()
733+ awaitLifecycleStage(t, blocking.returned, "restoration repository return")
734+ awaitLifecycleStage(t, callbackReturned, "Serve callback return after restoration")
735+ require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
736+ owner, err := lifecycle.TryAcquire(runtimeDir)
737+ require.NoError(t, err)
738+ require.NoError(t, owner.Release())
739+}
740+
741+// publishRestorableCheckpoint runs a complete daemon over the real catalogue and
742+// repository until one named session owns a committed checkpoint, then shuts it
743+// down. The returned name is restorable by any later daemon on the same state.
744+func publishRestorableCheckpoint(t *testing.T, stateDir string, repository *snapshot.Repository) string {
745+ t.Helper()
746+ const name = "restorable"
747+ runtimeDir := filepath.Join(t.TempDir(), "seed-runtime")
748+ ctx, cancel := context.WithCancel(context.Background())
749+ defer cancel()
750+
751+ opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
752+ StateDir: stateDir,
753+ Random: rand.Reader,
754+ SnapshotMigration: repository,
755+ })
756+ require.NoError(t, err)
757+ coordinator := recovery.NewCoordinator(opened.Catalogue, repository, recoveryfs.New(stateDir), rand.Reader)
758+ require.NoError(t, coordinator.Recover(ctx))
759+ listener, err := ipc.Listen(runtimeDir)
760+ require.NoError(t, err)
761+ d := daemon.New(
762+ pty.NewFactory(),
763+ clock.New(),
764+ discardLog(),
765+ daemon.WithShell("/bin/cat", nil),
766+ daemon.WithCatalogue(opened.Catalogue, opened.Records),
767+ daemon.WithSnapshotRepository(repository, nil),
768+ daemon.WithRecoveryCoordinator(coordinator),
769+ )
770+ served := make(chan error, 1)
771+ go func() { served <- d.Serve(ctx, listener) }()
772+
773+ tr, _ := attach(t, runtimeDir, ports.IntentNew, name, domain.Size{Cols: 80, Rows: 24})
774+ require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("checkpoint me\n")})}))
775+ require.Eventually(t, func() bool {
776+ record, ok := opened.Catalogue.Record(name)
777+ return ok && record.RecoveryState == domain.RecoveryHealthy && record.Committed != nil
778+ }, 5*time.Second, 10*time.Millisecond, "session never committed a checkpoint")
779+ require.NoError(t, tr.Close())
780+
781+ cancel()
782+ require.NoError(t, awaitLifecycleResult(t, served, "seed daemon shutdown"))
783+ return name
784+}
785+
786+type lifecycleBlockingRestore struct {
787+ ports.SnapshotRepository
788+ entered chan struct{}
789+ release chan struct{}
790+ returned chan struct{}
791+ enterOnce sync.Once
792+ returnOnce sync.Once
793+ mu sync.Mutex
794+ repaired string
795+}
796+
797+func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
798+ r.mu.Lock()
799+ r.repaired = name
800+ r.mu.Unlock()
801+ return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
802+}
803+
804+func (r *lifecycleBlockingRestore) RepairHEAD(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
805+ r.enterOnce.Do(func() { close(r.entered) })
806+ <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
807+ r.returnOnce.Do(func() { close(r.returned) })
808+ return r.SnapshotRepository.RepairHEAD(ctx, id, ref)
809+}
810+
811+func (r *lifecycleBlockingRestore) repairedName() string {
812+ r.mu.Lock()
813+ defer r.mu.Unlock()
814+ return r.repaired
815+}
816+
662817 type blockedSnapshotWriterShutdown struct {
663818 runtimeDir string
664819 callbackReturned <-chan struct{}
@@ -659,6 +659,161 @@ func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
659 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))659 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
660 }660 }
661 661
662+// Startup restoration repairs HEADs, promotes fallbacks, and replaces catalogue
663+// records. Lifecycle ownership must therefore outlive it: releasing the flock
664+// while a restoration repository call is still running would let a second
665+// daemon mutate the same durable state.
666+func TestLifecycleOwnershipOutlivesRestorationWriter(t *testing.T) {
667+ stateDir := filepath.Join(t.TempDir(), "state")
668+ repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
669+ checkpointed := publishRestorableCheckpoint(t, stateDir, repository)
670+
671+ runtimeDir := filepath.Join(t.TempDir(), "runtime")
672+ blocking := &lifecycleBlockingRestore{
673+ SnapshotRepository: repository,
674+ entered: make(chan struct{}),
675+ release: make(chan struct{}),
676+ returned: make(chan struct{}),
677+ }
678+ var releaseOnce sync.Once
679+ releaseRestore := func() { releaseOnce.Do(func() { close(blocking.release) }) }
680+ t.Cleanup(releaseRestore)
681+
682+ shutdownClock := newLifecycleShutdownClock()
683+ ctx, cancel := context.WithCancel(context.Background())
684+ t.Cleanup(cancel)
685+ listenerReady := make(chan struct{})
686+ callbackReturned := make(chan struct{})
687+ wrapperReturned := make(chan error, 1)
688+ go func() {
689+ wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
690+ defer close(callbackReturned)
691+ opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
692+ StateDir: stateDir,
693+ Random: rand.Reader,
694+ SnapshotMigration: repository,
695+ })
696+ if err != nil {
697+ return err
698+ }
699+ listener, err := ipc.Listen(runtimeDir)
700+ if err != nil {
701+ return errors.Join(err, opened.Catalogue.Close())
702+ }
703+ close(listenerReady)
704+ d := daemon.New(
705+ pty.NewFactory(),
706+ shutdownClock,
707+ discardLog(),
708+ daemon.WithShell("/bin/cat", nil),
709+ daemon.WithCatalogue(opened.Catalogue, opened.Records),
710+ daemon.WithSnapshotRepository(blocking, nil),
711+ )
712+ return d.Serve(ctx, listener)
713+ })
714+ }()
715+
716+ awaitLifecycleStage(t, listenerReady, "daemon listener")
717+ awaitLifecycleStage(t, blocking.entered, "restoration repository call")
718+ require.Equal(t, checkpointed, blocking.repairedName(), "restoration must target the checkpointed session")
719+
720+ cancel()
721+ shutdownClock.nextTimer(t).fire()
722+ select {
723+ case <-callbackReturned:
724+ t.Fatal("Serve callback returned after shutdown deadline while restoration was blocked")
725+ case <-time.After(100 * time.Millisecond):
726+ }
727+ assertLifecycleStagePending(t, callbackReturned, "Serve callback return while restoration is blocked")
728+ assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while restoration is blocked")
729+ _, err := lifecycle.TryAcquire(runtimeDir)
730+ require.ErrorIs(t, err, lifecycle.ErrBusy)
731+
732+ releaseRestore()
733+ awaitLifecycleStage(t, blocking.returned, "restoration repository return")
734+ awaitLifecycleStage(t, callbackReturned, "Serve callback return after restoration")
735+ require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
736+ owner, err := lifecycle.TryAcquire(runtimeDir)
737+ require.NoError(t, err)
738+ require.NoError(t, owner.Release())
739+}
740+
741+// publishRestorableCheckpoint runs a complete daemon over the real catalogue and
742+// repository until one named session owns a committed checkpoint, then shuts it
743+// down. The returned name is restorable by any later daemon on the same state.
744+func publishRestorableCheckpoint(t *testing.T, stateDir string, repository *snapshot.Repository) string {
745+ t.Helper()
746+ const name = "restorable"
747+ runtimeDir := filepath.Join(t.TempDir(), "seed-runtime")
748+ ctx, cancel := context.WithCancel(context.Background())
749+ defer cancel()
750+
751+ opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
752+ StateDir: stateDir,
753+ Random: rand.Reader,
754+ SnapshotMigration: repository,
755+ })
756+ require.NoError(t, err)
757+ coordinator := recovery.NewCoordinator(opened.Catalogue, repository, recoveryfs.New(stateDir), rand.Reader)
758+ require.NoError(t, coordinator.Recover(ctx))
759+ listener, err := ipc.Listen(runtimeDir)
760+ require.NoError(t, err)
761+ d := daemon.New(
762+ pty.NewFactory(),
763+ clock.New(),
764+ discardLog(),
765+ daemon.WithShell("/bin/cat", nil),
766+ daemon.WithCatalogue(opened.Catalogue, opened.Records),
767+ daemon.WithSnapshotRepository(repository, nil),
768+ daemon.WithRecoveryCoordinator(coordinator),
769+ )
770+ served := make(chan error, 1)
771+ go func() { served <- d.Serve(ctx, listener) }()
772+
773+ tr, _ := attach(t, runtimeDir, ports.IntentNew, name, domain.Size{Cols: 80, Rows: 24})
774+ require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("checkpoint me\n")})}))
775+ require.Eventually(t, func() bool {
776+ record, ok := opened.Catalogue.Record(name)
777+ return ok && record.RecoveryState == domain.RecoveryHealthy && record.Committed != nil
778+ }, 5*time.Second, 10*time.Millisecond, "session never committed a checkpoint")
779+ require.NoError(t, tr.Close())
780+
781+ cancel()
782+ require.NoError(t, awaitLifecycleResult(t, served, "seed daemon shutdown"))
783+ return name
784+}
785+
786+type lifecycleBlockingRestore struct {
787+ ports.SnapshotRepository
788+ entered chan struct{}
789+ release chan struct{}
790+ returned chan struct{}
791+ enterOnce sync.Once
792+ returnOnce sync.Once
793+ mu sync.Mutex
794+ repaired string
795+}
796+
797+func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
798+ r.mu.Lock()
799+ r.repaired = name
800+ r.mu.Unlock()
801+ return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
802+}
803+
804+func (r *lifecycleBlockingRestore) RepairHEAD(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
805+ r.enterOnce.Do(func() { close(r.entered) })
806+ <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
807+ r.returnOnce.Do(func() { close(r.returned) })
808+ return r.SnapshotRepository.RepairHEAD(ctx, id, ref)
809+}
810+
811+func (r *lifecycleBlockingRestore) repairedName() string {
812+ r.mu.Lock()
813+ defer r.mu.Unlock()
814+ return r.repaired
815+}
816+
662 type blockedSnapshotWriterShutdown struct {817 type blockedSnapshotWriterShutdown struct {
663 runtimeDir string818 runtimeDir string
664 callbackReturned <-chan struct{}819 callbackReturned <-chan struct{}
M internal/app/offline_kill_retry_test.go +30 -0
@@ -31,3 +31,33 @@ func TestRunKillOfflineUsesIncarnationDeletionProtocol(t *testing.T) {
3131 require.NoFileExists(t, filepath.Join(source, "sentinel"))
3232 require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))
3333 }
34+
35+// TestRunKillOfflineFindsSessionWhenOnlyPrevCandidateExists covers the same
36+// bug class as persist.LoadCatalogueReadOnly's fresh-install regression: a
37+// crash mid-compaction can leave sessions.kv absent with only a .prev (or
38+// .next) candidate on disk. `vev kill <name>` must still find and delete the
39+// session rather than reporting "no such session" from a bare os.Stat
40+// precheck on the fixed sessions.kv path.
41+func TestRunKillOfflineFindsSessionWhenOnlyPrevCandidateExists(t *testing.T) {
42+ stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
43+ t.Setenv("XDG_STATE_HOME", stateRoot)
44+ t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
45+
46+ id := domain.IncarnationID{1}
47+ p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
48+ now := time.Now().UnixNano()
49+ require.NoError(t, p.Save(persist.Record{Name: "crashed", IncarnationID: id, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
50+ require.NoError(t, p.Close())
51+ source := filepath.Join(stateRoot, "vev", "snapshots", "sessions", id.String())
52+ require.NoError(t, os.MkdirAll(source, 0o700))
53+ require.NoError(t, os.WriteFile(filepath.Join(source, "sentinel"), []byte("durable"), 0o600))
54+
55+ // Simulate a crash mid-compaction: current renamed away to .prev, no
56+ // sessions.kv on disk at all.
57+ storePath := persist.StorePath(filepath.Join(stateRoot, "vev"))
58+ require.NoError(t, os.Rename(storePath, storePath+".prev"))
59+
60+ require.NoError(t, runKill(context.Background(), "crashed", false, false))
61+ require.NoFileExists(t, filepath.Join(source, "sentinel"))
62+ require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))
63+}
@@ -31,3 +31,33 @@ func TestRunKillOfflineUsesIncarnationDeletionProtocol(t *testing.T) {
31 require.NoFileExists(t, filepath.Join(source, "sentinel"))31 require.NoFileExists(t, filepath.Join(source, "sentinel"))
32 require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))32 require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))
33 }33 }
34+
35+// TestRunKillOfflineFindsSessionWhenOnlyPrevCandidateExists covers the same
36+// bug class as persist.LoadCatalogueReadOnly's fresh-install regression: a
37+// crash mid-compaction can leave sessions.kv absent with only a .prev (or
38+// .next) candidate on disk. `vev kill <name>` must still find and delete the
39+// session rather than reporting "no such session" from a bare os.Stat
40+// precheck on the fixed sessions.kv path.
41+func TestRunKillOfflineFindsSessionWhenOnlyPrevCandidateExists(t *testing.T) {
42+ stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
43+ t.Setenv("XDG_STATE_HOME", stateRoot)
44+ t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
45+
46+ id := domain.IncarnationID{1}
47+ p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
48+ now := time.Now().UnixNano()
49+ require.NoError(t, p.Save(persist.Record{Name: "crashed", IncarnationID: id, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
50+ require.NoError(t, p.Close())
51+ source := filepath.Join(stateRoot, "vev", "snapshots", "sessions", id.String())
52+ require.NoError(t, os.MkdirAll(source, 0o700))
53+ require.NoError(t, os.WriteFile(filepath.Join(source, "sentinel"), []byte("durable"), 0o600))
54+
55+ // Simulate a crash mid-compaction: current renamed away to .prev, no
56+ // sessions.kv on disk at all.
57+ storePath := persist.StorePath(filepath.Join(stateRoot, "vev"))
58+ require.NoError(t, os.Rename(storePath, storePath+".prev"))
59+
60+ require.NoError(t, runKill(context.Background(), "crashed", false, false))
61+ require.NoFileExists(t, filepath.Join(source, "sentinel"))
62+ require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))
63+}
M internal/app/run.go +97 -8
@@ -506,8 +506,15 @@ func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, re
506506 log.Info("catalogue_compaction_recovery_complete", "recovery", recoveryMode)
507507 }
508508
509-func logTransactionRecovery(log *slog.Logger, examined int) {
510- log.Info("interrupted_transaction_recovery_complete", "sessions_examined", examined)
509+// logTransactionRecovery reports the recovery outcome. A startup that fenced
510+// any session is never reported as clean: the daemon comes up, but one or more
511+// sessions were skipped and still need an operator decision.
512+func logTransactionRecovery(log *slog.Logger, examined, fenced int) {
513+ outcome := "clean"
514+ if fenced > 0 {
515+ outcome = "fenced"
516+ }
517+ log.Info("interrupted_transaction_recovery_complete", "sessions_examined", examined, "sessions_fenced", fenced, "outcome", outcome)
511518 }
512519
513520 type interruptedRecoveryIdentity struct {
@@ -564,6 +571,76 @@ func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []inte
564571 return nil
565572 }
566573
574+// fencedRecoveryIdentity is one session that recovery fenced instead of failing
575+// startup for. Only the session identity and a sentinel-derived reason code are
576+// retained: raw causes can embed paths or operator input.
577+type fencedRecoveryIdentity struct {
578+ name string
579+ id domain.IncarnationID
580+ kind string
581+ reasonCode string
582+}
583+
584+// fencedRecoveries resolves each conflict against the post-recovery catalogue
585+// so a fenced session is reported with the incarnation it actually holds now.
586+func fencedRecoveries(conflicts []recovery.RecoveryConflict, catalogue ports.Catalogue) []fencedRecoveryIdentity {
587+ fenced := make([]fencedRecoveryIdentity, 0, len(conflicts))
588+ for _, conflict := range conflicts {
589+ identity := fencedRecoveryIdentity{name: conflict.Session, kind: conflict.Kind, reasonCode: fencedRecoveryReasonCode(conflict.Err)}
590+ if catalogue != nil {
591+ if record, ok := catalogue.Record(conflict.Session); ok {
592+ identity.id = record.IncarnationID
593+ }
594+ }
595+ fenced = append(fenced, identity)
596+ }
597+ return fenced
598+}
599+
600+func fencedRecoveryReasonCode(err error) string {
601+ switch {
602+ case errors.Is(err, recovery.ErrDiscardConflict):
603+ return "discard-intent-conflict"
604+ case errors.Is(err, recovery.ErrDiscardIntentInvalid):
605+ return "discard-intent-invalid"
606+ case errors.Is(err, recovery.ErrDeletionTombstoneConflict):
607+ return "deletion-tombstone-conflict"
608+ case errors.Is(err, recovery.ErrDeletionTombstoneInvalid):
609+ return "deletion-tombstone-invalid"
610+ case errors.Is(err, recovery.ErrSessionRecordInvalid):
611+ return "catalogue-record-invalid"
612+ default:
613+ return "unclassified"
614+ }
615+}
616+
617+func logFencedRecoveries(log *slog.Logger, fenced []fencedRecoveryIdentity) {
618+ for _, item := range fenced {
619+ log.Warn("interrupted_transaction_fenced",
620+ "session", item.name,
621+ "incarnation", item.id.String(),
622+ "kind", item.kind,
623+ "reason_code", item.reasonCode,
624+ )
625+ }
626+}
627+
628+func persistFencedRecoveryNotices(store ports.NoticeStore, fenced []fencedRecoveryIdentity, now time.Time) error {
629+ for _, item := range fenced {
630+ if err := store.Append(domain.Notification{
631+ Code: domain.NoticeSnapshotRestore,
632+ Severity: domain.NoticeWarn,
633+ Message: "skipped a durable session transaction that conflicts with its session state",
634+ Details: "session=" + item.name + " incarnation=" + item.id.String() + " kind=" + item.kind + " reason_code=" + item.reasonCode,
635+ Time: now,
636+ SessionID: domain.SessionID(item.id.String()),
637+ }); err != nil {
638+ return fmt.Errorf("persist fenced transaction notice for session %q incarnation %s: %w", item.name, item.id.String(), err)
639+ }
640+ }
641+ return nil
642+}
643+
567644 func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
568645 healthy, fresh, degraded := 0, 0, 0
569646 for _, record := range records {
@@ -662,11 +739,20 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
662739 if err := coordinator.Recover(ctx); err != nil {
663740 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
664741 }
742+ // Recovery fences a session instead of failing startup when only that
743+ // session's durable state is self-inconsistent. The daemon then comes up
744+ // looking healthy, so every fenced session must be reported and persisted.
745+ fenced := fencedRecoveries(coordinator.Conflicts(), opened.Catalogue)
746+ logFencedRecoveries(log, fenced)
665747 if err := persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now()); err != nil {
666748 log.Error("transaction_recovery_notice_failed", "err", err)
667749 return errors.Join(fmt.Errorf("vev: persist durable transaction recovery notice: %w", err), opened.Catalogue.Close())
668750 }
669- logTransactionRecovery(log, len(pendingRecoveries))
751+ if err := persistFencedRecoveryNotices(noticeStore, fenced, clk.Now()); err != nil {
752+ log.Error("transaction_recovery_notice_failed", "reason_code", "fenced-notice-failed")
753+ return errors.Join(fmt.Errorf("vev: persist fenced transaction notice: %w", err), opened.Catalogue.Close())
754+ }
755+ logTransactionRecovery(log, len(pendingRecoveries), len(fenced))
670756 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
671757 unresolved := make([]domain.IncarnationID, 0, len(pendingRecoveries))
672758 for _, pending := range pendingRecoveries {
@@ -1438,11 +1524,14 @@ func printSessions(w io.Writer, sessions []ports.SessionInfo) {
14381524 }
14391525
14401526 func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {
1441- if _, err := os.Stat(persist.StorePath(platform.StateDir())); errors.Is(err, os.ErrNotExist) {
1442- return fmt.Errorf("vev: no such session: %s", name)
1443- } else if err != nil {
1444- return fmt.Errorf("vev: reading stored sessions: %w", err)
1445- }
1527+ // LoadReadOnly already distinguishes a genuinely absent catalogue (empty
1528+ // slice, nil error) from a real IO error, and recovers the same
1529+ // fixed-path crash states LoadCatalogueReadOnly does, so a stray .next or
1530+ // .prev left by a crash mid-compaction no longer wrongly denies a session
1531+ // that still exists. A separate os.Stat precheck here was both redundant
1532+ // with that recovery and a regression of the same bug class: it treated
1533+ // "no sessions.kv" as "no such session" even when a recoverable .next or
1534+ // .prev candidate was present.
14461535 records, err := persist.LoadReadOnly(platform.StateDir())
14471536 if err != nil {
14481537 return fmt.Errorf("vev: reading stored sessions: %w", err)
@@ -506,8 +506,15 @@ func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, re
506 log.Info("catalogue_compaction_recovery_complete", "recovery", recoveryMode)506 log.Info("catalogue_compaction_recovery_complete", "recovery", recoveryMode)
507 }507 }
508 508
509-func logTransactionRecovery(log *slog.Logger, examined int) {509+// logTransactionRecovery reports the recovery outcome. A startup that fenced
510- log.Info("interrupted_transaction_recovery_complete", "sessions_examined", examined)510+// any session is never reported as clean: the daemon comes up, but one or more
511+// sessions were skipped and still need an operator decision.
512+func logTransactionRecovery(log *slog.Logger, examined, fenced int) {
513+ outcome := "clean"
514+ if fenced > 0 {
515+ outcome = "fenced"
516+ }
517+ log.Info("interrupted_transaction_recovery_complete", "sessions_examined", examined, "sessions_fenced", fenced, "outcome", outcome)
511 }518 }
512 519
513 type interruptedRecoveryIdentity struct {520 type interruptedRecoveryIdentity struct {
@@ -564,6 +571,76 @@ func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []inte
564 return nil571 return nil
565 }572 }
566 573
574+// fencedRecoveryIdentity is one session that recovery fenced instead of failing
575+// startup for. Only the session identity and a sentinel-derived reason code are
576+// retained: raw causes can embed paths or operator input.
577+type fencedRecoveryIdentity struct {
578+ name string
579+ id domain.IncarnationID
580+ kind string
581+ reasonCode string
582+}
583+
584+// fencedRecoveries resolves each conflict against the post-recovery catalogue
585+// so a fenced session is reported with the incarnation it actually holds now.
586+func fencedRecoveries(conflicts []recovery.RecoveryConflict, catalogue ports.Catalogue) []fencedRecoveryIdentity {
587+ fenced := make([]fencedRecoveryIdentity, 0, len(conflicts))
588+ for _, conflict := range conflicts {
589+ identity := fencedRecoveryIdentity{name: conflict.Session, kind: conflict.Kind, reasonCode: fencedRecoveryReasonCode(conflict.Err)}
590+ if catalogue != nil {
591+ if record, ok := catalogue.Record(conflict.Session); ok {
592+ identity.id = record.IncarnationID
593+ }
594+ }
595+ fenced = append(fenced, identity)
596+ }
597+ return fenced
598+}
599+
600+func fencedRecoveryReasonCode(err error) string {
601+ switch {
602+ case errors.Is(err, recovery.ErrDiscardConflict):
603+ return "discard-intent-conflict"
604+ case errors.Is(err, recovery.ErrDiscardIntentInvalid):
605+ return "discard-intent-invalid"
606+ case errors.Is(err, recovery.ErrDeletionTombstoneConflict):
607+ return "deletion-tombstone-conflict"
608+ case errors.Is(err, recovery.ErrDeletionTombstoneInvalid):
609+ return "deletion-tombstone-invalid"
610+ case errors.Is(err, recovery.ErrSessionRecordInvalid):
611+ return "catalogue-record-invalid"
612+ default:
613+ return "unclassified"
614+ }
615+}
616+
617+func logFencedRecoveries(log *slog.Logger, fenced []fencedRecoveryIdentity) {
618+ for _, item := range fenced {
619+ log.Warn("interrupted_transaction_fenced",
620+ "session", item.name,
621+ "incarnation", item.id.String(),
622+ "kind", item.kind,
623+ "reason_code", item.reasonCode,
624+ )
625+ }
626+}
627+
628+func persistFencedRecoveryNotices(store ports.NoticeStore, fenced []fencedRecoveryIdentity, now time.Time) error {
629+ for _, item := range fenced {
630+ if err := store.Append(domain.Notification{
631+ Code: domain.NoticeSnapshotRestore,
632+ Severity: domain.NoticeWarn,
633+ Message: "skipped a durable session transaction that conflicts with its session state",
634+ Details: "session=" + item.name + " incarnation=" + item.id.String() + " kind=" + item.kind + " reason_code=" + item.reasonCode,
635+ Time: now,
636+ SessionID: domain.SessionID(item.id.String()),
637+ }); err != nil {
638+ return fmt.Errorf("persist fenced transaction notice for session %q incarnation %s: %w", item.name, item.id.String(), err)
639+ }
640+ }
641+ return nil
642+}
643+
567 func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {644 func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
568 healthy, fresh, degraded := 0, 0, 0645 healthy, fresh, degraded := 0, 0, 0
569 for _, record := range records {646 for _, record := range records {
@@ -662,11 +739,20 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
662 if err := coordinator.Recover(ctx); err != nil {739 if err := coordinator.Recover(ctx); err != nil {
663 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())740 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
664 }741 }
742+ // Recovery fences a session instead of failing startup when only that
743+ // session's durable state is self-inconsistent. The daemon then comes up
744+ // looking healthy, so every fenced session must be reported and persisted.
745+ fenced := fencedRecoveries(coordinator.Conflicts(), opened.Catalogue)
746+ logFencedRecoveries(log, fenced)
665 if err := persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now()); err != nil {747 if err := persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now()); err != nil {
666 log.Error("transaction_recovery_notice_failed", "err", err)748 log.Error("transaction_recovery_notice_failed", "err", err)
667 return errors.Join(fmt.Errorf("vev: persist durable transaction recovery notice: %w", err), opened.Catalogue.Close())749 return errors.Join(fmt.Errorf("vev: persist durable transaction recovery notice: %w", err), opened.Catalogue.Close())
668 }750 }
669- logTransactionRecovery(log, len(pendingRecoveries))751+ if err := persistFencedRecoveryNotices(noticeStore, fenced, clk.Now()); err != nil {
752+ log.Error("transaction_recovery_notice_failed", "reason_code", "fenced-notice-failed")
753+ return errors.Join(fmt.Errorf("vev: persist fenced transaction notice: %w", err), opened.Catalogue.Close())
754+ }
755+ logTransactionRecovery(log, len(pendingRecoveries), len(fenced))
670 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))756 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
671 unresolved := make([]domain.IncarnationID, 0, len(pendingRecoveries))757 unresolved := make([]domain.IncarnationID, 0, len(pendingRecoveries))
672 for _, pending := range pendingRecoveries {758 for _, pending := range pendingRecoveries {
@@ -1438,11 +1524,14 @@ func printSessions(w io.Writer, sessions []ports.SessionInfo) {
1438 }1524 }
1439 1525
1440 func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {1526 func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {
1441- if _, err := os.Stat(persist.StorePath(platform.StateDir())); errors.Is(err, os.ErrNotExist) {1527+ // LoadReadOnly already distinguishes a genuinely absent catalogue (empty
1442- return fmt.Errorf("vev: no such session: %s", name)1528+ // slice, nil error) from a real IO error, and recovers the same
1443- } else if err != nil {1529+ // fixed-path crash states LoadCatalogueReadOnly does, so a stray .next or
1444- return fmt.Errorf("vev: reading stored sessions: %w", err)1530+ // .prev left by a crash mid-compaction no longer wrongly denies a session
1445- }1531+ // that still exists. A separate os.Stat precheck here was both redundant
1532+ // with that recovery and a regression of the same bug class: it treated
1533+ // "no sessions.kv" as "no such session" even when a recoverable .next or
1534+ // .prev candidate was present.
1446 records, err := persist.LoadReadOnly(platform.StateDir())1535 records, err := persist.LoadReadOnly(platform.StateDir())
1447 if err != nil {1536 if err != nil {
1448 return fmt.Errorf("vev: reading stored sessions: %w", err)1537 return fmt.Errorf("vev: reading stored sessions: %w", err)
M internal/app/run_test.go +77 -1
@@ -4,6 +4,7 @@ import (
44 "bufio"
55 "bytes"
66 "context"
7+ "crypto/rand"
78 "encoding/json"
89 "errors"
910 "fmt"
@@ -18,6 +19,7 @@ import (
1819 "time"
1920
2021 "github.com/bnema/vev/internal/adapters/lifecycle"
22+ "github.com/bnema/vev/internal/adapters/recoveryfs"
2123 "github.com/bnema/vev/internal/adapters/snapshot"
2224 "github.com/bnema/vev/internal/domain"
2325 "github.com/bnema/vev/internal/persist"
@@ -26,6 +28,7 @@ import (
2628 "github.com/bnema/vev/internal/usecase/client"
2729 "github.com/bnema/vev/internal/usecase/confirm"
2830 "github.com/bnema/vev/internal/usecase/daemon"
31+ "github.com/bnema/vev/internal/usecase/recovery"
2932 "github.com/bnema/vev/pkg/kv"
3033 "github.com/stretchr/testify/mock"
3134 "github.com/stretchr/testify/require"
@@ -87,7 +90,7 @@ func TestRecoveryObservability(t *testing.T) {
8790 noticeStore.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { recoveryNotice = n }).Return(nil).Once()
8891 require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {
8992 logCatalogueRecovery(log, records, "current")
90- logTransactionRecovery(log, 2)
93+ logTransactionRecovery(log, 2, 0)
9194 require.NoError(t, persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0)))
9295 logStartupRecoveryCounts(log, records, 0)
9396 return nil
@@ -102,6 +105,10 @@ func TestRecoveryObservability(t *testing.T) {
102105 require.EqualValues(t, 1, startup["fresh"])
103106 require.EqualValues(t, 0, startup["restoring"])
104107 require.EqualValues(t, 1, startup["degraded"])
108+ recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")
109+ require.EqualValues(t, 0, recovered["sessions_fenced"])
110+ require.Equal(t, "clean", recovered["outcome"])
111+ require.NotContains(t, eventNames(entries), "interrupted_transaction_fenced")
105112
106113 logBuffer.Reset()
107114 log.Error("catalogue_validation_failed", "path", "/state/vev/sessions.kv", "reason_code", "corrupt")
@@ -112,6 +119,75 @@ func TestRecoveryObservability(t *testing.T) {
112119 require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")
113120 }
114121
122+// A session-scoped conflict no longer aborts startup, so the daemon comes up
123+// looking healthy. Startup must therefore name the fenced session in the log,
124+// persist a notice for it, and never report the recovery as clean.
125+func TestFencedRecoveryObservability(t *testing.T) {
126+ ctx := context.Background()
127+ stateDir := filepath.Join(t.TempDir(), "vev")
128+ repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
129+ catalogue := newTestPersister(t, stateDir)
130+ t.Cleanup(func() { _ = catalogue.Close() })
131+
132+ live := domain.CatalogueRecord{
133+ Name: "work", IncarnationID: domain.IncarnationID{7}, Cwd: t.TempDir(),
134+ CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryHealthy,
135+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}},
136+ }
137+ require.NoError(t, catalogue.Save(live))
138+ require.NoError(t, repository.WriteDeletionTombstone(ctx, domain.DeletionTombstone{Name: live.Name, IncarnationID: live.IncarnationID}))
139+
140+ coordinator := recovery.NewCoordinator(catalogue, repository, recoveryfs.New(stateDir), rand.Reader)
141+ require.NoError(t, coordinator.Recover(ctx), "a session-scoped conflict must not abort startup")
142+ fenced := fencedRecoveries(coordinator.Conflicts(), catalogue)
143+ require.Equal(t, []fencedRecoveryIdentity{{
144+ name: live.Name, id: live.IncarnationID, kind: "deletion-tombstone", reasonCode: "deletion-tombstone-conflict",
145+ }}, fenced)
146+
147+ var logBuffer bytes.Buffer
148+ log := slog.New(slog.NewJSONHandler(&logBuffer, nil))
149+ logFencedRecoveries(log, fenced)
150+ logTransactionRecovery(log, 1, len(fenced))
151+ logStartupRecoveryCounts(log, catalogue.Records(), 0)
152+
153+ entries := decodeJSONLogs(t, logBuffer.Bytes())
154+ event := requireEvent(t, entries, "interrupted_transaction_fenced")
155+ require.Equal(t, "WARN", event["level"])
156+ require.Equal(t, live.Name, event["session"])
157+ require.Equal(t, live.IncarnationID.String(), event["incarnation"])
158+ require.Equal(t, "deletion-tombstone", event["kind"])
159+ require.Equal(t, "deletion-tombstone-conflict", event["reason_code"])
160+ require.NotContains(t, fmt.Sprint(event), "conflicts with catalogue", "raw cause text must not be logged")
161+ recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")
162+ require.EqualValues(t, 1, recovered["sessions_fenced"])
163+ require.Equal(t, "fenced", recovered["outcome"])
164+ require.Contains(t, eventNames(entries), "daemon_startup_complete")
165+
166+ store := portsmocks.NewMockNoticeStore(t)
167+ var notice domain.Notification
168+ store.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { notice = n }).Return(nil).Once()
169+ require.NoError(t, persistFencedRecoveryNotices(store, fenced, time.Unix(1, 0)))
170+ require.Equal(t, domain.NoticeSnapshotRestore, notice.Code)
171+ require.Equal(t, domain.NoticeWarn, notice.Severity)
172+ require.Equal(t, domain.SessionID(live.IncarnationID.String()), notice.SessionID)
173+ require.Contains(t, notice.Details, "session="+live.Name)
174+ require.Contains(t, notice.Details, "reason_code=deletion-tombstone-conflict")
175+
176+ degraded, ok := catalogue.Record(live.Name)
177+ require.True(t, ok)
178+ require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)
179+}
180+
181+func TestFencedRecoveryNoticeFailurePropagates(t *testing.T) {
182+ store := portsmocks.NewMockNoticeStore(t)
183+ store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()
184+ err := persistFencedRecoveryNotices(store, []fencedRecoveryIdentity{{
185+ name: "work", id: domain.IncarnationID{1}, kind: "discard-intent", reasonCode: "discard-intent-conflict",
186+ }}, time.Unix(1, 0))
187+ require.ErrorContains(t, err, "session \"work\"")
188+ require.ErrorContains(t, err, domain.IncarnationID{1}.String())
189+}
190+
115191 func TestTransactionRecoveryNoticeFailurePropagates(t *testing.T) {
116192 store := portsmocks.NewMockNoticeStore(t)
117193 store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()
@@ -4,6 +4,7 @@ import (
4 "bufio"4 "bufio"
5 "bytes"5 "bytes"
6 "context"6 "context"
7+ "crypto/rand"
7 "encoding/json"8 "encoding/json"
8 "errors"9 "errors"
9 "fmt"10 "fmt"
@@ -18,6 +19,7 @@ import (
18 "time"19 "time"
19 20
20 "github.com/bnema/vev/internal/adapters/lifecycle"21 "github.com/bnema/vev/internal/adapters/lifecycle"
22+ "github.com/bnema/vev/internal/adapters/recoveryfs"
21 "github.com/bnema/vev/internal/adapters/snapshot"23 "github.com/bnema/vev/internal/adapters/snapshot"
22 "github.com/bnema/vev/internal/domain"24 "github.com/bnema/vev/internal/domain"
23 "github.com/bnema/vev/internal/persist"25 "github.com/bnema/vev/internal/persist"
@@ -26,6 +28,7 @@ import (
26 "github.com/bnema/vev/internal/usecase/client"28 "github.com/bnema/vev/internal/usecase/client"
27 "github.com/bnema/vev/internal/usecase/confirm"29 "github.com/bnema/vev/internal/usecase/confirm"
28 "github.com/bnema/vev/internal/usecase/daemon"30 "github.com/bnema/vev/internal/usecase/daemon"
31+ "github.com/bnema/vev/internal/usecase/recovery"
29 "github.com/bnema/vev/pkg/kv"32 "github.com/bnema/vev/pkg/kv"
30 "github.com/stretchr/testify/mock"33 "github.com/stretchr/testify/mock"
31 "github.com/stretchr/testify/require"34 "github.com/stretchr/testify/require"
@@ -87,7 +90,7 @@ func TestRecoveryObservability(t *testing.T) {
87 noticeStore.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { recoveryNotice = n }).Return(nil).Once()90 noticeStore.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { recoveryNotice = n }).Return(nil).Once()
88 require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {91 require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {
89 logCatalogueRecovery(log, records, "current")92 logCatalogueRecovery(log, records, "current")
90- logTransactionRecovery(log, 2)93+ logTransactionRecovery(log, 2, 0)
91 require.NoError(t, persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0)))94 require.NoError(t, persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0)))
92 logStartupRecoveryCounts(log, records, 0)95 logStartupRecoveryCounts(log, records, 0)
93 return nil96 return nil
@@ -102,6 +105,10 @@ func TestRecoveryObservability(t *testing.T) {
102 require.EqualValues(t, 1, startup["fresh"])105 require.EqualValues(t, 1, startup["fresh"])
103 require.EqualValues(t, 0, startup["restoring"])106 require.EqualValues(t, 0, startup["restoring"])
104 require.EqualValues(t, 1, startup["degraded"])107 require.EqualValues(t, 1, startup["degraded"])
108+ recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")
109+ require.EqualValues(t, 0, recovered["sessions_fenced"])
110+ require.Equal(t, "clean", recovered["outcome"])
111+ require.NotContains(t, eventNames(entries), "interrupted_transaction_fenced")
105 112
106 logBuffer.Reset()113 logBuffer.Reset()
107 log.Error("catalogue_validation_failed", "path", "/state/vev/sessions.kv", "reason_code", "corrupt")114 log.Error("catalogue_validation_failed", "path", "/state/vev/sessions.kv", "reason_code", "corrupt")
@@ -112,6 +119,75 @@ func TestRecoveryObservability(t *testing.T) {
112 require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")119 require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")
113 }120 }
114 121
122+// A session-scoped conflict no longer aborts startup, so the daemon comes up
123+// looking healthy. Startup must therefore name the fenced session in the log,
124+// persist a notice for it, and never report the recovery as clean.
125+func TestFencedRecoveryObservability(t *testing.T) {
126+ ctx := context.Background()
127+ stateDir := filepath.Join(t.TempDir(), "vev")
128+ repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
129+ catalogue := newTestPersister(t, stateDir)
130+ t.Cleanup(func() { _ = catalogue.Close() })
131+
132+ live := domain.CatalogueRecord{
133+ Name: "work", IncarnationID: domain.IncarnationID{7}, Cwd: t.TempDir(),
134+ CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryHealthy,
135+ Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}},
136+ }
137+ require.NoError(t, catalogue.Save(live))
138+ require.NoError(t, repository.WriteDeletionTombstone(ctx, domain.DeletionTombstone{Name: live.Name, IncarnationID: live.IncarnationID}))
139+
140+ coordinator := recovery.NewCoordinator(catalogue, repository, recoveryfs.New(stateDir), rand.Reader)
141+ require.NoError(t, coordinator.Recover(ctx), "a session-scoped conflict must not abort startup")
142+ fenced := fencedRecoveries(coordinator.Conflicts(), catalogue)
143+ require.Equal(t, []fencedRecoveryIdentity{{
144+ name: live.Name, id: live.IncarnationID, kind: "deletion-tombstone", reasonCode: "deletion-tombstone-conflict",
145+ }}, fenced)
146+
147+ var logBuffer bytes.Buffer
148+ log := slog.New(slog.NewJSONHandler(&logBuffer, nil))
149+ logFencedRecoveries(log, fenced)
150+ logTransactionRecovery(log, 1, len(fenced))
151+ logStartupRecoveryCounts(log, catalogue.Records(), 0)
152+
153+ entries := decodeJSONLogs(t, logBuffer.Bytes())
154+ event := requireEvent(t, entries, "interrupted_transaction_fenced")
155+ require.Equal(t, "WARN", event["level"])
156+ require.Equal(t, live.Name, event["session"])
157+ require.Equal(t, live.IncarnationID.String(), event["incarnation"])
158+ require.Equal(t, "deletion-tombstone", event["kind"])
159+ require.Equal(t, "deletion-tombstone-conflict", event["reason_code"])
160+ require.NotContains(t, fmt.Sprint(event), "conflicts with catalogue", "raw cause text must not be logged")
161+ recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")
162+ require.EqualValues(t, 1, recovered["sessions_fenced"])
163+ require.Equal(t, "fenced", recovered["outcome"])
164+ require.Contains(t, eventNames(entries), "daemon_startup_complete")
165+
166+ store := portsmocks.NewMockNoticeStore(t)
167+ var notice domain.Notification
168+ store.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { notice = n }).Return(nil).Once()
169+ require.NoError(t, persistFencedRecoveryNotices(store, fenced, time.Unix(1, 0)))
170+ require.Equal(t, domain.NoticeSnapshotRestore, notice.Code)
171+ require.Equal(t, domain.NoticeWarn, notice.Severity)
172+ require.Equal(t, domain.SessionID(live.IncarnationID.String()), notice.SessionID)
173+ require.Contains(t, notice.Details, "session="+live.Name)
174+ require.Contains(t, notice.Details, "reason_code=deletion-tombstone-conflict")
175+
176+ degraded, ok := catalogue.Record(live.Name)
177+ require.True(t, ok)
178+ require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)
179+}
180+
181+func TestFencedRecoveryNoticeFailurePropagates(t *testing.T) {
182+ store := portsmocks.NewMockNoticeStore(t)
183+ store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()
184+ err := persistFencedRecoveryNotices(store, []fencedRecoveryIdentity{{
185+ name: "work", id: domain.IncarnationID{1}, kind: "discard-intent", reasonCode: "discard-intent-conflict",
186+ }}, time.Unix(1, 0))
187+ require.ErrorContains(t, err, "session \"work\"")
188+ require.ErrorContains(t, err, domain.IncarnationID{1}.String())
189+}
190+
115 func TestTransactionRecoveryNoticeFailurePropagates(t *testing.T) {191 func TestTransactionRecoveryNoticeFailurePropagates(t *testing.T) {
116 store := portsmocks.NewMockNoticeStore(t)192 store := portsmocks.NewMockNoticeStore(t)
117 store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()193 store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()

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