feat(snapshot): expose legacy snapshot imports
- commit
- b5f28c1724043fa9afb8a7b47658a8f75fde1bdb
- author
- brice · · 2026-07-21 20:36 UTC
- committer
- brice · · 2026-07-21 20:37 UTC
- parent
- be1ed56
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +131 -28
A internal/adapters/snapshot/legacy.go +68 -0
| @@ -0,0 +1,68 @@ | |||
| 1 | + | package snapshot | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "fmt" | |
| 7 | + | "os" | |
| 8 | + | "path/filepath" | |
| 9 | + | "sort" | |
| 10 | + | "strings" | |
| 11 | + | ||
| 12 | + | "github.com/bnema/vev/internal/ports" | |
| 13 | + | ) | |
| 14 | + | ||
| 15 | + | // LoadLegacy reads only pre-incremental root .snap files. Store remains the | |
| 16 | + | // legacy writer until the migration cutover, but is deliberately not used by | |
| 17 | + | // this read path so incremental repository directories are never traversed. | |
| 18 | + | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | |
| 19 | + | if err := ctx.Err(); err != nil { | |
| 20 | + | return nil, err | |
| 21 | + | } | |
| 22 | + | entries, err := os.ReadDir(r.dir) | |
| 23 | + | if errors.Is(err, os.ErrNotExist) { | |
| 24 | + | return nil, nil | |
| 25 | + | } | |
| 26 | + | if err != nil { | |
| 27 | + | return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err)) | |
| 28 | + | } | |
| 29 | + | sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() }) | |
| 30 | + | out := make([]ports.LegacySnapshot, 0, len(entries)) | |
| 31 | + | for _, entry := range entries { | |
| 32 | + | if err := ctx.Err(); err != nil { | |
| 33 | + | return nil, err | |
| 34 | + | } | |
| 35 | + | if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".snap") { | |
| 36 | + | continue | |
| 37 | + | } | |
| 38 | + | data, err := readBounded(filepath.Join(r.dir, entry.Name())) | |
| 39 | + | if err != nil { | |
| 40 | + | continue | |
| 41 | + | } | |
| 42 | + | name := strings.TrimSuffix(entry.Name(), ".snap") | |
| 43 | + | if strings.HasPrefix(name, "@") || !safeNameRE.MatchString(name) { | |
| 44 | + | name = "" | |
| 45 | + | } | |
| 46 | + | out = append(out, ports.LegacySnapshot{Name: name, Data: data}) | |
| 47 | + | } | |
| 48 | + | return out, nil | |
| 49 | + | } | |
| 50 | + | ||
| 51 | + | // DeleteLegacy deletes the deterministic v3 file and synchronizes the root. | |
| 52 | + | // A sync failure is returned even though the unlink may have succeeded; retrying | |
| 53 | + | // is safe because an already absent legacy file is considered deleted. | |
| 54 | + | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |
| 55 | + | if err := ctx.Err(); err != nil { | |
| 56 | + | return err | |
| 57 | + | } | |
| 58 | + | path := filepath.Join(r.dir, filenameForName(name)) | |
| 59 | + | if err := r.remove(path); errors.Is(err, os.ErrNotExist) { | |
| 60 | + | return nil | |
| 61 | + | } else if err != nil { | |
| 62 | + | return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err)) | |
| 63 | + | } | |
| 64 | + | if err := r.syncDirectory(r.dir); err != nil { | |
| 65 | + | return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err)) | |
| 66 | + | } | |
| 67 | + | return nil | |
| 68 | + | } | |
| @@ -0,0 +1,68 @@ | |||||
| 1 | + | package snapshot | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "fmt" | |||
| 7 | + | "os" | |||
| 8 | + | "path/filepath" | |||
| 9 | + | "sort" | |||
| 10 | + | "strings" | |||
| 11 | + | ||||
| 12 | + | "github.com/bnema/vev/internal/ports" | |||
| 13 | + | ) | |||
| 14 | + | ||||
| 15 | + | // LoadLegacy reads only pre-incremental root .snap files. Store remains the | |||
| 16 | + | // legacy writer until the migration cutover, but is deliberately not used by | |||
| 17 | + | // this read path so incremental repository directories are never traversed. | |||
| 18 | + | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | |||
| 19 | + | if err := ctx.Err(); err != nil { | |||
| 20 | + | return nil, err | |||
| 21 | + | } | |||
| 22 | + | entries, err := os.ReadDir(r.dir) | |||
| 23 | + | if errors.Is(err, os.ErrNotExist) { | |||
| 24 | + | return nil, nil | |||
| 25 | + | } | |||
| 26 | + | if err != nil { | |||
| 27 | + | return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err)) | |||
| 28 | + | } | |||
| 29 | + | sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() }) | |||
| 30 | + | out := make([]ports.LegacySnapshot, 0, len(entries)) | |||
| 31 | + | for _, entry := range entries { | |||
| 32 | + | if err := ctx.Err(); err != nil { | |||
| 33 | + | return nil, err | |||
| 34 | + | } | |||
| 35 | + | if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".snap") { | |||
| 36 | + | continue | |||
| 37 | + | } | |||
| 38 | + | data, err := readBounded(filepath.Join(r.dir, entry.Name())) | |||
| 39 | + | if err != nil { | |||
| 40 | + | continue | |||
| 41 | + | } | |||
| 42 | + | name := strings.TrimSuffix(entry.Name(), ".snap") | |||
| 43 | + | if strings.HasPrefix(name, "@") || !safeNameRE.MatchString(name) { | |||
| 44 | + | name = "" | |||
| 45 | + | } | |||
| 46 | + | out = append(out, ports.LegacySnapshot{Name: name, Data: data}) | |||
| 47 | + | } | |||
| 48 | + | return out, nil | |||
| 49 | + | } | |||
| 50 | + | ||||
| 51 | + | // DeleteLegacy deletes the deterministic v3 file and synchronizes the root. | |||
| 52 | + | // A sync failure is returned even though the unlink may have succeeded; retrying | |||
| 53 | + | // is safe because an already absent legacy file is considered deleted. | |||
| 54 | + | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |||
| 55 | + | if err := ctx.Err(); err != nil { | |||
| 56 | + | return err | |||
| 57 | + | } | |||
| 58 | + | path := filepath.Join(r.dir, filenameForName(name)) | |||
| 59 | + | if err := r.remove(path); errors.Is(err, os.ErrNotExist) { | |||
| 60 | + | return nil | |||
| 61 | + | } else if err != nil { | |||
| 62 | + | return fmt.Errorf("delete legacy snapshot: %w", safeFilesystemError(err)) | |||
| 63 | + | } | |||
| 64 | + | if err := r.syncDirectory(r.dir); err != nil { | |||
| 65 | + | return fmt.Errorf("sync legacy snapshot directory: %w", safeFilesystemError(err)) | |||
| 66 | + | } | |||
| 67 | + | return nil | |||
| 68 | + | } | |||
A internal/adapters/snapshot/legacy_test.go +59 -0
| @@ -0,0 +1,59 @@ | |||
| 1 | + | package snapshot | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "os" | |
| 7 | + | "path/filepath" | |
| 8 | + | "testing" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | func TestRepositoryLegacyReadsOnlyBoundedRootSnapshots(t *testing.T) { | |
| 12 | + | repo := NewRepository(privateDir(t)) | |
| 13 | + | if err := os.MkdirAll(filepath.Join(repo.dir, repositorySessionsDir), 0o700); err != nil { | |
| 14 | + | t.Fatal(err) | |
| 15 | + | } | |
| 16 | + | if err := os.WriteFile(filepath.Join(repo.dir, "named.snap"), []byte("legacy"), 0o600); err != nil { | |
| 17 | + | t.Fatal(err) | |
| 18 | + | } | |
| 19 | + | if err := os.WriteFile(filepath.Join(repo.dir, filenameForName("unsafe/name")), []byte("hashed"), 0o600); err != nil { | |
| 20 | + | t.Fatal(err) | |
| 21 | + | } | |
| 22 | + | if err := os.WriteFile(filepath.Join(repo.dir, repositorySessionsDir, "ignored.snap"), []byte("incremental"), 0o600); err != nil { | |
| 23 | + | t.Fatal(err) | |
| 24 | + | } | |
| 25 | + | ||
| 26 | + | got, err := repo.LoadLegacy(context.Background()) | |
| 27 | + | if err != nil { | |
| 28 | + | t.Fatal(err) | |
| 29 | + | } | |
| 30 | + | if len(got) != 2 || got[0].Name != "" || got[1].Name != "named" { | |
| 31 | + | t.Fatalf("LoadLegacy = %#v, want root safe and hashed snapshots", got) | |
| 32 | + | } | |
| 33 | + | } | |
| 34 | + | ||
| 35 | + | func TestRepositoryDeleteLegacyMissingFileDoesNotRequireRoot(t *testing.T) { | |
| 36 | + | repo := NewRepository(privateDir(t)) | |
| 37 | + | if err := repo.DeleteLegacy(context.Background(), "missing"); err != nil { | |
| 38 | + | t.Fatalf("DeleteLegacy missing file: %v", err) | |
| 39 | + | } | |
| 40 | + | } | |
| 41 | + | ||
| 42 | + | func TestRepositoryDeleteLegacySurfacesRootSyncAndCanRetry(t *testing.T) { | |
| 43 | + | repo := NewRepository(privateDir(t)) | |
| 44 | + | if err := os.Mkdir(repo.dir, 0o700); err != nil { | |
| 45 | + | t.Fatal(err) | |
| 46 | + | } | |
| 47 | + | if err := os.WriteFile(filepath.Join(repo.dir, filenameForName("named")), []byte("legacy"), 0o600); err != nil { | |
| 48 | + | t.Fatal(err) | |
| 49 | + | } | |
| 50 | + | injected := errors.New("root sync") | |
| 51 | + | repo.hooks.syncDirectory = func(string) error { return injected } | |
| 52 | + | if err := repo.DeleteLegacy(context.Background(), "named"); !errors.Is(err, injected) { | |
| 53 | + | t.Fatalf("DeleteLegacy error = %v, want root sync error", err) | |
| 54 | + | } | |
| 55 | + | repo.hooks.syncDirectory = nil | |
| 56 | + | if err := repo.DeleteLegacy(context.Background(), "named"); err != nil { | |
| 57 | + | t.Fatalf("retry DeleteLegacy: %v", err) | |
| 58 | + | } | |
| 59 | + | } | |
| @@ -0,0 +1,59 @@ | |||||
| 1 | + | package snapshot | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "os" | |||
| 7 | + | "path/filepath" | |||
| 8 | + | "testing" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | func TestRepositoryLegacyReadsOnlyBoundedRootSnapshots(t *testing.T) { | |||
| 12 | + | repo := NewRepository(privateDir(t)) | |||
| 13 | + | if err := os.MkdirAll(filepath.Join(repo.dir, repositorySessionsDir), 0o700); err != nil { | |||
| 14 | + | t.Fatal(err) | |||
| 15 | + | } | |||
| 16 | + | if err := os.WriteFile(filepath.Join(repo.dir, "named.snap"), []byte("legacy"), 0o600); err != nil { | |||
| 17 | + | t.Fatal(err) | |||
| 18 | + | } | |||
| 19 | + | if err := os.WriteFile(filepath.Join(repo.dir, filenameForName("unsafe/name")), []byte("hashed"), 0o600); err != nil { | |||
| 20 | + | t.Fatal(err) | |||
| 21 | + | } | |||
| 22 | + | if err := os.WriteFile(filepath.Join(repo.dir, repositorySessionsDir, "ignored.snap"), []byte("incremental"), 0o600); err != nil { | |||
| 23 | + | t.Fatal(err) | |||
| 24 | + | } | |||
| 25 | + | ||||
| 26 | + | got, err := repo.LoadLegacy(context.Background()) | |||
| 27 | + | if err != nil { | |||
| 28 | + | t.Fatal(err) | |||
| 29 | + | } | |||
| 30 | + | if len(got) != 2 || got[0].Name != "" || got[1].Name != "named" { | |||
| 31 | + | t.Fatalf("LoadLegacy = %#v, want root safe and hashed snapshots", got) | |||
| 32 | + | } | |||
| 33 | + | } | |||
| 34 | + | ||||
| 35 | + | func TestRepositoryDeleteLegacyMissingFileDoesNotRequireRoot(t *testing.T) { | |||
| 36 | + | repo := NewRepository(privateDir(t)) | |||
| 37 | + | if err := repo.DeleteLegacy(context.Background(), "missing"); err != nil { | |||
| 38 | + | t.Fatalf("DeleteLegacy missing file: %v", err) | |||
| 39 | + | } | |||
| 40 | + | } | |||
| 41 | + | ||||
| 42 | + | func TestRepositoryDeleteLegacySurfacesRootSyncAndCanRetry(t *testing.T) { | |||
| 43 | + | repo := NewRepository(privateDir(t)) | |||
| 44 | + | if err := os.Mkdir(repo.dir, 0o700); err != nil { | |||
| 45 | + | t.Fatal(err) | |||
| 46 | + | } | |||
| 47 | + | if err := os.WriteFile(filepath.Join(repo.dir, filenameForName("named")), []byte("legacy"), 0o600); err != nil { | |||
| 48 | + | t.Fatal(err) | |||
| 49 | + | } | |||
| 50 | + | injected := errors.New("root sync") | |||
| 51 | + | repo.hooks.syncDirectory = func(string) error { return injected } | |||
| 52 | + | if err := repo.DeleteLegacy(context.Background(), "named"); !errors.Is(err, injected) { | |||
| 53 | + | t.Fatalf("DeleteLegacy error = %v, want root sync error", err) | |||
| 54 | + | } | |||
| 55 | + | repo.hooks.syncDirectory = nil | |||
| 56 | + | if err := repo.DeleteLegacy(context.Background(), "named"); err != nil { | |||
| 57 | + | t.Fatalf("retry DeleteLegacy: %v", err) | |||
| 58 | + | } | |||
| 59 | + | } | |||
M internal/adapters/snapshot/repository.go +4 -28
| @@ -1,7 +1,6 @@ | |||
| 1 | 1 | package snapshot | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | - | "context" | |
| 5 | 4 | "sync" | |
| 6 | 5 | ||
| 7 | 6 | "github.com/bnema/vev/internal/ports" | |
| @@ -19,10 +18,9 @@ const ( | |||
| 19 | 18 | // Repository is the crash-safe, content-addressed session snapshot store. | |
| 20 | 19 | // Store remains available separately for the one-way legacy bridge. | |
| 21 | 20 | type Repository struct { | |
| 22 | - | dir string | |
| 23 | - | legacy *Store | |
| 24 | - | locks sync.Map // map[string]*sync.Mutex | |
| 25 | - | hooks repositoryHooks | |
| 21 | + | dir string | |
| 22 | + | locks sync.Map // map[string]*sync.Mutex | |
| 23 | + | hooks repositoryHooks | |
| 26 | 24 | } | |
| 27 | 25 | ||
| 28 | 26 | // repositoryHooks makes each persistence boundary fault-injectable. Hooks run | |
| @@ -47,31 +45,9 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil) | |||
| 47 | 45 | ||
| 48 | 46 | // NewRepository creates a repository rooted at dir. It does not create files | |
| 49 | 47 | // until the first publication, so merely constructing it is side-effect free. | |
| 50 | - | func NewRepository(dir string) *Repository { return &Repository{dir: dir, legacy: NewStore(dir)} } | |
| 48 | + | func NewRepository(dir string) *Repository { return &Repository{dir: dir} } | |
| 51 | 49 | ||
| 52 | 50 | func (r *Repository) sessionLock(key string) *sync.Mutex { | |
| 53 | 51 | lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{}) | |
| 54 | 52 | return lock.(*sync.Mutex) | |
| 55 | 53 | } | |
| 56 | - | ||
| 57 | - | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | |
| 58 | - | if err := ctx.Err(); err != nil { | |
| 59 | - | return nil, err | |
| 60 | - | } | |
| 61 | - | blobs, err := r.legacy.Load() | |
| 62 | - | if err != nil { | |
| 63 | - | return nil, err | |
| 64 | - | } | |
| 65 | - | out := make([]ports.LegacySnapshot, len(blobs)) | |
| 66 | - | for i, blob := range blobs { | |
| 67 | - | out[i] = ports.LegacySnapshot{Name: blob.Name, Data: append([]byte(nil), blob.Data...)} | |
| 68 | - | } | |
| 69 | - | return out, nil | |
| 70 | - | } | |
| 71 | - | ||
| 72 | - | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |
| 73 | - | if err := ctx.Err(); err != nil { | |
| 74 | - | return err | |
| 75 | - | } | |
| 76 | - | return r.legacy.Delete(name) | |
| 77 | - | } | |
| @@ -1,7 +1,6 @@ | |||||
| 1 | package snapshot | 1 | package snapshot | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | - | "context" | |||
| 5 | "sync" | 4 | "sync" | ||
| 6 | 5 | ||||
| 7 | "github.com/bnema/vev/internal/ports" | 6 | "github.com/bnema/vev/internal/ports" | ||
| @@ -19,10 +18,9 @@ const ( | |||||
| 19 | // Repository is the crash-safe, content-addressed session snapshot store. | 18 | // Repository is the crash-safe, content-addressed session snapshot store. | ||
| 20 | // Store remains available separately for the one-way legacy bridge. | 19 | // Store remains available separately for the one-way legacy bridge. | ||
| 21 | type Repository struct { | 20 | type Repository struct { | ||
| 22 | - | dir string | 21 | + | dir string |
| 23 | - | legacy *Store | 22 | + | locks sync.Map // map[string]*sync.Mutex |
| 24 | - | locks sync.Map // map[string]*sync.Mutex | 23 | + | hooks repositoryHooks |
| 25 | - | hooks repositoryHooks | |||
| 26 | } | 24 | } | ||
| 27 | 25 | ||||
| 28 | // repositoryHooks makes each persistence boundary fault-injectable. Hooks run | 26 | // repositoryHooks makes each persistence boundary fault-injectable. Hooks run | ||
| @@ -47,31 +45,9 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil) | |||||
| 47 | 45 | ||||
| 48 | // NewRepository creates a repository rooted at dir. It does not create files | 46 | // NewRepository creates a repository rooted at dir. It does not create files | ||
| 49 | // until the first publication, so merely constructing it is side-effect free. | 47 | // until the first publication, so merely constructing it is side-effect free. | ||
| 50 | - | func NewRepository(dir string) *Repository { return &Repository{dir: dir, legacy: NewStore(dir)} } | 48 | + | func NewRepository(dir string) *Repository { return &Repository{dir: dir} } |
| 51 | 49 | ||||
| 52 | func (r *Repository) sessionLock(key string) *sync.Mutex { | 50 | func (r *Repository) sessionLock(key string) *sync.Mutex { | ||
| 53 | lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{}) | 51 | lock, _ := r.locks.LoadOrStore(key, &sync.Mutex{}) | ||
| 54 | return lock.(*sync.Mutex) | 52 | return lock.(*sync.Mutex) | ||
| 55 | } | 53 | } | ||
| 56 | - | ||||
| 57 | - | func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) { | |||
| 58 | - | if err := ctx.Err(); err != nil { | |||
| 59 | - | return nil, err | |||
| 60 | - | } | |||
| 61 | - | blobs, err := r.legacy.Load() | |||
| 62 | - | if err != nil { | |||
| 63 | - | return nil, err | |||
| 64 | - | } | |||
| 65 | - | out := make([]ports.LegacySnapshot, len(blobs)) | |||
| 66 | - | for i, blob := range blobs { | |||
| 67 | - | out[i] = ports.LegacySnapshot{Name: blob.Name, Data: append([]byte(nil), blob.Data...)} | |||
| 68 | - | } | |||
| 69 | - | return out, nil | |||
| 70 | - | } | |||
| 71 | - | ||||
| 72 | - | func (r *Repository) DeleteLegacy(ctx context.Context, name string) error { | |||
| 73 | - | if err := ctx.Err(); err != nil { | |||
| 74 | - | return err | |||
| 75 | - | } | |||
| 76 | - | return r.legacy.Delete(name) | |||
| 77 | - | } | |||