clawbot
dd7a610c23
finalizeSnapshotMetadata marked the snapshot complete and then exported its metadata. A crash after completion but before/during the export left the local index showing the snapshot complete while the destination had no manifest or database, and PruneDatabase (which drops only NULL completed_at rows) kept it: a silently unrestorable snapshot. Reorder so completion is recorded last. CompleteSnapshot is split into PopulateSnapshotBlobs (before the export) and MarkSnapshotComplete (after it). An interrupted export now leaves the snapshot incomplete, so the next run PruneDatabase drops it and re-backs-up the data; the reverse tiny window leaves a restorable snapshot the index reports honestly as remote-only. Update REPOSTRUCTURE.md guarantee 4 and the ARCHITECTURE.md flow. Add a fault-injection test driving the full create path. Model: opus-4-8
830 lines
27 KiB
Go
830 lines
27 KiB
Go
package vaultik_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/spf13/afero"
|
|
"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/globals"
|
|
"sneak.berlin/go/vaultik/internal/log"
|
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
|
"sneak.berlin/go/vaultik/internal/storage"
|
|
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
|
"sneak.berlin/go/vaultik/internal/ui"
|
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
|
)
|
|
|
|
// These tests cover the failure modes a backup tool must survive:
|
|
// interrupted uploads, an interrupted metadata export, corrupt and
|
|
// truncated reads, a full restore disk, and a backend that reports
|
|
// success while storing nothing. Faults are injected through the
|
|
// storage.Storer seam (internal/storage/faultstore), never by patching
|
|
// production code. Each test asserts on the observable end state — what
|
|
// is in the index, what is at the destination, what the user is told —
|
|
// not merely that an error was returned. See
|
|
// https://git.eeqj.de/sneak/vaultik/issues/72.
|
|
//
|
|
// Object-level write atomicity (no partial blob object left behind) is
|
|
// covered by the file:// backend's atomic-write work
|
|
// (https://git.eeqj.de/sneak/vaultik/issues/130) and is not re-tested
|
|
// here; these tests target the layers above the backend.
|
|
//
|
|
// The tests run serially, not with t.Parallel: each calls
|
|
// log.Initialize, which replaces the package-global logger, and a
|
|
// backup or restore running concurrently reads that same logger. Under
|
|
// -race the two collide. Running one at a time is the same choice
|
|
// prune_count_test.go already makes for the same reason.
|
|
|
|
const (
|
|
faultChunkSize = int64(64 * 1024)
|
|
faultMaxBlobSize = int64(256 * 1024)
|
|
)
|
|
|
|
// faultTestConfig returns the config shared by the fault-injection
|
|
// tests: a real recipient/secret keypair so blobs are genuinely
|
|
// encrypted, and a blob size limit the restore sweeper can divide.
|
|
func faultTestConfig() *config.Config {
|
|
return &config.Config{
|
|
AgeRecipients: []string{testAgePublicKey},
|
|
AgeSecretKey: testAgeSecretKey,
|
|
CompressionLevel: 3,
|
|
Hostname: testHostname,
|
|
BlobSizeLimit: config.Size(faultMaxBlobSize),
|
|
}
|
|
}
|
|
|
|
// writeFaultSourceTree writes a spread of file sizes that forces several
|
|
// chunks across more than one blob, so a fault landing on a single blob
|
|
// still leaves other data intact. Returns the expected content by path.
|
|
func writeFaultSourceTree(
|
|
t *testing.T, fs afero.Fs, dataDir string,
|
|
) map[string][]byte {
|
|
t.Helper()
|
|
|
|
files := map[string][]byte{
|
|
filepath.Join(dataDir, "small.txt"): []byte("hello vaultik"),
|
|
filepath.Join(dataDir, "a.bin"): bytesPattern("a-", int(faultChunkSize*3)),
|
|
filepath.Join(dataDir, "sub", "b.bin"): bytesPattern("b-", int(faultChunkSize*3)),
|
|
filepath.Join(dataDir, "sub", "c.bin"): bytesPattern("c-", int(faultChunkSize*2)),
|
|
}
|
|
|
|
for path, content := range files {
|
|
require.NoError(t, fs.MkdirAll(filepath.Dir(path), 0o755))
|
|
require.NoError(t, afero.WriteFile(fs, path, content, 0o644))
|
|
}
|
|
|
|
return files
|
|
}
|
|
|
|
// newFaultScanner builds a scanner writing through the given storer.
|
|
func newFaultScanner(
|
|
fs afero.Fs, storer storage.Storer,
|
|
cfg *config.Config, repos *database.Repositories,
|
|
) *snapshot.Scanner {
|
|
return snapshot.NewScanner(snapshot.ScannerConfig{
|
|
FS: fs,
|
|
Storage: storer,
|
|
ChunkSize: faultChunkSize,
|
|
MaxBlobSize: faultMaxBlobSize,
|
|
CompressionLevel: cfg.CompressionLevel,
|
|
AgeRecipients: cfg.AgeRecipients,
|
|
Repositories: repos,
|
|
})
|
|
}
|
|
|
|
// newFaultSnapshotManager builds a snapshot manager writing through the
|
|
// given storer.
|
|
func newFaultSnapshotManager(
|
|
fs afero.Fs, storer storage.Storer,
|
|
cfg *config.Config, repos *database.Repositories,
|
|
) *snapshot.SnapshotManager {
|
|
sm := snapshot.NewSnapshotManager(snapshot.SnapshotManagerParams{
|
|
Repos: repos,
|
|
Storage: storer,
|
|
Config: cfg,
|
|
})
|
|
sm.SetFilesystem(fs)
|
|
|
|
return sm
|
|
}
|
|
|
|
// fullFaultBackup runs a complete backup (create, scan, complete,
|
|
// export) through storer and returns the snapshot ID.
|
|
func fullFaultBackup(
|
|
ctx context.Context, t *testing.T, fs afero.Fs, storer storage.Storer,
|
|
cfg *config.Config, repos *database.Repositories,
|
|
dataDir, dbPath, name string,
|
|
) string {
|
|
t.Helper()
|
|
|
|
sm := newFaultSnapshotManager(fs, storer, cfg, repos)
|
|
scanner := newFaultScanner(fs, storer, cfg, repos)
|
|
|
|
id, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, name, "v", "g")
|
|
require.NoError(t, err)
|
|
|
|
_, err = scanner.Scan(ctx, dataDir, id)
|
|
require.NoError(t, err)
|
|
|
|
require.NoError(t, sm.CompleteSnapshot(ctx, id))
|
|
require.NoError(t, sm.ExportSnapshotMetadata(ctx, dbPath, id))
|
|
|
|
return id
|
|
}
|
|
|
|
// newReaderVaultik builds a Vaultik that reads (restore/verify) through
|
|
// storer, with the given repositories (nil is fine for restore/verify,
|
|
// which read metadata from storage).
|
|
func newReaderVaultik(
|
|
ctx context.Context, cfg *config.Config, storer storage.Storer,
|
|
repos *database.Repositories, fs afero.Fs,
|
|
) *vaultik.Vaultik {
|
|
v := &vaultik.Vaultik{
|
|
Config: cfg,
|
|
Storage: storer,
|
|
Repositories: repos,
|
|
Fs: fs,
|
|
Stdout: io.Discard,
|
|
Stderr: io.Discard,
|
|
UI: ui.NewWithColor(io.Discard, false),
|
|
}
|
|
v.SetContext(ctx)
|
|
|
|
return v
|
|
}
|
|
|
|
// Scenario 3: a stored blob's bytes are flipped before restore reads
|
|
// them. Restore must fail loudly, and no file must be left on the
|
|
// restore target holding corrupt content.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestRestoreRejectsCorruptBlob(t *testing.T) {
|
|
assertRestoreRejectsDamagedBlob(t, faultstore.GetCorrupt, "corrupt")
|
|
}
|
|
|
|
// Scenario 4: a stored blob is truncated before restore reads it. Same
|
|
// contract as the corrupt case.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestRestoreRejectsTruncatedBlob(t *testing.T) {
|
|
assertRestoreRejectsDamagedBlob(t, faultstore.GetTruncate, "truncated")
|
|
}
|
|
|
|
// assertRestoreRejectsDamagedBlob backs up the source tree, then restores
|
|
// through a store that damages every blob read with the given fault, and
|
|
// asserts restore fails naming a blob and leaves no file on the target
|
|
// holding wrong bytes. Metadata reads are returned intact so the failure
|
|
// is isolated to the blob.
|
|
func assertRestoreRejectsDamagedBlob(
|
|
t *testing.T, fault faultstore.GetFault, name string,
|
|
) {
|
|
t.Helper()
|
|
log.Initialize(log.Config{})
|
|
|
|
fs := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
restoreDir := filepath.Join(tempDir, "restored")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
cfg := faultTestConfig()
|
|
testFiles := writeFaultSourceTree(t, fs, dataDir)
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
id := fullFaultBackup(ctx, t, fs, inner, cfg, repos, dataDir, dbPath, name)
|
|
require.NoError(t, db.Close())
|
|
|
|
faultStore := faultstore.New(inner)
|
|
faultStore.OnGet = func(key string) faultstore.GetFault {
|
|
if strings.HasPrefix(key, "blobs/") {
|
|
return fault
|
|
}
|
|
|
|
return faultstore.GetNormal
|
|
}
|
|
|
|
v := newReaderVaultik(ctx, cfg, faultStore, nil, fs)
|
|
err = v.Restore(&vaultik.RestoreOptions{SnapshotID: id, TargetDir: restoreDir})
|
|
|
|
require.Error(t, err, "restore must fail on a damaged blob")
|
|
assert.Contains(t, err.Error(), "blob",
|
|
"error should name the blob that failed")
|
|
assertNoCorruptFiles(t, fs, restoreDir, testFiles)
|
|
}
|
|
|
|
// Scenario 6: the backend accepts blob uploads and reports success but
|
|
// stores nothing. verify --deep must catch it.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestDeepVerifyCatchesLyingBackend(t *testing.T) {
|
|
log.Initialize(log.Config{})
|
|
|
|
fs := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
cfg := faultTestConfig()
|
|
|
|
writeFaultSourceTree(t, fs, dataDir)
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
// Blob uploads are swallowed; metadata uploads land, so verify can
|
|
// download the manifest and database and then discover the blobs are
|
|
// absent.
|
|
lying := faultstore.New(inner)
|
|
lying.OnPut = func(key string) faultstore.PutAction {
|
|
if strings.HasPrefix(key, "blobs/") {
|
|
return faultstore.PutSwallow
|
|
}
|
|
|
|
return faultstore.PutNormal
|
|
}
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
id := fullFaultBackup(ctx, t, fs, lying, cfg, repos, dataDir, dbPath, "lying")
|
|
require.NoError(t, db.Close())
|
|
|
|
// No blob objects were actually written.
|
|
blobKeys, err := inner.List(ctx, "blobs/")
|
|
require.NoError(t, err)
|
|
assert.Empty(t, blobKeys, "lying backend should have stored no blobs")
|
|
|
|
// Read back through the honest underlying store.
|
|
v := newReaderVaultik(ctx, cfg, inner, nil, fs)
|
|
err = v.VerifySnapshotWithOptions(id, &vaultik.VerifyOptions{Deep: true})
|
|
require.Error(t, err, "deep verify must catch a backend that stored nothing")
|
|
}
|
|
|
|
// Scenario 1a: a blob upload fails partway through. The interrupted run
|
|
// must not record the blob as uploaded, must not reference it from the
|
|
// snapshot, and must leave no blob object at the destination.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestInterruptedBlobUploadRecordsNoUploadedBlob(t *testing.T) {
|
|
log.Initialize(log.Config{})
|
|
|
|
fs := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
cfg := faultTestConfig()
|
|
|
|
writeFaultSourceTree(t, fs, dataDir)
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
defer func() { _ = db.Close() }()
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
// Every blob upload fails partway through. The scan must surface it.
|
|
fault := faultstore.New(inner)
|
|
fault.OnPut = func(key string) faultstore.PutAction {
|
|
if strings.HasPrefix(key, "blobs/") {
|
|
return faultstore.PutFail
|
|
}
|
|
|
|
return faultstore.PutNormal
|
|
}
|
|
|
|
sm := newFaultSnapshotManager(fs, fault, cfg, repos)
|
|
scanner := newFaultScanner(fs, fault, cfg, repos)
|
|
|
|
id, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "interrupted", "v", "g")
|
|
require.NoError(t, err)
|
|
|
|
_, err = scanner.Scan(ctx, dataDir, id)
|
|
require.Error(t, err, "scan must fail when a blob upload fails")
|
|
|
|
// No blob may claim to be uploaded.
|
|
blobs, err := repos.Blobs.GetAll(ctx)
|
|
require.NoError(t, err)
|
|
|
|
for _, b := range blobs {
|
|
assert.Nilf(t, b.UploadedTS,
|
|
"blob %s marked uploaded after a failed upload", b.Hash)
|
|
}
|
|
|
|
// The snapshot may reference no blobs, and the destination holds none.
|
|
hashes, err := repos.Snapshots.GetBlobHashes(ctx, id)
|
|
require.NoError(t, err)
|
|
assert.Empty(t, hashes, "interrupted snapshot must reference no blobs")
|
|
|
|
blobKeys, err := inner.List(ctx, "blobs/")
|
|
require.NoError(t, err)
|
|
assert.Empty(t, blobKeys, "no blob object may survive at the destination")
|
|
}
|
|
|
|
// Scenario 1b: after an interrupted upload, a retry on the same local
|
|
// index must produce a restorable snapshot. The interrupted run leaves
|
|
// the blob's chunk rows in the index; the fix for
|
|
// https://git.eeqj.de/sneak/vaultik/issues/148 discards those un-uploaded
|
|
// blob rows at the start of the next scan and deduplicates only against
|
|
// chunks in a blob that was actually uploaded, so the retry re-chunks and
|
|
// re-uploads the affected data instead of silently referencing data that
|
|
// never reached storage.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) {
|
|
log.Initialize(log.Config{})
|
|
|
|
fs := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
restoreDir := filepath.Join(tempDir, "restored")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
cfg := faultTestConfig()
|
|
testFiles := writeFaultSourceTree(t, fs, dataDir)
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
// Attempt 1: every blob upload fails.
|
|
fault := faultstore.New(inner)
|
|
fault.OnPut = func(key string) faultstore.PutAction {
|
|
if strings.HasPrefix(key, "blobs/") {
|
|
return faultstore.PutFail
|
|
}
|
|
|
|
return faultstore.PutNormal
|
|
}
|
|
|
|
sm := newFaultSnapshotManager(fs, fault, cfg, repos)
|
|
scanner := newFaultScanner(fs, fault, cfg, repos)
|
|
|
|
id1, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "interrupted", "v", "g")
|
|
require.NoError(t, err)
|
|
|
|
_, err = scanner.Scan(ctx, dataDir, id1)
|
|
require.Error(t, err)
|
|
|
|
// Retry on the same local index with a working backend.
|
|
id2 := fullFaultBackup(ctx, t, fs, inner, cfg, repos, dataDir, dbPath, "retry")
|
|
require.NoError(t, db.Close())
|
|
|
|
v := newReaderVaultik(ctx, cfg, inner, nil, fs)
|
|
require.NoError(t, v.Restore(&vaultik.RestoreOptions{
|
|
SnapshotID: id2,
|
|
TargetDir: restoreDir,
|
|
Verify: true,
|
|
}), "retry after an interrupted upload must produce a restorable snapshot")
|
|
|
|
assertRestoredTree(t, fs, restoreDir, testFiles)
|
|
}
|
|
|
|
// Scenario 2: the process dies during the metadata export, after the
|
|
// database is uploaded but before the manifest. The destination is left
|
|
// with blobs and a database but no manifest. verify and snapshot list
|
|
// must report the damage honestly rather than crashing or passing.
|
|
// Automatic detection and repair of this partial state on the next run is
|
|
// covered by TestBackupCompletesOnlyAfterMetadataExport
|
|
// (https://git.eeqj.de/sneak/vaultik/issues/177); this test exercises the
|
|
// lower-level export path in isolation.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestBackupSurvivesMetadataExportInterruption(t *testing.T) {
|
|
log.Initialize(log.Config{})
|
|
|
|
fs := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
cfg := faultTestConfig()
|
|
|
|
writeFaultSourceTree(t, fs, dataDir)
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
// Back up and complete with a working backend.
|
|
sm := newFaultSnapshotManager(fs, inner, cfg, repos)
|
|
scanner := newFaultScanner(fs, inner, cfg, repos)
|
|
|
|
id, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "export", "v", "g")
|
|
require.NoError(t, err)
|
|
|
|
_, err = scanner.Scan(ctx, dataDir, id)
|
|
require.NoError(t, err)
|
|
require.NoError(t, sm.CompleteSnapshot(ctx, id))
|
|
|
|
// Export through a backend that fails only the manifest upload. The
|
|
// database uploads first and lands; the manifest does not.
|
|
fault := faultstore.New(inner)
|
|
fault.OnPut = func(key string) faultstore.PutAction {
|
|
if strings.HasSuffix(key, "manifest.json.zst") {
|
|
return faultstore.PutFail
|
|
}
|
|
|
|
return faultstore.PutNormal
|
|
}
|
|
|
|
smFault := newFaultSnapshotManager(fs, fault, cfg, repos)
|
|
|
|
err = smFault.ExportSnapshotMetadata(ctx, dbPath, id)
|
|
require.Error(t, err, "export must fail when the manifest upload fails")
|
|
|
|
// The destination is in the partial state the scenario describes.
|
|
key := snapshot.RemoteSnapshotKey(id)
|
|
|
|
_, err = inner.Stat(ctx, "metadata/"+key+"/db.zst.age")
|
|
require.NoError(t, err, "database should have been uploaded before the manifest")
|
|
|
|
_, err = inner.Stat(ctx, "metadata/"+key+"/manifest.json.zst")
|
|
require.ErrorIs(t, err, storage.ErrNotFound, "manifest upload should not have landed")
|
|
|
|
// verify must fail loudly for this snapshot, in both modes.
|
|
reader := newReaderVaultik(ctx, cfg, inner, repos, fs)
|
|
|
|
deepOpts := &vaultik.VerifyOptions{Deep: true}
|
|
require.Error(t, reader.VerifySnapshotWithOptions(id, deepOpts),
|
|
"deep verify must report the missing manifest")
|
|
|
|
shallowOpts := &vaultik.VerifyOptions{Deep: false}
|
|
require.Error(t, reader.VerifySnapshotWithOptions(id, shallowOpts),
|
|
"shallow verify must report the missing manifest")
|
|
|
|
// snapshot list must not crash on the partial snapshot.
|
|
require.NoError(t, reader.ListSnapshots(false),
|
|
"snapshot list must tolerate a partially-exported snapshot")
|
|
}
|
|
|
|
// Scenario 2, repair: the process dies during the metadata export of a
|
|
// full backup run. Because completion is recorded only after the export
|
|
// succeeds (finalizeSnapshotMetadata), the interrupted snapshot is left
|
|
// incomplete rather than silently marked complete without metadata at the
|
|
// destination. Rerunning the backup must then prune the incomplete
|
|
// snapshot, produce a snapshot whose destination metadata and local index
|
|
// agree, and restore. See https://git.eeqj.de/sneak/vaultik/issues/177.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestBackupCompletesOnlyAfterMetadataExport(t *testing.T) {
|
|
log.Initialize(log.Config{})
|
|
|
|
fs := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
restoreDir := filepath.Join(tempDir, "restored")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
testFiles := writeFaultSourceTree(t, fs, dataDir)
|
|
|
|
// A full-backup config: the fault-test defaults plus the fields the
|
|
// production create path reads (index location, chunk size, and the
|
|
// named snapshot to back up).
|
|
cfg := faultTestConfig()
|
|
cfg.IndexPath = dbPath
|
|
cfg.ChunkSize = config.Size(faultChunkSize)
|
|
cfg.Snapshots = map[string]config.SnapshotConfig{
|
|
"data": {Paths: []string{dataDir}},
|
|
}
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
// failManifest is on for the first backup and off for the retry, so the
|
|
// manifest upload fails once — interrupting the export mid-way — then
|
|
// succeeds.
|
|
failManifest := true
|
|
store := faultstore.New(inner)
|
|
store.OnPut = func(key string) faultstore.PutAction {
|
|
if failManifest && strings.HasSuffix(key, "manifest.json.zst") {
|
|
return faultstore.PutFail
|
|
}
|
|
|
|
return faultstore.PutNormal
|
|
}
|
|
|
|
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
|
opts := &vaultik.SnapshotCreateOptions{Cron: true}
|
|
|
|
// First run: the export fails at the manifest upload, so the whole
|
|
// create fails and the snapshot is left incomplete.
|
|
require.Error(t, v.CreateSnapshot(opts),
|
|
"backup must fail when the metadata export is interrupted")
|
|
|
|
incompletes, err := repos.Snapshots.GetIncompleteSnapshots(ctx)
|
|
require.NoError(t, err)
|
|
require.Len(t, incompletes, 1,
|
|
"an interrupted export must leave exactly one incomplete snapshot")
|
|
|
|
afterFirst, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
|
require.NoError(t, err)
|
|
|
|
for _, s := range afterFirst {
|
|
require.Nil(t, s.CompletedAt,
|
|
"no snapshot may be marked complete before its metadata is exported")
|
|
}
|
|
|
|
// Second run on the same index and destination: the retry succeeds.
|
|
failManifest = false
|
|
|
|
require.NoError(t, v.CreateSnapshot(opts),
|
|
"a retry after an interrupted export must succeed")
|
|
|
|
assertRetryConsistentAndRestorable(
|
|
ctx, t, cfg, inner, repos, db, fs, restoreDir, testFiles)
|
|
}
|
|
|
|
// assertRetryConsistentAndRestorable checks the end state after the retry
|
|
// backup in TestBackupCompletesOnlyAfterMetadataExport: the interrupted
|
|
// snapshot is pruned, exactly one completed snapshot remains, its metadata
|
|
// is at the destination, and it restores to the original tree.
|
|
func assertRetryConsistentAndRestorable(
|
|
ctx context.Context, t *testing.T, cfg *config.Config,
|
|
inner storage.Storer, repos *database.Repositories, db *database.DB,
|
|
fs afero.Fs, restoreDir string, testFiles map[string][]byte,
|
|
) {
|
|
t.Helper()
|
|
|
|
incompletes, err := repos.Snapshots.GetIncompleteSnapshots(ctx)
|
|
require.NoError(t, err)
|
|
assert.Empty(t, incompletes,
|
|
"the next run's prune must drop the interrupted snapshot")
|
|
|
|
local, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
|
require.NoError(t, err)
|
|
require.Len(t, local, 1, "exactly one snapshot must remain after the retry")
|
|
|
|
final := local[0]
|
|
require.NotNil(t, final.CompletedAt, "the retry's snapshot must be complete")
|
|
|
|
// The destination and the local index agree: the completed snapshot has
|
|
// both its metadata objects at the destination.
|
|
key := snapshot.RemoteSnapshotKey(final.ID.String())
|
|
_, err = inner.Stat(ctx, "metadata/"+key+"/manifest.json.zst")
|
|
require.NoError(t, err, "the completed snapshot's manifest must be at the destination")
|
|
_, err = inner.Stat(ctx, "metadata/"+key+"/db.zst.age")
|
|
require.NoError(t, err, "the completed snapshot's database must be at the destination")
|
|
|
|
require.NoError(t, db.Close())
|
|
|
|
// The snapshot restores from the destination alone.
|
|
reader := newReaderVaultik(ctx, cfg, inner, nil, fs)
|
|
require.NoError(t, reader.Restore(&vaultik.RestoreOptions{
|
|
SnapshotID: final.ID.String(),
|
|
TargetDir: restoreDir,
|
|
Verify: true,
|
|
}), "the retry's snapshot must be restorable")
|
|
|
|
assertRestoredTree(t, fs, restoreDir, testFiles)
|
|
}
|
|
|
|
// listRecentTestLimit is a generous cap for the handful of snapshots these
|
|
// tests create when reading the local index directly.
|
|
const listRecentTestLimit = 100
|
|
|
|
// newBackupVaultik builds a Vaultik that runs the full create path
|
|
// (CreateSnapshot) writing through storer, wiring the same scanner factory
|
|
// and snapshot manager the production dependency graph provides.
|
|
func newBackupVaultik(
|
|
ctx context.Context, cfg *config.Config, storer storage.Storer,
|
|
repos *database.Repositories, db *database.DB, fs afero.Fs,
|
|
) *vaultik.Vaultik {
|
|
v := &vaultik.Vaultik{
|
|
Globals: &globals.Globals{Version: "v", Commit: "g"},
|
|
Config: cfg,
|
|
DB: db,
|
|
Repositories: repos,
|
|
Storage: storer,
|
|
SnapshotManager: newFaultSnapshotManager(fs, storer, cfg, repos),
|
|
ScannerFactory: faultScannerFactory(cfg, repos, storer),
|
|
Fs: fs,
|
|
Stdout: io.Discard,
|
|
Stderr: io.Discard,
|
|
UI: ui.NewWithColor(io.Discard, false),
|
|
}
|
|
v.SetContext(ctx)
|
|
|
|
return v
|
|
}
|
|
|
|
// faultScannerFactory mirrors the production provideScannerFactory, binding
|
|
// the scanner to the given store, repositories, and config so a full
|
|
// create-path backup writes through the fault-injecting store.
|
|
func faultScannerFactory(
|
|
cfg *config.Config, repos *database.Repositories, storer storage.Storer,
|
|
) snapshot.ScannerFactory {
|
|
return func(params snapshot.ScannerParams) *snapshot.Scanner {
|
|
return snapshot.NewScanner(snapshot.ScannerConfig{
|
|
FS: params.Fs,
|
|
Storage: storer,
|
|
ChunkSize: faultChunkSize,
|
|
MaxBlobSize: faultMaxBlobSize,
|
|
CompressionLevel: cfg.CompressionLevel,
|
|
AgeRecipients: cfg.AgeRecipients,
|
|
Repositories: repos,
|
|
EnableProgress: params.EnableProgress,
|
|
UI: params.UI,
|
|
Exclude: params.Exclude,
|
|
SkipErrors: params.SkipErrors,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Scenario 5: the restore target runs out of space mid-file. Restore
|
|
// must fail with an out-of-space error, and must not leave a truncated
|
|
// file at the target path presenting as a complete restore. Restore
|
|
// today writes each file straight to its final path and does not remove
|
|
// it when a write fails, so the truncated file survives; deleting it is
|
|
// tracked by https://git.eeqj.de/sneak/vaultik/issues/163. Skipped until
|
|
// that lands, so the destination assertion below is recorded rather than
|
|
// dropped.
|
|
//
|
|
//nolint:paralleltest // installs the global logger via log.Initialize
|
|
func TestRestoreReportsDiskFull(t *testing.T) {
|
|
t.Skip("blocked on https://git.eeqj.de/sneak/vaultik/issues/163: " +
|
|
"a disk-full write leaves a truncated file at the target path " +
|
|
"instead of removing it")
|
|
log.Initialize(log.Config{})
|
|
|
|
osFS := afero.NewOsFs()
|
|
tempDir := t.TempDir()
|
|
dataDir := filepath.Join(tempDir, "src")
|
|
storeDir := filepath.Join(tempDir, "remote")
|
|
restoreDir := filepath.Join(tempDir, "restored")
|
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
|
|
ctx := context.Background()
|
|
cfg := faultTestConfig()
|
|
|
|
testFiles := writeFaultSourceTree(t, osFS, dataDir)
|
|
|
|
inner, err := storage.NewFileStorer(storeDir)
|
|
require.NoError(t, err)
|
|
|
|
db, err := database.New(ctx, dbPath)
|
|
require.NoError(t, err)
|
|
|
|
repos := database.NewRepositories(db)
|
|
|
|
id := fullFaultBackup(ctx, t, osFS, inner, cfg, repos, dataDir, dbPath, "diskfull")
|
|
require.NoError(t, db.Close())
|
|
|
|
// Restore onto a filesystem that allows only a few bytes of file
|
|
// content: enough to create files, far too little to hold them.
|
|
budget := int64(8)
|
|
quota := "aFS{Fs: osFS, remaining: &budget}
|
|
|
|
v := newReaderVaultik(ctx, cfg, inner, nil, quota)
|
|
err = v.Restore(&vaultik.RestoreOptions{SnapshotID: id, TargetDir: restoreDir})
|
|
|
|
require.Error(t, err, "restore must fail when the target disk is full")
|
|
assert.Contains(t, err.Error(), errNoSpace.Error(),
|
|
"restore error should surface the out-of-space cause")
|
|
|
|
// The failure must not leave a truncated file behind presenting as a
|
|
// complete restore: any file at the target must hold the original
|
|
// bytes, or be absent.
|
|
assertNoCorruptFiles(t, osFS, restoreDir, testFiles)
|
|
}
|
|
|
|
// assertRestoredTree byte-compares every restored file against the
|
|
// original.
|
|
func assertRestoredTree(
|
|
t *testing.T, fs afero.Fs, restoreDir string, testFiles map[string][]byte,
|
|
) {
|
|
t.Helper()
|
|
|
|
for origPath, expected := range testFiles {
|
|
restoredPath := filepath.Join(restoreDir, origPath)
|
|
got, err := afero.ReadFile(fs, restoredPath)
|
|
require.NoErrorf(t, err, "restored file missing: %s", origPath)
|
|
require.Equalf(t, expected, got, "restored content mismatch for %s", origPath)
|
|
}
|
|
}
|
|
|
|
// errNoSpace is the out-of-space error quotaFS returns once its byte
|
|
// budget is exhausted, mirroring a real ENOSPC.
|
|
var errNoSpace = errors.New("no space left on device")
|
|
|
|
// quotaFS is an afero.Fs whose files may write only a fixed total number
|
|
// of content bytes before failing, simulating a full restore target. It
|
|
// wraps the interface so every method except Create delegates to the
|
|
// real filesystem; only file writes are capped.
|
|
type quotaFS struct {
|
|
afero.Fs
|
|
|
|
remaining *int64
|
|
}
|
|
|
|
//nolint:ireturn // afero.Fs.Create's signature requires returning afero.File.
|
|
func (q *quotaFS) Create(name string) (afero.File, error) {
|
|
f, err := q.Fs.Create(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return "aFile{File: f, remaining: q.remaining}, nil
|
|
}
|
|
|
|
// quotaFile fails writes once the shared byte budget is exhausted.
|
|
type quotaFile struct {
|
|
afero.File
|
|
|
|
remaining *int64
|
|
}
|
|
|
|
func (q *quotaFile) Write(p []byte) (int, error) {
|
|
if *q.remaining <= 0 {
|
|
return 0, errNoSpace
|
|
}
|
|
|
|
allowed := min(int64(len(p)), *q.remaining)
|
|
|
|
n, err := q.File.Write(p[:allowed])
|
|
*q.remaining -= int64(n)
|
|
|
|
if err != nil {
|
|
return n, err
|
|
}
|
|
|
|
if int64(n) < int64(len(p)) {
|
|
return n, errNoSpace
|
|
}
|
|
|
|
return n, nil
|
|
}
|
|
|
|
// assertNoCorruptFiles fails if any file that made it to the restore
|
|
// target holds content that differs from the original: a failed restore
|
|
// may leave a file absent, but must never leave wrong bytes presenting
|
|
// as the real file.
|
|
func assertNoCorruptFiles(
|
|
t *testing.T, fs afero.Fs, restoreDir string, testFiles map[string][]byte,
|
|
) {
|
|
t.Helper()
|
|
|
|
for origPath, expected := range testFiles {
|
|
restoredPath := filepath.Join(restoreDir, origPath)
|
|
|
|
got, err := afero.ReadFile(fs, restoredPath)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
continue
|
|
}
|
|
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
assert.Equalf(t, expected, got,
|
|
"restored file %s holds corrupt content", origPath)
|
|
}
|
|
}
|