Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2b61947eb0 |
No files matched your search
@@ -22,7 +22,6 @@ require (
|
|||||||
github.com/spf13/cobra v1.10.1
|
github.com/spf13/cobra v1.10.1
|
||||||
github.com/spf13/pflag v1.0.10
|
github.com/spf13/pflag v1.0.10
|
||||||
github.com/stretchr/testify v1.11.1
|
github.com/stretchr/testify v1.11.1
|
||||||
go.uber.org/dig v1.19.0
|
|
||||||
go.uber.org/fx v1.24.0
|
go.uber.org/fx v1.24.0
|
||||||
golang.org/x/sync v0.18.0
|
golang.org/x/sync v0.18.0
|
||||||
golang.org/x/sys v0.38.0
|
golang.org/x/sys v0.38.0
|
||||||
@@ -257,6 +256,7 @@ require (
|
|||||||
go.opentelemetry.io/otel/metric v1.38.0 // indirect
|
go.opentelemetry.io/otel/metric v1.38.0 // indirect
|
||||||
go.opentelemetry.io/otel/trace v1.38.0 // indirect
|
go.opentelemetry.io/otel/trace v1.38.0 // indirect
|
||||||
go.shabbyrobe.org/gocovmerge v0.0.0-20230507111327-fa4f82cfbf4d // indirect
|
go.shabbyrobe.org/gocovmerge v0.0.0-20230507111327-fa4f82cfbf4d // indirect
|
||||||
|
go.uber.org/dig v1.19.0 // indirect
|
||||||
go.uber.org/multierr v1.11.0 // indirect
|
go.uber.org/multierr v1.11.0 // indirect
|
||||||
go.uber.org/zap v1.27.0 // indirect
|
go.uber.org/zap v1.27.0 // indirect
|
||||||
go.yaml.in/yaml/v2 v2.4.3 // indirect
|
go.yaml.in/yaml/v2 v2.4.3 // indirect
|
||||||
|
|||||||
+10
-6
@@ -8,12 +8,12 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/adrg/xdg"
|
"github.com/adrg/xdg"
|
||||||
"github.com/spf13/cobra"
|
"github.com/spf13/cobra"
|
||||||
"go.uber.org/dig"
|
|
||||||
"go.uber.org/fx"
|
"go.uber.org/fx"
|
||||||
"sneak.berlin/go/vaultik/internal/config"
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
"sneak.berlin/go/vaultik/internal/database"
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
@@ -146,12 +146,16 @@ func (e *startupError) Error() string {
|
|||||||
// could not build arguments for function "Y" (file:line): failed to build U:
|
// could not build arguments for function "Y" (file:line): failed to build U:
|
||||||
// received non-nil error from function "Z" (file:line): <real error>
|
// received non-nil error from function "Z" (file:line): <real error>
|
||||||
//
|
//
|
||||||
// Users care about the real error, not the DI plumbing. dig.RootCause,
|
// Users care about the real error, not the DI plumbing. We strip everything
|
||||||
// from the library fx is built on, unwraps those layers and returns the
|
// up through the last "): " (which is always the close-paren of an fx
|
||||||
// real error whole. Cutting the text at "): " instead would also cut a
|
// function-location annotation followed by the wrapped error).
|
||||||
// real error that contains it, such as the blob_size_limit one.
|
|
||||||
func cleanStartupError(err error) error {
|
func cleanStartupError(err error) error {
|
||||||
return &startupError{msg: dig.RootCause(err).Error()}
|
msg := err.Error()
|
||||||
|
if idx := strings.LastIndex(msg, "): "); idx >= 0 {
|
||||||
|
msg = msg[idx+3:]
|
||||||
|
}
|
||||||
|
|
||||||
|
return &startupError{msg: msg}
|
||||||
}
|
}
|
||||||
|
|
||||||
// RunApp starts the fx application, blocks until it is asked to stop, and
|
// RunApp starts the fx application, blocks until it is asked to stop, and
|
||||||
|
|||||||
+40
-47
@@ -2,65 +2,58 @@ package cli //nolint:testpackage // needs access to unexported cleanStartupError
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"os"
|
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"go.uber.org/fx"
|
|
||||||
"sneak.berlin/go/vaultik/internal/config"
|
|
||||||
"sneak.berlin/go/vaultik/internal/pidlock"
|
"sneak.berlin/go/vaultik/internal/pidlock"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestCleanStartupError starts a real fx app whose config has a
|
|
||||||
// blob_size_limit below the minimum. fx puts annotations naming the
|
|
||||||
// functions it was building in front of the config error; only those
|
|
||||||
// may be removed, although the config error itself contains "): ".
|
|
||||||
func TestCleanStartupError(t *testing.T) {
|
func TestCleanStartupError(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
path := filepath.Join(t.TempDir(), "config.yml")
|
tests := []struct {
|
||||||
|
name string
|
||||||
err := os.WriteFile(path, []byte("snapshots:\n"+
|
in string
|
||||||
" test:\n"+
|
want string
|
||||||
" paths: [/tmp/vaultik-test-source]\n"+
|
}{
|
||||||
"storage_url: file:///tmp/vaultik-test-storage\n"+
|
{
|
||||||
"chunk_size: 10MB\n"+
|
name: "real fx error chain",
|
||||||
"blob_size_limit: 30MB\n"), 0o600)
|
in: `could not build arguments for function ` +
|
||||||
if err != nil {
|
`"sneak.berlin/go/vaultik/internal/cli".newSnapshotCreateCommand.func1.1 ` +
|
||||||
t.Fatalf("write config: %v", err)
|
`(/Users/user/dev/vaultik/internal/cli/snapshot.go:71): ` +
|
||||||
|
`failed to build *vaultik.Vaultik: ` +
|
||||||
|
`could not build arguments for function ` +
|
||||||
|
`"sneak.berlin/go/vaultik/internal/vaultik".New ` +
|
||||||
|
`(/Users/user/dev/vaultik/internal/vaultik/vaultik.go:59): ` +
|
||||||
|
`failed to build storage.Storer: ` +
|
||||||
|
`received non-nil error from function ` +
|
||||||
|
`"sneak.berlin/go/vaultik/internal/storage".NewStorer ` +
|
||||||
|
`(/Users/user/dev/vaultik/internal/storage/module.go:23): ` +
|
||||||
|
`creating base path: mkdir /Volumes/BACKUPS: permission denied`,
|
||||||
|
want: `creating base path: mkdir /Volumes/BACKUPS: permission denied`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "no fx wrapping",
|
||||||
|
in: "plain error",
|
||||||
|
want: "plain error",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "single fx wrapping",
|
||||||
|
in: `received non-nil error from function "foo" (file.go:1): underlying problem`,
|
||||||
|
want: "underlying problem",
|
||||||
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
app := fx.New(
|
for _, tt := range tests {
|
||||||
fx.NopLogger,
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
fx.Supply(config.Path(path)),
|
t.Parallel()
|
||||||
config.Module,
|
|
||||||
fx.Invoke(func(*config.Config) {}),
|
|
||||||
)
|
|
||||||
|
|
||||||
err = app.Start(t.Context())
|
//nolint:err113 // test constructs errors from table input
|
||||||
if err == nil {
|
got := cleanStartupError(errors.New(tt.in)).Error()
|
||||||
t.Fatal("app started; want a blob_size_limit error")
|
if got != tt.want {
|
||||||
}
|
t.Errorf("got %q, want %q", got, tt.want)
|
||||||
|
}
|
||||||
want := "failed to load config: invalid config: blob_size_limit must " +
|
})
|
||||||
"be at least the largest chunk the chunker can emit (chunk_size " +
|
|
||||||
"times the FastCDC size spread): need at least 40000000 bytes"
|
|
||||||
|
|
||||||
got := cleanStartupError(err).Error()
|
|
||||||
if got != want {
|
|
||||||
t.Errorf("got %q, want %q", got, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestCleanStartupErrorNotFromFx checks that an error fx did not wrap is
|
|
||||||
// shown unchanged.
|
|
||||||
func TestCleanStartupErrorNotFromFx(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
//nolint:err113 // a one-off error standing in for a startup failure
|
|
||||||
got := cleanStartupError(errors.New("plain error")).Error()
|
|
||||||
if got != "plain error" {
|
|
||||||
t.Errorf("got %q, want %q", got, "plain error")
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -39,9 +39,9 @@ type blobDiskCacheEntry struct {
|
|||||||
// internally and will fall back to LRU eviction if curBytes exceeds
|
// internally and will fall back to LRU eviction if curBytes exceeds
|
||||||
// maxBytes. Restore passes math.MaxInt64 as maxBytes and drives
|
// maxBytes. Restore passes math.MaxInt64 as maxBytes and drives
|
||||||
// eviction itself via Delete() through restoreSweeper, which deletes
|
// eviction itself via Delete() through restoreSweeper, which deletes
|
||||||
// each blob once no file still to be restored in this run needs it.
|
// each blob the moment every file that references its chunks has been
|
||||||
// LRU never fires under that configuration; it is kept as a safety net
|
// written. LRU never fires under that configuration; it is kept as a
|
||||||
// for callers that don't manage eviction themselves.
|
// safety net for callers that don't manage eviction themselves.
|
||||||
//
|
//
|
||||||
// Get/ReadAt/peak-Len counters are debugging instrumentation used by
|
// Get/ReadAt/peak-Len counters are debugging instrumentation used by
|
||||||
// tests to assert that the restore code path uses ReadAt rather than
|
// tests to assert that the restore code path uses ReadAt rather than
|
||||||
|
|||||||
@@ -1,17 +1,13 @@
|
|||||||
package vaultik //nolint:testpackage // exercises unexported count helpers
|
package vaultik //nolint:testpackage // exercises unexported count helpers
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"sneak.berlin/go/vaultik/internal/database"
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestTableCountForReportSurfacesReadFailure is the regression guard for
|
// TestTableCountForReportSurfacesReadFailure is the regression guard for
|
||||||
@@ -80,59 +76,3 @@ func TestCountDiffUnknownWhenEitherSideUnknown(t *testing.T) {
|
|||||||
assert.Nil(t, countDiff(&before, nil), "unknown after yields unknown delta")
|
assert.Nil(t, countDiff(&before, nil), "unknown after yields unknown delta")
|
||||||
assert.Nil(t, countDiff(nil, nil))
|
assert.Nil(t, countDiff(nil, nil))
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPruneDatabaseSummaryLabelsTotals removes the only incomplete
|
|
||||||
// snapshot from an index that also holds a complete one. The figure in
|
|
||||||
// brackets is every snapshot left in the local index, so the line must
|
|
||||||
// say so rather than read as incomplete snapshots that remain.
|
|
||||||
func TestPruneDatabaseSummaryLabelsTotals(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
db, err := database.New(ctx, ":memory:")
|
|
||||||
require.NoError(t, err)
|
|
||||||
t.Cleanup(func() { _ = db.Close() })
|
|
||||||
|
|
||||||
repos := database.NewRepositories(db)
|
|
||||||
|
|
||||||
startedAt := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
|
||||||
completedAt := startedAt.Add(time.Minute)
|
|
||||||
|
|
||||||
require.NoError(t, repos.Snapshots.Create(ctx, nil, &database.Snapshot{
|
|
||||||
ID: "host_complete_2026-01-02T03:04:05Z",
|
|
||||||
Hostname: "host",
|
|
||||||
StartedAt: startedAt,
|
|
||||||
CompletedAt: &completedAt,
|
|
||||||
}))
|
|
||||||
require.NoError(t, repos.Snapshots.Create(ctx, nil, &database.Snapshot{
|
|
||||||
ID: "host_incomplete_2026-01-02T03:04:05Z",
|
|
||||||
Hostname: "host",
|
|
||||||
StartedAt: startedAt,
|
|
||||||
}))
|
|
||||||
|
|
||||||
var out bytes.Buffer
|
|
||||||
|
|
||||||
v := &Vaultik{
|
|
||||||
DB: db,
|
|
||||||
Repositories: repos,
|
|
||||||
SnapshotManager: snapshot.NewSnapshotManager(
|
|
||||||
snapshot.SnapshotManagerParams{Repos: repos}),
|
|
||||||
UI: ui.NewWithColor(&out, false),
|
|
||||||
}
|
|
||||||
v.SetContext(ctx)
|
|
||||||
|
|
||||||
result, err := v.PruneDatabase()
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Equal(t, int64(1), result.SnapshotsDeleted)
|
|
||||||
|
|
||||||
assert.Contains(t, out.String(), "Incomplete snapshots: 1 removed "+
|
|
||||||
"(total snapshots left in the local index: 1).")
|
|
||||||
assert.Contains(t, out.String(), "Orphaned files: 0 removed "+
|
|
||||||
"(total files left in the local index: 0).")
|
|
||||||
assert.Contains(t, out.String(), "Orphaned chunks: 0 removed "+
|
|
||||||
"(total chunks left in the local index: 0).")
|
|
||||||
assert.Contains(t, out.String(), "Orphaned blobs: 0 removed "+
|
|
||||||
"(total blobs left in the local index: 0).")
|
|
||||||
}
|
|
||||||
@@ -323,6 +323,12 @@ func (v *Vaultik) restoreAllFiles(
|
|||||||
_ = blobCache.Close()
|
_ = blobCache.Close()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
// Per-restore sweep state: every blob_size_limit/100 bytes written,
|
||||||
|
// scan the cache and delete any blob whose remaining file references
|
||||||
|
// are all already restored.
|
||||||
|
sweeper := newRestoreSweeper(v.ctx, repos, blobCache,
|
||||||
|
v.Config.BlobSizeLimit.Int64()/sweepIntervalDivisor)
|
||||||
|
|
||||||
blobByHash, blobIDToHash, err := v.buildBlobIndexes(repos)
|
blobByHash, blobIDToHash, err := v.buildBlobIndexes(repos)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -334,12 +340,6 @@ func (v *Vaultik) restoreAllFiles(
|
|||||||
return nil, fmt.Errorf("building restore plan: %w", err)
|
return nil, fmt.Errorf("building restore plan: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Per-restore sweep state: every blob_size_limit/100 bytes written,
|
|
||||||
// scan the cache and delete any blob that no pending file in the
|
|
||||||
// plan references.
|
|
||||||
sweeper := newRestoreSweeper(plan, blobCache,
|
|
||||||
v.Config.BlobSizeLimit.Int64()/sweepIntervalDivisor)
|
|
||||||
|
|
||||||
// Ranging over files rather than the map reports them in a stable order.
|
// Ranging over files rather than the map reports them in a stable order.
|
||||||
for _, file := range files {
|
for _, file := range files {
|
||||||
if planErr, ok := unplanned[file.ID]; ok {
|
if planErr, ok := unplanned[file.ID]; ok {
|
||||||
@@ -427,8 +427,10 @@ func (v *Vaultik) runRestoreLoop(
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
// Drop the file from the plan's indexes so future picks ignore
|
// Record the file as restored so the sweeper can free blobs
|
||||||
// it and the sweeper can free blobs no pending file needs.
|
// once all referencing files are done, and drop it from the
|
||||||
|
// plan's indexes so future picks ignore it.
|
||||||
|
session.sweeper.fileRestored(fileID.String())
|
||||||
plan.finishFile(fileID)
|
plan.finishFile(fileID)
|
||||||
|
|
||||||
processed++
|
processed++
|
||||||
|
|||||||
@@ -74,8 +74,7 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
|
|||||||
buildLocalityFixture(t, fs, dataDir)
|
buildLocalityFixture(t, fs, dataDir)
|
||||||
|
|
||||||
cfg, storer, snapshotID := setupLocalityBackup(
|
cfg, storer, snapshotID := setupLocalityBackup(
|
||||||
context.Background(), t, fs, dataDir, storeDir, dbPath,
|
context.Background(), t, fs, dataDir, storeDir, dbPath)
|
||||||
localitySourceBytes)
|
|
||||||
|
|
||||||
// Point the "" temp paths (the blob cache directory and the
|
// Point the "" temp paths (the blob cache directory and the
|
||||||
// snapshot-database directory) at a private directory so the test can
|
// snapshot-database directory) at a private directory so the test can
|
||||||
|
|||||||
@@ -63,10 +63,6 @@ type localityCopy struct {
|
|||||||
data []byte
|
data []byte
|
||||||
}
|
}
|
||||||
|
|
||||||
// localitySourceBytes is the size of each source file of the locality
|
|
||||||
// fixture.
|
|
||||||
const localitySourceBytes = 1024 * 1024
|
|
||||||
|
|
||||||
// buildLocalityFixture writes the adversarial source layout described
|
// buildLocalityFixture writes the adversarial source layout described
|
||||||
// in TestRestoreLocalityAndReadAt: 15 sources of 1 MiB each (which the
|
// in TestRestoreLocalityAndReadAt: 15 sources of 1 MiB each (which the
|
||||||
// backup packs into 3 blobs of 5 chunks) plus 9 interleaved-name copies
|
// backup packs into 3 blobs of 5 chunks) plus 9 interleaved-name copies
|
||||||
@@ -77,6 +73,7 @@ func buildLocalityFixture(
|
|||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
srcBytes = 1024 * 1024
|
||||||
srcCount = 15
|
srcCount = 15
|
||||||
blobsCount = 3
|
blobsCount = 3
|
||||||
perBlob = srcCount / blobsCount
|
perBlob = srcCount / blobsCount
|
||||||
@@ -86,7 +83,7 @@ func buildLocalityFixture(
|
|||||||
for i := range srcCount {
|
for i := range srcCount {
|
||||||
s := &localitySource{
|
s := &localitySource{
|
||||||
path: fmt.Sprintf("src-%03d.bin", i+1),
|
path: fmt.Sprintf("src-%03d.bin", i+1),
|
||||||
data: randomBytes(t, localitySourceBytes),
|
data: randomBytes(t, srcBytes),
|
||||||
}
|
}
|
||||||
sources[i] = s
|
sources[i] = s
|
||||||
require.NoError(t, afero.WriteFile(
|
require.NoError(t, afero.WriteFile(
|
||||||
@@ -118,20 +115,19 @@ func buildLocalityFixture(
|
|||||||
|
|
||||||
// setupLocalityBackup runs the backup half of the locality test: a
|
// setupLocalityBackup runs the backup half of the locality test: a
|
||||||
// snapshot of dataDir into a file storer with sizes tuned so 15 one-chunk
|
// snapshot of dataDir into a file storer with sizes tuned so 15 one-chunk
|
||||||
// files of srcBytes pack into 3 blobs. The index database is closed
|
// files pack into 3 blobs. The index database is closed before returning
|
||||||
// before returning so the restore half runs from remote bytes only.
|
// so the restore half runs from remote bytes only.
|
||||||
func setupLocalityBackup(
|
func setupLocalityBackup(
|
||||||
ctx context.Context, t *testing.T, fs afero.Fs,
|
ctx context.Context, t *testing.T, fs afero.Fs,
|
||||||
dataDir, storeDir, dbPath string, srcBytes int64,
|
dataDir, storeDir, dbPath string,
|
||||||
) (*config.Config, *storage.FileStorer, string) {
|
) (*config.Config, *storage.FileStorer, string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
// An average chunk size of four times srcBytes makes the minimum
|
// chunkSize avg = 4 MiB makes minSize = 1 MiB, so a 1 MiB file
|
||||||
// chunk srcBytes, so each source file becomes one chunk. A blob limit
|
// becomes one chunk. maxBlobSize = 5 MiB packs exactly 5 chunks
|
||||||
// of five times srcBytes packs exactly 5 chunks per blob, yielding 3
|
// per blob, yielding 3 blobs from 15 source files.
|
||||||
// blobs from 15 source files.
|
chunkSize := int64(4 * 1024 * 1024)
|
||||||
chunkSize := 4 * srcBytes
|
maxBlobSize := int64(5 * 1024 * 1024)
|
||||||
maxBlobSize := 5 * srcBytes
|
|
||||||
|
|
||||||
storer, err := storage.NewFileStorer(storeDir)
|
storer, err := storage.NewFileStorer(storeDir)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -212,7 +208,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) {
|
|||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
cfg, storer, snapshotID := setupLocalityBackup(
|
cfg, storer, snapshotID := setupLocalityBackup(
|
||||||
ctx, t, fs, dataDir, storeDir, dbPath, localitySourceBytes)
|
ctx, t, fs, dataDir, storeDir, dbPath)
|
||||||
|
|
||||||
// Wrap the storer so we can count downloads per blob key.
|
// Wrap the storer so we can count downloads per blob key.
|
||||||
counter := newCountingStorer(storer)
|
counter := newCountingStorer(storer)
|
||||||
|
|||||||
@@ -31,8 +31,7 @@ var (
|
|||||||
// with no further downloads.
|
// with no further downloads.
|
||||||
// - blobFiles: for each blob, the set of pending files referencing
|
// - blobFiles: for each blob, the set of pending files referencing
|
||||||
// it. Used to short-circuit "when this blob lands, which files
|
// it. Used to short-circuit "when this blob lands, which files
|
||||||
// become ready" without a global scan, and by the sweeper to tell
|
// become ready" without a global scan.
|
||||||
// whether a cached blob is still needed.
|
|
||||||
type restorePlan struct {
|
type restorePlan struct {
|
||||||
fileBlobs map[types.FileID]map[string]struct{}
|
fileBlobs map[types.FileID]map[string]struct{}
|
||||||
blobFiles map[string]map[types.FileID]struct{}
|
blobFiles map[string]map[types.FileID]struct{}
|
||||||
@@ -238,12 +237,6 @@ func (p *restorePlan) filesReferencingBlob(blobHash string) []types.FileID {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// pendingFileUsesBlob reports whether any pending file references the
|
|
||||||
// named blob, whether or not the blob is cached.
|
|
||||||
func (p *restorePlan) pendingFileUsesBlob(blobHash string) bool {
|
|
||||||
return len(p.blobFiles[blobHash]) > 0
|
|
||||||
}
|
|
||||||
|
|
||||||
// hasPending reports whether any unfinished files remain.
|
// hasPending reports whether any unfinished files remain.
|
||||||
func (p *restorePlan) hasPending() bool {
|
func (p *restorePlan) hasPending() bool {
|
||||||
return len(p.fileBlobs) > 0
|
return len(p.fileBlobs) > 0
|
||||||
|
|||||||
@@ -63,6 +63,15 @@ func TestRunRestoreLoopFailsOnAbandonedFiles(t *testing.T) {
|
|||||||
|
|
||||||
t.Cleanup(func() { _ = cache.Close() })
|
t.Cleanup(func() { _ = cache.Close() })
|
||||||
|
|
||||||
|
v := &Vaultik{ctx: ctx}
|
||||||
|
session := &restoreSession{
|
||||||
|
v: v,
|
||||||
|
ctx: ctx,
|
||||||
|
repos: repos,
|
||||||
|
sweeper: newRestoreSweeper(ctx, repos, cache, 1),
|
||||||
|
result: &RestoreResult{},
|
||||||
|
}
|
||||||
|
|
||||||
// A file that is still pending but whose uncached-blob set is empty
|
// A file that is still pending but whose uncached-blob set is empty
|
||||||
// and which was never queued as ready: the loop can neither restore
|
// and which was never queued as ready: the loop can neither restore
|
||||||
// nor download it. This is the abandonment the guard must catch.
|
// nor download it. This is the abandonment the guard must catch.
|
||||||
@@ -74,15 +83,6 @@ func TestRunRestoreLoopFailsOnAbandonedFiles(t *testing.T) {
|
|||||||
cached: map[string]struct{}{},
|
cached: map[string]struct{}{},
|
||||||
}
|
}
|
||||||
|
|
||||||
v := &Vaultik{ctx: ctx}
|
|
||||||
session := &restoreSession{
|
|
||||||
v: v,
|
|
||||||
ctx: ctx,
|
|
||||||
repos: repos,
|
|
||||||
sweeper: newRestoreSweeper(plan, cache, 1),
|
|
||||||
result: &RestoreResult{},
|
|
||||||
}
|
|
||||||
|
|
||||||
err = v.runRestoreLoop(session, plan, map[types.FileID]*database.File{}, 0)
|
err = v.runRestoreLoop(session, plan, map[types.FileID]*database.File{}, 0)
|
||||||
require.ErrorIs(t, err, errRestoreIncomplete)
|
require.ErrorIs(t, err, errRestoreIncomplete)
|
||||||
}
|
}
|
||||||
@@ -1,30 +1,46 @@
|
|||||||
package vaultik
|
package vaultik
|
||||||
|
|
||||||
// restoreSweeper frees cached blobs once no file still to be restored in
|
import (
|
||||||
// this run needs them. It works as follows:
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
)
|
||||||
|
|
||||||
|
// restoreSweeper frees cached blobs once all files that reference any of
|
||||||
|
// their chunks have been restored. It works as follows:
|
||||||
//
|
//
|
||||||
// 1. After each chunk is restored, chunkRestored accumulates a running
|
// 1. Callers add a file's ID to an in-memory restored set via
|
||||||
|
// fileRestored once the file is fully written to disk.
|
||||||
|
// 2. After each chunk is restored, chunkRestored accumulates a running
|
||||||
// byte count.
|
// byte count.
|
||||||
// 2. When the accumulator crosses a threshold (one hundredth of the
|
// 3. When the accumulator crosses a threshold (one hundredth of the
|
||||||
// configured blob size — so a sweep runs about a hundred times per
|
// configured blob size — so a sweep runs about a hundred times per
|
||||||
// blob's worth of restored bytes), the sweeper iterates every key in
|
// blob's worth of restored bytes), the sweeper iterates every key in
|
||||||
// the cache and deletes each blob that the restore plan no longer
|
// the cache. For each cached blob it asks the DB which files
|
||||||
// needs.
|
// reference any chunk in that blob, then compares that list against
|
||||||
|
// the in-memory restored set. If any referencing file is missing
|
||||||
|
// from the set the blob is kept; otherwise the cache entry is
|
||||||
|
// deleted.
|
||||||
//
|
//
|
||||||
// The plan drops a file once it is restored or has failed, and holds
|
// All DB reads happen against the snapshot's temporary metadata DB,
|
||||||
// only the files this run restores, so files outside the requested paths
|
// which is local, indexed, and not under contention — the queries are
|
||||||
// never keep a blob.
|
// cheap and run at most once per blob per sweep interval.
|
||||||
type restoreSweeper struct {
|
type restoreSweeper struct {
|
||||||
plan *restorePlan
|
ctx context.Context //nolint:containedctx // ctx bound at construction by design
|
||||||
|
repos *database.Repositories
|
||||||
cache *blobDiskCache
|
cache *blobDiskCache
|
||||||
threshold int64
|
threshold int64
|
||||||
bytesAccum int64
|
bytesAccum int64
|
||||||
|
restored map[string]struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
// newRestoreSweeper returns a sweeper that triggers eviction every
|
// newRestoreSweeper returns a sweeper that triggers eviction every
|
||||||
// `threshold` bytes restored. Callers should pass blob_size_limit/100.
|
// `threshold` bytes restored. Callers should pass blob_size_limit/100.
|
||||||
func newRestoreSweeper(
|
func newRestoreSweeper(
|
||||||
plan *restorePlan,
|
ctx context.Context,
|
||||||
|
repos *database.Repositories,
|
||||||
cache *blobDiskCache,
|
cache *blobDiskCache,
|
||||||
threshold int64,
|
threshold int64,
|
||||||
) *restoreSweeper {
|
) *restoreSweeper {
|
||||||
@@ -33,12 +49,21 @@ func newRestoreSweeper(
|
|||||||
}
|
}
|
||||||
|
|
||||||
return &restoreSweeper{
|
return &restoreSweeper{
|
||||||
plan: plan,
|
ctx: ctx,
|
||||||
|
repos: repos,
|
||||||
cache: cache,
|
cache: cache,
|
||||||
threshold: threshold,
|
threshold: threshold,
|
||||||
|
restored: make(map[string]struct{}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// fileRestored records a file as fully restored. After this call, any
|
||||||
|
// blob whose only remaining references come from files in the restored
|
||||||
|
// set may be evicted on the next sweep.
|
||||||
|
func (s *restoreSweeper) fileRestored(fileID string) {
|
||||||
|
s.restored[fileID] = struct{}{}
|
||||||
|
}
|
||||||
|
|
||||||
// chunkRestored accounts n bytes against the sweep threshold and runs a
|
// chunkRestored accounts n bytes against the sweep threshold and runs a
|
||||||
// sweep if the threshold has been crossed since the last sweep.
|
// sweep if the threshold has been crossed since the last sweep.
|
||||||
func (s *restoreSweeper) chunkRestored(n int64) {
|
func (s *restoreSweeper) chunkRestored(n int64) {
|
||||||
@@ -51,11 +76,60 @@ func (s *restoreSweeper) chunkRestored(n int64) {
|
|||||||
s.sweep()
|
s.sweep()
|
||||||
}
|
}
|
||||||
|
|
||||||
// sweep deletes every cached blob that no pending file references.
|
// sweep deletes any cached blob whose chunks are no longer referenced
|
||||||
|
// by an unrestored file. Per-blob DB failures are logged and the blob
|
||||||
|
// is kept — we'd rather hold a blob longer than risk a re-download.
|
||||||
func (s *restoreSweeper) sweep() {
|
func (s *restoreSweeper) sweep() {
|
||||||
for _, blobHash := range s.cache.Keys() {
|
for _, blobHash := range s.cache.Keys() {
|
||||||
if !s.plan.pendingFileUsesBlob(blobHash) {
|
needed, err := s.blobStillNeeded(blobHash)
|
||||||
|
if err != nil {
|
||||||
|
log.Debug("sweeper referencing-files query failed",
|
||||||
|
"blob_hash", blobHash[:16], "error", err)
|
||||||
|
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
if !needed {
|
||||||
s.cache.Delete(blobHash)
|
s.cache.Delete(blobHash)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// blobStillNeeded returns true if any file that references a chunk in
|
||||||
|
// this blob has not yet been restored. On any error the function
|
||||||
|
// returns true — keeping the blob is always the safe answer because we
|
||||||
|
// can't prove we're done with it.
|
||||||
|
func (s *restoreSweeper) blobStillNeeded(blobHash string) (bool, error) {
|
||||||
|
rows, err := s.repos.DB().Conn().QueryContext(s.ctx, `
|
||||||
|
SELECT DISTINCT fc.file_id
|
||||||
|
FROM file_chunks fc
|
||||||
|
JOIN blob_chunks bc ON bc.chunk_hash = fc.chunk_hash
|
||||||
|
JOIN blobs b ON b.id = bc.blob_id
|
||||||
|
WHERE b.blob_hash = ?
|
||||||
|
`, blobHash)
|
||||||
|
if err != nil {
|
||||||
|
return true, fmt.Errorf("querying referencing files: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
defer func() { _ = rows.Close() }()
|
||||||
|
|
||||||
|
for rows.Next() {
|
||||||
|
var fileID string
|
||||||
|
|
||||||
|
err := rows.Scan(&fileID)
|
||||||
|
if err != nil {
|
||||||
|
return true, fmt.Errorf("scanning file_id: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, ok := s.restored[fileID]; !ok {
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
err = rows.Err()
|
||||||
|
if err != nil {
|
||||||
|
return true, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
@@ -1,163 +0,0 @@
|
|||||||
package vaultik //nolint:testpackage // reads the restore blob cache's peak size
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"path/filepath"
|
|
||||||
"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/log"
|
|
||||||
"sneak.berlin/go/vaultik/internal/storage"
|
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
|
||||||
)
|
|
||||||
|
|
||||||
// TestRestoreFreesBlobsNoPendingFileNeeds asserts that the restore cache
|
|
||||||
// frees a blob once no file still to be restored in this run needs it.
|
|
||||||
// Every original file has a copy with the same content, so every blob
|
|
||||||
// holds chunks of copies: a restore that leaves the copies out of its
|
|
||||||
// paths, or fails on them, must not keep the blobs for them.
|
|
||||||
func TestRestoreFreesBlobsNoPendingFileNeeds(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
// Small files keep the backup and both restores quick.
|
|
||||||
const fileBytes = 16 * 1024
|
|
||||||
|
|
||||||
fs := afero.NewOsFs()
|
|
||||||
tempDir := t.TempDir()
|
|
||||||
|
|
||||||
dataDir := filepath.Join(tempDir, "source")
|
|
||||||
originalsDir := filepath.Join(dataDir, "originals")
|
|
||||||
copiesDir := filepath.Join(dataDir, "copies")
|
|
||||||
storeDir := filepath.Join(tempDir, "remote")
|
|
||||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
||||||
|
|
||||||
originals := writeFilesWithCopies(t, fs, originalsDir, copiesDir, fileBytes)
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
cfg, storer, snapshotID := setupLocalityBackup(
|
|
||||||
ctx, t, fs, dataDir, storeDir, dbPath, fileBytes)
|
|
||||||
|
|
||||||
t.Run("restore by path", func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
restoreDir := filepath.Join(tempDir, "by-path")
|
|
||||||
|
|
||||||
peak, err := restorePeakCachedBlobs(ctx, t, fs, cfg, storer,
|
|
||||||
&RestoreOptions{
|
|
||||||
SnapshotID: snapshotID,
|
|
||||||
TargetDir: restoreDir,
|
|
||||||
Paths: []string{originalsDir},
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
requireRestoredFiles(t, fs, restoreDir, originals)
|
|
||||||
assert.LessOrEqualf(t, peak, 1,
|
|
||||||
"peak cached blobs was %d; blobs were kept for unselected files",
|
|
||||||
peak)
|
|
||||||
})
|
|
||||||
|
|
||||||
t.Run("restore with failed files", func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
restoreDir := filepath.Join(tempDir, "with-failures")
|
|
||||||
|
|
||||||
// A regular file where the copies directory belongs makes every
|
|
||||||
// copy fail to restore.
|
|
||||||
blocker := filepath.Join(restoreDir, copiesDir)
|
|
||||||
require.NoError(t, fs.MkdirAll(filepath.Dir(blocker), 0o755))
|
|
||||||
require.NoError(t, afero.WriteFile(fs, blocker, nil, 0o644))
|
|
||||||
|
|
||||||
peak, err := restorePeakCachedBlobs(ctx, t, fs, cfg, storer,
|
|
||||||
&RestoreOptions{
|
|
||||||
SnapshotID: snapshotID,
|
|
||||||
TargetDir: restoreDir,
|
|
||||||
SkipErrors: true,
|
|
||||||
})
|
|
||||||
require.ErrorIs(t, err, errFilesFailedRestore)
|
|
||||||
|
|
||||||
requireRestoredFiles(t, fs, restoreDir, originals)
|
|
||||||
assert.LessOrEqualf(t, peak, 1,
|
|
||||||
"peak cached blobs was %d; blobs were kept for failed files", peak)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// writeFilesWithCopies writes 15 random files of fileBytes into
|
|
||||||
// originalsDir and a copy of each into copiesDir. With
|
|
||||||
// setupLocalityBackup given the same size, each file is one chunk and the
|
|
||||||
// 15 chunks fill 3 blobs. Returns the originals' content by path.
|
|
||||||
func writeFilesWithCopies(
|
|
||||||
t *testing.T, fs afero.Fs, originalsDir, copiesDir string, fileBytes int,
|
|
||||||
) map[string][]byte {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
const fileCount = 15
|
|
||||||
|
|
||||||
require.NoError(t, fs.MkdirAll(originalsDir, 0o755))
|
|
||||||
require.NoError(t, fs.MkdirAll(copiesDir, 0o755))
|
|
||||||
|
|
||||||
originals := make(map[string][]byte, fileCount)
|
|
||||||
|
|
||||||
for i := range fileCount {
|
|
||||||
name := fmt.Sprintf("file-%02d.bin", i)
|
|
||||||
data := randomBytes(t, fileBytes)
|
|
||||||
original := filepath.Join(originalsDir, name)
|
|
||||||
originals[original] = data
|
|
||||||
|
|
||||||
require.NoError(t, afero.WriteFile(fs, original, data, 0o644))
|
|
||||||
require.NoError(t, afero.WriteFile(
|
|
||||||
fs, filepath.Join(copiesDir, name), data, 0o644))
|
|
||||||
}
|
|
||||||
|
|
||||||
return originals
|
|
||||||
}
|
|
||||||
|
|
||||||
// requireRestoredFiles requires every file in want to have been restored
|
|
||||||
// under restoreDir with its content intact.
|
|
||||||
func requireRestoredFiles(
|
|
||||||
t *testing.T, fs afero.Fs, restoreDir string, want map[string][]byte,
|
|
||||||
) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
for path, data := range want {
|
|
||||||
got, err := afero.ReadFile(fs, filepath.Join(restoreDir, path))
|
|
||||||
require.NoErrorf(t, err, "file missing after restore: %s", path)
|
|
||||||
require.Equalf(t, data, got, "byte mismatch for %s", path)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// restorePeakCachedBlobs runs a restore and returns the most blobs its
|
|
||||||
// cache held at once, together with the restore's error.
|
|
||||||
func restorePeakCachedBlobs(
|
|
||||||
ctx context.Context, t *testing.T, fs afero.Fs,
|
|
||||||
cfg *config.Config, storer storage.Storer, opts *RestoreOptions,
|
|
||||||
) (int, error) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
var cache *blobDiskCache
|
|
||||||
|
|
||||||
v := &Vaultik{
|
|
||||||
Config: cfg,
|
|
||||||
Storage: storer,
|
|
||||||
Fs: fs,
|
|
||||||
Stdout: io.Discard,
|
|
||||||
Stderr: io.Discard,
|
|
||||||
UI: ui.NewWithColor(io.Discard, false),
|
|
||||||
restoreCacheObserver: func(c *blobDiskCache) {
|
|
||||||
cache = c
|
|
||||||
},
|
|
||||||
}
|
|
||||||
v.SetContext(ctx)
|
|
||||||
|
|
||||||
err := v.Restore(opts)
|
|
||||||
|
|
||||||
require.NotNil(t, cache, "restoreCacheObserver must fire during restore")
|
|
||||||
|
|
||||||
return cache.PeakLen(), err
|
|
||||||
}
|
|
||||||
@@ -1905,17 +1905,13 @@ func (v *Vaultik) PruneDatabase() (*PruneResult, error) {
|
|||||||
snapshotsRemain := countDiff(snapshotCountBefore, &result.SnapshotsDeleted)
|
snapshotsRemain := countDiff(snapshotCountBefore, &result.SnapshotsDeleted)
|
||||||
|
|
||||||
v.UI.Completef("Pruned local index database.")
|
v.UI.Completef("Pruned local index database.")
|
||||||
v.UI.Detailf("Incomplete snapshots: %s removed "+
|
v.UI.Detailf("Incomplete snapshots: %s removed (%s remain).",
|
||||||
"(total snapshots left in the local index: %s).",
|
|
||||||
countText(&result.SnapshotsDeleted), countText(snapshotsRemain))
|
countText(&result.SnapshotsDeleted), countText(snapshotsRemain))
|
||||||
v.UI.Detailf("Orphaned files: %s removed "+
|
v.UI.Detailf("Orphaned files: %s removed (%s remain).",
|
||||||
"(total files left in the local index: %s).",
|
|
||||||
countText(result.FilesDeleted), countText(fileCountAfter))
|
countText(result.FilesDeleted), countText(fileCountAfter))
|
||||||
v.UI.Detailf("Orphaned chunks: %s removed "+
|
v.UI.Detailf("Orphaned chunks: %s removed (%s remain).",
|
||||||
"(total chunks left in the local index: %s).",
|
|
||||||
countText(result.ChunksDeleted), countText(chunkCountAfter))
|
countText(result.ChunksDeleted), countText(chunkCountAfter))
|
||||||
v.UI.Detailf("Orphaned blobs: %s removed "+
|
v.UI.Detailf("Orphaned blobs: %s removed (%s remain).",
|
||||||
"(total blobs left in the local index: %s).",
|
|
||||||
countText(result.BlobsDeleted), countText(blobCountAfter))
|
countText(result.BlobsDeleted), countText(blobCountAfter))
|
||||||
|
|
||||||
return result, nil
|
return result, nil
|
||||||
|
|||||||
Reference in new issue
Block a user