detached 00a4e55

feat(persist): open or migrate durable state once

commit
00a4e55886918230af1838e5b51477ef49e6160a
author
brice · · 2026-07-25 20:57 UTC
parent
f90cadc
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
5 files · +1210 -0
Unified
UnifiedSplit
A internal/adapters/snapshot/migration_v0.go +289 -0
@@ -0,0 +1,289 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "encoding/binary"
7+ "encoding/hex"
8+ "errors"
9+ "fmt"
10+ "hash/crc32"
11+ "io"
12+ "os"
13+ "path/filepath"
14+ "strings"
15+
16+ "github.com/bnema/vev/internal/domain"
17+ "github.com/bnema/vev/internal/ports"
18+ codec "github.com/bnema/vev/internal/usecase/snapshot"
19+)
20+
21+var ErrLegacyTraversalBudgetExceeded = errors.New("snapshot: legacy presence traversal budget exceeded")
22+
23+const (
24+ legacyPresenceEntryLimit = 4096
25+ legacyIncrementalSessionsDir = "sessions"
26+ legacyManifestHeaderSize = 16
27+)
28+
29+type legacyManifestV1 struct {
30+ Name string
31+ Generation uint64
32+ Objects []legacyObjectV1
33+}
34+
35+type legacyObjectV1 struct {
36+ Kind uint8
37+ Digest ports.SnapshotDigest
38+ Size uint64
39+}
40+
41+func isLegacyIncrementalSessionNamespace(name string) bool { return canonicalSessionKey(name) }
42+
43+func decodeManifestV1(data []byte) (legacyManifestV1, error) {
44+ manifest, _, err := upgradeManifestV1(data, domain.IncarnationID{1})
45+ if err != nil {
46+ return legacyManifestV1{}, err
47+ }
48+ seen := make(map[ports.SnapshotDigest]struct{})
49+ objects := make([]legacyObjectV1, 0)
50+ add := func(ref codec.ObjectRef) error {
51+ if _, ok := seen[ref.Digest]; ok {
52+ return errors.New("snapshot: duplicate legacy object digest")
53+ }
54+ seen[ref.Digest] = struct{}{}
55+ objects = append(objects, legacyObjectV1{Kind: uint8(ref.Kind), Digest: ref.Digest, Size: uint64(ref.Size)})
56+ return nil
57+ }
58+ for _, tab := range manifest.Tabs {
59+ for _, pane := range tab.Panes {
60+ for _, ref := range pane.Sealed {
61+ if err := add(ref); err != nil {
62+ return legacyManifestV1{}, err
63+ }
64+ }
65+ if err := add(pane.Tail); err != nil {
66+ return legacyManifestV1{}, err
67+ }
68+ if err := add(pane.Visible); err != nil {
69+ return legacyManifestV1{}, err
70+ }
71+ }
72+ }
73+ return legacyManifestV1{Name: manifest.Name, Generation: manifest.Generation, Objects: objects}, nil
74+}
75+
76+func upgradeManifestV1(data []byte, id domain.IncarnationID) (codec.Manifest, []byte, error) {
77+ if len(data) < legacyManifestHeaderSize {
78+ return codec.Manifest{}, nil, errors.New("snapshot: short legacy manifest")
79+ }
80+ if string(data[:4]) != "VEVM" || binary.BigEndian.Uint16(data[4:6]) != 1 || binary.BigEndian.Uint16(data[6:8]) != 0 {
81+ return codec.Manifest{}, nil, errors.New("snapshot: invalid legacy manifest header")
82+ }
83+ n := binary.BigEndian.Uint32(data[8:12])
84+ if uint64(len(data)) != legacyManifestHeaderSize+uint64(n) {
85+ return codec.Manifest{}, nil, errors.New("snapshot: invalid legacy manifest length")
86+ }
87+ body := data[legacyManifestHeaderSize:]
88+ if crc32.ChecksumIEEE(body) != binary.BigEndian.Uint32(data[12:16]) || len(body) < 8 {
89+ return codec.Manifest{}, nil, errors.New("snapshot: invalid legacy manifest checksum")
90+ }
91+ upgradedBody := make([]byte, 0, len(body)+17)
92+ upgradedBody = append(upgradedBody, body[:8]...)
93+ upgradedBody = append(upgradedBody, id[:]...)
94+ upgradedBody = append(upgradedBody, 0)
95+ upgradedBody = append(upgradedBody, body[8:]...)
96+ upgraded := make([]byte, legacyManifestHeaderSize, legacyManifestHeaderSize+len(upgradedBody))
97+ copy(upgraded, "VEVM")
98+ binary.BigEndian.PutUint16(upgraded[4:6], codec.ManifestVersion)
99+ binary.BigEndian.PutUint32(upgraded[8:12], uint32(len(upgradedBody)))
100+ binary.BigEndian.PutUint32(upgraded[12:16], crc32.ChecksumIEEE(upgradedBody))
101+ upgraded = append(upgraded, upgradedBody...)
102+ manifest, err := codec.UnmarshalManifest(upgraded)
103+ if err != nil {
104+ return codec.Manifest{}, nil, fmt.Errorf("snapshot: decode legacy manifest: %w", err)
105+ }
106+ return manifest, upgraded, nil
107+}
108+
109+func (r *Repository) HasLegacyState(ctx context.Context) (found bool, err error) {
110+ visited := 0
111+ visit := func(dir string, recognize func(os.DirEntry, os.FileInfo) bool) (found bool, err error) {
112+ f, err := r.openLegacyDirectory(dir)
113+ if errors.Is(err, os.ErrNotExist) {
114+ return false, nil
115+ }
116+ if err != nil {
117+ return false, err
118+ }
119+ defer func() {
120+ if closeErr := f.Close(); err == nil && closeErr != nil {
121+ err = closeErr
122+ }
123+ }()
124+ for {
125+ if err := ctx.Err(); err != nil {
126+ return false, err
127+ }
128+ entries, readErr := f.ReadDir(directoryTraversalBatch)
129+ for _, entry := range entries {
130+ if visited >= legacyPresenceEntryLimit {
131+ return false, ErrLegacyTraversalBudgetExceeded
132+ }
133+ visited++
134+ info, infoErr := entry.Info()
135+ if infoErr != nil {
136+ return false, infoErr
137+ }
138+ if recognize(entry, info) {
139+ return true, nil
140+ }
141+ }
142+ if errors.Is(readErr, io.EOF) {
143+ return false, nil
144+ }
145+ if readErr != nil {
146+ return false, readErr
147+ }
148+ }
149+ }
150+ found, err = visit(r.dir, func(e os.DirEntry, info os.FileInfo) bool {
151+ return info.Mode().IsRegular() && isLegacyBlobName(e.Name())
152+ })
153+ if found || err != nil {
154+ return found, err
155+ }
156+ return visit(filepath.Join(r.dir, legacyIncrementalSessionsDir), func(e os.DirEntry, info os.FileInfo) bool {
157+ return info.IsDir() && isLegacyIncrementalSessionNamespace(e.Name())
158+ })
159+}
160+
161+func isLegacyBlobName(name string) bool {
162+ if !strings.HasSuffix(name, ".snap") {
163+ return false
164+ }
165+ key := strings.TrimSuffix(name, ".snap")
166+ if safeNameRE.MatchString(key) && key != "." && key != ".." {
167+ return true
168+ }
169+ if len(key) != 41 || key[0] != '@' || strings.ToLower(key) != key {
170+ return false
171+ }
172+ _, err := hex.DecodeString(key[1:])
173+ return err == nil
174+}
175+
176+func (r *Repository) ReadLegacyHEAD(ctx context.Context, name string) (domain.CheckpointRef, error) {
177+ if err := ctx.Err(); err != nil {
178+ return domain.CheckpointRef{}, err
179+ }
180+ key := sessionKey(name)
181+ generation, digest, err := r.readHead(key)
182+ if err != nil {
183+ if errors.Is(err, os.ErrNotExist) || errors.Is(err, ErrInvalidHEAD) {
184+ return domain.CheckpointRef{}, uncertainLegacyError("read HEAD", err)
185+ }
186+ return domain.CheckpointRef{}, fmt.Errorf("snapshot: read legacy HEAD: %w", err)
187+ }
188+ data, err := r.readBounded(r.manifestPath(key, generation))
189+ if err != nil {
190+ if errors.Is(err, os.ErrNotExist) {
191+ return domain.CheckpointRef{}, uncertainLegacyError("read referenced manifest", err)
192+ }
193+ return domain.CheckpointRef{}, err
194+ }
195+ if sha256.Sum256(data) != digest {
196+ return domain.CheckpointRef{}, uncertainLegacyError("HEAD digest mismatch", nil)
197+ }
198+ legacy, err := decodeManifestV1(data)
199+ if err != nil {
200+ return domain.CheckpointRef{}, uncertainLegacyError("decode referenced manifest", err)
201+ }
202+ if legacy.Name != name || legacy.Generation != generation {
203+ return domain.CheckpointRef{}, uncertainLegacyError("manifest identity mismatch", nil)
204+ }
205+ return domain.CheckpointRef{Generation: generation, ManifestDigest: digest}, nil
206+}
207+
208+func uncertainLegacyError(operation string, cause error) error {
209+ if cause == nil {
210+ return fmt.Errorf("%w: %s", ports.ErrLegacySnapshotUncertain, operation)
211+ }
212+ return fmt.Errorf("%w: %s: %v", ports.ErrLegacySnapshotUncertain, operation, cause)
213+}
214+
215+func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {
216+ if err := ctx.Err(); err != nil {
217+ return domain.CheckpointRef{}, err
218+ }
219+ if req.IncarnationID == (domain.IncarnationID{}) || req.LegacyRef.Generation == 0 {
220+ return domain.CheckpointRef{}, errors.New("snapshot: invalid migration request")
221+ }
222+ legacyKey := sessionKey(req.LegacyName)
223+ data, err := r.readBounded(r.manifestPath(legacyKey, req.LegacyRef.Generation))
224+ if err != nil {
225+ if errors.Is(err, os.ErrNotExist) {
226+ return domain.CheckpointRef{}, uncertainLegacyError("referenced manifest missing", err)
227+ }
228+ return domain.CheckpointRef{}, err
229+ }
230+ if sha256.Sum256(data) != req.LegacyRef.ManifestDigest {
231+ return domain.CheckpointRef{}, uncertainLegacyError("manifest digest mismatch", nil)
232+ }
233+ manifest, upgraded, err := upgradeManifestV1(data, req.IncarnationID)
234+ if err != nil {
235+ return domain.CheckpointRef{}, uncertainLegacyError("decode manifest", err)
236+ }
237+ decoded, err := decodeManifestV1(data)
238+ if err != nil {
239+ return domain.CheckpointRef{}, uncertainLegacyError("decode manifest objects", err)
240+ }
241+ if manifest.Name != req.LegacyName || manifest.Generation != req.LegacyRef.Generation {
242+ return domain.CheckpointRef{}, uncertainLegacyError("manifest identity mismatch", nil)
243+ }
244+ key, _ := incarnationKey(req.IncarnationID)
245+ lock := r.lockSession(key)
246+ defer r.unlockSession(lock)
247+ if err := r.ensureSession(key); err != nil {
248+ return domain.CheckpointRef{}, err
249+ }
250+ for _, ref := range decoded.Objects {
251+ if err := ctx.Err(); err != nil {
252+ return domain.CheckpointRef{}, err
253+ }
254+ object, err := r.readBounded(r.objectPath(legacyKey, ref.Digest))
255+ if err != nil {
256+ if errors.Is(err, os.ErrNotExist) {
257+ return domain.CheckpointRef{}, uncertainLegacyError("referenced object missing", err)
258+ }
259+ return domain.CheckpointRef{}, err
260+ }
261+ if sha256.Sum256(object) != ref.Digest || uint64(len(object)) != ref.Size {
262+ return domain.CheckpointRef{}, uncertainLegacyError("invalid object", nil)
263+ }
264+ if err := r.writeImmutable(r.objectPath(req.IncarnationID, ref.Digest), object, func(existing []byte) error {
265+ if sha256.Sum256(existing) != ref.Digest {
266+ return uncertainLegacyError("migrated object conflict", nil)
267+ }
268+ return nil
269+ }); err != nil {
270+ return domain.CheckpointRef{}, err
271+ }
272+ }
273+ path := r.manifestPath(req.IncarnationID, manifest.Generation)
274+ if err := r.writeImmutable(path, upgraded, func(existing []byte) error {
275+ if sha256.Sum256(existing) != sha256.Sum256(upgraded) {
276+ return uncertainLegacyError("migrated manifest conflict", nil)
277+ }
278+ return nil
279+ }); err != nil {
280+ return domain.CheckpointRef{}, err
281+ }
282+ ref := domain.CheckpointRef{Generation: manifest.Generation, ManifestDigest: sha256.Sum256(upgraded)}
283+ if err := r.writeMutable(r.headPath(req.IncarnationID), marshalHead(ref.Generation, ref.ManifestDigest)); err != nil {
284+ return domain.CheckpointRef{}, err
285+ }
286+ return ref, nil
287+}
288+
289+var _ ports.SnapshotMigration = (*Repository)(nil)
@@ -0,0 +1,289 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "encoding/binary"
7+ "encoding/hex"
8+ "errors"
9+ "fmt"
10+ "hash/crc32"
11+ "io"
12+ "os"
13+ "path/filepath"
14+ "strings"
15+
16+ "github.com/bnema/vev/internal/domain"
17+ "github.com/bnema/vev/internal/ports"
18+ codec "github.com/bnema/vev/internal/usecase/snapshot"
19+)
20+
21+var ErrLegacyTraversalBudgetExceeded = errors.New("snapshot: legacy presence traversal budget exceeded")
22+
23+const (
24+ legacyPresenceEntryLimit = 4096
25+ legacyIncrementalSessionsDir = "sessions"
26+ legacyManifestHeaderSize = 16
27+)
28+
29+type legacyManifestV1 struct {
30+ Name string
31+ Generation uint64
32+ Objects []legacyObjectV1
33+}
34+
35+type legacyObjectV1 struct {
36+ Kind uint8
37+ Digest ports.SnapshotDigest
38+ Size uint64
39+}
40+
41+func isLegacyIncrementalSessionNamespace(name string) bool { return canonicalSessionKey(name) }
42+
43+func decodeManifestV1(data []byte) (legacyManifestV1, error) {
44+ manifest, _, err := upgradeManifestV1(data, domain.IncarnationID{1})
45+ if err != nil {
46+ return legacyManifestV1{}, err
47+ }
48+ seen := make(map[ports.SnapshotDigest]struct{})
49+ objects := make([]legacyObjectV1, 0)
50+ add := func(ref codec.ObjectRef) error {
51+ if _, ok := seen[ref.Digest]; ok {
52+ return errors.New("snapshot: duplicate legacy object digest")
53+ }
54+ seen[ref.Digest] = struct{}{}
55+ objects = append(objects, legacyObjectV1{Kind: uint8(ref.Kind), Digest: ref.Digest, Size: uint64(ref.Size)})
56+ return nil
57+ }
58+ for _, tab := range manifest.Tabs {
59+ for _, pane := range tab.Panes {
60+ for _, ref := range pane.Sealed {
61+ if err := add(ref); err != nil {
62+ return legacyManifestV1{}, err
63+ }
64+ }
65+ if err := add(pane.Tail); err != nil {
66+ return legacyManifestV1{}, err
67+ }
68+ if err := add(pane.Visible); err != nil {
69+ return legacyManifestV1{}, err
70+ }
71+ }
72+ }
73+ return legacyManifestV1{Name: manifest.Name, Generation: manifest.Generation, Objects: objects}, nil
74+}
75+
76+func upgradeManifestV1(data []byte, id domain.IncarnationID) (codec.Manifest, []byte, error) {
77+ if len(data) < legacyManifestHeaderSize {
78+ return codec.Manifest{}, nil, errors.New("snapshot: short legacy manifest")
79+ }
80+ if string(data[:4]) != "VEVM" || binary.BigEndian.Uint16(data[4:6]) != 1 || binary.BigEndian.Uint16(data[6:8]) != 0 {
81+ return codec.Manifest{}, nil, errors.New("snapshot: invalid legacy manifest header")
82+ }
83+ n := binary.BigEndian.Uint32(data[8:12])
84+ if uint64(len(data)) != legacyManifestHeaderSize+uint64(n) {
85+ return codec.Manifest{}, nil, errors.New("snapshot: invalid legacy manifest length")
86+ }
87+ body := data[legacyManifestHeaderSize:]
88+ if crc32.ChecksumIEEE(body) != binary.BigEndian.Uint32(data[12:16]) || len(body) < 8 {
89+ return codec.Manifest{}, nil, errors.New("snapshot: invalid legacy manifest checksum")
90+ }
91+ upgradedBody := make([]byte, 0, len(body)+17)
92+ upgradedBody = append(upgradedBody, body[:8]...)
93+ upgradedBody = append(upgradedBody, id[:]...)
94+ upgradedBody = append(upgradedBody, 0)
95+ upgradedBody = append(upgradedBody, body[8:]...)
96+ upgraded := make([]byte, legacyManifestHeaderSize, legacyManifestHeaderSize+len(upgradedBody))
97+ copy(upgraded, "VEVM")
98+ binary.BigEndian.PutUint16(upgraded[4:6], codec.ManifestVersion)
99+ binary.BigEndian.PutUint32(upgraded[8:12], uint32(len(upgradedBody)))
100+ binary.BigEndian.PutUint32(upgraded[12:16], crc32.ChecksumIEEE(upgradedBody))
101+ upgraded = append(upgraded, upgradedBody...)
102+ manifest, err := codec.UnmarshalManifest(upgraded)
103+ if err != nil {
104+ return codec.Manifest{}, nil, fmt.Errorf("snapshot: decode legacy manifest: %w", err)
105+ }
106+ return manifest, upgraded, nil
107+}
108+
109+func (r *Repository) HasLegacyState(ctx context.Context) (found bool, err error) {
110+ visited := 0
111+ visit := func(dir string, recognize func(os.DirEntry, os.FileInfo) bool) (found bool, err error) {
112+ f, err := r.openLegacyDirectory(dir)
113+ if errors.Is(err, os.ErrNotExist) {
114+ return false, nil
115+ }
116+ if err != nil {
117+ return false, err
118+ }
119+ defer func() {
120+ if closeErr := f.Close(); err == nil && closeErr != nil {
121+ err = closeErr
122+ }
123+ }()
124+ for {
125+ if err := ctx.Err(); err != nil {
126+ return false, err
127+ }
128+ entries, readErr := f.ReadDir(directoryTraversalBatch)
129+ for _, entry := range entries {
130+ if visited >= legacyPresenceEntryLimit {
131+ return false, ErrLegacyTraversalBudgetExceeded
132+ }
133+ visited++
134+ info, infoErr := entry.Info()
135+ if infoErr != nil {
136+ return false, infoErr
137+ }
138+ if recognize(entry, info) {
139+ return true, nil
140+ }
141+ }
142+ if errors.Is(readErr, io.EOF) {
143+ return false, nil
144+ }
145+ if readErr != nil {
146+ return false, readErr
147+ }
148+ }
149+ }
150+ found, err = visit(r.dir, func(e os.DirEntry, info os.FileInfo) bool {
151+ return info.Mode().IsRegular() && isLegacyBlobName(e.Name())
152+ })
153+ if found || err != nil {
154+ return found, err
155+ }
156+ return visit(filepath.Join(r.dir, legacyIncrementalSessionsDir), func(e os.DirEntry, info os.FileInfo) bool {
157+ return info.IsDir() && isLegacyIncrementalSessionNamespace(e.Name())
158+ })
159+}
160+
161+func isLegacyBlobName(name string) bool {
162+ if !strings.HasSuffix(name, ".snap") {
163+ return false
164+ }
165+ key := strings.TrimSuffix(name, ".snap")
166+ if safeNameRE.MatchString(key) && key != "." && key != ".." {
167+ return true
168+ }
169+ if len(key) != 41 || key[0] != '@' || strings.ToLower(key) != key {
170+ return false
171+ }
172+ _, err := hex.DecodeString(key[1:])
173+ return err == nil
174+}
175+
176+func (r *Repository) ReadLegacyHEAD(ctx context.Context, name string) (domain.CheckpointRef, error) {
177+ if err := ctx.Err(); err != nil {
178+ return domain.CheckpointRef{}, err
179+ }
180+ key := sessionKey(name)
181+ generation, digest, err := r.readHead(key)
182+ if err != nil {
183+ if errors.Is(err, os.ErrNotExist) || errors.Is(err, ErrInvalidHEAD) {
184+ return domain.CheckpointRef{}, uncertainLegacyError("read HEAD", err)
185+ }
186+ return domain.CheckpointRef{}, fmt.Errorf("snapshot: read legacy HEAD: %w", err)
187+ }
188+ data, err := r.readBounded(r.manifestPath(key, generation))
189+ if err != nil {
190+ if errors.Is(err, os.ErrNotExist) {
191+ return domain.CheckpointRef{}, uncertainLegacyError("read referenced manifest", err)
192+ }
193+ return domain.CheckpointRef{}, err
194+ }
195+ if sha256.Sum256(data) != digest {
196+ return domain.CheckpointRef{}, uncertainLegacyError("HEAD digest mismatch", nil)
197+ }
198+ legacy, err := decodeManifestV1(data)
199+ if err != nil {
200+ return domain.CheckpointRef{}, uncertainLegacyError("decode referenced manifest", err)
201+ }
202+ if legacy.Name != name || legacy.Generation != generation {
203+ return domain.CheckpointRef{}, uncertainLegacyError("manifest identity mismatch", nil)
204+ }
205+ return domain.CheckpointRef{Generation: generation, ManifestDigest: digest}, nil
206+}
207+
208+func uncertainLegacyError(operation string, cause error) error {
209+ if cause == nil {
210+ return fmt.Errorf("%w: %s", ports.ErrLegacySnapshotUncertain, operation)
211+ }
212+ return fmt.Errorf("%w: %s: %v", ports.ErrLegacySnapshotUncertain, operation, cause)
213+}
214+
215+func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {
216+ if err := ctx.Err(); err != nil {
217+ return domain.CheckpointRef{}, err
218+ }
219+ if req.IncarnationID == (domain.IncarnationID{}) || req.LegacyRef.Generation == 0 {
220+ return domain.CheckpointRef{}, errors.New("snapshot: invalid migration request")
221+ }
222+ legacyKey := sessionKey(req.LegacyName)
223+ data, err := r.readBounded(r.manifestPath(legacyKey, req.LegacyRef.Generation))
224+ if err != nil {
225+ if errors.Is(err, os.ErrNotExist) {
226+ return domain.CheckpointRef{}, uncertainLegacyError("referenced manifest missing", err)
227+ }
228+ return domain.CheckpointRef{}, err
229+ }
230+ if sha256.Sum256(data) != req.LegacyRef.ManifestDigest {
231+ return domain.CheckpointRef{}, uncertainLegacyError("manifest digest mismatch", nil)
232+ }
233+ manifest, upgraded, err := upgradeManifestV1(data, req.IncarnationID)
234+ if err != nil {
235+ return domain.CheckpointRef{}, uncertainLegacyError("decode manifest", err)
236+ }
237+ decoded, err := decodeManifestV1(data)
238+ if err != nil {
239+ return domain.CheckpointRef{}, uncertainLegacyError("decode manifest objects", err)
240+ }
241+ if manifest.Name != req.LegacyName || manifest.Generation != req.LegacyRef.Generation {
242+ return domain.CheckpointRef{}, uncertainLegacyError("manifest identity mismatch", nil)
243+ }
244+ key, _ := incarnationKey(req.IncarnationID)
245+ lock := r.lockSession(key)
246+ defer r.unlockSession(lock)
247+ if err := r.ensureSession(key); err != nil {
248+ return domain.CheckpointRef{}, err
249+ }
250+ for _, ref := range decoded.Objects {
251+ if err := ctx.Err(); err != nil {
252+ return domain.CheckpointRef{}, err
253+ }
254+ object, err := r.readBounded(r.objectPath(legacyKey, ref.Digest))
255+ if err != nil {
256+ if errors.Is(err, os.ErrNotExist) {
257+ return domain.CheckpointRef{}, uncertainLegacyError("referenced object missing", err)
258+ }
259+ return domain.CheckpointRef{}, err
260+ }
261+ if sha256.Sum256(object) != ref.Digest || uint64(len(object)) != ref.Size {
262+ return domain.CheckpointRef{}, uncertainLegacyError("invalid object", nil)
263+ }
264+ if err := r.writeImmutable(r.objectPath(req.IncarnationID, ref.Digest), object, func(existing []byte) error {
265+ if sha256.Sum256(existing) != ref.Digest {
266+ return uncertainLegacyError("migrated object conflict", nil)
267+ }
268+ return nil
269+ }); err != nil {
270+ return domain.CheckpointRef{}, err
271+ }
272+ }
273+ path := r.manifestPath(req.IncarnationID, manifest.Generation)
274+ if err := r.writeImmutable(path, upgraded, func(existing []byte) error {
275+ if sha256.Sum256(existing) != sha256.Sum256(upgraded) {
276+ return uncertainLegacyError("migrated manifest conflict", nil)
277+ }
278+ return nil
279+ }); err != nil {
280+ return domain.CheckpointRef{}, err
281+ }
282+ ref := domain.CheckpointRef{Generation: manifest.Generation, ManifestDigest: sha256.Sum256(upgraded)}
283+ if err := r.writeMutable(r.headPath(req.IncarnationID), marshalHead(ref.Generation, ref.ManifestDigest)); err != nil {
284+ return domain.CheckpointRef{}, err
285+ }
286+ return ref, nil
287+}
288+
289+var _ ports.SnapshotMigration = (*Repository)(nil)
A internal/adapters/snapshot/migration_v0_test.go +109 -0
@@ -0,0 +1,109 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "encoding/binary"
7+ "fmt"
8+ "hash/crc32"
9+ "os"
10+ "path/filepath"
11+ "testing"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+ codec "github.com/bnema/vev/internal/usecase/snapshot"
16+ "github.com/stretchr/testify/require"
17+)
18+
19+func legacyManifestBytes(name string, generation uint64) []byte {
20+ body := binary.BigEndian.AppendUint64(nil, generation)
21+ body = binary.BigEndian.AppendUint16(body, uint16(len(name)))
22+ body = append(body, name...)
23+ body = binary.BigEndian.AppendUint64(body, 1)
24+ body = binary.BigEndian.AppendUint16(body, 0)
25+ body = binary.BigEndian.AppendUint16(body, 0)
26+ out := make([]byte, 16, 16+len(body))
27+ copy(out, "VEVM")
28+ binary.BigEndian.PutUint16(out[4:6], 1)
29+ binary.BigEndian.PutUint32(out[8:12], uint32(len(body)))
30+ binary.BigEndian.PutUint32(out[12:16], crc32.ChecksumIEEE(body))
31+ return append(out, body...)
32+}
33+
34+func TestLegacyManifestV1(t *testing.T) {
35+ encoded := legacyManifestBytes("work", 7)
36+ got, err := decodeManifestV1(encoded)
37+ require.NoError(t, err)
38+ require.Equal(t, "work", got.Name)
39+ require.Equal(t, uint64(7), got.Generation)
40+ for _, invalid := range [][]byte{encoded[:5], append(append([]byte(nil), encoded...), 0)} {
41+ _, err := decodeManifestV1(invalid)
42+ require.Error(t, err)
43+ }
44+}
45+
46+func TestHasLegacyState(t *testing.T) {
47+ for _, tc := range []struct {
48+ name string
49+ setup func(string)
50+ want bool
51+ budget bool
52+ }{
53+ {"absence", func(string) {}, false, false},
54+ {"top level blob", func(dir string) {
55+ require.NoError(t, os.WriteFile(filepath.Join(dir, "work.snap"), []byte("x"), 0o600))
56+ }, true, false},
57+ {"incremental child", func(dir string) { require.NoError(t, os.MkdirAll(filepath.Join(dir, "sessions", "work"), 0o700)) }, true, false},
58+ {"budget exhaustion", func(dir string) {
59+ for i := 0; i <= legacyPresenceEntryLimit; i++ {
60+ require.NoError(t, os.WriteFile(filepath.Join(dir, fmt.Sprintf("x-%04d", i)), nil, 0o600))
61+ }
62+ }, false, true},
63+ } {
64+ t.Run(tc.name, func(t *testing.T) {
65+ dir := t.TempDir()
66+ require.NoError(t, os.Chmod(dir, 0o700))
67+ tc.setup(dir)
68+ got, err := NewRepository(dir).HasLegacyState(context.Background())
69+ if tc.budget {
70+ require.ErrorIs(t, err, ErrLegacyTraversalBudgetExceeded)
71+ } else {
72+ require.NoError(t, err)
73+ require.Equal(t, tc.want, got)
74+ }
75+ })
76+ }
77+}
78+
79+func TestMigrationOver4096(t *testing.T) {
80+ dir := t.TempDir()
81+ require.NoError(t, os.Chmod(dir, 0o700))
82+ repo := NewRepository(dir)
83+ name := "work"
84+ key := sessionKey(name)
85+ generations := filepath.Join(dir, "sessions", key, "generations")
86+ require.NoError(t, os.MkdirAll(generations, 0o700))
87+ for i := 1; i <= 5000; i++ {
88+ require.NoError(t, os.WriteFile(filepath.Join(generations, generationFilename(uint64(i))), []byte("stale"), 0o600))
89+ }
90+ legacy := legacyManifestBytes(name, 5001)
91+ legacyDigest := sha256.Sum256(legacy)
92+ require.NoError(t, os.WriteFile(filepath.Join(generations, generationFilename(5001)), legacy, 0o600))
93+ require.NoError(t, os.WriteFile(filepath.Join(dir, "sessions", key, "HEAD"), marshalHead(5001, legacyDigest), 0o600))
94+ ref, err := repo.ReadLegacyHEAD(context.Background(), name)
95+ require.NoError(t, err)
96+ require.Equal(t, uint64(5001), ref.Generation)
97+ id := domain.IncarnationID{1}
98+ migrated, err := repo.MigrateV1Checkpoint(context.Background(), ports.SnapshotMigrationRequest{LegacyName: name, IncarnationID: id, LegacyRef: ref})
99+ require.NoError(t, err)
100+ got, err := repo.LoadCheckpoint(context.Background(), id, name, migrated)
101+ require.NoError(t, err)
102+ manifest, err := codec.UnmarshalManifest(got.Manifest)
103+ require.NoError(t, err)
104+ require.Equal(t, id, manifest.IncarnationID)
105+ require.Nil(t, manifest.ParentCheckpoint)
106+ for i := 1; i <= 5000; i++ {
107+ require.FileExists(t, filepath.Join(generations, generationFilename(uint64(i))))
108+ }
109+}
@@ -0,0 +1,109 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "encoding/binary"
7+ "fmt"
8+ "hash/crc32"
9+ "os"
10+ "path/filepath"
11+ "testing"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+ codec "github.com/bnema/vev/internal/usecase/snapshot"
16+ "github.com/stretchr/testify/require"
17+)
18+
19+func legacyManifestBytes(name string, generation uint64) []byte {
20+ body := binary.BigEndian.AppendUint64(nil, generation)
21+ body = binary.BigEndian.AppendUint16(body, uint16(len(name)))
22+ body = append(body, name...)
23+ body = binary.BigEndian.AppendUint64(body, 1)
24+ body = binary.BigEndian.AppendUint16(body, 0)
25+ body = binary.BigEndian.AppendUint16(body, 0)
26+ out := make([]byte, 16, 16+len(body))
27+ copy(out, "VEVM")
28+ binary.BigEndian.PutUint16(out[4:6], 1)
29+ binary.BigEndian.PutUint32(out[8:12], uint32(len(body)))
30+ binary.BigEndian.PutUint32(out[12:16], crc32.ChecksumIEEE(body))
31+ return append(out, body...)
32+}
33+
34+func TestLegacyManifestV1(t *testing.T) {
35+ encoded := legacyManifestBytes("work", 7)
36+ got, err := decodeManifestV1(encoded)
37+ require.NoError(t, err)
38+ require.Equal(t, "work", got.Name)
39+ require.Equal(t, uint64(7), got.Generation)
40+ for _, invalid := range [][]byte{encoded[:5], append(append([]byte(nil), encoded...), 0)} {
41+ _, err := decodeManifestV1(invalid)
42+ require.Error(t, err)
43+ }
44+}
45+
46+func TestHasLegacyState(t *testing.T) {
47+ for _, tc := range []struct {
48+ name string
49+ setup func(string)
50+ want bool
51+ budget bool
52+ }{
53+ {"absence", func(string) {}, false, false},
54+ {"top level blob", func(dir string) {
55+ require.NoError(t, os.WriteFile(filepath.Join(dir, "work.snap"), []byte("x"), 0o600))
56+ }, true, false},
57+ {"incremental child", func(dir string) { require.NoError(t, os.MkdirAll(filepath.Join(dir, "sessions", "work"), 0o700)) }, true, false},
58+ {"budget exhaustion", func(dir string) {
59+ for i := 0; i <= legacyPresenceEntryLimit; i++ {
60+ require.NoError(t, os.WriteFile(filepath.Join(dir, fmt.Sprintf("x-%04d", i)), nil, 0o600))
61+ }
62+ }, false, true},
63+ } {
64+ t.Run(tc.name, func(t *testing.T) {
65+ dir := t.TempDir()
66+ require.NoError(t, os.Chmod(dir, 0o700))
67+ tc.setup(dir)
68+ got, err := NewRepository(dir).HasLegacyState(context.Background())
69+ if tc.budget {
70+ require.ErrorIs(t, err, ErrLegacyTraversalBudgetExceeded)
71+ } else {
72+ require.NoError(t, err)
73+ require.Equal(t, tc.want, got)
74+ }
75+ })
76+ }
77+}
78+
79+func TestMigrationOver4096(t *testing.T) {
80+ dir := t.TempDir()
81+ require.NoError(t, os.Chmod(dir, 0o700))
82+ repo := NewRepository(dir)
83+ name := "work"
84+ key := sessionKey(name)
85+ generations := filepath.Join(dir, "sessions", key, "generations")
86+ require.NoError(t, os.MkdirAll(generations, 0o700))
87+ for i := 1; i <= 5000; i++ {
88+ require.NoError(t, os.WriteFile(filepath.Join(generations, generationFilename(uint64(i))), []byte("stale"), 0o600))
89+ }
90+ legacy := legacyManifestBytes(name, 5001)
91+ legacyDigest := sha256.Sum256(legacy)
92+ require.NoError(t, os.WriteFile(filepath.Join(generations, generationFilename(5001)), legacy, 0o600))
93+ require.NoError(t, os.WriteFile(filepath.Join(dir, "sessions", key, "HEAD"), marshalHead(5001, legacyDigest), 0o600))
94+ ref, err := repo.ReadLegacyHEAD(context.Background(), name)
95+ require.NoError(t, err)
96+ require.Equal(t, uint64(5001), ref.Generation)
97+ id := domain.IncarnationID{1}
98+ migrated, err := repo.MigrateV1Checkpoint(context.Background(), ports.SnapshotMigrationRequest{LegacyName: name, IncarnationID: id, LegacyRef: ref})
99+ require.NoError(t, err)
100+ got, err := repo.LoadCheckpoint(context.Background(), id, name, migrated)
101+ require.NoError(t, err)
102+ manifest, err := codec.UnmarshalManifest(got.Manifest)
103+ require.NoError(t, err)
104+ require.Equal(t, id, manifest.IncarnationID)
105+ require.Nil(t, manifest.ParentCheckpoint)
106+ for i := 1; i <= 5000; i++ {
107+ require.FileExists(t, filepath.Join(generations, generationFilename(uint64(i))))
108+ }
109+}
A internal/persist/migration.go +593 -0
@@ -0,0 +1,593 @@
1+package persist
2+
3+import (
4+ "context"
5+ "encoding/binary"
6+ "errors"
7+ "fmt"
8+ "io"
9+ "os"
10+ "path/filepath"
11+ "sort"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+ "github.com/bnema/vev/pkg/kv"
16+ "github.com/bnema/vev/pkg/safedir"
17+)
18+
19+type MigrationPhase uint8
20+
21+const (
22+ MigrationBackedUp MigrationPhase = iota + 1
23+ MigrationAssigned
24+ MigrationValidated
25+ MigrationInstalled
26+ MigrationComplete
27+)
28+
29+type MigrationRecord struct {
30+ Phase MigrationPhase
31+ Assignments map[string]domain.IncarnationID
32+ Validated map[string]*domain.CheckpointRef
33+ Degraded map[string]string
34+}
35+
36+type OpenDeps struct {
37+ StateDir string
38+ Random io.Reader
39+ SnapshotMigration ports.SnapshotMigration
40+}
41+
42+type OpenResult struct {
43+ Catalogue *Persister
44+ Records []domain.CatalogueRecord
45+ NewInstall bool
46+ Migrated bool
47+}
48+
49+type legacyCatalogueRecordV0 struct {
50+ Name string
51+ Cwd string
52+ CreatedAt int64
53+ UpdatedAt int64
54+ LastUsedSeq uint64
55+ TabNames []string
56+}
57+
58+const (
59+ migrationDirName = "migration"
60+ migrationIntentName = "durable-session-v1.intent"
61+ migrationCompleteName = "durable-session-v1.complete"
62+ migrationBackupName = "legacy-catalogue.backup"
63+ catalogueFormatName = "durable-session-v1.format"
64+ migrationRecordMagic = "VEVJ"
65+ migrationRecordVersion uint16 = 1
66+)
67+
68+func OpenOrMigrate(ctx context.Context, deps OpenDeps) (OpenResult, error) {
69+ if err := ctx.Err(); err != nil {
70+ return OpenResult{}, err
71+ }
72+ if deps.StateDir == "" || deps.Random == nil || deps.SnapshotMigration == nil {
73+ return OpenResult{}, errors.New("persist: incomplete open or migrate dependencies")
74+ }
75+ path := StorePath(deps.StateDir)
76+ present, err := catalogueCandidatesPresent(path)
77+ if err != nil {
78+ return OpenResult{}, err
79+ }
80+ migrationDir := filepath.Join(deps.StateDir, migrationDirName)
81+ intentPath := filepath.Join(migrationDir, migrationIntentName)
82+ if !present {
83+ if _, err := os.Stat(intentPath); err == nil {
84+ return resumeLegacyMigration(ctx, deps)
85+ } else if !errors.Is(err, os.ErrNotExist) {
86+ return OpenResult{}, err
87+ }
88+ legacy, err := deps.SnapshotMigration.HasLegacyState(ctx)
89+ if err != nil {
90+ return OpenResult{}, err
91+ }
92+ if legacy {
93+ return OpenResult{}, errors.New("persist: legacy snapshot state exists without a migratable catalogue")
94+ }
95+ if err := atomicMigrationWrite(filepath.Join(migrationDir, catalogueFormatName), []byte("v1\n")); err != nil {
96+ return OpenResult{}, err
97+ }
98+ catalogue, records, err := openCurrentCatalogue(deps.StateDir, true)
99+ if err != nil {
100+ return OpenResult{}, err
101+ }
102+ return OpenResult{Catalogue: catalogue, Records: records, NewInstall: true}, nil
103+ }
104+ // Opening and closing first applies fixed-path WAL recovery before format detection.
105+ store, err := OpenStore(path)
106+ if err != nil {
107+ return OpenResult{}, err
108+ }
109+ if err := store.Close(); err != nil {
110+ return OpenResult{}, err
111+ }
112+ if complete, err := completedMigration(migrationDir, intentPath); err != nil {
113+ return OpenResult{}, err
114+ } else if complete {
115+ data, err := kv.Replay(path)
116+ if err != nil {
117+ return OpenResult{}, err
118+ }
119+ current, legacy, err := classifyCatalogue(data)
120+ if err != nil || !current || legacy {
121+ return OpenResult{}, errors.Join(errors.New("persist: completed migration has invalid current catalogue"), err)
122+ }
123+ catalogue, records, err := openCurrentCatalogue(deps.StateDir, false)
124+ return OpenResult{Catalogue: catalogue, Records: records}, err
125+ }
126+ if _, err := os.Stat(intentPath); err == nil {
127+ return resumeLegacyMigration(ctx, deps)
128+ } else if !errors.Is(err, os.ErrNotExist) {
129+ return OpenResult{}, err
130+ }
131+ data, err := kv.Replay(path)
132+ if err != nil {
133+ return OpenResult{}, err
134+ }
135+ current, legacy, err := classifyCatalogue(data)
136+ if err != nil {
137+ if len(data) != 0 || !catalogueFormatProven(migrationDir) {
138+ return OpenResult{}, err
139+ }
140+ current = true
141+ }
142+ if current {
143+ if err := atomicMigrationWrite(filepath.Join(migrationDir, catalogueFormatName), []byte("v1\n")); err != nil {
144+ return OpenResult{}, err
145+ }
146+ catalogue, records, err := openCurrentCatalogue(deps.StateDir, false)
147+ return OpenResult{Catalogue: catalogue, Records: records}, err
148+ }
149+ if !legacy {
150+ return OpenResult{}, errors.New("persist: ambiguous catalogue format")
151+ }
152+ return startLegacyMigration(ctx, deps)
153+}
154+
155+func catalogueFormatProven(dir string) bool {
156+ data, err := os.ReadFile(filepath.Join(dir, catalogueFormatName))
157+ return err == nil && string(data) == "v1\n"
158+}
159+
160+func completedMigration(dir, intentPath string) (bool, error) {
161+ completePath := filepath.Join(dir, migrationCompleteName)
162+ data, err := os.ReadFile(completePath)
163+ if errors.Is(err, os.ErrNotExist) {
164+ return false, nil
165+ }
166+ if err != nil {
167+ return false, err
168+ }
169+ if string(data) != "complete\n" {
170+ return false, errors.New("persist: invalid migration completion marker")
171+ }
172+ record, err := readMigrationRecord(intentPath)
173+ if err != nil {
174+ return false, err
175+ }
176+ if record.Phase != MigrationComplete {
177+ return false, errors.New("persist: completion marker precedes completed intent")
178+ }
179+ return true, nil
180+}
181+
182+func catalogueCandidatesPresent(path string) (bool, error) {
183+ present := false
184+ for _, candidate := range []string{path, path + ".next", path + ".prev"} {
185+ _, err := os.Stat(candidate)
186+ if err == nil {
187+ present = true
188+ } else if !errors.Is(err, os.ErrNotExist) {
189+ return false, err
190+ }
191+ }
192+ return present, nil
193+}
194+func classifyCatalogue(data map[string][]byte) (current, legacy bool, err error) {
195+ if len(data) == 0 {
196+ return false, false, errors.New("persist: empty existing catalogue has unknown provenance")
197+ }
198+ for name, value := range data {
199+ if len(value) >= 6 && string(value[:4]) == string(catalogueMagic[:]) {
200+ if _, e := decodeRecordValue(name, value); e != nil {
201+ return false, false, e
202+ }
203+ current = true
204+ } else {
205+ if _, e := decodeLegacyRecordV0(name, value); e != nil {
206+ return false, false, e
207+ }
208+ legacy = true
209+ }
210+ }
211+ if current && legacy {
212+ return false, false, errors.New("persist: mixed catalogue formats")
213+ }
214+ return current, legacy, nil
215+}
216+
217+func startLegacyMigration(ctx context.Context, deps OpenDeps) (OpenResult, error) {
218+ migrationDir := filepath.Join(deps.StateDir, migrationDirName)
219+ if err := safedir.EnsurePrivate(migrationDir); err != nil {
220+ return OpenResult{}, err
221+ }
222+ backup := filepath.Join(migrationDir, migrationBackupName)
223+ source, err := os.ReadFile(StorePath(deps.StateDir))
224+ if err != nil {
225+ return OpenResult{}, err
226+ }
227+ if err := atomicMigrationWrite(backup, source); err != nil {
228+ return OpenResult{}, err
229+ }
230+ record := MigrationRecord{Phase: MigrationBackedUp, Assignments: map[string]domain.IncarnationID{}, Validated: map[string]*domain.CheckpointRef{}, Degraded: map[string]string{}}
231+ if err := writeMigrationRecord(filepath.Join(migrationDir, migrationIntentName), record); err != nil {
232+ return OpenResult{}, err
233+ }
234+ return resumeLegacyMigration(ctx, deps)
235+}
236+
237+func resumeLegacyMigration(ctx context.Context, deps OpenDeps) (OpenResult, error) {
238+ migrationDir := filepath.Join(deps.StateDir, migrationDirName)
239+ intent := filepath.Join(migrationDir, migrationIntentName)
240+ record, err := readMigrationRecord(intent)
241+ if err != nil {
242+ return OpenResult{}, err
243+ }
244+ backup := filepath.Join(migrationDir, migrationBackupName)
245+ legacyData, err := kv.Replay(backup)
246+ if err != nil {
247+ return OpenResult{}, fmt.Errorf("persist: replay legacy backup: %w", err)
248+ }
249+ legacyRecords, err := decodeLegacyMap(legacyData)
250+ if err != nil {
251+ return OpenResult{}, err
252+ }
253+ if record.Phase < MigrationAssigned {
254+ for _, legacy := range legacyRecords {
255+ if _, ok := record.Assignments[legacy.Name]; ok {
256+ continue
257+ }
258+ id, err := domain.NewIncarnationID(deps.Random)
259+ if err != nil {
260+ return OpenResult{}, err
261+ }
262+ record.Assignments[legacy.Name] = id
263+ if err := writeMigrationRecord(intent, record); err != nil {
264+ return OpenResult{}, err
265+ }
266+ }
267+ record.Phase = MigrationAssigned
268+ if err := writeMigrationRecord(intent, record); err != nil {
269+ return OpenResult{}, err
270+ }
271+ }
272+ if record.Phase < MigrationValidated {
273+ for _, legacy := range legacyRecords {
274+ if _, ok := record.Validated[legacy.Name]; ok {
275+ continue
276+ }
277+ if _, ok := record.Degraded[legacy.Name]; ok {
278+ continue
279+ }
280+ ref, err := deps.SnapshotMigration.ReadLegacyHEAD(ctx, legacy.Name)
281+ if err == nil {
282+ var migrated domain.CheckpointRef
283+ migrated, err = deps.SnapshotMigration.MigrateV1Checkpoint(ctx, ports.SnapshotMigrationRequest{LegacyName: legacy.Name, IncarnationID: record.Assignments[legacy.Name], LegacyRef: ref})
284+ if err == nil {
285+ record.Validated[legacy.Name] = &migrated
286+ }
287+ }
288+ if err != nil {
289+ if !errors.Is(err, ports.ErrLegacySnapshotUncertain) {
290+ return OpenResult{}, fmt.Errorf("persist: migrate legacy snapshot %q: %w", legacy.Name, err)
291+ }
292+ if ref.Generation != 0 && ref.ManifestDigest != ([32]byte{}) {
293+ legacyRef := ref
294+ record.Validated[legacy.Name] = &legacyRef
295+ }
296+ record.Degraded[legacy.Name] = err.Error()
297+ }
298+ if err := writeMigrationRecord(intent, record); err != nil {
299+ return OpenResult{}, err
300+ }
301+ }
302+ record.Phase = MigrationValidated
303+ if err := writeMigrationRecord(intent, record); err != nil {
304+ return OpenResult{}, err
305+ }
306+ }
307+ records := make([]domain.CatalogueRecord, 0, len(legacyRecords))
308+ for _, legacy := range legacyRecords {
309+ next := domain.CatalogueRecord{Name: legacy.Name, IncarnationID: record.Assignments[legacy.Name], Cwd: legacy.Cwd, CreatedAt: legacy.CreatedAt, UpdatedAt: legacy.UpdatedAt, LastUsedSeq: legacy.LastUsedSeq, TabNames: append([]string(nil), legacy.TabNames...)}
310+ if ref := record.Validated[legacy.Name]; ref != nil {
311+ copyRef := *ref
312+ next.Committed = &copyRef
313+ }
314+ if reason := record.Degraded[legacy.Name]; reason != "" {
315+ next.RecoveryState = domain.RecoveryDegraded
316+ next.DegradedReason = reason
317+ } else {
318+ next.RecoveryState = domain.RecoveryHealthy
319+ }
320+ records = append(records, next)
321+ }
322+ if record.Phase < MigrationInstalled {
323+ if err := installMigratedCatalogue(deps.StateDir, records); err != nil {
324+ return OpenResult{}, err
325+ }
326+ record.Phase = MigrationInstalled
327+ if err := writeMigrationRecord(intent, record); err != nil {
328+ return OpenResult{}, err
329+ }
330+ }
331+ catalogue, opened, err := openCurrentCatalogue(deps.StateDir, false)
332+ if err != nil {
333+ return OpenResult{}, err
334+ }
335+ if !catalogueRecordsEqual(records, opened) {
336+ _ = catalogue.Close()
337+ return OpenResult{}, errors.New("persist: installed catalogue validation mismatch")
338+ }
339+ if record.Phase < MigrationComplete {
340+ record.Phase = MigrationComplete
341+ if err := writeMigrationRecord(intent, record); err != nil {
342+ _ = catalogue.Close()
343+ return OpenResult{}, err
344+ }
345+ if err := atomicMigrationWrite(filepath.Join(migrationDir, migrationCompleteName), []byte("complete\n")); err != nil {
346+ _ = catalogue.Close()
347+ return OpenResult{}, err
348+ }
349+ }
350+ return OpenResult{Catalogue: catalogue, Records: opened, Migrated: true}, nil
351+}
352+
353+func installMigratedCatalogue(stateDir string, records []domain.CatalogueRecord) error {
354+ stage := filepath.Join(stateDir, migrationDirName, "install")
355+ _ = os.RemoveAll(stage)
356+ if err := safedir.EnsurePrivate(stage); err != nil {
357+ return err
358+ }
359+ p, _, err := openCurrentCatalogue(stage, true)
360+ if err != nil {
361+ return err
362+ }
363+ changes := make(map[string]*domain.CatalogueRecord, len(records))
364+ for i := range records {
365+ r := records[i]
366+ changes[r.Name] = &r
367+ }
368+ if len(changes) > 0 {
369+ if err := p.Apply(changes); err != nil {
370+ _ = p.Close()
371+ return err
372+ }
373+ }
374+ if err := p.Close(); err != nil {
375+ return err
376+ }
377+ source := StorePath(stage)
378+ target := StorePath(stateDir)
379+ if err := os.Rename(source, target); err != nil {
380+ return err
381+ }
382+ if err := syncMigrationDir(stateDir); err != nil {
383+ return err
384+ }
385+ return atomicMigrationWrite(filepath.Join(stateDir, migrationDirName, catalogueFormatName), []byte("v1\n"))
386+}
387+
388+func decodeLegacyCatalogueV0(data []byte) ([]legacyCatalogueRecordV0, error) {
389+ record, err := decodeLegacyRecordV0("legacy", data)
390+ if err != nil {
391+ return nil, err
392+ }
393+ return []legacyCatalogueRecordV0{record}, nil
394+}
395+func decodeLegacyMap(data map[string][]byte) ([]legacyCatalogueRecordV0, error) {
396+ out := make([]legacyCatalogueRecordV0, 0, len(data))
397+ for name, value := range data {
398+ record, err := decodeLegacyRecordV0(name, value)
399+ if err != nil {
400+ return nil, err
401+ }
402+ out = append(out, record)
403+ }
404+ sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
405+ return out, nil
406+}
407+func decodeLegacyRecordV0(name string, value []byte) (legacyCatalogueRecordV0, error) {
408+ if err := domain.ValidateSessionName(name); err != nil {
409+ return legacyCatalogueRecordV0{}, err
410+ }
411+ r := valueReader{data: value}
412+ cwd, ok := r.str()
413+ if !ok {
414+ return legacyCatalogueRecordV0{}, errMalformedRecord
415+ }
416+ created, ok := r.u64()
417+ if !ok {
418+ return legacyCatalogueRecordV0{}, errMalformedRecord
419+ }
420+ updated, ok := r.u64()
421+ if !ok {
422+ return legacyCatalogueRecordV0{}, errMalformedRecord
423+ }
424+ record := legacyCatalogueRecordV0{Name: name, Cwd: cwd, CreatedAt: int64(created), UpdatedAt: int64(updated)}
425+ if r.remaining() == 0 {
426+ return record, nil
427+ }
428+ if record.LastUsedSeq, ok = r.u64(); !ok {
429+ return legacyCatalogueRecordV0{}, errMalformedRecord
430+ }
431+ if r.remaining() == 0 {
432+ return record, nil
433+ }
434+ count, ok := r.u32()
435+ if !ok || uint64(count) > uint64(r.remaining())/4 {
436+ return legacyCatalogueRecordV0{}, errMalformedRecord
437+ }
438+ for range count {
439+ tab, ok := r.str()
440+ if !ok {
441+ return legacyCatalogueRecordV0{}, errMalformedRecord
442+ }
443+ record.TabNames = append(record.TabNames, tab)
444+ }
445+ if r.remaining() != 0 {
446+ return legacyCatalogueRecordV0{}, errMalformedRecord
447+ }
448+ return record, nil
449+}
450+
451+func writeMigrationRecord(path string, record MigrationRecord) error {
452+ return atomicMigrationWrite(path, encodeMigrationRecord(record))
453+}
454+func encodeMigrationRecord(record MigrationRecord) []byte {
455+ out := append([]byte(migrationRecordMagic), byte(migrationRecordVersion>>8), byte(migrationRecordVersion), byte(record.Phase))
456+ names := make([]string, 0, len(record.Assignments))
457+ for name := range record.Assignments {
458+ names = append(names, name)
459+ }
460+ sort.Strings(names)
461+ out = binary.BigEndian.AppendUint32(out, uint32(len(names)))
462+ for _, name := range names {
463+ out = appendMigrationString(out, name)
464+ id := record.Assignments[name]
465+ out = append(out, id[:]...)
466+ if ref := record.Validated[name]; ref != nil {
467+ out = append(out, 1)
468+ out = binary.BigEndian.AppendUint64(out, ref.Generation)
469+ out = append(out, ref.ManifestDigest[:]...)
470+ } else {
471+ out = append(out, 0)
472+ }
473+ out = appendMigrationString(out, record.Degraded[name])
474+ }
475+ return out
476+}
477+func readMigrationRecord(path string) (MigrationRecord, error) {
478+ data, err := os.ReadFile(path)
479+ if err != nil {
480+ return MigrationRecord{}, err
481+ }
482+ if len(data) < 11 || string(data[:4]) != migrationRecordMagic || binary.BigEndian.Uint16(data[4:6]) != migrationRecordVersion {
483+ return MigrationRecord{}, errors.New("persist: invalid migration journal")
484+ }
485+ record := MigrationRecord{Phase: MigrationPhase(data[6]), Assignments: map[string]domain.IncarnationID{}, Validated: map[string]*domain.CheckpointRef{}, Degraded: map[string]string{}}
486+ r := valueReader{data: data[7:]}
487+ count, ok := r.u32()
488+ if !ok {
489+ return MigrationRecord{}, errMalformedRecord
490+ }
491+ for range count {
492+ name, ok := r.str()
493+ if !ok {
494+ return MigrationRecord{}, errMalformedRecord
495+ }
496+ idBytes, ok := r.take(16)
497+ if !ok {
498+ return MigrationRecord{}, errMalformedRecord
499+ }
500+ var id domain.IncarnationID
501+ copy(id[:], idBytes)
502+ if _, exists := record.Assignments[name]; exists || id == (domain.IncarnationID{}) {
503+ return MigrationRecord{}, errMalformedRecord
504+ }
505+ record.Assignments[name] = id
506+ present, ok := r.byte()
507+ if !ok {
508+ return MigrationRecord{}, errMalformedRecord
509+ }
510+ if present == 1 {
511+ generation, ok := r.u64()
512+ if !ok {
513+ return MigrationRecord{}, errMalformedRecord
514+ }
515+ digest, ok := r.take(32)
516+ if !ok {
517+ return MigrationRecord{}, errMalformedRecord
518+ }
519+ ref := &domain.CheckpointRef{Generation: generation}
520+ copy(ref.ManifestDigest[:], digest)
521+ record.Validated[name] = ref
522+ } else if present != 0 {
523+ return MigrationRecord{}, errMalformedRecord
524+ }
525+ reason, ok := r.str()
526+ if !ok {
527+ return MigrationRecord{}, errMalformedRecord
528+ }
529+ if reason != "" {
530+ record.Degraded[name] = reason
531+ }
532+ }
533+ if r.remaining() != 0 || record.Phase < MigrationBackedUp || record.Phase > MigrationComplete {
534+ return MigrationRecord{}, errMalformedRecord
535+ }
536+ return record, nil
537+}
538+func appendMigrationString(out []byte, value string) []byte {
539+ out = binary.BigEndian.AppendUint32(out, uint32(len(value)))
540+ return append(out, value...)
541+}
542+func atomicMigrationWrite(path string, data []byte) error {
543+ dir := filepath.Dir(path)
544+ if err := safedir.EnsurePrivate(dir); err != nil {
545+ return err
546+ }
547+ tmp, err := os.CreateTemp(dir, ".migration-*")
548+ if err != nil {
549+ return err
550+ }
551+ tmpName := tmp.Name()
552+ defer os.Remove(tmpName)
553+ if err := tmp.Chmod(0o600); err != nil {
554+ tmp.Close()
555+ return err
556+ }
557+ if _, err := tmp.Write(data); err != nil {
558+ tmp.Close()
559+ return err
560+ }
561+ if err := tmp.Sync(); err != nil {
562+ tmp.Close()
563+ return err
564+ }
565+ if err := tmp.Close(); err != nil {
566+ return err
567+ }
568+ if err := os.Rename(tmpName, path); err != nil {
569+ return err
570+ }
571+ return syncMigrationDir(dir)
572+}
573+func syncMigrationDir(dir string) error {
574+ f, err := os.Open(dir)
575+ if err != nil {
576+ return err
577+ }
578+ defer f.Close()
579+ return f.Sync()
580+}
581+func catalogueRecordsEqual(a, b []domain.CatalogueRecord) bool {
582+ if len(a) != len(b) {
583+ return false
584+ }
585+ sort.Slice(a, func(i, j int) bool { return a[i].Name < a[j].Name })
586+ sort.Slice(b, func(i, j int) bool { return b[i].Name < b[j].Name })
587+ for i := range a {
588+ if a[i].Name != b[i].Name || a[i].IncarnationID != b[i].IncarnationID || a[i].RecoveryState != b[i].RecoveryState {
589+ return false
590+ }
591+ }
592+ return true
593+}
@@ -0,0 +1,593 @@
1+package persist
2+
3+import (
4+ "context"
5+ "encoding/binary"
6+ "errors"
7+ "fmt"
8+ "io"
9+ "os"
10+ "path/filepath"
11+ "sort"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+ "github.com/bnema/vev/pkg/kv"
16+ "github.com/bnema/vev/pkg/safedir"
17+)
18+
19+type MigrationPhase uint8
20+
21+const (
22+ MigrationBackedUp MigrationPhase = iota + 1
23+ MigrationAssigned
24+ MigrationValidated
25+ MigrationInstalled
26+ MigrationComplete
27+)
28+
29+type MigrationRecord struct {
30+ Phase MigrationPhase
31+ Assignments map[string]domain.IncarnationID
32+ Validated map[string]*domain.CheckpointRef
33+ Degraded map[string]string
34+}
35+
36+type OpenDeps struct {
37+ StateDir string
38+ Random io.Reader
39+ SnapshotMigration ports.SnapshotMigration
40+}
41+
42+type OpenResult struct {
43+ Catalogue *Persister
44+ Records []domain.CatalogueRecord
45+ NewInstall bool
46+ Migrated bool
47+}
48+
49+type legacyCatalogueRecordV0 struct {
50+ Name string
51+ Cwd string
52+ CreatedAt int64
53+ UpdatedAt int64
54+ LastUsedSeq uint64
55+ TabNames []string
56+}
57+
58+const (
59+ migrationDirName = "migration"
60+ migrationIntentName = "durable-session-v1.intent"
61+ migrationCompleteName = "durable-session-v1.complete"
62+ migrationBackupName = "legacy-catalogue.backup"
63+ catalogueFormatName = "durable-session-v1.format"
64+ migrationRecordMagic = "VEVJ"
65+ migrationRecordVersion uint16 = 1
66+)
67+
68+func OpenOrMigrate(ctx context.Context, deps OpenDeps) (OpenResult, error) {
69+ if err := ctx.Err(); err != nil {
70+ return OpenResult{}, err
71+ }
72+ if deps.StateDir == "" || deps.Random == nil || deps.SnapshotMigration == nil {
73+ return OpenResult{}, errors.New("persist: incomplete open or migrate dependencies")
74+ }
75+ path := StorePath(deps.StateDir)
76+ present, err := catalogueCandidatesPresent(path)
77+ if err != nil {
78+ return OpenResult{}, err
79+ }
80+ migrationDir := filepath.Join(deps.StateDir, migrationDirName)
81+ intentPath := filepath.Join(migrationDir, migrationIntentName)
82+ if !present {
83+ if _, err := os.Stat(intentPath); err == nil {
84+ return resumeLegacyMigration(ctx, deps)
85+ } else if !errors.Is(err, os.ErrNotExist) {
86+ return OpenResult{}, err
87+ }
88+ legacy, err := deps.SnapshotMigration.HasLegacyState(ctx)
89+ if err != nil {
90+ return OpenResult{}, err
91+ }
92+ if legacy {
93+ return OpenResult{}, errors.New("persist: legacy snapshot state exists without a migratable catalogue")
94+ }
95+ if err := atomicMigrationWrite(filepath.Join(migrationDir, catalogueFormatName), []byte("v1\n")); err != nil {
96+ return OpenResult{}, err
97+ }
98+ catalogue, records, err := openCurrentCatalogue(deps.StateDir, true)
99+ if err != nil {
100+ return OpenResult{}, err
101+ }
102+ return OpenResult{Catalogue: catalogue, Records: records, NewInstall: true}, nil
103+ }
104+ // Opening and closing first applies fixed-path WAL recovery before format detection.
105+ store, err := OpenStore(path)
106+ if err != nil {
107+ return OpenResult{}, err
108+ }
109+ if err := store.Close(); err != nil {
110+ return OpenResult{}, err
111+ }
112+ if complete, err := completedMigration(migrationDir, intentPath); err != nil {
113+ return OpenResult{}, err
114+ } else if complete {
115+ data, err := kv.Replay(path)
116+ if err != nil {
117+ return OpenResult{}, err
118+ }
119+ current, legacy, err := classifyCatalogue(data)
120+ if err != nil || !current || legacy {
121+ return OpenResult{}, errors.Join(errors.New("persist: completed migration has invalid current catalogue"), err)
122+ }
123+ catalogue, records, err := openCurrentCatalogue(deps.StateDir, false)
124+ return OpenResult{Catalogue: catalogue, Records: records}, err
125+ }
126+ if _, err := os.Stat(intentPath); err == nil {
127+ return resumeLegacyMigration(ctx, deps)
128+ } else if !errors.Is(err, os.ErrNotExist) {
129+ return OpenResult{}, err
130+ }
131+ data, err := kv.Replay(path)
132+ if err != nil {
133+ return OpenResult{}, err
134+ }
135+ current, legacy, err := classifyCatalogue(data)
136+ if err != nil {
137+ if len(data) != 0 || !catalogueFormatProven(migrationDir) {
138+ return OpenResult{}, err
139+ }
140+ current = true
141+ }
142+ if current {
143+ if err := atomicMigrationWrite(filepath.Join(migrationDir, catalogueFormatName), []byte("v1\n")); err != nil {
144+ return OpenResult{}, err
145+ }
146+ catalogue, records, err := openCurrentCatalogue(deps.StateDir, false)
147+ return OpenResult{Catalogue: catalogue, Records: records}, err
148+ }
149+ if !legacy {
150+ return OpenResult{}, errors.New("persist: ambiguous catalogue format")
151+ }
152+ return startLegacyMigration(ctx, deps)
153+}
154+
155+func catalogueFormatProven(dir string) bool {
156+ data, err := os.ReadFile(filepath.Join(dir, catalogueFormatName))
157+ return err == nil && string(data) == "v1\n"
158+}
159+
160+func completedMigration(dir, intentPath string) (bool, error) {
161+ completePath := filepath.Join(dir, migrationCompleteName)
162+ data, err := os.ReadFile(completePath)
163+ if errors.Is(err, os.ErrNotExist) {
164+ return false, nil
165+ }
166+ if err != nil {
167+ return false, err
168+ }
169+ if string(data) != "complete\n" {
170+ return false, errors.New("persist: invalid migration completion marker")
171+ }
172+ record, err := readMigrationRecord(intentPath)
173+ if err != nil {
174+ return false, err
175+ }
176+ if record.Phase != MigrationComplete {
177+ return false, errors.New("persist: completion marker precedes completed intent")
178+ }
179+ return true, nil
180+}
181+
182+func catalogueCandidatesPresent(path string) (bool, error) {
183+ present := false
184+ for _, candidate := range []string{path, path + ".next", path + ".prev"} {
185+ _, err := os.Stat(candidate)
186+ if err == nil {
187+ present = true
188+ } else if !errors.Is(err, os.ErrNotExist) {
189+ return false, err
190+ }
191+ }
192+ return present, nil
193+}
194+func classifyCatalogue(data map[string][]byte) (current, legacy bool, err error) {
195+ if len(data) == 0 {
196+ return false, false, errors.New("persist: empty existing catalogue has unknown provenance")
197+ }
198+ for name, value := range data {
199+ if len(value) >= 6 && string(value[:4]) == string(catalogueMagic[:]) {
200+ if _, e := decodeRecordValue(name, value); e != nil {
201+ return false, false, e
202+ }
203+ current = true
204+ } else {
205+ if _, e := decodeLegacyRecordV0(name, value); e != nil {
206+ return false, false, e
207+ }
208+ legacy = true
209+ }
210+ }
211+ if current && legacy {
212+ return false, false, errors.New("persist: mixed catalogue formats")
213+ }
214+ return current, legacy, nil
215+}
216+
217+func startLegacyMigration(ctx context.Context, deps OpenDeps) (OpenResult, error) {
218+ migrationDir := filepath.Join(deps.StateDir, migrationDirName)
219+ if err := safedir.EnsurePrivate(migrationDir); err != nil {
220+ return OpenResult{}, err
221+ }
222+ backup := filepath.Join(migrationDir, migrationBackupName)
223+ source, err := os.ReadFile(StorePath(deps.StateDir))
224+ if err != nil {
225+ return OpenResult{}, err
226+ }
227+ if err := atomicMigrationWrite(backup, source); err != nil {
228+ return OpenResult{}, err
229+ }
230+ record := MigrationRecord{Phase: MigrationBackedUp, Assignments: map[string]domain.IncarnationID{}, Validated: map[string]*domain.CheckpointRef{}, Degraded: map[string]string{}}
231+ if err := writeMigrationRecord(filepath.Join(migrationDir, migrationIntentName), record); err != nil {
232+ return OpenResult{}, err
233+ }
234+ return resumeLegacyMigration(ctx, deps)
235+}
236+
237+func resumeLegacyMigration(ctx context.Context, deps OpenDeps) (OpenResult, error) {
238+ migrationDir := filepath.Join(deps.StateDir, migrationDirName)
239+ intent := filepath.Join(migrationDir, migrationIntentName)
240+ record, err := readMigrationRecord(intent)
241+ if err != nil {
242+ return OpenResult{}, err
243+ }
244+ backup := filepath.Join(migrationDir, migrationBackupName)
245+ legacyData, err := kv.Replay(backup)
246+ if err != nil {
247+ return OpenResult{}, fmt.Errorf("persist: replay legacy backup: %w", err)
248+ }
249+ legacyRecords, err := decodeLegacyMap(legacyData)
250+ if err != nil {
251+ return OpenResult{}, err
252+ }
253+ if record.Phase < MigrationAssigned {
254+ for _, legacy := range legacyRecords {
255+ if _, ok := record.Assignments[legacy.Name]; ok {
256+ continue
257+ }
258+ id, err := domain.NewIncarnationID(deps.Random)
259+ if err != nil {
260+ return OpenResult{}, err
261+ }
262+ record.Assignments[legacy.Name] = id
263+ if err := writeMigrationRecord(intent, record); err != nil {
264+ return OpenResult{}, err
265+ }
266+ }
267+ record.Phase = MigrationAssigned
268+ if err := writeMigrationRecord(intent, record); err != nil {
269+ return OpenResult{}, err
270+ }
271+ }
272+ if record.Phase < MigrationValidated {
273+ for _, legacy := range legacyRecords {
274+ if _, ok := record.Validated[legacy.Name]; ok {
275+ continue
276+ }
277+ if _, ok := record.Degraded[legacy.Name]; ok {
278+ continue
279+ }
280+ ref, err := deps.SnapshotMigration.ReadLegacyHEAD(ctx, legacy.Name)
281+ if err == nil {
282+ var migrated domain.CheckpointRef
283+ migrated, err = deps.SnapshotMigration.MigrateV1Checkpoint(ctx, ports.SnapshotMigrationRequest{LegacyName: legacy.Name, IncarnationID: record.Assignments[legacy.Name], LegacyRef: ref})
284+ if err == nil {
285+ record.Validated[legacy.Name] = &migrated
286+ }
287+ }
288+ if err != nil {
289+ if !errors.Is(err, ports.ErrLegacySnapshotUncertain) {
290+ return OpenResult{}, fmt.Errorf("persist: migrate legacy snapshot %q: %w", legacy.Name, err)
291+ }
292+ if ref.Generation != 0 && ref.ManifestDigest != ([32]byte{}) {
293+ legacyRef := ref
294+ record.Validated[legacy.Name] = &legacyRef
295+ }
296+ record.Degraded[legacy.Name] = err.Error()
297+ }
298+ if err := writeMigrationRecord(intent, record); err != nil {
299+ return OpenResult{}, err
300+ }
301+ }
302+ record.Phase = MigrationValidated
303+ if err := writeMigrationRecord(intent, record); err != nil {
304+ return OpenResult{}, err
305+ }
306+ }
307+ records := make([]domain.CatalogueRecord, 0, len(legacyRecords))
308+ for _, legacy := range legacyRecords {
309+ next := domain.CatalogueRecord{Name: legacy.Name, IncarnationID: record.Assignments[legacy.Name], Cwd: legacy.Cwd, CreatedAt: legacy.CreatedAt, UpdatedAt: legacy.UpdatedAt, LastUsedSeq: legacy.LastUsedSeq, TabNames: append([]string(nil), legacy.TabNames...)}
310+ if ref := record.Validated[legacy.Name]; ref != nil {
311+ copyRef := *ref
312+ next.Committed = &copyRef
313+ }
314+ if reason := record.Degraded[legacy.Name]; reason != "" {
315+ next.RecoveryState = domain.RecoveryDegraded
316+ next.DegradedReason = reason
317+ } else {
318+ next.RecoveryState = domain.RecoveryHealthy
319+ }
320+ records = append(records, next)
321+ }
322+ if record.Phase < MigrationInstalled {
323+ if err := installMigratedCatalogue(deps.StateDir, records); err != nil {
324+ return OpenResult{}, err
325+ }
326+ record.Phase = MigrationInstalled
327+ if err := writeMigrationRecord(intent, record); err != nil {
328+ return OpenResult{}, err
329+ }
330+ }
331+ catalogue, opened, err := openCurrentCatalogue(deps.StateDir, false)
332+ if err != nil {
333+ return OpenResult{}, err
334+ }
335+ if !catalogueRecordsEqual(records, opened) {
336+ _ = catalogue.Close()
337+ return OpenResult{}, errors.New("persist: installed catalogue validation mismatch")
338+ }
339+ if record.Phase < MigrationComplete {
340+ record.Phase = MigrationComplete
341+ if err := writeMigrationRecord(intent, record); err != nil {
342+ _ = catalogue.Close()
343+ return OpenResult{}, err
344+ }
345+ if err := atomicMigrationWrite(filepath.Join(migrationDir, migrationCompleteName), []byte("complete\n")); err != nil {
346+ _ = catalogue.Close()
347+ return OpenResult{}, err
348+ }
349+ }
350+ return OpenResult{Catalogue: catalogue, Records: opened, Migrated: true}, nil
351+}
352+
353+func installMigratedCatalogue(stateDir string, records []domain.CatalogueRecord) error {
354+ stage := filepath.Join(stateDir, migrationDirName, "install")
355+ _ = os.RemoveAll(stage)
356+ if err := safedir.EnsurePrivate(stage); err != nil {
357+ return err
358+ }
359+ p, _, err := openCurrentCatalogue(stage, true)
360+ if err != nil {
361+ return err
362+ }
363+ changes := make(map[string]*domain.CatalogueRecord, len(records))
364+ for i := range records {
365+ r := records[i]
366+ changes[r.Name] = &r
367+ }
368+ if len(changes) > 0 {
369+ if err := p.Apply(changes); err != nil {
370+ _ = p.Close()
371+ return err
372+ }
373+ }
374+ if err := p.Close(); err != nil {
375+ return err
376+ }
377+ source := StorePath(stage)
378+ target := StorePath(stateDir)
379+ if err := os.Rename(source, target); err != nil {
380+ return err
381+ }
382+ if err := syncMigrationDir(stateDir); err != nil {
383+ return err
384+ }
385+ return atomicMigrationWrite(filepath.Join(stateDir, migrationDirName, catalogueFormatName), []byte("v1\n"))
386+}
387+
388+func decodeLegacyCatalogueV0(data []byte) ([]legacyCatalogueRecordV0, error) {
389+ record, err := decodeLegacyRecordV0("legacy", data)
390+ if err != nil {
391+ return nil, err
392+ }
393+ return []legacyCatalogueRecordV0{record}, nil
394+}
395+func decodeLegacyMap(data map[string][]byte) ([]legacyCatalogueRecordV0, error) {
396+ out := make([]legacyCatalogueRecordV0, 0, len(data))
397+ for name, value := range data {
398+ record, err := decodeLegacyRecordV0(name, value)
399+ if err != nil {
400+ return nil, err
401+ }
402+ out = append(out, record)
403+ }
404+ sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
405+ return out, nil
406+}
407+func decodeLegacyRecordV0(name string, value []byte) (legacyCatalogueRecordV0, error) {
408+ if err := domain.ValidateSessionName(name); err != nil {
409+ return legacyCatalogueRecordV0{}, err
410+ }
411+ r := valueReader{data: value}
412+ cwd, ok := r.str()
413+ if !ok {
414+ return legacyCatalogueRecordV0{}, errMalformedRecord
415+ }
416+ created, ok := r.u64()
417+ if !ok {
418+ return legacyCatalogueRecordV0{}, errMalformedRecord
419+ }
420+ updated, ok := r.u64()
421+ if !ok {
422+ return legacyCatalogueRecordV0{}, errMalformedRecord
423+ }
424+ record := legacyCatalogueRecordV0{Name: name, Cwd: cwd, CreatedAt: int64(created), UpdatedAt: int64(updated)}
425+ if r.remaining() == 0 {
426+ return record, nil
427+ }
428+ if record.LastUsedSeq, ok = r.u64(); !ok {
429+ return legacyCatalogueRecordV0{}, errMalformedRecord
430+ }
431+ if r.remaining() == 0 {
432+ return record, nil
433+ }
434+ count, ok := r.u32()
435+ if !ok || uint64(count) > uint64(r.remaining())/4 {
436+ return legacyCatalogueRecordV0{}, errMalformedRecord
437+ }
438+ for range count {
439+ tab, ok := r.str()
440+ if !ok {
441+ return legacyCatalogueRecordV0{}, errMalformedRecord
442+ }
443+ record.TabNames = append(record.TabNames, tab)
444+ }
445+ if r.remaining() != 0 {
446+ return legacyCatalogueRecordV0{}, errMalformedRecord
447+ }
448+ return record, nil
449+}
450+
451+func writeMigrationRecord(path string, record MigrationRecord) error {
452+ return atomicMigrationWrite(path, encodeMigrationRecord(record))
453+}
454+func encodeMigrationRecord(record MigrationRecord) []byte {
455+ out := append([]byte(migrationRecordMagic), byte(migrationRecordVersion>>8), byte(migrationRecordVersion), byte(record.Phase))
456+ names := make([]string, 0, len(record.Assignments))
457+ for name := range record.Assignments {
458+ names = append(names, name)
459+ }
460+ sort.Strings(names)
461+ out = binary.BigEndian.AppendUint32(out, uint32(len(names)))
462+ for _, name := range names {
463+ out = appendMigrationString(out, name)
464+ id := record.Assignments[name]
465+ out = append(out, id[:]...)
466+ if ref := record.Validated[name]; ref != nil {
467+ out = append(out, 1)
468+ out = binary.BigEndian.AppendUint64(out, ref.Generation)
469+ out = append(out, ref.ManifestDigest[:]...)
470+ } else {
471+ out = append(out, 0)
472+ }
473+ out = appendMigrationString(out, record.Degraded[name])
474+ }
475+ return out
476+}
477+func readMigrationRecord(path string) (MigrationRecord, error) {
478+ data, err := os.ReadFile(path)
479+ if err != nil {
480+ return MigrationRecord{}, err
481+ }
482+ if len(data) < 11 || string(data[:4]) != migrationRecordMagic || binary.BigEndian.Uint16(data[4:6]) != migrationRecordVersion {
483+ return MigrationRecord{}, errors.New("persist: invalid migration journal")
484+ }
485+ record := MigrationRecord{Phase: MigrationPhase(data[6]), Assignments: map[string]domain.IncarnationID{}, Validated: map[string]*domain.CheckpointRef{}, Degraded: map[string]string{}}
486+ r := valueReader{data: data[7:]}
487+ count, ok := r.u32()
488+ if !ok {
489+ return MigrationRecord{}, errMalformedRecord
490+ }
491+ for range count {
492+ name, ok := r.str()
493+ if !ok {
494+ return MigrationRecord{}, errMalformedRecord
495+ }
496+ idBytes, ok := r.take(16)
497+ if !ok {
498+ return MigrationRecord{}, errMalformedRecord
499+ }
500+ var id domain.IncarnationID
501+ copy(id[:], idBytes)
502+ if _, exists := record.Assignments[name]; exists || id == (domain.IncarnationID{}) {
503+ return MigrationRecord{}, errMalformedRecord
504+ }
505+ record.Assignments[name] = id
506+ present, ok := r.byte()
507+ if !ok {
508+ return MigrationRecord{}, errMalformedRecord
509+ }
510+ if present == 1 {
511+ generation, ok := r.u64()
512+ if !ok {
513+ return MigrationRecord{}, errMalformedRecord
514+ }
515+ digest, ok := r.take(32)
516+ if !ok {
517+ return MigrationRecord{}, errMalformedRecord
518+ }
519+ ref := &domain.CheckpointRef{Generation: generation}
520+ copy(ref.ManifestDigest[:], digest)
521+ record.Validated[name] = ref
522+ } else if present != 0 {
523+ return MigrationRecord{}, errMalformedRecord
524+ }
525+ reason, ok := r.str()
526+ if !ok {
527+ return MigrationRecord{}, errMalformedRecord
528+ }
529+ if reason != "" {
530+ record.Degraded[name] = reason
531+ }
532+ }
533+ if r.remaining() != 0 || record.Phase < MigrationBackedUp || record.Phase > MigrationComplete {
534+ return MigrationRecord{}, errMalformedRecord
535+ }
536+ return record, nil
537+}
538+func appendMigrationString(out []byte, value string) []byte {
539+ out = binary.BigEndian.AppendUint32(out, uint32(len(value)))
540+ return append(out, value...)
541+}
542+func atomicMigrationWrite(path string, data []byte) error {
543+ dir := filepath.Dir(path)
544+ if err := safedir.EnsurePrivate(dir); err != nil {
545+ return err
546+ }
547+ tmp, err := os.CreateTemp(dir, ".migration-*")
548+ if err != nil {
549+ return err
550+ }
551+ tmpName := tmp.Name()
552+ defer os.Remove(tmpName)
553+ if err := tmp.Chmod(0o600); err != nil {
554+ tmp.Close()
555+ return err
556+ }
557+ if _, err := tmp.Write(data); err != nil {
558+ tmp.Close()
559+ return err
560+ }
561+ if err := tmp.Sync(); err != nil {
562+ tmp.Close()
563+ return err
564+ }
565+ if err := tmp.Close(); err != nil {
566+ return err
567+ }
568+ if err := os.Rename(tmpName, path); err != nil {
569+ return err
570+ }
571+ return syncMigrationDir(dir)
572+}
573+func syncMigrationDir(dir string) error {
574+ f, err := os.Open(dir)
575+ if err != nil {
576+ return err
577+ }
578+ defer f.Close()
579+ return f.Sync()
580+}
581+func catalogueRecordsEqual(a, b []domain.CatalogueRecord) bool {
582+ if len(a) != len(b) {
583+ return false
584+ }
585+ sort.Slice(a, func(i, j int) bool { return a[i].Name < a[j].Name })
586+ sort.Slice(b, func(i, j int) bool { return b[i].Name < b[j].Name })
587+ for i := range a {
588+ if a[i].Name != b[i].Name || a[i].IncarnationID != b[i].IncarnationID || a[i].RecoveryState != b[i].RecoveryState {
589+ return false
590+ }
591+ }
592+ return true
593+}
A internal/persist/migration_test.go +216 -0
@@ -0,0 +1,216 @@
1+package persist
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "encoding/binary"
7+ "errors"
8+ "io"
9+ "os"
10+ "path/filepath"
11+ "testing"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+ "github.com/bnema/vev/pkg/kv"
16+ "github.com/stretchr/testify/require"
17+)
18+
19+type migrationStub struct {
20+ legacy bool
21+ probeErr error
22+ heads map[string]domain.CheckpointRef
23+ migrateErr map[string]error
24+ assignments map[string]domain.IncarnationID
25+}
26+
27+func (s *migrationStub) HasLegacyState(context.Context) (bool, error) { return s.legacy, s.probeErr }
28+func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.CheckpointRef, error) {
29+ ref, ok := s.heads[name]
30+ if !ok {
31+ return domain.CheckpointRef{}, errors.New("uncertain legacy checkpoint")
32+ }
33+ return ref, nil
34+}
35+func (s *migrationStub) MigrateV1Checkpoint(_ context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {
36+ if err := s.migrateErr[req.LegacyName]; err != nil {
37+ return domain.CheckpointRef{}, err
38+ }
39+ if s.assignments == nil {
40+ s.assignments = map[string]domain.IncarnationID{}
41+ }
42+ s.assignments[req.LegacyName] = req.IncarnationID
43+ ref := req.LegacyRef
44+ ref.ManifestDigest = sha256.Sum256(append(ref.ManifestDigest[:], req.IncarnationID[:]...))
45+ return ref, nil
46+}
47+
48+func encodeLegacyRecordV0(r legacyCatalogueRecordV0) []byte {
49+ out := binary.BigEndian.AppendUint32(nil, uint32(len(r.Cwd)))
50+ out = append(out, r.Cwd...)
51+ out = binary.BigEndian.AppendUint64(out, uint64(r.CreatedAt))
52+ out = binary.BigEndian.AppendUint64(out, uint64(r.UpdatedAt))
53+ out = binary.BigEndian.AppendUint64(out, r.LastUsedSeq)
54+ out = binary.BigEndian.AppendUint32(out, uint32(len(r.TabNames)))
55+ for _, tab := range r.TabNames {
56+ out = binary.BigEndian.AppendUint32(out, uint32(len(tab)))
57+ out = append(out, tab...)
58+ }
59+ return out
60+}
61+
62+func writeLegacyCatalogue(t *testing.T, dir string, records ...legacyCatalogueRecordV0) string {
63+ t.Helper()
64+ require.NoError(t, os.Chmod(dir, 0o700))
65+ require.NoError(t, os.MkdirAll(dir, 0o700))
66+ store, err := kv.Open(StorePath(dir))
67+ require.NoError(t, err)
68+ for _, record := range records {
69+ require.NoError(t, store.Set([]byte(record.Name), encodeLegacyRecordV0(record)))
70+ }
71+ require.NoError(t, store.Sync())
72+ require.NoError(t, store.Close())
73+ return StorePath(dir)
74+}
75+
76+func TestMigrationExistingState(t *testing.T) {
77+ dir := t.TempDir()
78+ legacy := writeLegacyCatalogue(t, dir,
79+ legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 2, TabNames: []string{"one"}},
80+ legacyCatalogueRecordV0{Name: "uncertain", Cwd: "/tmp", CreatedAt: 3, UpdatedAt: 4},
81+ )
82+ ref := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{1}}
83+ migration := &migrationStub{heads: map[string]domain.CheckpointRef{"work": ref, "uncertain": ref}, migrateErr: map[string]error{"uncertain": errors.Join(ports.ErrLegacySnapshotUncertain, errors.New("invalid legacy manifest"))}}
84+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: migration})
85+ require.NoError(t, err)
86+ require.True(t, result.Migrated)
87+ require.False(t, result.NewInstall)
88+ require.FileExists(t, filepath.Join(dir, "migration", "legacy-catalogue.backup"))
89+ require.FileExists(t, legacy)
90+ require.Len(t, result.Records, 2)
91+ require.Equal(t, domain.RecoveryHealthy, recordByName(t, result.Records, "work").RecoveryState)
92+ require.Equal(t, domain.RecoveryDegraded, recordByName(t, result.Records, "uncertain").RecoveryState)
93+ require.NoError(t, result.Catalogue.Close())
94+ reopened, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: io.LimitReader(&countingReader{}, 0), SnapshotMigration: migration})
95+ require.NoError(t, err)
96+ require.False(t, reopened.Migrated, "completed migration is one-shot")
97+ require.NoError(t, reopened.Catalogue.Close())
98+}
99+
100+func TestMigrationOver4096(t *testing.T) {
101+ dir := t.TempDir()
102+ writeLegacyCatalogue(t, dir, legacyCatalogueRecordV0{Name: "large", Cwd: "/tmp"})
103+ migration := &migrationStub{heads: map[string]domain.CheckpointRef{"large": {Generation: 5001, ManifestDigest: [32]byte{1}}}, migrateErr: map[string]error{}}
104+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: migration})
105+ require.NoError(t, err)
106+ require.NoError(t, result.Catalogue.Close())
107+ require.Equal(t, uint64(5001), result.Records[0].Committed.Generation)
108+}
109+
110+func TestMigrationResumeMatrix(t *testing.T) {
111+ dir := t.TempDir()
112+ writeLegacyCatalogue(t, dir,
113+ legacyCatalogueRecordV0{Name: "alpha", Cwd: "/tmp"},
114+ legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"},
115+ )
116+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
117+ first := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}
118+ _, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: io.LimitReader(&countingReader{}, 16), SnapshotMigration: first})
119+ require.Error(t, err)
120+ journal, err := readMigrationRecord(filepath.Join(dir, "migration", "durable-session-v1.intent"))
121+ require.NoError(t, err)
122+ assigned := journal.Assignments["alpha"]
123+ require.NotEqual(t, domain.IncarnationID{}, assigned)
124+ second := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}
125+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{next: 16}, SnapshotMigration: second})
126+ require.NoError(t, err)
127+ require.NoError(t, result.Catalogue.Close())
128+ require.Equal(t, assigned, second.assignments["alpha"])
129+ require.FileExists(t, filepath.Join(dir, "migration", "durable-session-v1.complete"))
130+}
131+
132+func TestMigrationPreservesUncertain(t *testing.T) {
133+ dir := t.TempDir()
134+ legacy := writeLegacyCatalogue(t, dir, legacyCatalogueRecordV0{Name: "broken", Cwd: "/tmp"})
135+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
136+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: &migrationStub{heads: map[string]domain.CheckpointRef{"broken": ref}, migrateErr: map[string]error{"broken": errors.Join(ports.ErrLegacySnapshotUncertain, errors.New("invalid legacy manifest"))}}})
137+ require.NoError(t, err)
138+ require.NoError(t, result.Catalogue.Close())
139+ require.FileExists(t, legacy)
140+ require.Equal(t, domain.RecoveryDegraded, result.Records[0].RecoveryState)
141+ require.NotEmpty(t, result.Records[0].DegradedReason)
142+
143+ failClosedDir := t.TempDir()
144+ writeLegacyCatalogue(t, failClosedDir, legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"})
145+ _, err = OpenOrMigrate(context.Background(), OpenDeps{StateDir: failClosedDir, Random: &countingReader{}, SnapshotMigration: &migrationStub{heads: map[string]domain.CheckpointRef{"work": ref}, migrateErr: map[string]error{"work": errors.New("repository unavailable")}}})
146+ require.ErrorContains(t, err, "repository unavailable")
147+ data, replayErr := kv.Replay(StorePath(failClosedDir))
148+ require.NoError(t, replayErr)
149+ _, legacyFormat, classifyErr := classifyCatalogue(data)
150+ require.NoError(t, classifyErr)
151+ require.True(t, legacyFormat, "migration-port failure must not install a new catalogue")
152+}
153+
154+func TestOpenOrMigrateLegacyProbe(t *testing.T) {
155+ probeErr := errors.New("probe uncertain")
156+ for _, tc := range []struct {
157+ name string
158+ stub *migrationStub
159+ wantNew bool
160+ wantErr error
161+ }{
162+ {"proven absent", &migrationStub{}, true, nil},
163+ {"legacy blob", &migrationStub{legacy: true}, false, nil},
164+ {"incremental namespace", &migrationStub{legacy: true}, false, nil},
165+ {"probe error", &migrationStub{probeErr: probeErr}, false, probeErr},
166+ } {
167+ t.Run(tc.name, func(t *testing.T) {
168+ dir := t.TempDir()
169+ require.NoError(t, os.Chmod(dir, 0o700))
170+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: tc.stub})
171+ if tc.wantErr != nil {
172+ require.ErrorIs(t, err, tc.wantErr)
173+ } else if tc.stub.legacy {
174+ require.Error(t, err)
175+ } else {
176+ require.NoError(t, err)
177+ require.Equal(t, tc.wantNew, result.NewInstall)
178+ require.NoError(t, result.Catalogue.Close())
179+ if tc.wantNew {
180+ reopened, reopenErr := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: tc.stub})
181+ require.NoError(t, reopenErr)
182+ require.False(t, reopened.NewInstall)
183+ require.False(t, reopened.Migrated)
184+ require.NoError(t, reopened.Catalogue.Close())
185+ }
186+ }
187+ if tc.stub.legacy || tc.wantErr != nil {
188+ _, statErr := os.Stat(StorePath(dir))
189+ require.ErrorIs(t, statErr, os.ErrNotExist)
190+ }
191+ })
192+ }
193+}
194+
195+type countingReader struct{ next byte }
196+
197+func (r *countingReader) Read(p []byte) (int, error) {
198+ for i := range p {
199+ r.next++
200+ if r.next == 0 {
201+ r.next = 1
202+ }
203+ p[i] = r.next
204+ }
205+ return len(p), nil
206+}
207+func recordByName(t *testing.T, records []domain.CatalogueRecord, name string) domain.CatalogueRecord {
208+ t.Helper()
209+ for _, r := range records {
210+ if r.Name == name {
211+ return r
212+ }
213+ }
214+ t.Fatalf("record %q not found", name)
215+ return domain.CatalogueRecord{}
216+}
@@ -0,0 +1,216 @@
1+package persist
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "encoding/binary"
7+ "errors"
8+ "io"
9+ "os"
10+ "path/filepath"
11+ "testing"
12+
13+ "github.com/bnema/vev/internal/domain"
14+ "github.com/bnema/vev/internal/ports"
15+ "github.com/bnema/vev/pkg/kv"
16+ "github.com/stretchr/testify/require"
17+)
18+
19+type migrationStub struct {
20+ legacy bool
21+ probeErr error
22+ heads map[string]domain.CheckpointRef
23+ migrateErr map[string]error
24+ assignments map[string]domain.IncarnationID
25+}
26+
27+func (s *migrationStub) HasLegacyState(context.Context) (bool, error) { return s.legacy, s.probeErr }
28+func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.CheckpointRef, error) {
29+ ref, ok := s.heads[name]
30+ if !ok {
31+ return domain.CheckpointRef{}, errors.New("uncertain legacy checkpoint")
32+ }
33+ return ref, nil
34+}
35+func (s *migrationStub) MigrateV1Checkpoint(_ context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {
36+ if err := s.migrateErr[req.LegacyName]; err != nil {
37+ return domain.CheckpointRef{}, err
38+ }
39+ if s.assignments == nil {
40+ s.assignments = map[string]domain.IncarnationID{}
41+ }
42+ s.assignments[req.LegacyName] = req.IncarnationID
43+ ref := req.LegacyRef
44+ ref.ManifestDigest = sha256.Sum256(append(ref.ManifestDigest[:], req.IncarnationID[:]...))
45+ return ref, nil
46+}
47+
48+func encodeLegacyRecordV0(r legacyCatalogueRecordV0) []byte {
49+ out := binary.BigEndian.AppendUint32(nil, uint32(len(r.Cwd)))
50+ out = append(out, r.Cwd...)
51+ out = binary.BigEndian.AppendUint64(out, uint64(r.CreatedAt))
52+ out = binary.BigEndian.AppendUint64(out, uint64(r.UpdatedAt))
53+ out = binary.BigEndian.AppendUint64(out, r.LastUsedSeq)
54+ out = binary.BigEndian.AppendUint32(out, uint32(len(r.TabNames)))
55+ for _, tab := range r.TabNames {
56+ out = binary.BigEndian.AppendUint32(out, uint32(len(tab)))
57+ out = append(out, tab...)
58+ }
59+ return out
60+}
61+
62+func writeLegacyCatalogue(t *testing.T, dir string, records ...legacyCatalogueRecordV0) string {
63+ t.Helper()
64+ require.NoError(t, os.Chmod(dir, 0o700))
65+ require.NoError(t, os.MkdirAll(dir, 0o700))
66+ store, err := kv.Open(StorePath(dir))
67+ require.NoError(t, err)
68+ for _, record := range records {
69+ require.NoError(t, store.Set([]byte(record.Name), encodeLegacyRecordV0(record)))
70+ }
71+ require.NoError(t, store.Sync())
72+ require.NoError(t, store.Close())
73+ return StorePath(dir)
74+}
75+
76+func TestMigrationExistingState(t *testing.T) {
77+ dir := t.TempDir()
78+ legacy := writeLegacyCatalogue(t, dir,
79+ legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 2, TabNames: []string{"one"}},
80+ legacyCatalogueRecordV0{Name: "uncertain", Cwd: "/tmp", CreatedAt: 3, UpdatedAt: 4},
81+ )
82+ ref := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{1}}
83+ migration := &migrationStub{heads: map[string]domain.CheckpointRef{"work": ref, "uncertain": ref}, migrateErr: map[string]error{"uncertain": errors.Join(ports.ErrLegacySnapshotUncertain, errors.New("invalid legacy manifest"))}}
84+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: migration})
85+ require.NoError(t, err)
86+ require.True(t, result.Migrated)
87+ require.False(t, result.NewInstall)
88+ require.FileExists(t, filepath.Join(dir, "migration", "legacy-catalogue.backup"))
89+ require.FileExists(t, legacy)
90+ require.Len(t, result.Records, 2)
91+ require.Equal(t, domain.RecoveryHealthy, recordByName(t, result.Records, "work").RecoveryState)
92+ require.Equal(t, domain.RecoveryDegraded, recordByName(t, result.Records, "uncertain").RecoveryState)
93+ require.NoError(t, result.Catalogue.Close())
94+ reopened, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: io.LimitReader(&countingReader{}, 0), SnapshotMigration: migration})
95+ require.NoError(t, err)
96+ require.False(t, reopened.Migrated, "completed migration is one-shot")
97+ require.NoError(t, reopened.Catalogue.Close())
98+}
99+
100+func TestMigrationOver4096(t *testing.T) {
101+ dir := t.TempDir()
102+ writeLegacyCatalogue(t, dir, legacyCatalogueRecordV0{Name: "large", Cwd: "/tmp"})
103+ migration := &migrationStub{heads: map[string]domain.CheckpointRef{"large": {Generation: 5001, ManifestDigest: [32]byte{1}}}, migrateErr: map[string]error{}}
104+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: migration})
105+ require.NoError(t, err)
106+ require.NoError(t, result.Catalogue.Close())
107+ require.Equal(t, uint64(5001), result.Records[0].Committed.Generation)
108+}
109+
110+func TestMigrationResumeMatrix(t *testing.T) {
111+ dir := t.TempDir()
112+ writeLegacyCatalogue(t, dir,
113+ legacyCatalogueRecordV0{Name: "alpha", Cwd: "/tmp"},
114+ legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"},
115+ )
116+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
117+ first := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}
118+ _, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: io.LimitReader(&countingReader{}, 16), SnapshotMigration: first})
119+ require.Error(t, err)
120+ journal, err := readMigrationRecord(filepath.Join(dir, "migration", "durable-session-v1.intent"))
121+ require.NoError(t, err)
122+ assigned := journal.Assignments["alpha"]
123+ require.NotEqual(t, domain.IncarnationID{}, assigned)
124+ second := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}
125+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{next: 16}, SnapshotMigration: second})
126+ require.NoError(t, err)
127+ require.NoError(t, result.Catalogue.Close())
128+ require.Equal(t, assigned, second.assignments["alpha"])
129+ require.FileExists(t, filepath.Join(dir, "migration", "durable-session-v1.complete"))
130+}
131+
132+func TestMigrationPreservesUncertain(t *testing.T) {
133+ dir := t.TempDir()
134+ legacy := writeLegacyCatalogue(t, dir, legacyCatalogueRecordV0{Name: "broken", Cwd: "/tmp"})
135+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
136+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: &migrationStub{heads: map[string]domain.CheckpointRef{"broken": ref}, migrateErr: map[string]error{"broken": errors.Join(ports.ErrLegacySnapshotUncertain, errors.New("invalid legacy manifest"))}}})
137+ require.NoError(t, err)
138+ require.NoError(t, result.Catalogue.Close())
139+ require.FileExists(t, legacy)
140+ require.Equal(t, domain.RecoveryDegraded, result.Records[0].RecoveryState)
141+ require.NotEmpty(t, result.Records[0].DegradedReason)
142+
143+ failClosedDir := t.TempDir()
144+ writeLegacyCatalogue(t, failClosedDir, legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"})
145+ _, err = OpenOrMigrate(context.Background(), OpenDeps{StateDir: failClosedDir, Random: &countingReader{}, SnapshotMigration: &migrationStub{heads: map[string]domain.CheckpointRef{"work": ref}, migrateErr: map[string]error{"work": errors.New("repository unavailable")}}})
146+ require.ErrorContains(t, err, "repository unavailable")
147+ data, replayErr := kv.Replay(StorePath(failClosedDir))
148+ require.NoError(t, replayErr)
149+ _, legacyFormat, classifyErr := classifyCatalogue(data)
150+ require.NoError(t, classifyErr)
151+ require.True(t, legacyFormat, "migration-port failure must not install a new catalogue")
152+}
153+
154+func TestOpenOrMigrateLegacyProbe(t *testing.T) {
155+ probeErr := errors.New("probe uncertain")
156+ for _, tc := range []struct {
157+ name string
158+ stub *migrationStub
159+ wantNew bool
160+ wantErr error
161+ }{
162+ {"proven absent", &migrationStub{}, true, nil},
163+ {"legacy blob", &migrationStub{legacy: true}, false, nil},
164+ {"incremental namespace", &migrationStub{legacy: true}, false, nil},
165+ {"probe error", &migrationStub{probeErr: probeErr}, false, probeErr},
166+ } {
167+ t.Run(tc.name, func(t *testing.T) {
168+ dir := t.TempDir()
169+ require.NoError(t, os.Chmod(dir, 0o700))
170+ result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: tc.stub})
171+ if tc.wantErr != nil {
172+ require.ErrorIs(t, err, tc.wantErr)
173+ } else if tc.stub.legacy {
174+ require.Error(t, err)
175+ } else {
176+ require.NoError(t, err)
177+ require.Equal(t, tc.wantNew, result.NewInstall)
178+ require.NoError(t, result.Catalogue.Close())
179+ if tc.wantNew {
180+ reopened, reopenErr := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: tc.stub})
181+ require.NoError(t, reopenErr)
182+ require.False(t, reopened.NewInstall)
183+ require.False(t, reopened.Migrated)
184+ require.NoError(t, reopened.Catalogue.Close())
185+ }
186+ }
187+ if tc.stub.legacy || tc.wantErr != nil {
188+ _, statErr := os.Stat(StorePath(dir))
189+ require.ErrorIs(t, statErr, os.ErrNotExist)
190+ }
191+ })
192+ }
193+}
194+
195+type countingReader struct{ next byte }
196+
197+func (r *countingReader) Read(p []byte) (int, error) {
198+ for i := range p {
199+ r.next++
200+ if r.next == 0 {
201+ r.next = 1
202+ }
203+ p[i] = r.next
204+ }
205+ return len(p), nil
206+}
207+func recordByName(t *testing.T, records []domain.CatalogueRecord, name string) domain.CatalogueRecord {
208+ t.Helper()
209+ for _, r := range records {
210+ if r.Name == name {
211+ return r
212+ }
213+ }
214+ t.Fatalf("record %q not found", name)
215+ return domain.CatalogueRecord{}
216+}
M internal/ports/recovery.go +3 -0
@@ -2,6 +2,7 @@ package ports
22
33 import (
44 "context"
5+ "errors"
56
67 "github.com/bnema/vev/internal/domain"
78 )
@@ -99,6 +100,8 @@ type ReconcileDecision struct {
99100 ReasonCode string
100101 }
101102
103+var ErrLegacySnapshotUncertain = errors.New("legacy snapshot state is uncertain")
104+
102105 type SnapshotMigrationRequest struct {
103106 LegacyName string
104107 IncarnationID domain.IncarnationID
@@ -2,6 +2,7 @@ package ports
2 2
3 import (3 import (
4 "context"4 "context"
5+ "errors"
5 6
6 "github.com/bnema/vev/internal/domain"7 "github.com/bnema/vev/internal/domain"
7 )8 )
@@ -99,6 +100,8 @@ type ReconcileDecision struct {
99 ReasonCode string100 ReasonCode string
100 }101 }
101 102
103+var ErrLegacySnapshotUncertain = errors.New("legacy snapshot state is uncertain")
104+
102 type SnapshotMigrationRequest struct {105 type SnapshotMigrationRequest struct {
103 LegacyName string106 LegacyName string
104 IncarnationID domain.IncarnationID107 IncarnationID domain.IncarnationID

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