Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7fcb78013c |
@@ -25,18 +25,16 @@ release" is exactly the contradiction
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-09-21: Stopped an interrupted blob upload from making a later
|
||||
backup deduplicate against data that was never stored
|
||||
([issue #148](https://git.eeqj.de/sneak/vaultik/issues/148)). The
|
||||
packer commits a blob's `chunks`, `blob_chunks`, and `blobs` rows
|
||||
before the upload is attempted, so a failed upload left chunk rows
|
||||
behind and the next run skipped re-uploading them, producing a
|
||||
snapshot that reported success but could not be restored. A run now
|
||||
deduplicates only against chunks held by a blob whose `uploaded_ts` is
|
||||
set, and at startup drops any un-uploaded blob rows (and the chunks
|
||||
they orphan) so the affected data is re-chunked and re-uploaded. Blobs
|
||||
recorded with no remote backend are marked uploaded so this invariant
|
||||
holds uniformly.
|
||||
- 2026-09-21: Stopped `--json` from silencing stderr diagnostics
|
||||
([issue #112](https://git.eeqj.de/sneak/vaultik/issues/112)). `--json`
|
||||
used to be folded into `Quiet`, which pinned the log level to `WARN`,
|
||||
so `prune --json` gave a machine consumer no record of the local index
|
||||
rows it deleted even under `--verbose`. `--json` now quiets only the
|
||||
stdout UI (the JSON document must stay clean, per
|
||||
[issue #108](https://git.eeqj.de/sneak/vaultik/issues/108)); the stderr
|
||||
log level follows `--verbose`/`--debug` again. The coupling was
|
||||
removed the same way for `snapshot verify`, `snapshot remove`, and
|
||||
`remote info`, which carried it for the same outdated reason.
|
||||
|
||||
- 2026-09-21: Stopped `prune` from reporting a failed row count as 0
|
||||
([issue #96](https://git.eeqj.de/sneak/vaultik/issues/96)). The seven
|
||||
|
||||
+12
-5
@@ -49,6 +49,11 @@ type AppOptions struct {
|
||||
// silenced — per the documented convention that --quiet suppresses
|
||||
// non-error output only. The startup banner is printed by Entry
|
||||
// before cobra parses arguments, gated by the same arg-level check.
|
||||
//
|
||||
// --json quiets the UI here too, because stdout then carries a JSON
|
||||
// document and human narration would corrupt it. Unlike Quiet it does
|
||||
// not lower the stderr log level (issue #112), so --verbose/--debug
|
||||
// still surface diagnostics alongside the document.
|
||||
func setupGlobals(
|
||||
lc fx.Lifecycle, g *globals.Globals, v *vaultik.Vaultik, opts log.Options,
|
||||
) {
|
||||
@@ -56,7 +61,7 @@ func setupGlobals(
|
||||
OnStart: func(_ context.Context) error {
|
||||
g.StartTime = time.Now().UTC()
|
||||
|
||||
if opts.Cron || opts.Quiet {
|
||||
if opts.Cron || opts.Quiet || opts.JSON {
|
||||
v.UI.SetQuiet(true)
|
||||
}
|
||||
|
||||
@@ -279,10 +284,11 @@ func RunOperation(
|
||||
// shared by the list/purge/verify/remove/remote-info subcommands:
|
||||
// resolve the config, then run op against the Vaultik instance through
|
||||
// RunOperation, reporting a failure prefixed with failMsg (suppressed
|
||||
// while suppressErrors is true, e.g. under --json). extraQuiet is OR-ed
|
||||
// into LogOptions.Quiet (e.g. --json output modes).
|
||||
// while suppressErrors is true, e.g. under --json). jsonOutput marks a
|
||||
// command whose stdout is a JSON document: it quiets the UI but, unlike
|
||||
// Quiet, leaves the stderr log level alone.
|
||||
func runVaultikApp(
|
||||
cmd *cobra.Command, extraQuiet, suppressErrors bool,
|
||||
cmd *cobra.Command, jsonOutput, suppressErrors bool,
|
||||
failMsg string, op func(v *vaultik.Vaultik) error,
|
||||
) error {
|
||||
configPath, err := ResolveConfigPath()
|
||||
@@ -297,7 +303,8 @@ func runVaultikApp(
|
||||
LogOptions: log.Options{
|
||||
Verbose: rootFlags.Verbose,
|
||||
Debug: rootFlags.Debug,
|
||||
Quiet: rootFlags.Quiet || extraQuiet,
|
||||
Quiet: rootFlags.Quiet,
|
||||
JSON: jsonOutput,
|
||||
},
|
||||
}, op, func(err error) {
|
||||
if suppressErrors {
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
package cli //nolint:testpackage // shares the prune fixtures and capture helpers
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// staleRecordLogMessage is the local-cleanup audit line CleanupLocalSnapshots
|
||||
// logs for each stale record. It is exactly the signal issue #112 says a
|
||||
// machine consumer lost under --json: gated off stdout, and pinned below
|
||||
// the log level on stderr because --json used to force Quiet.
|
||||
const staleRecordLogMessage = "Removing stale local snapshot record"
|
||||
|
||||
// TestEntryPruneJSONStderrHonoursVerbosity is the end-to-end regression
|
||||
// guard for issue #112. Under --json the log level must still follow
|
||||
// --verbose/--debug rather than being pinned to WARN, so the
|
||||
// local-cleanup records reach stderr under --verbose while stdout stays
|
||||
// exactly one JSON document; without --verbose they stay below the
|
||||
// level, as they do without --json.
|
||||
//
|
||||
// Both halves are asserted together on the same run, because the fix has
|
||||
// to keep the document clean (issue #108) while freeing stderr.
|
||||
//
|
||||
// Not parallel: it replaces os.Args, os.Stdout, os.Stderr and the xdg
|
||||
// globals.
|
||||
//
|
||||
//nolint:paralleltest // replaces os.Args, os.Stdout, os.Stderr and the xdg globals
|
||||
func TestEntryPruneJSONStderrHonoursVerbosity(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
verbose bool
|
||||
wantOnStderr bool
|
||||
}{
|
||||
{
|
||||
name: "verbose json surfaces the cleanup record on stderr",
|
||||
verbose: true,
|
||||
wantOnStderr: true,
|
||||
},
|
||||
{
|
||||
name: "json alone keeps the cleanup record below the level",
|
||||
verbose: false,
|
||||
wantOnStderr: false,
|
||||
},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
configPath := writeHermeticPruneConfig(t, true)
|
||||
|
||||
previousArgs := os.Args
|
||||
|
||||
t.Cleanup(func() {
|
||||
os.Args = previousArgs
|
||||
rootFlags = RootFlags{}
|
||||
})
|
||||
|
||||
args := []string{
|
||||
programName, flagConfig, configPath, cmdPrune, flagJSON,
|
||||
}
|
||||
if testCase.verbose {
|
||||
args = append(args, "--verbose")
|
||||
}
|
||||
|
||||
os.Args = args
|
||||
|
||||
stdout, stderr := captureProcessStdoutAndStderr(t,
|
||||
func() { _ = Entry() })
|
||||
|
||||
// The document stays clean in both cases: freeing stderr must
|
||||
// not regress issue #108.
|
||||
requireExactlyOneJSONDocument(t, stdout)
|
||||
|
||||
if testCase.wantOnStderr {
|
||||
assert.Contains(t, stderr, staleRecordLogMessage,
|
||||
"--verbose --json must emit the cleanup record on stderr")
|
||||
assert.Contains(t, stderr, stalePruneSnapshotID,
|
||||
"the record must name the snapshot it removed")
|
||||
} else {
|
||||
assert.NotContains(t, stderr, staleRecordLogMessage,
|
||||
"without --verbose the record stays below the log level")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// captureProcessStdoutAndStderr redirects both of the process's own
|
||||
// standard streams to pipes for the duration of fn and returns what was
|
||||
// written to each. The redirection is at the file-descriptor level
|
||||
// because the logger binds os.Stderr when it initializes inside fn, and
|
||||
// the JSON document reaches os.Stdout independently; the point is to see
|
||||
// where each actually lands.
|
||||
//
|
||||
// Not parallel-safe: os.Stdout and os.Stderr are process-global.
|
||||
func captureProcessStdoutAndStderr(t *testing.T, fn func()) (string, string) {
|
||||
t.Helper()
|
||||
|
||||
outReader, outWriter, err := os.Pipe()
|
||||
require.NoError(t, err)
|
||||
|
||||
errReader, errWriter, err := os.Pipe()
|
||||
require.NoError(t, err)
|
||||
|
||||
previousOut, previousErr := os.Stdout, os.Stderr
|
||||
os.Stdout, os.Stderr = outWriter, errWriter
|
||||
|
||||
capturedOut := drain(outReader)
|
||||
capturedErr := drain(errReader)
|
||||
|
||||
fn()
|
||||
|
||||
os.Stdout, os.Stderr = previousOut, previousErr
|
||||
|
||||
require.NoError(t, outWriter.Close())
|
||||
require.NoError(t, errWriter.Close())
|
||||
|
||||
out, errOut := <-capturedOut, <-capturedErr
|
||||
|
||||
require.NoError(t, outReader.Close())
|
||||
require.NoError(t, errReader.Close())
|
||||
|
||||
return out, errOut
|
||||
}
|
||||
|
||||
// drain copies a reader to a string on a goroutine and delivers the
|
||||
// result once the writer end is closed.
|
||||
func drain(reader io.Reader) <-chan string {
|
||||
captured := make(chan string, 1)
|
||||
|
||||
go func() {
|
||||
var buf bytes.Buffer
|
||||
|
||||
_, _ = io.Copy(&buf, reader)
|
||||
captured <- buf.String()
|
||||
}()
|
||||
|
||||
return captured
|
||||
}
|
||||
@@ -41,7 +41,8 @@ work (e.g. after a crashed backup or to reclaim storage).`,
|
||||
LogOptions: log.Options{
|
||||
Verbose: rootFlags.Verbose,
|
||||
Debug: rootFlags.Debug,
|
||||
Quiet: rootFlags.Quiet || opts.JSON,
|
||||
Quiet: rootFlags.Quiet,
|
||||
JSON: opts.JSON,
|
||||
},
|
||||
}, func(v *vaultik.Vaultik) error {
|
||||
return v.Prune(opts)
|
||||
|
||||
@@ -85,7 +85,8 @@ func newRemoteInfoCommand() *cobra.Command {
|
||||
LogOptions: log.Options{
|
||||
Verbose: rootFlags.Verbose,
|
||||
Debug: rootFlags.Debug,
|
||||
Quiet: rootFlags.Quiet || jsonOutput,
|
||||
Quiet: rootFlags.Quiet,
|
||||
JSON: jsonOutput,
|
||||
},
|
||||
}, func(v *vaultik.Vaultik) error {
|
||||
return v.RemoteInfo(jsonOutput)
|
||||
|
||||
@@ -209,7 +209,8 @@ func newSnapshotVerifyCommand() *cobra.Command {
|
||||
LogOptions: log.Options{
|
||||
Verbose: rootFlags.Verbose,
|
||||
Debug: rootFlags.Debug,
|
||||
Quiet: rootFlags.Quiet || opts.JSON,
|
||||
Quiet: rootFlags.Quiet,
|
||||
JSON: opts.JSON,
|
||||
},
|
||||
}, func(v *vaultik.Vaultik) error {
|
||||
return v.VerifySnapshotWithOptions(snapshotID, opts)
|
||||
|
||||
@@ -208,30 +208,6 @@ func (r *BlobRepository) DeleteOrphaned(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteUnuploaded deletes blob rows whose upload never completed
|
||||
// (uploaded_ts IS NULL) and returns how many were removed. Their
|
||||
// blob_chunks rows are removed by the ON DELETE CASCADE foreign key.
|
||||
// A blob is only ever attached to a snapshot once its upload has been
|
||||
// recorded, so an un-uploaded blob is never referenced by a completed
|
||||
// snapshot: dropping it discards chunk rows that point at data which
|
||||
// was never stored remotely, so the affected content is re-chunked and
|
||||
// re-uploaded on the next run.
|
||||
func (r *BlobRepository) DeleteUnuploaded(ctx context.Context) (int64, error) {
|
||||
query := `DELETE FROM blobs WHERE uploaded_ts IS NULL`
|
||||
|
||||
result, err := r.db.ExecWithLog(ctx, query)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("deleting un-uploaded blobs: %w", err)
|
||||
}
|
||||
|
||||
rowsAffected, _ := result.RowsAffected()
|
||||
if rowsAffected > 0 {
|
||||
log.Debug("Deleted un-uploaded blobs", "count", rowsAffected)
|
||||
}
|
||||
|
||||
return rowsAffected, nil
|
||||
}
|
||||
|
||||
// getOne fetches a single blob row matched on the given column, or
|
||||
// (nil, nil) when no row matches.
|
||||
func (r *BlobRepository) getOne(
|
||||
|
||||
@@ -7,32 +7,12 @@ import (
|
||||
|
||||
// List returns every chunk in the index, ordered by chunk hash.
|
||||
func (r *ChunkRepository) List(ctx context.Context) ([]*Chunk, error) {
|
||||
return r.list(ctx, `
|
||||
query := `
|
||||
SELECT chunk_hash, size
|
||||
FROM chunks
|
||||
ORDER BY chunk_hash
|
||||
`)
|
||||
}
|
||||
`
|
||||
|
||||
// ListInUploadedBlobs returns the chunks that are stored in a blob whose
|
||||
// upload has completed (uploaded_ts set), ordered by chunk hash. These
|
||||
// are the only chunks a backup may safely deduplicate against: a chunk
|
||||
// recorded solely in a blob that was never uploaded refers to data that
|
||||
// is not in remote storage, so trusting it would silently drop that data
|
||||
// from later snapshots.
|
||||
func (r *ChunkRepository) ListInUploadedBlobs(ctx context.Context) ([]*Chunk, error) {
|
||||
return r.list(ctx, `
|
||||
SELECT DISTINCT c.chunk_hash, c.size
|
||||
FROM chunks c
|
||||
JOIN blob_chunks bc ON c.chunk_hash = bc.chunk_hash
|
||||
JOIN blobs b ON bc.blob_id = b.id
|
||||
WHERE b.uploaded_ts IS NOT NULL
|
||||
ORDER BY c.chunk_hash
|
||||
`)
|
||||
}
|
||||
|
||||
// list runs a chunk-selecting query and scans the (chunk_hash, size) rows.
|
||||
func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, error) {
|
||||
rows, err := r.db.conn.QueryContext(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||
|
||||
+15
-1
@@ -14,8 +14,18 @@ var Module = fx.Module("log",
|
||||
)
|
||||
|
||||
// New creates a new logger configuration from provided options.
|
||||
//
|
||||
// JSON is intentionally not carried into Config: a command emitting a
|
||||
// JSON document on stdout must keep its stderr log level under
|
||||
// --verbose/--debug, so --json must not lower it (issue #112). JSON
|
||||
// silences the stdout UI in setupGlobals instead.
|
||||
func New(opts Options) Config {
|
||||
return Config(opts)
|
||||
return Config{
|
||||
Verbose: opts.Verbose,
|
||||
Debug: opts.Debug,
|
||||
Cron: opts.Cron,
|
||||
Quiet: opts.Quiet,
|
||||
}
|
||||
}
|
||||
|
||||
// Options are provided by the CLI.
|
||||
@@ -24,4 +34,8 @@ type Options struct {
|
||||
Debug bool
|
||||
Cron bool
|
||||
Quiet bool
|
||||
// JSON marks a command whose stdout carries a machine-readable
|
||||
// document. It silences the human UI on stdout (see setupGlobals),
|
||||
// but unlike Quiet it leaves the stderr log level alone.
|
||||
JSON bool
|
||||
}
|
||||
|
||||
@@ -220,14 +220,7 @@ func (s *Scanner) Scan(
|
||||
defer s.progress.Stop()
|
||||
}
|
||||
|
||||
// Phase 0: Repair any state left by an interrupted previous run, then
|
||||
// load known files and chunks from the database into memory for fast
|
||||
// lookup.
|
||||
err := s.repairInterruptedBlobs(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Phase 0: Load known files and chunks from database into memory for fast lookup
|
||||
knownFiles, err := s.loadDatabaseState(ctx, path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -324,38 +317,6 @@ func (s *Scanner) loadDatabaseState(
|
||||
return knownFiles, nil
|
||||
}
|
||||
|
||||
// repairInterruptedBlobs discards blob rows left by a previous run whose
|
||||
// upload never completed. Such a blob has its chunks, blob_chunks, and
|
||||
// blobs rows committed to the local index before the upload is attempted,
|
||||
// so a crash or dropped connection mid-upload leaves them behind while the
|
||||
// data never reaches remote storage. Deduplicating against those chunks on
|
||||
// a later run would produce a snapshot that reports success but cannot be
|
||||
// restored. Dropping the un-uploaded blobs (their blob_chunks cascade) and
|
||||
// then any chunks left unreferenced forces the affected data to be
|
||||
// re-chunked and re-uploaded this run. A blob is attached to a snapshot
|
||||
// only once its upload is recorded, so this never touches a completed
|
||||
// snapshot's data.
|
||||
func (s *Scanner) repairInterruptedBlobs(ctx context.Context) error {
|
||||
removed, err := s.repos.Blobs.DeleteUnuploaded(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("removing un-uploaded blob records: %w", err)
|
||||
}
|
||||
|
||||
if removed == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Warn("Discarded blob records from an interrupted previous run; "+
|
||||
"their data will be re-uploaded", "blobs", removed)
|
||||
|
||||
err = s.repos.Chunks.DeleteOrphaned(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("removing orphaned chunks: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// summarizeScanPhase calculates total size to process, updates progress tracking,
|
||||
// and prints the scan phase summary with file counts and sizes
|
||||
func (s *Scanner) summarizeScanPhase(
|
||||
@@ -431,14 +392,11 @@ func (s *Scanner) loadKnownFiles(
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// loadKnownChunks loads the chunk hashes safe to deduplicate against into
|
||||
// an in-memory map for fast lookup, avoiding per-chunk database queries
|
||||
// during file processing. Only chunks held by a blob whose upload
|
||||
// completed are loaded: a chunk left behind by an interrupted upload
|
||||
// refers to data that never reached remote storage, and deduplicating
|
||||
// against it would silently produce an unrestorable snapshot.
|
||||
// loadKnownChunks loads all known chunk hashes from the database into a
|
||||
// map for fast lookup. This avoids per-chunk database queries during file
|
||||
// processing.
|
||||
func (s *Scanner) loadKnownChunks(ctx context.Context) error {
|
||||
chunks, err := s.repos.Chunks.ListInUploadedBlobs(ctx)
|
||||
chunks, err := s.repos.Chunks.List(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listing chunks: %w", err)
|
||||
}
|
||||
@@ -1443,17 +1401,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
|
||||
return fmt.Errorf("parsing blob ID: %w", err)
|
||||
}
|
||||
|
||||
// With no remote backend the blob's lifecycle ends here, so
|
||||
// mark it uploaded in the same transaction that attaches it to
|
||||
// the snapshot. This keeps the invariant that any blob a
|
||||
// snapshot references has uploaded_ts set, so deduplication and
|
||||
// interrupted-run repair treat these blobs as trustworthy.
|
||||
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||
err := s.repos.Blobs.UpdateUploaded(ctx, tx, b.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marking blob uploaded: %w", err)
|
||||
}
|
||||
|
||||
return s.repos.Snapshots.AddBlob(ctx, tx, s.snapshotID, blobID,
|
||||
types.BlobHash(b.Hash))
|
||||
})
|
||||
|
||||
@@ -1,198 +0,0 @@
|
||||
package vaultik //nolint:testpackage // constructs Vaultik with unexported fields
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"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/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
"sneak.berlin/go/vaultik/internal/ui"
|
||||
)
|
||||
|
||||
// errUploadInterrupted stands in for a dropped connection or kill -9
|
||||
// partway through a blob upload.
|
||||
var errUploadInterrupted = errors.New("simulated interrupted blob upload")
|
||||
|
||||
// interruptedBlobStorer wraps a real Storer but fails every blob upload,
|
||||
// modelling a run that dies mid-blob after the packer has already
|
||||
// committed the blob's chunk rows to the local index.
|
||||
type interruptedBlobStorer struct {
|
||||
storage.Storer
|
||||
}
|
||||
|
||||
func (s *interruptedBlobStorer) PutWithProgress(
|
||||
ctx context.Context, key string, reader io.Reader,
|
||||
size int64, cb storage.ProgressCallback,
|
||||
) error {
|
||||
if strings.HasPrefix(key, "blobs/") {
|
||||
return errUploadInterrupted
|
||||
}
|
||||
|
||||
return s.Storer.PutWithProgress(ctx, key, reader, size, cb)
|
||||
}
|
||||
|
||||
func (s *interruptedBlobStorer) Put(
|
||||
ctx context.Context, key string, reader io.Reader,
|
||||
) error {
|
||||
if strings.HasPrefix(key, "blobs/") {
|
||||
return errUploadInterrupted
|
||||
}
|
||||
|
||||
return s.Storer.Put(ctx, key, reader)
|
||||
}
|
||||
|
||||
// TestBackupRetryAfterInterruptedUploadIsRestorable reproduces the
|
||||
// silent-data-loss defect in
|
||||
// https://git.eeqj.de/sneak/vaultik/issues/148: a blob upload is
|
||||
// interrupted, leaving chunk rows in the local index for data that never
|
||||
// reached storage. The retry run reuses the same index. Before the fix it
|
||||
// deduplicated against those orphaned chunks, uploaded nothing for them,
|
||||
// and produced a snapshot that reported success but could not be
|
||||
// restored. The retried snapshot must instead restore byte-for-byte.
|
||||
func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
tempDir := t.TempDir()
|
||||
|
||||
dataDir := filepath.Join(tempDir, "source")
|
||||
storeDir := filepath.Join(tempDir, "remote")
|
||||
restoreDir := filepath.Join(tempDir, "restored")
|
||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||
|
||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||
|
||||
// Random content forces real chunks; small blobs guarantee at least
|
||||
// one blob is finalized (and its upload attempted) during the run.
|
||||
sources := map[string][]byte{
|
||||
"a.bin": randomBytes(t, 128*1024),
|
||||
"b.bin": randomBytes(t, 128*1024),
|
||||
"c.bin": randomBytes(t, 128*1024),
|
||||
}
|
||||
for name, data := range sources {
|
||||
require.NoError(t, afero.WriteFile(
|
||||
fs, filepath.Join(dataDir, name), data, 0o644))
|
||||
}
|
||||
|
||||
cfg := &config.Config{
|
||||
AgeRecipients: []string{"age1ezrjmfpwsc95svdg0y54mums3zevgzu0x0ecq2" +
|
||||
"f7tp8a05gl0sjq9q9wjg"},
|
||||
AgeSecretKey: "AGE-SECRET-KEY-19CR5YSFW59HM4TLD6GXVEDMZFTVVF7PPHKU" +
|
||||
"T68TXSFPK7APHXA2QS2NJA5",
|
||||
CompressionLevel: 3,
|
||||
Hostname: "test-host",
|
||||
ChunkSize: config.Size(16 * 1024),
|
||||
BlobSizeLimit: config.Size(64 * 1024),
|
||||
}
|
||||
|
||||
working, err := storage.NewFileStorer(storeDir)
|
||||
require.NoError(t, err)
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Run 1: the upload is interrupted, so the scan fails but leaves the
|
||||
// interrupted blob's chunk rows committed in the index.
|
||||
_, err = runBackup(ctx, fs, cfg, &interruptedBlobStorer{Storer: working},
|
||||
dataDir, dbPath, "interrupted")
|
||||
require.Error(t, err, "interrupted upload must fail the run")
|
||||
|
||||
// Run 2: retry on the same index with a working backend. This must
|
||||
// succeed and produce a fully restorable snapshot.
|
||||
snapshotID, err := runBackup(ctx, fs, cfg, working, dataDir, dbPath, "retry")
|
||||
require.NoError(t, err, "retry backup must succeed")
|
||||
|
||||
v := &Vaultik{
|
||||
Config: cfg,
|
||||
Storage: working,
|
||||
Fs: fs,
|
||||
Stdout: io.Discard,
|
||||
Stderr: io.Discard,
|
||||
UI: ui.NewWithColor(io.Discard, false),
|
||||
}
|
||||
v.SetContext(ctx)
|
||||
|
||||
require.NoError(t, v.Restore(&RestoreOptions{
|
||||
SnapshotID: snapshotID,
|
||||
TargetDir: restoreDir,
|
||||
}), "the retried snapshot must be restorable")
|
||||
|
||||
for name, data := range sources {
|
||||
restored := filepath.Join(restoreDir, dataDir, name)
|
||||
got, err := afero.ReadFile(fs, restored)
|
||||
require.NoErrorf(t, err, "restored file missing: %s", name)
|
||||
require.Truef(t, bytes.Equal(got, data),
|
||||
"restored bytes differ from original for %s", name)
|
||||
}
|
||||
}
|
||||
|
||||
// runBackup performs one backup of dataDir into a fresh snapshot on the
|
||||
// index at dbPath, returning the snapshot ID. When the scan succeeds it
|
||||
// also completes and exports the snapshot metadata so the result can be
|
||||
// restored. The index database is always closed before returning.
|
||||
func runBackup(
|
||||
ctx context.Context,
|
||||
fs afero.Fs,
|
||||
cfg *config.Config,
|
||||
storer storage.Storer,
|
||||
dataDir, dbPath, name string,
|
||||
) (string, error) {
|
||||
db, err := database.New(ctx, dbPath)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
defer func() { _ = db.Close() }()
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
|
||||
sm := snapshot.NewSnapshotManager(snapshot.SnapshotManagerParams{
|
||||
Repos: repos,
|
||||
Storage: storer,
|
||||
Config: cfg,
|
||||
})
|
||||
sm.SetFilesystem(fs)
|
||||
|
||||
scanner := snapshot.NewScanner(snapshot.ScannerConfig{
|
||||
FS: fs,
|
||||
Storage: storer,
|
||||
ChunkSize: cfg.ChunkSize.Int64(),
|
||||
MaxBlobSize: cfg.BlobSizeLimit.Int64(),
|
||||
CompressionLevel: cfg.CompressionLevel,
|
||||
AgeRecipients: cfg.AgeRecipients,
|
||||
Repositories: repos,
|
||||
})
|
||||
|
||||
snapshotID, err := sm.CreateSnapshotWithName(
|
||||
ctx, cfg.Hostname, name, "test-version", "test-git")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
_, err = scanner.Scan(ctx, dataDir, snapshotID)
|
||||
if err != nil {
|
||||
return snapshotID, err
|
||||
}
|
||||
|
||||
err = sm.CompleteSnapshot(ctx, snapshotID)
|
||||
if err != nil {
|
||||
return snapshotID, err
|
||||
}
|
||||
|
||||
err = sm.ExportSnapshotMetadata(ctx, dbPath, snapshotID)
|
||||
if err != nil {
|
||||
return snapshotID, err
|
||||
}
|
||||
|
||||
return snapshotID, nil
|
||||
}
|
||||
Reference in New Issue
Block a user