detached 8759629

test(recovery): exercise durable crash boundaries

commit
8759629ebf4cabfd3e4f1ab31bbe15b68faae2dd
author
brice · · 2026-07-26 01:19 UTC
committer
brice · · 2026-07-26 01:19 UTC
parent
0fe8a6c
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +382 -128
Split
UnifiedSplit
A internal/app/durable_crash_matrix_test.go +382 -0
@@ -0,0 +1,382 @@
1+package app
2+
3+import (
4+ "bytes"
5+ "context"
6+ "crypto/sha256"
7+ "errors"
8+ "os"
9+ "path/filepath"
10+ "testing"
11+
12+ "github.com/bnema/vev/internal/adapters/recoveryfs"
13+ "github.com/bnema/vev/internal/adapters/snapshot"
14+ "github.com/bnema/vev/internal/domain"
15+ "github.com/bnema/vev/internal/persist"
16+ "github.com/bnema/vev/internal/ports"
17+ "github.com/bnema/vev/internal/usecase/recovery"
18+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
19+ "github.com/stretchr/testify/require"
20+)
21+
22+var errInjectedCrash = errors.New("injected crash after durable boundary")
23+
24+type durableCrashHarness struct {
25+ stateDir string
26+ snapshotDir string
27+}
28+
29+func newDurableCrashHarness(t *testing.T) durableCrashHarness {
30+ t.Helper()
31+ stateDir := t.TempDir()
32+ require.NoError(t, os.Chmod(stateDir, 0o700))
33+ return durableCrashHarness{stateDir: stateDir, snapshotDir: filepath.Join(stateDir, "snapshots")}
34+}
35+
36+func (h durableCrashHarness) open(t *testing.T) (*persist.Persister, *snapshot.Repository, *recoveryfs.Journal) {
37+ t.Helper()
38+ store, err := persist.OpenStore(persist.StorePath(h.stateDir))
39+ require.NoError(t, err)
40+ return persist.New(store), snapshot.NewRepository(h.snapshotDir), recoveryfs.New(h.stateDir)
41+}
42+
43+func (h durableCrashHarness) recoverTwice(t *testing.T, assert func(*persist.Persister, *snapshot.Repository, *recoveryfs.Journal)) {
44+ t.Helper()
45+ for attempt := range 2 {
46+ catalogue, repository, journal := h.open(t)
47+ coordinator := recovery.NewCoordinator(catalogue, repository, journal, nil)
48+ require.NoError(t, coordinator.Recover(context.Background()), "recovery attempt %d", attempt+1)
49+ assert(catalogue, repository, journal)
50+ require.NoError(t, catalogue.Close())
51+ }
52+}
53+
54+type durableBoundary struct {
55+ want string
56+ triggered bool
57+}
58+
59+func (b *durableBoundary) after(name string, err error) error {
60+ if err != nil || name != b.want {
61+ return err
62+ }
63+ b.triggered = true
64+ return errInjectedCrash
65+}
66+
67+type crashCatalogue struct {
68+ ports.Catalogue
69+ boundary *durableBoundary
70+}
71+
72+func (c crashCatalogue) Create(record domain.CatalogueRecord) error {
73+ return c.boundary.after("catalogue-create", c.Catalogue.Create(record))
74+}
75+func (c crashCatalogue) Replace(name string, record domain.CatalogueRecord) error {
76+ return c.boundary.after("catalogue-replace", c.Catalogue.Replace(name, record))
77+}
78+func (c crashCatalogue) Rename(oldName string, record domain.CatalogueRecord) error {
79+ return c.boundary.after("catalogue-rename", c.Catalogue.Rename(oldName, record))
80+}
81+func (c crashCatalogue) Delete(name string) error {
82+ return c.boundary.after("catalogue-delete", c.Catalogue.Delete(name))
83+}
84+
85+type crashRepository struct {
86+ ports.SnapshotRepository
87+ boundary *durableBoundary
88+}
89+
90+func (r crashRepository) Publish(ctx context.Context, publication ports.SnapshotPublication) error {
91+ return r.boundary.after("snapshot-publish", r.SnapshotRepository.Publish(ctx, publication))
92+}
93+func (r crashRepository) WriteDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {
94+ return r.boundary.after("deletion-tombstone", r.SnapshotRepository.WriteDeletionTombstone(ctx, tombstone))
95+}
96+func (r crashRepository) QuarantineDeletionSources(ctx context.Context, tombstone domain.DeletionTombstone, includeLegacy bool) error {
97+ return r.boundary.after("deletion-quarantine", r.SnapshotRepository.QuarantineDeletionSources(ctx, tombstone, includeLegacy))
98+}
99+func (r crashRepository) DeleteDeletionTombstone(ctx context.Context, id domain.IncarnationID) error {
100+ return r.boundary.after("deletion-tombstone-delete", r.SnapshotRepository.DeleteDeletionTombstone(ctx, id))
101+}
102+func (r crashRepository) SaveQuarantineDescriptor(ctx context.Context, descriptor domain.QuarantineDescriptor) error {
103+ return r.boundary.after("discard-descriptor", r.SnapshotRepository.SaveQuarantineDescriptor(ctx, descriptor))
104+}
105+func (r crashRepository) QuarantineIncarnation(ctx context.Context, id domain.IncarnationID) error {
106+ return r.boundary.after("discard-quarantine", r.SnapshotRepository.QuarantineIncarnation(ctx, id))
107+}
108+
109+type crashJournal struct {
110+ ports.RecoveryJournal
111+ boundary *durableBoundary
112+}
113+
114+func (j crashJournal) SaveDiscard(ctx context.Context, intent domain.DiscardIntent) error {
115+ return j.boundary.after("discard-intent", j.RecoveryJournal.SaveDiscard(ctx, intent))
116+}
117+func (j crashJournal) DeleteDiscard(ctx context.Context, id domain.IncarnationID) error {
118+ return j.boundary.after("discard-intent-delete", j.RecoveryJournal.DeleteDiscard(ctx, id))
119+}
120+
121+func TestDurableTransactionCrashMatrix(t *testing.T) {
122+ ctx := context.Background()
123+
124+ t.Run("create/catalogue-create", func(t *testing.T) {
125+ h := newDurableCrashHarness(t)
126+ catalogue, repository, journal := h.open(t)
127+ boundary := &durableBoundary{want: "catalogue-create"}
128+ coordinator := recovery.NewCoordinator(crashCatalogue{Catalogue: catalogue, boundary: boundary}, repository, journal, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
129+ _, err := coordinator.Create(ctx, domain.CatalogueRecord{Name: "work", Cwd: "/tmp"})
130+ require.ErrorIs(t, err, errInjectedCrash)
131+ require.True(t, boundary.triggered)
132+ require.NoError(t, catalogue.Close())
133+
134+ h.recoverTwice(t, func(catalogue *persist.Persister, _ *snapshot.Repository, _ *recoveryfs.Journal) {
135+ record, exists := catalogue.Record("work")
136+ require.True(t, exists)
137+ require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
138+ require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
139+ })
140+ })
141+
142+ for _, boundaryName := range []string{"snapshot-publish", "catalogue-replace"} {
143+ t.Run("checkpoint/"+boundaryName, func(t *testing.T) {
144+ h := newDurableCrashHarness(t)
145+ catalogue, repository, journal := h.open(t)
146+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryFresh}
147+ require.NoError(t, catalogue.Create(record))
148+ publication := durablePublication(t, record.Name, record.IncarnationID, []byte("checkpoint"))
149+ boundary := &durableBoundary{want: boundaryName}
150+ coordinator := recovery.NewCoordinator(
151+ crashCatalogue{Catalogue: catalogue, boundary: boundary},
152+ crashRepository{SnapshotRepository: repository, boundary: boundary},
153+ journal,
154+ nil,
155+ )
156+ _, err := coordinator.PublishCheckpoint(ctx, record.Name, publication)
157+ require.ErrorIs(t, err, errInjectedCrash)
158+ require.True(t, boundary.triggered)
159+ require.NoError(t, catalogue.Close())
160+
161+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(publication.Manifest)}
162+ h.recoverTwice(t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
163+ stored, exists := catalogue.Record(record.Name)
164+ require.True(t, exists)
165+ if boundaryName == "snapshot-publish" {
166+ require.Equal(t, domain.RecoveryFresh, stored.RecoveryState)
167+ require.Nil(t, stored.Committed)
168+ } else {
169+ require.Equal(t, domain.RecoveryHealthy, stored.RecoveryState)
170+ require.Equal(t, ref, *stored.Committed)
171+ }
172+ generation, err := repository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
173+ require.NoError(t, err)
174+ require.Equal(t, publication.Manifest, generation.Manifest)
175+ })
176+ })
177+ }
178+
179+ t.Run("rename/catalogue-rename", func(t *testing.T) {
180+ h := newDurableCrashHarness(t)
181+ catalogue, repository, journal := h.open(t)
182+ record := domain.CatalogueRecord{Name: "old", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh}
183+ require.NoError(t, catalogue.Create(record))
184+ boundary := &durableBoundary{want: "catalogue-rename"}
185+ coordinator := recovery.NewCoordinator(crashCatalogue{Catalogue: catalogue, boundary: boundary}, repository, journal, nil)
186+ _, err := coordinator.Rename(ctx, "old", "new")
187+ require.ErrorIs(t, err, errInjectedCrash)
188+ require.True(t, boundary.triggered)
189+ require.NoError(t, catalogue.Close())
190+
191+ h.recoverTwice(t, func(catalogue *persist.Persister, _ *snapshot.Repository, _ *recoveryfs.Journal) {
192+ _, oldExists := catalogue.Record("old")
193+ renamed, newExists := catalogue.Record("new")
194+ require.False(t, oldExists)
195+ require.True(t, newExists)
196+ require.Equal(t, record.IncarnationID, renamed.IncarnationID)
197+ })
198+ })
199+
200+ deleteBoundaries := []string{"catalogue-replace", "deletion-tombstone", "deletion-quarantine", "catalogue-delete", "deletion-tombstone-delete", "transaction-complete"}
201+ for _, boundaryName := range deleteBoundaries {
202+ t.Run("delete/"+boundaryName, func(t *testing.T) {
203+ h := newDurableCrashHarness(t)
204+ catalogue, repository, journal := h.open(t)
205+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryFresh}
206+ require.NoError(t, catalogue.Create(record))
207+ publication := durablePublication(t, record.Name, record.IncarnationID, []byte("delete-source"))
208+ require.NoError(t, repository.Publish(ctx, publication))
209+ legacy := []byte{0, 1, 2, 3, 0xff}
210+ require.NoError(t, os.WriteFile(filepath.Join(h.snapshotDir, "work.snap"), legacy, 0o600))
211+ sourceBytes := readTree(t, filepath.Join(h.snapshotDir, "sessions", record.IncarnationID.String()))
212+
213+ boundary := &durableBoundary{want: boundaryName}
214+ coordinator := recovery.NewCoordinator(
215+ crashCatalogue{Catalogue: catalogue, boundary: boundary},
216+ crashRepository{SnapshotRepository: repository, boundary: boundary},
217+ journal,
218+ nil,
219+ )
220+ err := coordinator.Delete(ctx, record.Name)
221+ if boundaryName == "transaction-complete" {
222+ require.NoError(t, err)
223+ boundary.triggered = true // interruption before the caller transfers lifecycle ownership
224+ } else {
225+ require.ErrorIs(t, err, errInjectedCrash)
226+ }
227+ require.True(t, boundary.triggered)
228+ require.NoError(t, catalogue.Close())
229+
230+ h.recoverTwice(t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
231+ _, exists := catalogue.Record(record.Name)
232+ require.False(t, exists)
233+ assertNoTombstones(t, repository)
234+ require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", record.IncarnationID.String()))
235+ require.Equal(t, sourceBytes, readTree(t, filepath.Join(h.snapshotDir, "quarantine", record.IncarnationID.String(), "snapshot")))
236+ require.Equal(t, legacy, mustReadFile(t, filepath.Join(h.snapshotDir, "quarantine", record.IncarnationID.String(), "legacy.snap")))
237+ })
238+ })
239+ }
240+}
241+
242+func TestDurableDiscardCrashMatrix(t *testing.T) {
243+ ctx := context.Background()
244+ oldID := domain.IncarnationID{5}
245+ newID := domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
246+ boundaries := []string{"discard-intent", "discard-descriptor", "discard-quarantine", "catalogue-replace", "discard-intent-delete", "runtime-exposure"}
247+
248+ for _, boundaryName := range boundaries {
249+ t.Run(boundaryName, func(t *testing.T) {
250+ h := newDurableCrashHarness(t)
251+ catalogue, repository, journal := h.open(t)
252+ publication := durablePublication(t, "broken", oldID, []byte("discard-source"))
253+ require.NoError(t, repository.Publish(ctx, publication))
254+ ref := &domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(publication.Manifest)}
255+ old := domain.CatalogueRecord{Name: "broken", IncarnationID: oldID, RecoveryState: domain.RecoveryDegraded, Committed: ref, DegradedReason: "checkpoint unreadable"}
256+ require.NoError(t, catalogue.Create(old))
257+ sourceBytes := readTree(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
258+
259+ boundary := &durableBoundary{want: boundaryName}
260+ coordinator := recovery.NewCoordinator(
261+ crashCatalogue{Catalogue: catalogue, boundary: boundary},
262+ crashRepository{SnapshotRepository: repository, boundary: boundary},
263+ crashJournal{RecoveryJournal: journal, boundary: boundary},
264+ bytes.NewReader(newID[:]),
265+ )
266+ replacement, err := coordinator.Discard(ctx, old.Name, "explicit discard")
267+ if boundaryName == "runtime-exposure" {
268+ require.NoError(t, err)
269+ require.Equal(t, newID, replacement.IncarnationID)
270+ boundary.triggered = true // crash before the returned runtime is installed
271+ } else {
272+ require.ErrorIs(t, err, errInjectedCrash)
273+ }
274+ require.True(t, boundary.triggered)
275+ require.NoError(t, catalogue.Close())
276+
277+ h.recoverTwice(t, func(catalogue *persist.Persister, _ *snapshot.Repository, journal *recoveryfs.Journal) {
278+ got, exists := catalogue.Record(old.Name)
279+ require.True(t, exists)
280+ require.Equal(t, newID, got.IncarnationID)
281+ require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
282+ require.Nil(t, got.Committed)
283+ require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
284+ require.Equal(t, sourceBytes, readTree(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "snapshot")))
285+ intents, err := journal.ListDiscards(ctx)
286+ require.NoError(t, err)
287+ require.Empty(t, intents)
288+ require.FileExists(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "record"))
289+ })
290+ })
291+ }
292+}
293+
294+func TestDeletionRecoveryProtectsReusedNameFilesystemBytes(t *testing.T) {
295+ ctx := context.Background()
296+ h := newDurableCrashHarness(t)
297+ catalogue, repository, _ := h.open(t)
298+ oldID := domain.IncarnationID{6}
299+ replacementID := domain.IncarnationID{7}
300+ oldPublication := durablePublication(t, "work", oldID, []byte("old-incarnation"))
301+ replacementPublication := durablePublication(t, "work", replacementID, []byte("replacement-incarnation"))
302+ require.NoError(t, repository.Publish(ctx, oldPublication))
303+ require.NoError(t, repository.Publish(ctx, replacementPublication))
304+ replacement := domain.CatalogueRecord{Name: "work", IncarnationID: replacementID, RecoveryState: domain.RecoveryFresh}
305+ require.NoError(t, catalogue.Create(replacement))
306+ require.NoError(t, repository.WriteDeletionTombstone(ctx, domain.DeletionTombstone{Name: "work", IncarnationID: oldID}))
307+ legacy := []byte{0xde, 0xad, 0x00, 0xbe, 0xef}
308+ require.NoError(t, os.WriteFile(filepath.Join(h.snapshotDir, "work.snap"), legacy, 0o600))
309+ oldBytes := readTree(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
310+ replacementPath := filepath.Join(h.snapshotDir, "sessions", replacementID.String())
311+ replacementBytes := readTree(t, replacementPath)
312+ require.NoError(t, catalogue.Close())
313+
314+ h.recoverTwice(t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
315+ got, exists := catalogue.Record("work")
316+ require.True(t, exists)
317+ require.Equal(t, replacement, got)
318+ require.Equal(t, replacementBytes, readTree(t, replacementPath))
319+ require.Equal(t, legacy, mustReadFile(t, filepath.Join(h.snapshotDir, "work.snap")))
320+ require.Equal(t, oldBytes, readTree(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "snapshot")))
321+ require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
322+ assertNoTombstones(t, repository)
323+ })
324+}
325+
326+func durablePublication(t *testing.T, name string, id domain.IncarnationID, payload []byte) ports.SnapshotPublication {
327+ t.Helper()
328+ tail, err := snapcodec.MarshalObject(snapcodec.HistoryTail, payload)
329+ require.NoError(t, err)
330+ visible, err := snapcodec.MarshalObject(snapcodec.Visible, []byte("visible"))
331+ require.NoError(t, err)
332+ manifest, err := snapcodec.MarshalManifest(snapcodec.Manifest{
333+ Generation: 1, IncarnationID: id, Name: name,
334+ Tabs: []snapcodec.ManifestTab{{Cols: 1, Rows: 1, Panes: []snapcodec.ManifestPane{{
335+ ID: "p", Tail: snapcodec.ObjectRef{Kind: snapcodec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))},
336+ Visible: snapcodec.ObjectRef{Kind: snapcodec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))},
337+ }}}},
338+ })
339+ require.NoError(t, err)
340+ return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: 1, Manifest: manifest, Objects: []ports.SnapshotObject{tail, visible}}
341+}
342+
343+func assertNoTombstones(t *testing.T, repository *snapshot.Repository) {
344+ t.Helper()
345+ page, err := repository.ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10})
346+ require.NoError(t, err)
347+ require.True(t, page.Done)
348+ require.Empty(t, page.Tombstones)
349+}
350+
351+func mustReadFile(t *testing.T, path string) []byte {
352+ t.Helper()
353+ data, err := os.ReadFile(path)
354+ require.NoError(t, err)
355+ return data
356+}
357+
358+func readTree(t *testing.T, root string) map[string][]byte {
359+ t.Helper()
360+ files := make(map[string][]byte)
361+ err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error {
362+ if err != nil {
363+ return err
364+ }
365+ if entry.IsDir() {
366+ return nil
367+ }
368+ relative, err := filepath.Rel(root, path)
369+ if err != nil {
370+ return err
371+ }
372+ data, err := os.ReadFile(path)
373+ if err != nil {
374+ return err
375+ }
376+ files[relative] = data
377+ return nil
378+ })
379+ require.NoError(t, err)
380+ require.NotEmpty(t, files)
381+ return files
382+}
@@ -0,0 +1,382 @@
1+package app
2+
3+import (
4+ "bytes"
5+ "context"
6+ "crypto/sha256"
7+ "errors"
8+ "os"
9+ "path/filepath"
10+ "testing"
11+
12+ "github.com/bnema/vev/internal/adapters/recoveryfs"
13+ "github.com/bnema/vev/internal/adapters/snapshot"
14+ "github.com/bnema/vev/internal/domain"
15+ "github.com/bnema/vev/internal/persist"
16+ "github.com/bnema/vev/internal/ports"
17+ "github.com/bnema/vev/internal/usecase/recovery"
18+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
19+ "github.com/stretchr/testify/require"
20+)
21+
22+var errInjectedCrash = errors.New("injected crash after durable boundary")
23+
24+type durableCrashHarness struct {
25+ stateDir string
26+ snapshotDir string
27+}
28+
29+func newDurableCrashHarness(t *testing.T) durableCrashHarness {
30+ t.Helper()
31+ stateDir := t.TempDir()
32+ require.NoError(t, os.Chmod(stateDir, 0o700))
33+ return durableCrashHarness{stateDir: stateDir, snapshotDir: filepath.Join(stateDir, "snapshots")}
34+}
35+
36+func (h durableCrashHarness) open(t *testing.T) (*persist.Persister, *snapshot.Repository, *recoveryfs.Journal) {
37+ t.Helper()
38+ store, err := persist.OpenStore(persist.StorePath(h.stateDir))
39+ require.NoError(t, err)
40+ return persist.New(store), snapshot.NewRepository(h.snapshotDir), recoveryfs.New(h.stateDir)
41+}
42+
43+func (h durableCrashHarness) recoverTwice(t *testing.T, assert func(*persist.Persister, *snapshot.Repository, *recoveryfs.Journal)) {
44+ t.Helper()
45+ for attempt := range 2 {
46+ catalogue, repository, journal := h.open(t)
47+ coordinator := recovery.NewCoordinator(catalogue, repository, journal, nil)
48+ require.NoError(t, coordinator.Recover(context.Background()), "recovery attempt %d", attempt+1)
49+ assert(catalogue, repository, journal)
50+ require.NoError(t, catalogue.Close())
51+ }
52+}
53+
54+type durableBoundary struct {
55+ want string
56+ triggered bool
57+}
58+
59+func (b *durableBoundary) after(name string, err error) error {
60+ if err != nil || name != b.want {
61+ return err
62+ }
63+ b.triggered = true
64+ return errInjectedCrash
65+}
66+
67+type crashCatalogue struct {
68+ ports.Catalogue
69+ boundary *durableBoundary
70+}
71+
72+func (c crashCatalogue) Create(record domain.CatalogueRecord) error {
73+ return c.boundary.after("catalogue-create", c.Catalogue.Create(record))
74+}
75+func (c crashCatalogue) Replace(name string, record domain.CatalogueRecord) error {
76+ return c.boundary.after("catalogue-replace", c.Catalogue.Replace(name, record))
77+}
78+func (c crashCatalogue) Rename(oldName string, record domain.CatalogueRecord) error {
79+ return c.boundary.after("catalogue-rename", c.Catalogue.Rename(oldName, record))
80+}
81+func (c crashCatalogue) Delete(name string) error {
82+ return c.boundary.after("catalogue-delete", c.Catalogue.Delete(name))
83+}
84+
85+type crashRepository struct {
86+ ports.SnapshotRepository
87+ boundary *durableBoundary
88+}
89+
90+func (r crashRepository) Publish(ctx context.Context, publication ports.SnapshotPublication) error {
91+ return r.boundary.after("snapshot-publish", r.SnapshotRepository.Publish(ctx, publication))
92+}
93+func (r crashRepository) WriteDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {
94+ return r.boundary.after("deletion-tombstone", r.SnapshotRepository.WriteDeletionTombstone(ctx, tombstone))
95+}
96+func (r crashRepository) QuarantineDeletionSources(ctx context.Context, tombstone domain.DeletionTombstone, includeLegacy bool) error {
97+ return r.boundary.after("deletion-quarantine", r.SnapshotRepository.QuarantineDeletionSources(ctx, tombstone, includeLegacy))
98+}
99+func (r crashRepository) DeleteDeletionTombstone(ctx context.Context, id domain.IncarnationID) error {
100+ return r.boundary.after("deletion-tombstone-delete", r.SnapshotRepository.DeleteDeletionTombstone(ctx, id))
101+}
102+func (r crashRepository) SaveQuarantineDescriptor(ctx context.Context, descriptor domain.QuarantineDescriptor) error {
103+ return r.boundary.after("discard-descriptor", r.SnapshotRepository.SaveQuarantineDescriptor(ctx, descriptor))
104+}
105+func (r crashRepository) QuarantineIncarnation(ctx context.Context, id domain.IncarnationID) error {
106+ return r.boundary.after("discard-quarantine", r.SnapshotRepository.QuarantineIncarnation(ctx, id))
107+}
108+
109+type crashJournal struct {
110+ ports.RecoveryJournal
111+ boundary *durableBoundary
112+}
113+
114+func (j crashJournal) SaveDiscard(ctx context.Context, intent domain.DiscardIntent) error {
115+ return j.boundary.after("discard-intent", j.RecoveryJournal.SaveDiscard(ctx, intent))
116+}
117+func (j crashJournal) DeleteDiscard(ctx context.Context, id domain.IncarnationID) error {
118+ return j.boundary.after("discard-intent-delete", j.RecoveryJournal.DeleteDiscard(ctx, id))
119+}
120+
121+func TestDurableTransactionCrashMatrix(t *testing.T) {
122+ ctx := context.Background()
123+
124+ t.Run("create/catalogue-create", func(t *testing.T) {
125+ h := newDurableCrashHarness(t)
126+ catalogue, repository, journal := h.open(t)
127+ boundary := &durableBoundary{want: "catalogue-create"}
128+ coordinator := recovery.NewCoordinator(crashCatalogue{Catalogue: catalogue, boundary: boundary}, repository, journal, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
129+ _, err := coordinator.Create(ctx, domain.CatalogueRecord{Name: "work", Cwd: "/tmp"})
130+ require.ErrorIs(t, err, errInjectedCrash)
131+ require.True(t, boundary.triggered)
132+ require.NoError(t, catalogue.Close())
133+
134+ h.recoverTwice(t, func(catalogue *persist.Persister, _ *snapshot.Repository, _ *recoveryfs.Journal) {
135+ record, exists := catalogue.Record("work")
136+ require.True(t, exists)
137+ require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
138+ require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
139+ })
140+ })
141+
142+ for _, boundaryName := range []string{"snapshot-publish", "catalogue-replace"} {
143+ t.Run("checkpoint/"+boundaryName, func(t *testing.T) {
144+ h := newDurableCrashHarness(t)
145+ catalogue, repository, journal := h.open(t)
146+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryFresh}
147+ require.NoError(t, catalogue.Create(record))
148+ publication := durablePublication(t, record.Name, record.IncarnationID, []byte("checkpoint"))
149+ boundary := &durableBoundary{want: boundaryName}
150+ coordinator := recovery.NewCoordinator(
151+ crashCatalogue{Catalogue: catalogue, boundary: boundary},
152+ crashRepository{SnapshotRepository: repository, boundary: boundary},
153+ journal,
154+ nil,
155+ )
156+ _, err := coordinator.PublishCheckpoint(ctx, record.Name, publication)
157+ require.ErrorIs(t, err, errInjectedCrash)
158+ require.True(t, boundary.triggered)
159+ require.NoError(t, catalogue.Close())
160+
161+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(publication.Manifest)}
162+ h.recoverTwice(t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
163+ stored, exists := catalogue.Record(record.Name)
164+ require.True(t, exists)
165+ if boundaryName == "snapshot-publish" {
166+ require.Equal(t, domain.RecoveryFresh, stored.RecoveryState)
167+ require.Nil(t, stored.Committed)
168+ } else {
169+ require.Equal(t, domain.RecoveryHealthy, stored.RecoveryState)
170+ require.Equal(t, ref, *stored.Committed)
171+ }
172+ generation, err := repository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
173+ require.NoError(t, err)
174+ require.Equal(t, publication.Manifest, generation.Manifest)
175+ })
176+ })
177+ }
178+
179+ t.Run("rename/catalogue-rename", func(t *testing.T) {
180+ h := newDurableCrashHarness(t)
181+ catalogue, repository, journal := h.open(t)
182+ record := domain.CatalogueRecord{Name: "old", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh}
183+ require.NoError(t, catalogue.Create(record))
184+ boundary := &durableBoundary{want: "catalogue-rename"}
185+ coordinator := recovery.NewCoordinator(crashCatalogue{Catalogue: catalogue, boundary: boundary}, repository, journal, nil)
186+ _, err := coordinator.Rename(ctx, "old", "new")
187+ require.ErrorIs(t, err, errInjectedCrash)
188+ require.True(t, boundary.triggered)
189+ require.NoError(t, catalogue.Close())
190+
191+ h.recoverTwice(t, func(catalogue *persist.Persister, _ *snapshot.Repository, _ *recoveryfs.Journal) {
192+ _, oldExists := catalogue.Record("old")
193+ renamed, newExists := catalogue.Record("new")
194+ require.False(t, oldExists)
195+ require.True(t, newExists)
196+ require.Equal(t, record.IncarnationID, renamed.IncarnationID)
197+ })
198+ })
199+
200+ deleteBoundaries := []string{"catalogue-replace", "deletion-tombstone", "deletion-quarantine", "catalogue-delete", "deletion-tombstone-delete", "transaction-complete"}
201+ for _, boundaryName := range deleteBoundaries {
202+ t.Run("delete/"+boundaryName, func(t *testing.T) {
203+ h := newDurableCrashHarness(t)
204+ catalogue, repository, journal := h.open(t)
205+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryFresh}
206+ require.NoError(t, catalogue.Create(record))
207+ publication := durablePublication(t, record.Name, record.IncarnationID, []byte("delete-source"))
208+ require.NoError(t, repository.Publish(ctx, publication))
209+ legacy := []byte{0, 1, 2, 3, 0xff}
210+ require.NoError(t, os.WriteFile(filepath.Join(h.snapshotDir, "work.snap"), legacy, 0o600))
211+ sourceBytes := readTree(t, filepath.Join(h.snapshotDir, "sessions", record.IncarnationID.String()))
212+
213+ boundary := &durableBoundary{want: boundaryName}
214+ coordinator := recovery.NewCoordinator(
215+ crashCatalogue{Catalogue: catalogue, boundary: boundary},
216+ crashRepository{SnapshotRepository: repository, boundary: boundary},
217+ journal,
218+ nil,
219+ )
220+ err := coordinator.Delete(ctx, record.Name)
221+ if boundaryName == "transaction-complete" {
222+ require.NoError(t, err)
223+ boundary.triggered = true // interruption before the caller transfers lifecycle ownership
224+ } else {
225+ require.ErrorIs(t, err, errInjectedCrash)
226+ }
227+ require.True(t, boundary.triggered)
228+ require.NoError(t, catalogue.Close())
229+
230+ h.recoverTwice(t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
231+ _, exists := catalogue.Record(record.Name)
232+ require.False(t, exists)
233+ assertNoTombstones(t, repository)
234+ require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", record.IncarnationID.String()))
235+ require.Equal(t, sourceBytes, readTree(t, filepath.Join(h.snapshotDir, "quarantine", record.IncarnationID.String(), "snapshot")))
236+ require.Equal(t, legacy, mustReadFile(t, filepath.Join(h.snapshotDir, "quarantine", record.IncarnationID.String(), "legacy.snap")))
237+ })
238+ })
239+ }
240+}
241+
242+func TestDurableDiscardCrashMatrix(t *testing.T) {
243+ ctx := context.Background()
244+ oldID := domain.IncarnationID{5}
245+ newID := domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
246+ boundaries := []string{"discard-intent", "discard-descriptor", "discard-quarantine", "catalogue-replace", "discard-intent-delete", "runtime-exposure"}
247+
248+ for _, boundaryName := range boundaries {
249+ t.Run(boundaryName, func(t *testing.T) {
250+ h := newDurableCrashHarness(t)
251+ catalogue, repository, journal := h.open(t)
252+ publication := durablePublication(t, "broken", oldID, []byte("discard-source"))
253+ require.NoError(t, repository.Publish(ctx, publication))
254+ ref := &domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(publication.Manifest)}
255+ old := domain.CatalogueRecord{Name: "broken", IncarnationID: oldID, RecoveryState: domain.RecoveryDegraded, Committed: ref, DegradedReason: "checkpoint unreadable"}
256+ require.NoError(t, catalogue.Create(old))
257+ sourceBytes := readTree(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
258+
259+ boundary := &durableBoundary{want: boundaryName}
260+ coordinator := recovery.NewCoordinator(
261+ crashCatalogue{Catalogue: catalogue, boundary: boundary},
262+ crashRepository{SnapshotRepository: repository, boundary: boundary},
263+ crashJournal{RecoveryJournal: journal, boundary: boundary},
264+ bytes.NewReader(newID[:]),
265+ )
266+ replacement, err := coordinator.Discard(ctx, old.Name, "explicit discard")
267+ if boundaryName == "runtime-exposure" {
268+ require.NoError(t, err)
269+ require.Equal(t, newID, replacement.IncarnationID)
270+ boundary.triggered = true // crash before the returned runtime is installed
271+ } else {
272+ require.ErrorIs(t, err, errInjectedCrash)
273+ }
274+ require.True(t, boundary.triggered)
275+ require.NoError(t, catalogue.Close())
276+
277+ h.recoverTwice(t, func(catalogue *persist.Persister, _ *snapshot.Repository, journal *recoveryfs.Journal) {
278+ got, exists := catalogue.Record(old.Name)
279+ require.True(t, exists)
280+ require.Equal(t, newID, got.IncarnationID)
281+ require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
282+ require.Nil(t, got.Committed)
283+ require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
284+ require.Equal(t, sourceBytes, readTree(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "snapshot")))
285+ intents, err := journal.ListDiscards(ctx)
286+ require.NoError(t, err)
287+ require.Empty(t, intents)
288+ require.FileExists(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "record"))
289+ })
290+ })
291+ }
292+}
293+
294+func TestDeletionRecoveryProtectsReusedNameFilesystemBytes(t *testing.T) {
295+ ctx := context.Background()
296+ h := newDurableCrashHarness(t)
297+ catalogue, repository, _ := h.open(t)
298+ oldID := domain.IncarnationID{6}
299+ replacementID := domain.IncarnationID{7}
300+ oldPublication := durablePublication(t, "work", oldID, []byte("old-incarnation"))
301+ replacementPublication := durablePublication(t, "work", replacementID, []byte("replacement-incarnation"))
302+ require.NoError(t, repository.Publish(ctx, oldPublication))
303+ require.NoError(t, repository.Publish(ctx, replacementPublication))
304+ replacement := domain.CatalogueRecord{Name: "work", IncarnationID: replacementID, RecoveryState: domain.RecoveryFresh}
305+ require.NoError(t, catalogue.Create(replacement))
306+ require.NoError(t, repository.WriteDeletionTombstone(ctx, domain.DeletionTombstone{Name: "work", IncarnationID: oldID}))
307+ legacy := []byte{0xde, 0xad, 0x00, 0xbe, 0xef}
308+ require.NoError(t, os.WriteFile(filepath.Join(h.snapshotDir, "work.snap"), legacy, 0o600))
309+ oldBytes := readTree(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
310+ replacementPath := filepath.Join(h.snapshotDir, "sessions", replacementID.String())
311+ replacementBytes := readTree(t, replacementPath)
312+ require.NoError(t, catalogue.Close())
313+
314+ h.recoverTwice(t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
315+ got, exists := catalogue.Record("work")
316+ require.True(t, exists)
317+ require.Equal(t, replacement, got)
318+ require.Equal(t, replacementBytes, readTree(t, replacementPath))
319+ require.Equal(t, legacy, mustReadFile(t, filepath.Join(h.snapshotDir, "work.snap")))
320+ require.Equal(t, oldBytes, readTree(t, filepath.Join(h.snapshotDir, "quarantine", oldID.String(), "snapshot")))
321+ require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", oldID.String()))
322+ assertNoTombstones(t, repository)
323+ })
324+}
325+
326+func durablePublication(t *testing.T, name string, id domain.IncarnationID, payload []byte) ports.SnapshotPublication {
327+ t.Helper()
328+ tail, err := snapcodec.MarshalObject(snapcodec.HistoryTail, payload)
329+ require.NoError(t, err)
330+ visible, err := snapcodec.MarshalObject(snapcodec.Visible, []byte("visible"))
331+ require.NoError(t, err)
332+ manifest, err := snapcodec.MarshalManifest(snapcodec.Manifest{
333+ Generation: 1, IncarnationID: id, Name: name,
334+ Tabs: []snapcodec.ManifestTab{{Cols: 1, Rows: 1, Panes: []snapcodec.ManifestPane{{
335+ ID: "p", Tail: snapcodec.ObjectRef{Kind: snapcodec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))},
336+ Visible: snapcodec.ObjectRef{Kind: snapcodec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))},
337+ }}}},
338+ })
339+ require.NoError(t, err)
340+ return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: 1, Manifest: manifest, Objects: []ports.SnapshotObject{tail, visible}}
341+}
342+
343+func assertNoTombstones(t *testing.T, repository *snapshot.Repository) {
344+ t.Helper()
345+ page, err := repository.ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10})
346+ require.NoError(t, err)
347+ require.True(t, page.Done)
348+ require.Empty(t, page.Tombstones)
349+}
350+
351+func mustReadFile(t *testing.T, path string) []byte {
352+ t.Helper()
353+ data, err := os.ReadFile(path)
354+ require.NoError(t, err)
355+ return data
356+}
357+
358+func readTree(t *testing.T, root string) map[string][]byte {
359+ t.Helper()
360+ files := make(map[string][]byte)
361+ err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error {
362+ if err != nil {
363+ return err
364+ }
365+ if entry.IsDir() {
366+ return nil
367+ }
368+ relative, err := filepath.Rel(root, path)
369+ if err != nil {
370+ return err
371+ }
372+ data, err := os.ReadFile(path)
373+ if err != nil {
374+ return err
375+ }
376+ files[relative] = data
377+ return nil
378+ })
379+ require.NoError(t, err)
380+ require.NotEmpty(t, files)
381+ return files
382+}
M internal/usecase/recovery/transactions_test.go +0 -128
@@ -5,7 +5,6 @@ import (
55 "context"
66 "crypto/sha256"
77 "errors"
8- "strconv"
98 "sync"
109 "testing"
1110
@@ -174,78 +173,6 @@ func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id do
174173 return nil
175174 }
176175
177-func TestTransactionCrashMatrix(t *testing.T) {
178- ctx := context.Background()
179- t.Run("create", func(t *testing.T) {
180- catalogue := newTransactionCatalogue()
181- coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
182- record, err := coordinator.Create(ctx, domain.CatalogueRecord{Name: "work", Cwd: "/tmp"})
183- require.NoError(t, err)
184- require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
185- require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
186- require.Equal(t, record, catalogue.records["work"])
187- })
188- t.Run("checkpoint", func(t *testing.T) {
189- record := checkpointRecord()
190- catalogue := newTransactionCatalogue(record)
191- repository := &checkpointRepository{}
192- coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
193- next, err := coordinator.PublishCheckpoint(ctx, record.Name, checkpointPublication(record, 1))
194- require.NoError(t, err)
195- require.Equal(t, next, catalogue.records[record.Name])
196- })
197- t.Run("rename", func(t *testing.T) {
198- ref := &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{1}}
199- record := domain.CatalogueRecord{Name: "old", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryHealthy, Committed: ref}
200- catalogue := newTransactionCatalogue(record)
201- coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, nil)
202- next, err := coordinator.Rename(ctx, "old", "new")
203- require.NoError(t, err)
204- _, oldExists := catalogue.records["old"]
205- require.False(t, oldExists)
206- require.Equal(t, record.IncarnationID, catalogue.records["new"].IncarnationID)
207- require.Equal(t, record.Committed, next.Committed)
208- })
209- t.Run("delete", func(t *testing.T) {
210- record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh}
211- tombstone := domain.DeletionTombstone{Name: record.Name, IncarnationID: record.IncarnationID}
212- for crashAfter := 1; crashAfter <= 6; crashAfter++ {
213- t.Run(string(rune('0'+crashAfter)), func(t *testing.T) {
214- catalogue := newTransactionCatalogue()
215- repository := newTransactionRepository()
216- switch crashAfter {
217- case 1:
218- deleting := record
219- deleting.RecoveryState = domain.RecoveryDeleting
220- catalogue.records[record.Name] = deleting
221- case 2, 3, 4:
222- deleting := record
223- deleting.RecoveryState = domain.RecoveryDeleting
224- catalogue.records[record.Name] = deleting
225- repository.tombstones[record.IncarnationID] = tombstone
226- case 5:
227- repository.tombstones[record.IncarnationID] = tombstone
228- case 6:
229- // The complete transaction has no durable work left.
230- }
231- coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
232- require.NoError(t, coordinator.Recover(ctx))
233- _, exists := catalogue.records[record.Name]
234- require.False(t, exists)
235- require.Empty(t, repository.tombstones)
236- })
237- }
238-
239- catalogue := newTransactionCatalogue(record)
240- repository := newTransactionRepository()
241- coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
242- require.NoError(t, coordinator.Delete(ctx, "work"))
243- require.Equal(t, []string{"replace:work", "delete:work"}, catalogue.events)
244- require.Equal(t, []string{"tombstone", "quarantine", "remove-tombstone"}, repository.events)
245- require.Equal(t, []bool{true}, repository.quarantine)
246- })
247-}
248-
249176 type transactionJournal struct {
250177 mu sync.Mutex
251178 intents map[domain.IncarnationID]domain.DiscardIntent
@@ -353,61 +280,6 @@ func TestDegradedExportIsReadOnly(t *testing.T) {
353280 require.Empty(t, catalogue.events)
354281 }
355282
356-func TestDiscardCrashMatrix(t *testing.T) {
357- ctx := context.Background()
358- old := degradedTransactionRecord()
359- newID := domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
360- intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: newID, SessionName: old.Name, Reason: "explicit discard"}
361-
362- t.Run("commit order", func(t *testing.T) {
363- catalogue := newTransactionCatalogue(old)
364- repository := newTransactionRepository()
365- journal := newTransactionJournal()
366- coordinator := NewCoordinator(catalogue, repository, journal, bytes.NewReader(newID[:]))
367- got, err := coordinator.Discard(ctx, old.Name, intent.Reason)
368- require.NoError(t, err)
369- require.Equal(t, freshReplacement(intent), got)
370- require.Equal(t, []string{"intent", "delete-intent"}, journal.events)
371- require.Equal(t, []string{"descriptor", "quarantine-incarnation"}, repository.events)
372- require.Equal(t, []string{"replace:" + old.Name}, catalogue.events)
373- require.Empty(t, journal.intents)
374- })
375-
376- for step := 1; step <= 6; step++ {
377- t.Run(strconv.Itoa(step), func(t *testing.T) {
378- catalogue := newTransactionCatalogue(old)
379- repository := newTransactionRepository()
380- journal := newTransactionJournal()
381- if step >= 1 {
382- journal.intents[old.IncarnationID] = intent
383- }
384- if step >= 2 {
385- repository.descriptors = append(repository.descriptors, quarantineDescriptor(intent))
386- }
387- if step >= 3 {
388- repository.quarantinedIDs = append(repository.quarantinedIDs, old.IncarnationID)
389- }
390- if step >= 4 {
391- catalogue.records[old.Name] = freshReplacement(intent)
392- }
393- if step >= 5 {
394- delete(journal.intents, old.IncarnationID)
395- }
396- if step == 6 {
397- // Runtime exposure is deliberately outside durable recovery.
398- }
399- coordinator := NewCoordinator(catalogue, repository, journal, nil)
400- require.NoError(t, coordinator.Recover(ctx))
401- got := catalogue.records[old.Name]
402- require.Equal(t, newID, got.IncarnationID)
403- require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
404- require.NotEmpty(t, repository.descriptors)
405- require.NotEmpty(t, repository.quarantinedIDs)
406- require.Empty(t, journal.intents)
407- })
408- }
409-}
410-
411283 func TestDiscardIncarnationConflict(t *testing.T) {
412284 old := degradedTransactionRecord()
413285 intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}
@@ -5,7 +5,6 @@ import (
5 "context"5 "context"
6 "crypto/sha256"6 "crypto/sha256"
7 "errors"7 "errors"
8- "strconv"
9 "sync"8 "sync"
10 "testing"9 "testing"
11 10
@@ -174,78 +173,6 @@ func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id do
174 return nil173 return nil
175 }174 }
176 175
177-func TestTransactionCrashMatrix(t *testing.T) {
178- ctx := context.Background()
179- t.Run("create", func(t *testing.T) {
180- catalogue := newTransactionCatalogue()
181- coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
182- record, err := coordinator.Create(ctx, domain.CatalogueRecord{Name: "work", Cwd: "/tmp"})
183- require.NoError(t, err)
184- require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
185- require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
186- require.Equal(t, record, catalogue.records["work"])
187- })
188- t.Run("checkpoint", func(t *testing.T) {
189- record := checkpointRecord()
190- catalogue := newTransactionCatalogue(record)
191- repository := &checkpointRepository{}
192- coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
193- next, err := coordinator.PublishCheckpoint(ctx, record.Name, checkpointPublication(record, 1))
194- require.NoError(t, err)
195- require.Equal(t, next, catalogue.records[record.Name])
196- })
197- t.Run("rename", func(t *testing.T) {
198- ref := &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{1}}
199- record := domain.CatalogueRecord{Name: "old", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryHealthy, Committed: ref}
200- catalogue := newTransactionCatalogue(record)
201- coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, nil)
202- next, err := coordinator.Rename(ctx, "old", "new")
203- require.NoError(t, err)
204- _, oldExists := catalogue.records["old"]
205- require.False(t, oldExists)
206- require.Equal(t, record.IncarnationID, catalogue.records["new"].IncarnationID)
207- require.Equal(t, record.Committed, next.Committed)
208- })
209- t.Run("delete", func(t *testing.T) {
210- record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh}
211- tombstone := domain.DeletionTombstone{Name: record.Name, IncarnationID: record.IncarnationID}
212- for crashAfter := 1; crashAfter <= 6; crashAfter++ {
213- t.Run(string(rune('0'+crashAfter)), func(t *testing.T) {
214- catalogue := newTransactionCatalogue()
215- repository := newTransactionRepository()
216- switch crashAfter {
217- case 1:
218- deleting := record
219- deleting.RecoveryState = domain.RecoveryDeleting
220- catalogue.records[record.Name] = deleting
221- case 2, 3, 4:
222- deleting := record
223- deleting.RecoveryState = domain.RecoveryDeleting
224- catalogue.records[record.Name] = deleting
225- repository.tombstones[record.IncarnationID] = tombstone
226- case 5:
227- repository.tombstones[record.IncarnationID] = tombstone
228- case 6:
229- // The complete transaction has no durable work left.
230- }
231- coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
232- require.NoError(t, coordinator.Recover(ctx))
233- _, exists := catalogue.records[record.Name]
234- require.False(t, exists)
235- require.Empty(t, repository.tombstones)
236- })
237- }
238-
239- catalogue := newTransactionCatalogue(record)
240- repository := newTransactionRepository()
241- coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
242- require.NoError(t, coordinator.Delete(ctx, "work"))
243- require.Equal(t, []string{"replace:work", "delete:work"}, catalogue.events)
244- require.Equal(t, []string{"tombstone", "quarantine", "remove-tombstone"}, repository.events)
245- require.Equal(t, []bool{true}, repository.quarantine)
246- })
247-}
248-
249 type transactionJournal struct {176 type transactionJournal struct {
250 mu sync.Mutex177 mu sync.Mutex
251 intents map[domain.IncarnationID]domain.DiscardIntent178 intents map[domain.IncarnationID]domain.DiscardIntent
@@ -353,61 +280,6 @@ func TestDegradedExportIsReadOnly(t *testing.T) {
353 require.Empty(t, catalogue.events)280 require.Empty(t, catalogue.events)
354 }281 }
355 282
356-func TestDiscardCrashMatrix(t *testing.T) {
357- ctx := context.Background()
358- old := degradedTransactionRecord()
359- newID := domain.IncarnationID{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
360- intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: newID, SessionName: old.Name, Reason: "explicit discard"}
361-
362- t.Run("commit order", func(t *testing.T) {
363- catalogue := newTransactionCatalogue(old)
364- repository := newTransactionRepository()
365- journal := newTransactionJournal()
366- coordinator := NewCoordinator(catalogue, repository, journal, bytes.NewReader(newID[:]))
367- got, err := coordinator.Discard(ctx, old.Name, intent.Reason)
368- require.NoError(t, err)
369- require.Equal(t, freshReplacement(intent), got)
370- require.Equal(t, []string{"intent", "delete-intent"}, journal.events)
371- require.Equal(t, []string{"descriptor", "quarantine-incarnation"}, repository.events)
372- require.Equal(t, []string{"replace:" + old.Name}, catalogue.events)
373- require.Empty(t, journal.intents)
374- })
375-
376- for step := 1; step <= 6; step++ {
377- t.Run(strconv.Itoa(step), func(t *testing.T) {
378- catalogue := newTransactionCatalogue(old)
379- repository := newTransactionRepository()
380- journal := newTransactionJournal()
381- if step >= 1 {
382- journal.intents[old.IncarnationID] = intent
383- }
384- if step >= 2 {
385- repository.descriptors = append(repository.descriptors, quarantineDescriptor(intent))
386- }
387- if step >= 3 {
388- repository.quarantinedIDs = append(repository.quarantinedIDs, old.IncarnationID)
389- }
390- if step >= 4 {
391- catalogue.records[old.Name] = freshReplacement(intent)
392- }
393- if step >= 5 {
394- delete(journal.intents, old.IncarnationID)
395- }
396- if step == 6 {
397- // Runtime exposure is deliberately outside durable recovery.
398- }
399- coordinator := NewCoordinator(catalogue, repository, journal, nil)
400- require.NoError(t, coordinator.Recover(ctx))
401- got := catalogue.records[old.Name]
402- require.Equal(t, newID, got.IncarnationID)
403- require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
404- require.NotEmpty(t, repository.descriptors)
405- require.NotEmpty(t, repository.quarantinedIDs)
406- require.Empty(t, journal.intents)
407- })
408- }
409-}
410-
411 func TestDiscardIncarnationConflict(t *testing.T) {283 func TestDiscardIncarnationConflict(t *testing.T) {
412 old := degradedTransactionRecord()284 old := degradedTransactionRecord()
413 intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}285 intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}

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