detached 1c282c1

test(recovery): cover final transaction crash points

commit
1c282c13280ec1901828c19fafed7d843b1efffd
author
brice · · 2026-07-26 01:23 UTC
parent
6f62b03
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +154 -0
Unified
UnifiedSplit
M internal/adapters/snapshot/maintenance_delete.go +10 -0
@@ -527,10 +527,20 @@ func (r *Repository) QuarantineDeletionSources(ctx context.Context, tombstone do
527527 if err := r.quarantineSource(r.sessionPath(tombstone.IncarnationID), filepath.Join(quarantineDir, "snapshot")); err != nil {
528528 return err
529529 }
530+ if hook := r.hooks.afterDeletionIncarnationQuarantine; hook != nil {
531+ if err := hook(); err != nil {
532+ return err
533+ }
534+ }
530535 if includeLegacyName {
531536 if err := r.quarantineSource(filepath.Join(r.dir, filenameForName(tombstone.Name)), filepath.Join(quarantineDir, "legacy.snap")); err != nil {
532537 return err
533538 }
539+ if hook := r.hooks.afterDeletionLegacyQuarantine; hook != nil {
540+ if err := hook(); err != nil {
541+ return err
542+ }
543+ }
534544 }
535545 return nil
536546 }
@@ -527,10 +527,20 @@ func (r *Repository) QuarantineDeletionSources(ctx context.Context, tombstone do
527 if err := r.quarantineSource(r.sessionPath(tombstone.IncarnationID), filepath.Join(quarantineDir, "snapshot")); err != nil {527 if err := r.quarantineSource(r.sessionPath(tombstone.IncarnationID), filepath.Join(quarantineDir, "snapshot")); err != nil {
528 return err528 return err
529 }529 }
530+ if hook := r.hooks.afterDeletionIncarnationQuarantine; hook != nil {
531+ if err := hook(); err != nil {
532+ return err
533+ }
534+ }
530 if includeLegacyName {535 if includeLegacyName {
531 if err := r.quarantineSource(filepath.Join(r.dir, filenameForName(tombstone.Name)), filepath.Join(quarantineDir, "legacy.snap")); err != nil {536 if err := r.quarantineSource(filepath.Join(r.dir, filenameForName(tombstone.Name)), filepath.Join(quarantineDir, "legacy.snap")); err != nil {
532 return err537 return err
533 }538 }
539+ if hook := r.hooks.afterDeletionLegacyQuarantine; hook != nil {
540+ if err := hook(); err != nil {
541+ return err
542+ }
543+ }
534 }544 }
535 return nil545 return nil
536 }546 }
A internal/adapters/snapshot/recovery_crash_test.go +140 -0
@@ -0,0 +1,140 @@
1+package snapshot
2+
3+import (
4+ "bytes"
5+ "context"
6+ "errors"
7+ "os"
8+ "path/filepath"
9+ "testing"
10+
11+ "github.com/bnema/vev/internal/adapters/recoveryfs"
12+ "github.com/bnema/vev/internal/domain"
13+ "github.com/bnema/vev/internal/persist"
14+ "github.com/bnema/vev/internal/ports"
15+ "github.com/bnema/vev/internal/usecase/recovery"
16+ "github.com/stretchr/testify/require"
17+)
18+
19+var errDeletionBoundaryCrash = errors.New("simulated process interruption")
20+
21+type recoveryCrashHarness struct {
22+ stateDir string
23+ snapshotDir string
24+}
25+
26+func newRecoveryCrashHarness(t *testing.T) recoveryCrashHarness {
27+ t.Helper()
28+ stateDir := t.TempDir()
29+ require.NoError(t, os.Chmod(stateDir, 0o700))
30+ return recoveryCrashHarness{stateDir: stateDir, snapshotDir: filepath.Join(stateDir, "snapshots")}
31+}
32+
33+func (h recoveryCrashHarness) open(t *testing.T) (*persist.Persister, *Repository, *recoveryfs.Journal) {
34+ t.Helper()
35+ store, err := persist.OpenStore(persist.StorePath(h.stateDir))
36+ require.NoError(t, err)
37+ return persist.New(store), NewRepository(h.snapshotDir), recoveryfs.New(h.stateDir)
38+}
39+
40+func TestDeletionRestartsAtInternalQuarantineBoundaries(t *testing.T) {
41+ ctx := context.Background()
42+ for _, tc := range []struct {
43+ name string
44+ inject func(*Repository)
45+ }{
46+ {name: "after incarnation namespace", inject: func(repository *Repository) {
47+ repository.hooks.afterDeletionIncarnationQuarantine = func() error { return errDeletionBoundaryCrash }
48+ }},
49+ {name: "after legacy name", inject: func(repository *Repository) {
50+ repository.hooks.afterDeletionLegacyQuarantine = func() error { return errDeletionBoundaryCrash }
51+ }},
52+ } {
53+ t.Run(tc.name, func(t *testing.T) {
54+ h := newRecoveryCrashHarness(t)
55+ catalogue, repository, journal := h.open(t)
56+ oldID := domain.IncarnationID{1}
57+ record := domain.CatalogueRecord{Name: "work", IncarnationID: oldID, RecoveryState: domain.RecoveryFresh}
58+ require.NoError(t, catalogue.Create(record))
59+ sessionPath := repository.sessionPath(oldID)
60+ require.NoError(t, os.MkdirAll(sessionPath, 0o700))
61+ require.NoError(t, os.WriteFile(filepath.Join(sessionPath, "durable"), []byte("old namespace"), 0o600))
62+ require.NoError(t, os.WriteFile(filepath.Join(h.snapshotDir, filenameForName(record.Name)), []byte("old legacy"), 0o600))
63+ tc.inject(repository)
64+
65+ coordinator := recovery.NewCoordinator(catalogue, repository, journal, nil)
66+ require.ErrorIs(t, coordinator.Delete(ctx, record.Name), errDeletionBoundaryCrash)
67+ require.NoError(t, catalogue.Close())
68+
69+ // Simulate a new process: none of the adapter or coordinator state survives.
70+ catalogue, repository, journal = h.open(t)
71+ require.NoError(t, recovery.NewCoordinator(catalogue, repository, journal, nil).Recover(ctx))
72+ _, exists := catalogue.Record(record.Name)
73+ require.False(t, exists)
74+ require.NoDirExists(t, repository.sessionPath(oldID))
75+ require.Equal(t, []byte("old namespace"), mustReadCrashFile(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "snapshot", "durable")))
76+ require.Equal(t, []byte("old legacy"), mustReadCrashFile(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "legacy.snap")))
77+ page, err := repository.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 16, Bytes: 4096})
78+ require.NoError(t, err)
79+ require.Empty(t, page.Tombstones)
80+
81+ // Reusing the name after roll-forward must be safe on another restart.
82+ newID := domain.IncarnationID{2}
83+ created, err := recovery.NewCoordinator(catalogue, repository, journal, bytes.NewReader(newID[:])).Create(ctx, domain.CatalogueRecord{Name: record.Name})
84+ require.NoError(t, err)
85+ require.Equal(t, newID, created.IncarnationID)
86+ newPath := repository.sessionPath(newID)
87+ require.NoError(t, os.MkdirAll(newPath, 0o700))
88+ require.NoError(t, os.WriteFile(filepath.Join(newPath, "durable"), []byte("new namespace"), 0o600))
89+ require.NoError(t, catalogue.Close())
90+
91+ catalogue, repository, journal = h.open(t)
92+ require.NoError(t, recovery.NewCoordinator(catalogue, repository, journal, nil).Recover(ctx))
93+ got, exists := catalogue.Record(record.Name)
94+ require.True(t, exists)
95+ require.Equal(t, newID, got.IncarnationID)
96+ require.Equal(t, []byte("new namespace"), mustReadCrashFile(t, filepath.Join(repository.sessionPath(newID), "durable")))
97+ require.NoError(t, catalogue.Close())
98+ })
99+ }
100+}
101+
102+type rejectCreateCatalogue struct {
103+ ports.Catalogue
104+}
105+
106+func (c rejectCreateCatalogue) Create(domain.CatalogueRecord) error {
107+ return errDeletionBoundaryCrash
108+}
109+
110+func TestCreatePreCommitRestartLeavesNoAdoptableNamespace(t *testing.T) {
111+ ctx := context.Background()
112+ h := newRecoveryCrashHarness(t)
113+ catalogue, repository, journal := h.open(t)
114+ interruptedID := domain.IncarnationID{3}
115+ _, err := recovery.NewCoordinator(rejectCreateCatalogue{Catalogue: catalogue}, repository, journal, bytes.NewReader(interruptedID[:])).Create(ctx, domain.CatalogueRecord{Name: "work"})
116+ require.ErrorIs(t, err, errDeletionBoundaryCrash)
117+ require.Empty(t, catalogue.Records())
118+ require.NoDirExists(t, filepath.Join(h.snapshotDir, repositorySessionsDir))
119+ require.NoError(t, catalogue.Close())
120+
121+ catalogue, repository, journal = h.open(t)
122+ require.NoError(t, recovery.NewCoordinator(catalogue, repository, journal, nil).Recover(ctx))
123+ require.Empty(t, catalogue.Records())
124+ require.NoDirExists(t, repository.sessionPath(interruptedID))
125+
126+ retryID := domain.IncarnationID{4}
127+ created, err := recovery.NewCoordinator(catalogue, repository, journal, bytes.NewReader(retryID[:])).Create(ctx, domain.CatalogueRecord{Name: "work"})
128+ require.NoError(t, err)
129+ require.Equal(t, retryID, created.IncarnationID)
130+ require.Len(t, catalogue.Records(), 1)
131+ require.NoDirExists(t, repository.sessionPath(interruptedID))
132+ require.NoError(t, catalogue.Close())
133+}
134+
135+func mustReadCrashFile(t *testing.T, path string) []byte {
136+ t.Helper()
137+ data, err := os.ReadFile(path)
138+ require.NoError(t, err)
139+ return data
140+}
@@ -0,0 +1,140 @@
1+package snapshot
2+
3+import (
4+ "bytes"
5+ "context"
6+ "errors"
7+ "os"
8+ "path/filepath"
9+ "testing"
10+
11+ "github.com/bnema/vev/internal/adapters/recoveryfs"
12+ "github.com/bnema/vev/internal/domain"
13+ "github.com/bnema/vev/internal/persist"
14+ "github.com/bnema/vev/internal/ports"
15+ "github.com/bnema/vev/internal/usecase/recovery"
16+ "github.com/stretchr/testify/require"
17+)
18+
19+var errDeletionBoundaryCrash = errors.New("simulated process interruption")
20+
21+type recoveryCrashHarness struct {
22+ stateDir string
23+ snapshotDir string
24+}
25+
26+func newRecoveryCrashHarness(t *testing.T) recoveryCrashHarness {
27+ t.Helper()
28+ stateDir := t.TempDir()
29+ require.NoError(t, os.Chmod(stateDir, 0o700))
30+ return recoveryCrashHarness{stateDir: stateDir, snapshotDir: filepath.Join(stateDir, "snapshots")}
31+}
32+
33+func (h recoveryCrashHarness) open(t *testing.T) (*persist.Persister, *Repository, *recoveryfs.Journal) {
34+ t.Helper()
35+ store, err := persist.OpenStore(persist.StorePath(h.stateDir))
36+ require.NoError(t, err)
37+ return persist.New(store), NewRepository(h.snapshotDir), recoveryfs.New(h.stateDir)
38+}
39+
40+func TestDeletionRestartsAtInternalQuarantineBoundaries(t *testing.T) {
41+ ctx := context.Background()
42+ for _, tc := range []struct {
43+ name string
44+ inject func(*Repository)
45+ }{
46+ {name: "after incarnation namespace", inject: func(repository *Repository) {
47+ repository.hooks.afterDeletionIncarnationQuarantine = func() error { return errDeletionBoundaryCrash }
48+ }},
49+ {name: "after legacy name", inject: func(repository *Repository) {
50+ repository.hooks.afterDeletionLegacyQuarantine = func() error { return errDeletionBoundaryCrash }
51+ }},
52+ } {
53+ t.Run(tc.name, func(t *testing.T) {
54+ h := newRecoveryCrashHarness(t)
55+ catalogue, repository, journal := h.open(t)
56+ oldID := domain.IncarnationID{1}
57+ record := domain.CatalogueRecord{Name: "work", IncarnationID: oldID, RecoveryState: domain.RecoveryFresh}
58+ require.NoError(t, catalogue.Create(record))
59+ sessionPath := repository.sessionPath(oldID)
60+ require.NoError(t, os.MkdirAll(sessionPath, 0o700))
61+ require.NoError(t, os.WriteFile(filepath.Join(sessionPath, "durable"), []byte("old namespace"), 0o600))
62+ require.NoError(t, os.WriteFile(filepath.Join(h.snapshotDir, filenameForName(record.Name)), []byte("old legacy"), 0o600))
63+ tc.inject(repository)
64+
65+ coordinator := recovery.NewCoordinator(catalogue, repository, journal, nil)
66+ require.ErrorIs(t, coordinator.Delete(ctx, record.Name), errDeletionBoundaryCrash)
67+ require.NoError(t, catalogue.Close())
68+
69+ // Simulate a new process: none of the adapter or coordinator state survives.
70+ catalogue, repository, journal = h.open(t)
71+ require.NoError(t, recovery.NewCoordinator(catalogue, repository, journal, nil).Recover(ctx))
72+ _, exists := catalogue.Record(record.Name)
73+ require.False(t, exists)
74+ require.NoDirExists(t, repository.sessionPath(oldID))
75+ require.Equal(t, []byte("old namespace"), mustReadCrashFile(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "snapshot", "durable")))
76+ require.Equal(t, []byte("old legacy"), mustReadCrashFile(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "legacy.snap")))
77+ page, err := repository.ListDeletionTombstones(ctx, ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 16, Bytes: 4096})
78+ require.NoError(t, err)
79+ require.Empty(t, page.Tombstones)
80+
81+ // Reusing the name after roll-forward must be safe on another restart.
82+ newID := domain.IncarnationID{2}
83+ created, err := recovery.NewCoordinator(catalogue, repository, journal, bytes.NewReader(newID[:])).Create(ctx, domain.CatalogueRecord{Name: record.Name})
84+ require.NoError(t, err)
85+ require.Equal(t, newID, created.IncarnationID)
86+ newPath := repository.sessionPath(newID)
87+ require.NoError(t, os.MkdirAll(newPath, 0o700))
88+ require.NoError(t, os.WriteFile(filepath.Join(newPath, "durable"), []byte("new namespace"), 0o600))
89+ require.NoError(t, catalogue.Close())
90+
91+ catalogue, repository, journal = h.open(t)
92+ require.NoError(t, recovery.NewCoordinator(catalogue, repository, journal, nil).Recover(ctx))
93+ got, exists := catalogue.Record(record.Name)
94+ require.True(t, exists)
95+ require.Equal(t, newID, got.IncarnationID)
96+ require.Equal(t, []byte("new namespace"), mustReadCrashFile(t, filepath.Join(repository.sessionPath(newID), "durable")))
97+ require.NoError(t, catalogue.Close())
98+ })
99+ }
100+}
101+
102+type rejectCreateCatalogue struct {
103+ ports.Catalogue
104+}
105+
106+func (c rejectCreateCatalogue) Create(domain.CatalogueRecord) error {
107+ return errDeletionBoundaryCrash
108+}
109+
110+func TestCreatePreCommitRestartLeavesNoAdoptableNamespace(t *testing.T) {
111+ ctx := context.Background()
112+ h := newRecoveryCrashHarness(t)
113+ catalogue, repository, journal := h.open(t)
114+ interruptedID := domain.IncarnationID{3}
115+ _, err := recovery.NewCoordinator(rejectCreateCatalogue{Catalogue: catalogue}, repository, journal, bytes.NewReader(interruptedID[:])).Create(ctx, domain.CatalogueRecord{Name: "work"})
116+ require.ErrorIs(t, err, errDeletionBoundaryCrash)
117+ require.Empty(t, catalogue.Records())
118+ require.NoDirExists(t, filepath.Join(h.snapshotDir, repositorySessionsDir))
119+ require.NoError(t, catalogue.Close())
120+
121+ catalogue, repository, journal = h.open(t)
122+ require.NoError(t, recovery.NewCoordinator(catalogue, repository, journal, nil).Recover(ctx))
123+ require.Empty(t, catalogue.Records())
124+ require.NoDirExists(t, repository.sessionPath(interruptedID))
125+
126+ retryID := domain.IncarnationID{4}
127+ created, err := recovery.NewCoordinator(catalogue, repository, journal, bytes.NewReader(retryID[:])).Create(ctx, domain.CatalogueRecord{Name: "work"})
128+ require.NoError(t, err)
129+ require.Equal(t, retryID, created.IncarnationID)
130+ require.Len(t, catalogue.Records(), 1)
131+ require.NoDirExists(t, repository.sessionPath(interruptedID))
132+ require.NoError(t, catalogue.Close())
133+}
134+
135+func mustReadCrashFile(t *testing.T, path string) []byte {
136+ t.Helper()
137+ data, err := os.ReadFile(path)
138+ require.NoError(t, err)
139+ return data
140+}
M internal/adapters/snapshot/repository.go +4 -0
@@ -78,6 +78,10 @@ type repositoryHooks struct {
7878 openMaintenanceDirectory func(string) (maintenanceDirectory, error)
7979 beforeDirectoryRead func(string)
8080 beforeDeletionTombstoneRead func(string)
81+ // afterDeletionIncarnationQuarantine and afterDeletionLegacyQuarantine expose
82+ // the two completed deletion rename+sync boundaries to package tests.
83+ afterDeletionIncarnationQuarantine func() error
84+ afterDeletionLegacyQuarantine func() error
8185 // beforeMaintenanceWork observes each budgeted quarantine filesystem step.
8286 // It is test-only instrumentation for hostile traversal bounds.
8387 beforeMaintenanceWork func(string)
@@ -78,6 +78,10 @@ type repositoryHooks struct {
78 openMaintenanceDirectory func(string) (maintenanceDirectory, error)78 openMaintenanceDirectory func(string) (maintenanceDirectory, error)
79 beforeDirectoryRead func(string)79 beforeDirectoryRead func(string)
80 beforeDeletionTombstoneRead func(string)80 beforeDeletionTombstoneRead func(string)
81+ // afterDeletionIncarnationQuarantine and afterDeletionLegacyQuarantine expose
82+ // the two completed deletion rename+sync boundaries to package tests.
83+ afterDeletionIncarnationQuarantine func() error
84+ afterDeletionLegacyQuarantine func() error
81 // beforeMaintenanceWork observes each budgeted quarantine filesystem step.85 // beforeMaintenanceWork observes each budgeted quarantine filesystem step.
82 // It is test-only instrumentation for hostile traversal bounds.86 // It is test-only instrumentation for hostile traversal bounds.
83 beforeMaintenanceWork func(string)87 beforeMaintenanceWork func(string)

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