Re-chunk a known file whose chunks no uploaded blob holds (closes #214)
File rows are shared by every snapshot and updated in place, while a blob row is deleted once no snapshot references it. Removing the newest snapshot, or the prune after an interrupted run, could drop the only blob holding a changed file's current chunks while an older snapshot kept the file row. The next backup compared metadata only, skipped the file, and completed a snapshot that could not restore it. The scanner now loads the IDs of known files that list a chunk no uploaded blob holds and re-chunks them even when their metadata is unchanged. The tests append to a file, so the file keeps its first chunk in a blob the first snapshot still references. Each backup run gets its own snapshot name, so the second-precision snapshot IDs differ without sleeping. Model: opus-5-5
This commit was merged in pull request #234.
This commit is contained in:
@@ -0,0 +1,234 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/chunker"
|
||||
"sneak.berlin/go/vaultik/internal/config"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// These tests cover https://git.eeqj.de/sneak/vaultik/issues/214: once
|
||||
// the only snapshot referencing a changed file's current blob is
|
||||
// dropped, an older snapshot still keeps the file's row, and the next
|
||||
// backup must re-chunk the file instead of treating it as unchanged.
|
||||
//
|
||||
// Each backup run uses its own snapshot name over the same source
|
||||
// directory, so the snapshot IDs differ without waiting for their
|
||||
// one-second timestamps to tick over. File rows are keyed by path, so
|
||||
// the runs share them exactly as runs of one snapshot name would.
|
||||
|
||||
// changedFileConfig returns a full-backup config with the snapshot names
|
||||
// "first", "second" and "third", all backing up dataDir.
|
||||
func changedFileConfig(dataDir, dbPath string) *config.Config {
|
||||
source := config.SnapshotConfig{Paths: []string{dataDir}}
|
||||
|
||||
cfg := faultTestConfig()
|
||||
cfg.IndexPath = dbPath
|
||||
cfg.ChunkSize = config.Size(faultChunkSize)
|
||||
cfg.Snapshots = map[string]config.SnapshotConfig{
|
||||
"first": source,
|
||||
"second": source,
|
||||
"third": source,
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
||||
// backUp runs a full backup of the named snapshot.
|
||||
func backUp(v *vaultik.Vaultik, name string) error {
|
||||
return v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||
Cron: true,
|
||||
Snapshots: []string{name},
|
||||
})
|
||||
}
|
||||
|
||||
// appendedFileSize is the size of the file before the tests append to it.
|
||||
// It is twice the largest chunk the chunker cuts, so the file's first
|
||||
// chunk ends before the appended bytes and stays in a blob the first
|
||||
// snapshot still references, while its new chunks are only in the blob
|
||||
// that gets dropped.
|
||||
const appendedFileSize = 2 * chunker.ChunkSizeSpread * faultChunkSize
|
||||
|
||||
// appendRandomBytes appends n random bytes to the file at path, creating
|
||||
// the file if it does not exist, and records the new content in files.
|
||||
// Random content gives the chunker real cut points.
|
||||
func appendRandomBytes(
|
||||
t *testing.T, fs afero.Fs, files map[string][]byte, path string, n int64,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
added := make([]byte, n)
|
||||
_, err := rand.Read(added)
|
||||
require.NoError(t, err)
|
||||
|
||||
content := slices.Concat(files[path], added)
|
||||
require.NoError(t, afero.WriteFile(fs, path, content, 0o644))
|
||||
|
||||
files[path] = content
|
||||
}
|
||||
|
||||
// localSnapshotID returns the ID of the one snapshot in the local index
|
||||
// that was backed up under name.
|
||||
func localSnapshotID(
|
||||
ctx context.Context, t *testing.T,
|
||||
repos *database.Repositories, name string,
|
||||
) string {
|
||||
t.Helper()
|
||||
|
||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||
require.NoError(t, err)
|
||||
|
||||
var ids []string
|
||||
|
||||
for _, s := range snapshots {
|
||||
if strings.Contains(s.ID.String(), "_"+name+"_") {
|
||||
ids = append(ids, s.ID.String())
|
||||
}
|
||||
}
|
||||
|
||||
require.Lenf(t, ids, 1, "expected one local snapshot named %q", name)
|
||||
|
||||
return ids[0]
|
||||
}
|
||||
|
||||
// assertThirdSnapshotRestores closes the local index, then restores the
|
||||
// snapshot named "third" from store alone and byte-compares every file
|
||||
// against files.
|
||||
func assertThirdSnapshotRestores(
|
||||
ctx context.Context, t *testing.T, cfg *config.Config,
|
||||
store storage.Storer, repos *database.Repositories, db *database.DB,
|
||||
fs afero.Fs, restoreDir string, files map[string][]byte,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
id := localSnapshotID(ctx, t, repos, "third")
|
||||
require.NoError(t, db.Close())
|
||||
|
||||
reader := newReaderVaultik(ctx, cfg, store, nil, fs)
|
||||
require.NoError(t, reader.Restore(&vaultik.RestoreOptions{
|
||||
SnapshotID: id,
|
||||
TargetDir: restoreDir,
|
||||
Verify: true,
|
||||
}), "the backup after the drop must restore the changed file")
|
||||
|
||||
assertRestoredTree(t, fs, restoreDir, files)
|
||||
}
|
||||
|
||||
// Trigger 1: bytes are appended to the file, a second snapshot backs it
|
||||
// up, and that snapshot is removed. The first snapshot keeps the file row,
|
||||
// which now lists the appended content's chunks, while removal drops the
|
||||
// blob that held them.
|
||||
//
|
||||
//nolint:paralleltest // installs the global logger via log.Initialize
|
||||
func TestBackupAfterRemovingNewestSnapshotRestoresChangedFile(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")
|
||||
changedPath := filepath.Join(dataDir, "appended.bin")
|
||||
|
||||
ctx := context.Background()
|
||||
files := writeFaultSourceTree(t, fs, dataDir)
|
||||
cfg := changedFileConfig(dataDir, dbPath)
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, appendedFileSize)
|
||||
|
||||
store, err := storage.NewFileStorer(storeDir)
|
||||
require.NoError(t, err)
|
||||
|
||||
db, err := database.New(ctx, dbPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||
|
||||
require.NoError(t, backUp(v, "first"))
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||
require.NoError(t, backUp(v, "second"))
|
||||
|
||||
_, err = v.RemoveSnapshot(localSnapshotID(ctx, t, repos, "second"),
|
||||
&vaultik.RemoveOptions{Force: true})
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, backUp(v, "third"))
|
||||
|
||||
assertThirdSnapshotRestores(
|
||||
ctx, t, cfg, store, repos, db, fs, restoreDir, files)
|
||||
}
|
||||
|
||||
// Trigger 2: bytes are appended to the file and the run that backs it up
|
||||
// is interrupted at the manifest upload, after its blobs were uploaded.
|
||||
// The next run's prune drops that incomplete snapshot and its blob, while
|
||||
// the first snapshot keeps the file row, which now lists the appended
|
||||
// content's chunks.
|
||||
//
|
||||
//nolint:paralleltest // installs the global logger via log.Initialize
|
||||
func TestBackupAfterInterruptedRunRestoresChangedFile(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")
|
||||
changedPath := filepath.Join(dataDir, "appended.bin")
|
||||
|
||||
ctx := context.Background()
|
||||
files := writeFaultSourceTree(t, fs, dataDir)
|
||||
cfg := changedFileConfig(dataDir, dbPath)
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, appendedFileSize)
|
||||
|
||||
inner, err := storage.NewFileStorer(storeDir)
|
||||
require.NoError(t, err)
|
||||
|
||||
failManifest := false
|
||||
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
|
||||
}
|
||||
|
||||
db, err := database.New(ctx, dbPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||
|
||||
require.NoError(t, backUp(v, "first"))
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||
|
||||
failManifest = true
|
||||
|
||||
require.Error(t, backUp(v, "second"),
|
||||
"the run must fail when its manifest upload fails")
|
||||
|
||||
failManifest = false
|
||||
|
||||
require.NoError(t, backUp(v, "third"))
|
||||
|
||||
assertThirdSnapshotRestores(
|
||||
ctx, t, cfg, inner, repos, db, fs, restoreDir, files)
|
||||
}
|
||||
Reference in New Issue
Block a user