Merge branch 'fix/bind-local-index-to-storage-url'
All checks were successful
check / check (push) Successful in 2m18s
All checks were successful
check / check (push) Successful in 2m18s
This commit is contained in:
53
internal/database/local_meta.go
Normal file
53
internal/database/local_meta.go
Normal file
@@ -0,0 +1,53 @@
|
|||||||
|
package database
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
)
|
||||||
|
|
||||||
|
// LocalMetaKeyStorageURL is the key under which the destination store's
|
||||||
|
// URL is recorded when a mutating command first binds the local index
|
||||||
|
// to a specific backup destination.
|
||||||
|
const LocalMetaKeyStorageURL = "storage_url"
|
||||||
|
|
||||||
|
// LocalMetaRepository provides keyed access to host-local settings
|
||||||
|
// stored in the local_meta table.
|
||||||
|
type LocalMetaRepository struct {
|
||||||
|
db *DB
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewLocalMetaRepository(db *DB) *LocalMetaRepository {
|
||||||
|
return &LocalMetaRepository{db: db}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get returns the value stored at key, or the empty string if the key
|
||||||
|
// is not set. A missing key is not an error — the caller distinguishes
|
||||||
|
// "unset" (bind on first use) from "set to something" (compare).
|
||||||
|
func (r *LocalMetaRepository) Get(ctx context.Context, key string) (string, error) {
|
||||||
|
var value string
|
||||||
|
err := r.db.conn.QueryRowContext(ctx,
|
||||||
|
"SELECT value FROM local_meta WHERE key = ?", key,
|
||||||
|
).Scan(&value)
|
||||||
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("reading local_meta %q: %w", key, err)
|
||||||
|
}
|
||||||
|
return value, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set writes key=value, replacing any prior value.
|
||||||
|
func (r *LocalMetaRepository) Set(ctx context.Context, key, value string) error {
|
||||||
|
_, err := r.db.ExecWithLog(ctx,
|
||||||
|
`INSERT INTO local_meta (key, value) VALUES (?, ?)
|
||||||
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
|
||||||
|
key, value,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("writing local_meta %q: %w", key, err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
52
internal/database/local_meta_test.go
Normal file
52
internal/database/local_meta_test.go
Normal file
@@ -0,0 +1,52 @@
|
|||||||
|
package database_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestLocalMetaEmptyOnFresh(t *testing.T) {
|
||||||
|
db, err := database.NewTestDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
got, err := repos.LocalMeta.Get(context.Background(), database.LocalMetaKeyStorageURL)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, "", got, "fresh DB must return empty for unset keys, not error")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLocalMetaSetGetRoundTrip(t *testing.T) {
|
||||||
|
db, err := database.NewTestDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
require.NoError(t, repos.LocalMeta.Set(ctx, database.LocalMetaKeyStorageURL, "file:///mnt/backups"))
|
||||||
|
|
||||||
|
got, err := repos.LocalMeta.Get(ctx, database.LocalMetaKeyStorageURL)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, "file:///mnt/backups", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLocalMetaSetOverwrites(t *testing.T) {
|
||||||
|
db, err := database.NewTestDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
require.NoError(t, repos.LocalMeta.Set(ctx, database.LocalMetaKeyStorageURL, "s3://old"))
|
||||||
|
require.NoError(t, repos.LocalMeta.Set(ctx, database.LocalMetaKeyStorageURL, "s3://new"))
|
||||||
|
|
||||||
|
got, err := repos.LocalMeta.Get(ctx, database.LocalMetaKeyStorageURL)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, "s3://new", got)
|
||||||
|
}
|
||||||
@@ -19,6 +19,7 @@ type Repositories struct {
|
|||||||
ChunkFiles *ChunkFileRepository
|
ChunkFiles *ChunkFileRepository
|
||||||
Snapshots *SnapshotRepository
|
Snapshots *SnapshotRepository
|
||||||
Uploads *UploadRepository
|
Uploads *UploadRepository
|
||||||
|
LocalMeta *LocalMetaRepository
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewRepositories creates a new Repositories instance with all repository types.
|
// NewRepositories creates a new Repositories instance with all repository types.
|
||||||
@@ -34,6 +35,7 @@ func NewRepositories(db *DB) *Repositories {
|
|||||||
ChunkFiles: NewChunkFileRepository(db),
|
ChunkFiles: NewChunkFileRepository(db),
|
||||||
Snapshots: NewSnapshotRepository(db),
|
Snapshots: NewSnapshotRepository(db),
|
||||||
Uploads: NewUploadRepository(db.conn),
|
Uploads: NewUploadRepository(db.conn),
|
||||||
|
LocalMeta: NewLocalMetaRepository(db),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -133,3 +133,17 @@ CREATE TABLE IF NOT EXISTS uploads (
|
|||||||
|
|
||||||
-- Index for efficient snapshot lookups
|
-- Index for efficient snapshot lookups
|
||||||
CREATE INDEX IF NOT EXISTS idx_uploads_snapshot_id ON uploads(snapshot_id);
|
CREATE INDEX IF NOT EXISTS idx_uploads_snapshot_id ON uploads(snapshot_id);
|
||||||
|
|
||||||
|
-- Local metadata: keyed, host-local settings that bind the state of the
|
||||||
|
-- local index database to external context. The primary use is
|
||||||
|
-- storage_url: once a backup writes blobs to a destination, the local
|
||||||
|
-- index is only valid against that destination — if the configured
|
||||||
|
-- storage_url later changes, the scanner would silently think already-
|
||||||
|
-- known chunks are still on the new (empty) destination and skip
|
||||||
|
-- uploading them, corrupting future snapshots. On every mutating
|
||||||
|
-- command startup, we compare the configured storage_url to the stored
|
||||||
|
-- one and refuse to proceed on mismatch.
|
||||||
|
CREATE TABLE IF NOT EXISTS local_meta (
|
||||||
|
key TEXT PRIMARY KEY,
|
||||||
|
value TEXT NOT NULL
|
||||||
|
);
|
||||||
|
|||||||
@@ -55,6 +55,9 @@ type PruneBlobsResult struct {
|
|||||||
// prefer this method over PruneDatabase or PruneBlobs individually
|
// prefer this method over PruneDatabase or PruneBlobs individually
|
||||||
// unless it specifically wants one half.
|
// unless it specifically wants one half.
|
||||||
func (v *Vaultik) Prune(opts *PruneOptions) error {
|
func (v *Vaultik) Prune(opts *PruneOptions) error {
|
||||||
|
if err := v.EnsureStorageBinding(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
if _, err := v.PruneDatabase(); err != nil {
|
if _, err := v.PruneDatabase(); err != nil {
|
||||||
return fmt.Errorf("pruning local database: %w", err)
|
return fmt.Errorf("pruning local database: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -36,6 +36,10 @@ func (v *Vaultik) CreateSnapshot(opts *SnapshotCreateOptions) error {
|
|||||||
"index_path", v.Config.IndexPath,
|
"index_path", v.Config.IndexPath,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if err := v.EnsureStorageBinding(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
// Clean up incomplete snapshots FIRST, before any scanning
|
// Clean up incomplete snapshots FIRST, before any scanning
|
||||||
// This is critical for data safety - see CleanupIncompleteSnapshots for details
|
// This is critical for data safety - see CleanupIncompleteSnapshots for details
|
||||||
hostname := v.Config.Hostname
|
hostname := v.Config.Hostname
|
||||||
@@ -581,6 +585,9 @@ type SnapshotPurgeOptions struct {
|
|||||||
// snapshot name, not the latest globally. This prevents `home` and `system`
|
// snapshot name, not the latest globally. This prevents `home` and `system`
|
||||||
// snapshots from cannibalizing each other.
|
// snapshots from cannibalizing each other.
|
||||||
func (v *Vaultik) PurgeSnapshotsWithOptions(opts *SnapshotPurgeOptions) error {
|
func (v *Vaultik) PurgeSnapshotsWithOptions(opts *SnapshotPurgeOptions) error {
|
||||||
|
if err := v.EnsureStorageBinding(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
// Sync with remote first
|
// Sync with remote first
|
||||||
if err := v.syncWithRemote(); err != nil {
|
if err := v.syncWithRemote(); err != nil {
|
||||||
return fmt.Errorf("syncing with remote: %w", err)
|
return fmt.Errorf("syncing with remote: %w", err)
|
||||||
@@ -861,6 +868,9 @@ func (v *Vaultik) outputVerifyJSON(result *VerifyResult) error {
|
|||||||
// human ID is hashed via RemoteSnapshotKey and compared against the
|
// human ID is hashed via RemoteSnapshotKey and compared against the
|
||||||
// remote listing.
|
// remote listing.
|
||||||
func (v *Vaultik) CleanupLocalSnapshots() error {
|
func (v *Vaultik) CleanupLocalSnapshots() error {
|
||||||
|
if err := v.EnsureStorageBinding(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
remoteKeys, err := v.listAllRemoteSnapshotKeys()
|
remoteKeys, err := v.listAllRemoteSnapshotKeys()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -1006,6 +1016,10 @@ func (v *Vaultik) RemoveSnapshot(snapshotID string, opts *RemoveOptions) (*Remov
|
|||||||
SnapshotID: snapshotID,
|
SnapshotID: snapshotID,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := v.EnsureStorageBinding(); err != nil {
|
||||||
|
return result, err
|
||||||
|
}
|
||||||
|
|
||||||
if opts.DryRun {
|
if opts.DryRun {
|
||||||
result.DryRun = true
|
result.DryRun = true
|
||||||
if !opts.JSON {
|
if !opts.JSON {
|
||||||
@@ -1086,6 +1100,9 @@ func (v *Vaultik) RemoveSnapshot(snapshotID string, opts *RemoveOptions) (*Remov
|
|||||||
// "remove --all" leaves nothing behind, even when the local DB and
|
// "remove --all" leaves nothing behind, even when the local DB and
|
||||||
// remote storage have diverged.
|
// remote storage have diverged.
|
||||||
func (v *Vaultik) RemoveAllSnapshots(opts *RemoveOptions) (*RemoveResult, error) {
|
func (v *Vaultik) RemoveAllSnapshots(opts *RemoveOptions) (*RemoveResult, error) {
|
||||||
|
if err := v.EnsureStorageBinding(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
localSnaps, err := v.localSnapshotIDs()
|
localSnaps, err := v.localSnapshotIDs()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("listing local snapshots: %w", err)
|
return nil, fmt.Errorf("listing local snapshots: %w", err)
|
||||||
|
|||||||
91
internal/vaultik/storage_bind.go
Normal file
91
internal/vaultik/storage_bind.go
Normal file
@@ -0,0 +1,91 @@
|
|||||||
|
package vaultik
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
)
|
||||||
|
|
||||||
|
// EnsureStorageBinding guarantees that the local index database is
|
||||||
|
// bound to the currently-configured storage destination. Every mutating
|
||||||
|
// command must call this before touching either the local index or the
|
||||||
|
// destination store, because the two live in lockstep: the local index
|
||||||
|
// records which chunks/blobs already exist at the destination, and
|
||||||
|
// mismatched destination + local index produces silent corruption (the
|
||||||
|
// scanner sees "known" chunks and skips uploads, then writes snapshots
|
||||||
|
// whose manifests reference blobs that aren't on the new destination).
|
||||||
|
//
|
||||||
|
// Behaviour:
|
||||||
|
// - On first use (empty stored value), record the configured
|
||||||
|
// storage_url and log the binding.
|
||||||
|
// - When the stored value matches the configured storage_url, do
|
||||||
|
// nothing and return nil.
|
||||||
|
// - When the two differ, refuse with an error that tells the user
|
||||||
|
// how to recover (revert the config, or run `vaultik database
|
||||||
|
// purge` to discard the local index and rebuild against the new
|
||||||
|
// destination on the next backup).
|
||||||
|
//
|
||||||
|
// Read-only inspection commands (remote info, snapshot list, etc.)
|
||||||
|
// deliberately don't call this: they can be run against a bare
|
||||||
|
// destination store without any binding state.
|
||||||
|
func (v *Vaultik) EnsureStorageBinding() error {
|
||||||
|
if v.Repositories == nil || v.Config == nil {
|
||||||
|
// NewForTesting builds a Vaultik with no DB or config;
|
||||||
|
// there's nothing to bind and no binding to check. Callers
|
||||||
|
// exercising the bind path use a real DB and populate Config
|
||||||
|
// explicitly (see storage_bind_test.go).
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
configured := v.Config.StorageURL
|
||||||
|
if configured == "" {
|
||||||
|
// Some legacy configs still use the split s3.* keys instead of
|
||||||
|
// storage_url. Falling back to a synthetic URL for those would
|
||||||
|
// hide the fact that the binding is loose. Instead, treat
|
||||||
|
// unset as "nothing to bind against" — the check is
|
||||||
|
// necessarily best-effort for those configs.
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
stored, err := v.Repositories.LocalMeta.Get(v.ctx, database.LocalMetaKeyStorageURL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("reading local storage binding: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if stored == "" {
|
||||||
|
if err := v.Repositories.LocalMeta.Set(v.ctx, database.LocalMetaKeyStorageURL, configured); err != nil {
|
||||||
|
return fmt.Errorf("recording local storage binding: %w", err)
|
||||||
|
}
|
||||||
|
log.Info("Bound local index to storage destination", "storage_url", configured)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if stored == configured {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return fmt.Errorf("%s", buildBindingMismatchMessage(stored, configured))
|
||||||
|
}
|
||||||
|
|
||||||
|
// buildBindingMismatchMessage assembles the multi-line explanation
|
||||||
|
// shown when the local index is bound to a different destination than
|
||||||
|
// the currently-configured one. Kept as a separate function so the
|
||||||
|
// lint-flagged multi-line format string is expressed as a plain string
|
||||||
|
// literal rather than a fmt.Errorf argument (staticcheck ST1005
|
||||||
|
// disallows trailing punctuation on error format strings).
|
||||||
|
func buildBindingMismatchMessage(stored, configured string) string {
|
||||||
|
return "local index is bound to a different backup destination\n" +
|
||||||
|
" local index bound to: " + stored + "\n" +
|
||||||
|
" currently configured: " + configured + "\n" +
|
||||||
|
"\n" +
|
||||||
|
"The local index database tracks which chunks and blobs already exist at the\n" +
|
||||||
|
"destination store. Using it against a different destination would silently\n" +
|
||||||
|
"skip uploads (the scanner would treat every chunk as already present), leaving\n" +
|
||||||
|
"future snapshots referencing blobs that don't exist at the new destination.\n" +
|
||||||
|
"\n" +
|
||||||
|
"To proceed, either:\n" +
|
||||||
|
" - revert storage_url in your config to the bound destination, or\n" +
|
||||||
|
" - run 'vaultik database purge' to discard the local index and rebuild it\n" +
|
||||||
|
" from a fresh full backup against the new destination"
|
||||||
|
}
|
||||||
72
internal/vaultik/storage_bind_test.go
Normal file
72
internal/vaultik/storage_bind_test.go
Normal file
@@ -0,0 +1,72 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// buildBindTestVaultik returns a minimal Vaultik wired with a real
|
||||||
|
// in-memory DB and a config carrying the given StorageURL. Enough to
|
||||||
|
// exercise EnsureStorageBinding without spinning up storage or fx.
|
||||||
|
func buildBindTestVaultik(t *testing.T, storageURL string) (*vaultik.TestVaultik, *database.Repositories) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
db, err := database.NewTestDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
tv := vaultik.NewForTesting(nil)
|
||||||
|
tv.Repositories = repos
|
||||||
|
tv.DB = db
|
||||||
|
tv.Config = &config.Config{StorageURL: storageURL}
|
||||||
|
tv.SetContext(context.Background())
|
||||||
|
|
||||||
|
return tv, repos
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsureStorageBinding_FreshDBRecordsURL(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
tv, repos := buildBindTestVaultik(t, "file:///mnt/backups/new")
|
||||||
|
|
||||||
|
require.NoError(t, tv.EnsureStorageBinding())
|
||||||
|
|
||||||
|
got, err := repos.LocalMeta.Get(context.Background(), database.LocalMetaKeyStorageURL)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, "file:///mnt/backups/new", got, "first call must record the configured URL")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsureStorageBinding_MatchingURLPasses(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
tv, repos := buildBindTestVaultik(t, "s3://bucket/prefix")
|
||||||
|
|
||||||
|
require.NoError(t, repos.LocalMeta.Set(context.Background(),
|
||||||
|
database.LocalMetaKeyStorageURL, "s3://bucket/prefix"))
|
||||||
|
|
||||||
|
require.NoError(t, tv.EnsureStorageBinding())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnsureStorageBinding_MismatchRefuses(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
tv, repos := buildBindTestVaultik(t, "file:///mnt/backups/new")
|
||||||
|
|
||||||
|
require.NoError(t, repos.LocalMeta.Set(context.Background(),
|
||||||
|
database.LocalMetaKeyStorageURL, "file:///mnt/backups/old"))
|
||||||
|
|
||||||
|
err := tv.EnsureStorageBinding()
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.Contains(t, err.Error(), "file:///mnt/backups/old")
|
||||||
|
assert.Contains(t, err.Error(), "file:///mnt/backups/new")
|
||||||
|
assert.Contains(t, err.Error(), "vaultik database purge")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user