Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1ad3faa8ee |
+3
-24
@@ -5,16 +5,7 @@
|
||||
FROM golangci/golangci-lint:v2.14.0@sha256:ad862ba6b3798cbe0fd9fd7408d498fd74fbd2623a92406b2fd3898faf0bf98f AS lint
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
# Without the time limit a stalled connection to the module proxy holds
|
||||
# this step, and every build on the host sharing it, until the kernel
|
||||
# drops the connection, which can take an hour. The limit is well above
|
||||
# the three minutes a working download can take on a busy build host.
|
||||
RUN for attempt in 1 2 3; do \
|
||||
timeout 300 go mod download && exit 0; \
|
||||
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
|
||||
done; \
|
||||
echo "module download failed: giving up after 3 attempts" >&2; \
|
||||
exit 1
|
||||
RUN go mod download
|
||||
COPY . .
|
||||
# `golangci-lint run` silently ignores an unknown top-level key in
|
||||
# .golangci.yml, such as a misspelt `linters:`; `config verify` fails on it.
|
||||
@@ -27,13 +18,7 @@ RUN golangci-lint run --config .golangci.yml ./...
|
||||
FROM golang:1.26.1@sha256:cd78d88e00afadbedd272f977d375a6247455f3a4b1178f8ae8bbcb201743a8a AS test
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
# Limited and retried as in the lint phase above.
|
||||
RUN for attempt in 1 2 3; do \
|
||||
timeout 300 go mod download && exit 0; \
|
||||
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
|
||||
done; \
|
||||
echo "module download failed: giving up after 3 attempts" >&2; \
|
||||
exit 1
|
||||
RUN go mod download
|
||||
COPY . .
|
||||
RUN go test -timeout 90s -race -cover ./... || \
|
||||
{ echo "--- Rerunning with -v for details ---"; \
|
||||
@@ -51,13 +36,7 @@ RUN apk add --no-cache git
|
||||
RUN git config --system --add safe.directory /src
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
# Limited and retried as in the lint phase above.
|
||||
RUN for attempt in 1 2 3; do \
|
||||
timeout 300 go mod download && exit 0; \
|
||||
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
|
||||
done; \
|
||||
echo "module download failed: giving up after 3 attempts" >&2; \
|
||||
exit 1
|
||||
RUN go mod download
|
||||
COPY . .
|
||||
|
||||
# The VERSION build arg when one is given, otherwise
|
||||
|
||||
@@ -294,8 +294,7 @@ needed.
|
||||
of the local index and the backup destination store — with timestamps
|
||||
and three sizes per snapshot (compressed remote size; total
|
||||
uncompressed chunk size; size of the chunks that snapshot stored that
|
||||
were not stored before). Both chunk sizes are recorded when the
|
||||
snapshot's backup runs.
|
||||
were not stored before, recorded when its backup ran).
|
||||
|
||||
Listing the destination store does **not** require the age secret key,
|
||||
so it works in vaultik's intended configuration, where the backed-up
|
||||
@@ -339,9 +338,7 @@ local index alone, and still exits zero.
|
||||
|
||||
**`snapshot purge`**: Remove old snapshots based on criteria. Retention is
|
||||
per-snapshot-name (`--keep-latest` keeps the latest of each name, not the
|
||||
latest globally). If a snapshot cannot be removed from the local index, the
|
||||
command stops there, leaves that snapshot's metadata on the destination store
|
||||
and exits non-zero; `snapshot create --prune` fails the same way.
|
||||
latest globally).
|
||||
* `--keep-latest`: Keep only the most recent snapshot of each name
|
||||
* `--older-than <duration>`: Remove snapshots older than duration (e.g. `30d`,
|
||||
`4w`, `6mo`, `1y`; `m` is minutes, `mo` is months)
|
||||
@@ -378,9 +375,7 @@ on the destination in one go, use `vaultik remote nuke --force`.
|
||||
* Optional path arguments to restore specific files/directories (default: all)
|
||||
* Preserves file permissions, timestamps, ownership (ownership requires root),
|
||||
symlinks, and empty directories
|
||||
* `--verify`: After restoring, verify every restored file's chunk hashes
|
||||
match. A file skipped under `--skip-errors` is reported as failed to
|
||||
restore and is not verified.
|
||||
* `--verify`: After restoring, verify every file's chunk hashes match
|
||||
|
||||
**`prune`**: Tidy up everything that isn't needed. Runs three passes:
|
||||
(1) reconcile the local index against the destination store — any
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
package main_test
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestDockerfileBoundsEveryModuleDownload fails unless every
|
||||
// `go mod download` in the Dockerfile runs under a time limit, is tried
|
||||
// a fixed number of times, and then fails the build with a message that
|
||||
// names the module download (issue #351). Like the version tests, this
|
||||
// parses the Dockerfile; that a stalled download really is cut off and
|
||||
// retried was checked by hand.
|
||||
func TestDockerfileBoundsEveryModuleDownload(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var downloads []string
|
||||
|
||||
for _, instruction := range instructions(t, productDockerfile) {
|
||||
if strings.Contains(instruction, "go mod download") {
|
||||
downloads = append(downloads, instruction)
|
||||
}
|
||||
}
|
||||
|
||||
require.NotEmpty(t, downloads, "%s must download modules",
|
||||
productDockerfile)
|
||||
|
||||
for _, download := range downloads {
|
||||
assert.Contains(t, download, "timeout 300 go mod download",
|
||||
"%s: a module download must have a time limit: %s",
|
||||
productDockerfile, download)
|
||||
assert.Contains(t, download, "for attempt in 1 2 3",
|
||||
"%s: a module download must be retried: %s",
|
||||
productDockerfile, download)
|
||||
assert.Contains(t, download, "module download failed",
|
||||
"%s: a module download that never succeeds must say so: %s",
|
||||
productDockerfile, download)
|
||||
assert.True(t, strings.HasSuffix(download, "exit 1"),
|
||||
"%s: a module download that never succeeds must fail the"+
|
||||
" build: %s", productDockerfile, download)
|
||||
}
|
||||
}
|
||||
+2
-3
@@ -90,7 +90,7 @@ Stores information about packed, compressed, and encrypted blob files.
|
||||
- `created_ts` (INTEGER NOT NULL) - Creation timestamp
|
||||
- `finished_ts` (INTEGER) - Finalization timestamp (NULL if in progress)
|
||||
- `uncompressed_size` (INTEGER NOT NULL DEFAULT 0) - Total size of chunks before compression
|
||||
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption. Once uploaded, the size of the object in storage: when the blob was already stored, under another compression level or other recipients, this can differ from the size of the local copy
|
||||
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption
|
||||
- `uploaded_ts` (INTEGER) - Upload completion timestamp (NULL if not uploaded)
|
||||
|
||||
**Purpose:** Tracks blob lifecycle from creation through upload. The UUID primary key allows immediate association of chunks with blobs.
|
||||
@@ -122,7 +122,6 @@ Tracks backup snapshots.
|
||||
- `new_chunk_size` (INTEGER) - Total size of those chunks before compression; `snapshot list` shows it as NEW CHUNK SIZE
|
||||
- `blob_count` (INTEGER) - Number of blobs this snapshot created
|
||||
- `total_size` (INTEGER) - Total size of all files
|
||||
- `uncompressed_chunk_size` (INTEGER) - Total size before compression of the distinct chunks the snapshot's files reference, recorded when its backup ran; `snapshot list` shows it as UNCOMPRESSED SIZE
|
||||
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs
|
||||
- `blob_uncompressed_size` (INTEGER) - Total uncompressed size of all referenced blobs
|
||||
- `compression_ratio` (REAL) - Compression ratio achieved
|
||||
@@ -185,7 +184,7 @@ Tracks blob upload metrics.
|
||||
- `UPDATE blobs SET blob_hash = ?, finished_ts = ?` - Finalize blob after packing
|
||||
|
||||
4. **Upload**
|
||||
- `UPDATE blobs SET uploaded_ts = ?, compressed_size = ?` - Mark blob as uploaded, with the size of the object in storage (the upload is skipped when the blob is already stored)
|
||||
- `UPDATE blobs SET uploaded_ts = ?` - Mark blob as uploaded
|
||||
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
||||
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
||||
|
||||
|
||||
@@ -1,90 +0,0 @@
|
||||
package cli //nolint:testpackage // shares runEntry and the hermetic config helpers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/types"
|
||||
)
|
||||
|
||||
// The `snapshot purge` command word and flags used below.
|
||||
const (
|
||||
cmdPurge = "purge"
|
||||
flagKeepLatest = "--keep-latest"
|
||||
flagForce = "--force"
|
||||
)
|
||||
|
||||
// TestEntrySnapshotPurgeKeepsMetadataWhenLocalIndexDeleteFails makes
|
||||
// removing a snapshot from the local index fail and checks that
|
||||
// `snapshot purge` exits non-zero and leaves that snapshot's metadata on
|
||||
// the destination store. It used to log the error, delete the metadata
|
||||
// anyway and exit 0.
|
||||
//
|
||||
//nolint:paralleltest // replaces os.Args, os.Stdout, os.Stderr and the xdg globals
|
||||
func TestEntrySnapshotPurgeKeepsMetadataWhenLocalIndexDeleteFails(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
configPath := writeHermeticPruneConfig(t, false)
|
||||
|
||||
// writeHermeticPruneConfig puts the index and the store directory
|
||||
// beside the config file.
|
||||
dir := filepath.Dir(configPath)
|
||||
|
||||
// Two snapshots of the same name: --keep-latest purges the older.
|
||||
olderStart := time.Date(2026, 5, 1, 9, 0, 0, 0, time.UTC)
|
||||
newerStart := olderStart.Add(time.Hour)
|
||||
olderID := "test-host_test_" + olderStart.Format(time.RFC3339)
|
||||
newerID := "test-host_test_" + newerStart.Format(time.RFC3339)
|
||||
|
||||
db, err := database.New(ctx, filepath.Join(dir, "index.sqlite"))
|
||||
require.NoError(t, err)
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
|
||||
manifestPaths := make(map[string]string)
|
||||
|
||||
for id, startedAt := range map[string]time.Time{
|
||||
olderID: olderStart,
|
||||
newerID: newerStart,
|
||||
} {
|
||||
completedAt := startedAt.Add(time.Minute)
|
||||
snap := &database.Snapshot{
|
||||
ID: types.SnapshotID(id),
|
||||
Hostname: "test-host",
|
||||
VaultikVersion: "test",
|
||||
StartedAt: startedAt,
|
||||
CompletedAt: &completedAt,
|
||||
}
|
||||
|
||||
err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||
return repos.Snapshots.Create(ctx, tx, snap)
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
manifestPaths[id] = filepath.Join(dir, "store", "metadata",
|
||||
snapshot.RemoteSnapshotKey(id), "manifest.json.zst")
|
||||
require.NoError(t, os.MkdirAll(filepath.Dir(manifestPaths[id]), 0o750))
|
||||
require.NoError(t, os.WriteFile(manifestPaths[id], []byte("stub"), 0o600))
|
||||
}
|
||||
|
||||
_, err = db.Conn().ExecContext(ctx, `
|
||||
CREATE TRIGGER refuse_snapshot_delete BEFORE DELETE ON snapshots
|
||||
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.Close())
|
||||
|
||||
code, _, stderr := runEntry(t,
|
||||
flagConfig, configPath, cmdSnapshot, cmdPurge, flagKeepLatest, flagForce)
|
||||
|
||||
assert.Equal(t, 1, code)
|
||||
assert.Contains(t, stderr, "simulated index error")
|
||||
assert.FileExists(t, manifestPaths[olderID],
|
||||
"a snapshot still in the local index keeps its metadata")
|
||||
}
|
||||
@@ -58,7 +58,12 @@ func (r *BlobChunkRepository) GetByBlobID(
|
||||
return nil, fmt.Errorf("querying blob chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var blobChunks []*BlobChunk
|
||||
|
||||
|
||||
+12
-10
@@ -83,7 +83,12 @@ func (r *BlobRepository) GetAll(ctx context.Context) (map[string]*Blob, error) {
|
||||
return nil, fmt.Errorf("querying blobs: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
out := make(map[string]*Blob)
|
||||
|
||||
@@ -154,16 +159,13 @@ func (r *BlobRepository) UpdateFinished(
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateUploaded marks a blob as uploaded and records the size of the
|
||||
// stored object as its compressed size. That size can differ from the
|
||||
// local copy's when an identical blob was already stored with another
|
||||
// compression level or other recipients.
|
||||
// UpdateUploaded marks a blob as uploaded
|
||||
func (r *BlobRepository) UpdateUploaded(
|
||||
ctx context.Context, tx *sql.Tx, id string, storedSize int64,
|
||||
ctx context.Context, tx *sql.Tx, id string,
|
||||
) error {
|
||||
query := `
|
||||
UPDATE blobs
|
||||
SET uploaded_ts = ?, compressed_size = ?
|
||||
UPDATE blobs
|
||||
SET uploaded_ts = ?
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
@@ -171,9 +173,9 @@ func (r *BlobRepository) UpdateUploaded(
|
||||
|
||||
var err error
|
||||
if tx != nil {
|
||||
_, err = tx.ExecContext(ctx, query, now, storedSize, id)
|
||||
_, err = tx.ExecContext(ctx, query, now, id)
|
||||
} else {
|
||||
_, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
|
||||
_, err = r.db.ExecWithLog(ctx, query, now, id)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
|
||||
@@ -122,8 +122,8 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
||||
t.Errorf("expected compressed size 500, got %d", updated.CompressedSize)
|
||||
}
|
||||
|
||||
// Test UpdateUploaded, with a stored object larger than the local copy
|
||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
|
||||
// Test UpdateUploaded
|
||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to update blob as uploaded: %v", err)
|
||||
}
|
||||
@@ -141,10 +141,6 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
||||
if uploaded.UploadedTS.Before(now.Add(-1 * time.Second)) {
|
||||
t.Error("uploaded timestamp should be around test time")
|
||||
}
|
||||
|
||||
if uploaded.CompressedSize != 600 {
|
||||
t.Errorf("expected the stored size 600, got %d", uploaded.CompressedSize)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlobRepositoryDuplicate(t *testing.T) {
|
||||
|
||||
@@ -61,7 +61,12 @@ func (r *ChunkFileRepository) GetByChunkHash(
|
||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanChunkFiles(rows)
|
||||
}
|
||||
@@ -82,7 +87,12 @@ func (r *ChunkFileRepository) GetByFilePath(
|
||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanChunkFiles(rows)
|
||||
}
|
||||
@@ -102,7 +112,12 @@ func (r *ChunkFileRepository) GetByFileID(
|
||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanChunkFiles(rows)
|
||||
}
|
||||
|
||||
@@ -107,7 +107,12 @@ func (r *ChunkRepository) GetByHashes(
|
||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var chunks []*Chunk
|
||||
|
||||
@@ -147,7 +152,12 @@ func (r *ChunkRepository) ListUnpacked(
|
||||
return nil, fmt.Errorf("querying unpacked chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var chunks []*Chunk
|
||||
|
||||
|
||||
@@ -38,7 +38,12 @@ func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, err
|
||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var chunks []*Chunk
|
||||
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
)
|
||||
|
||||
// Fatalf prints an error message to stderr and exits with status 1
|
||||
func Fatalf(format string, args ...any) {
|
||||
fmt.Fprintf(os.Stderr, "FATAL: "+format+"\n", args...)
|
||||
os.Exit(1)
|
||||
}
|
||||
@@ -62,7 +62,12 @@ func (r *FileChunkRepository) GetByPath(
|
||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanFileChunks(rows)
|
||||
}
|
||||
@@ -83,7 +88,12 @@ func (r *FileChunkRepository) GetByFileID(
|
||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanFileChunks(rows)
|
||||
}
|
||||
@@ -107,7 +117,12 @@ func (r *FileChunkRepository) GetByPathTx(
|
||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
fileChunks, err := r.scanFileChunks(rows)
|
||||
LogSQL("GetByPathTx", "Complete", path, "count", len(fileChunks))
|
||||
|
||||
@@ -175,7 +175,12 @@ func (r *FileRepository) ListModifiedSince(
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var files []*File
|
||||
|
||||
@@ -251,7 +256,12 @@ func (r *FileRepository) ListUnderPath(
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var files []*File
|
||||
|
||||
@@ -294,7 +304,12 @@ func (r *FileRepository) ListIDsWithChunksNotInUploadedBlobs(
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var ids []types.FileID
|
||||
|
||||
@@ -325,7 +340,12 @@ func (r *FileRepository) ListAll(ctx context.Context) ([]*File, error) {
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var files []*File
|
||||
|
||||
|
||||
@@ -106,24 +106,22 @@ func (r *Repositories) WithReadTx(ctx context.Context, fn TxFunc) error {
|
||||
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
// The panic is the failure to report, so a rollback error
|
||||
// is dropped.
|
||||
_ = tx.Rollback()
|
||||
rollbackErr := tx.Rollback()
|
||||
if rollbackErr != nil {
|
||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
||||
}
|
||||
|
||||
panic(p)
|
||||
} else if err != nil {
|
||||
rollbackErr := tx.Rollback()
|
||||
if rollbackErr != nil {
|
||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
err = fn(ctx, tx)
|
||||
if err != nil {
|
||||
// Once ctx is cancelled, database/sql rolls the transaction
|
||||
// back itself and Rollback returns sql.ErrTxDone.
|
||||
rollbackErr := tx.Rollback()
|
||||
if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
|
||||
return fmt.Errorf("%w; rolling back transaction: %w",
|
||||
err, rollbackErr)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
@@ -411,41 +411,3 @@ func TestRepositoriesReadTransaction(t *testing.T) {
|
||||
t.Error("expected to retrieve file in read transaction")
|
||||
}
|
||||
}
|
||||
|
||||
// When the context is cancelled during a read transaction's body,
|
||||
// database/sql rolls the transaction back itself, so WithReadTx's own
|
||||
// rollback finds it already done. WithReadTx must return the body's error
|
||||
// alone, without that rollback.
|
||||
func TestRepositoriesReadTransactionContextCancelled(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db, cleanup := setupTestDB(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
|
||||
err := repos.WithReadTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||
cancel()
|
||||
|
||||
// database/sql rolls back from a goroutine of its own; wait
|
||||
// until it has, so the test does not depend on that race.
|
||||
for {
|
||||
_, err := tx.ExecContext(context.WithoutCancel(ctx), "SELECT 1")
|
||||
if errors.Is(err, sql.ErrTxDone) {
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
})
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("expected context.Canceled, got: %v", err)
|
||||
}
|
||||
|
||||
if errors.Is(err, sql.ErrTxDone) {
|
||||
t.Fatalf("the already-done rollback was reported: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -89,7 +89,6 @@ CREATE TABLE IF NOT EXISTS snapshots (
|
||||
new_chunk_size INTEGER NOT NULL DEFAULT 0,
|
||||
blob_count INTEGER NOT NULL DEFAULT 0,
|
||||
total_size INTEGER NOT NULL DEFAULT 0,
|
||||
uncompressed_chunk_size INTEGER NOT NULL DEFAULT 0,
|
||||
blob_size INTEGER NOT NULL DEFAULT 0,
|
||||
blob_uncompressed_size INTEGER NOT NULL DEFAULT 0,
|
||||
compression_ratio REAL NOT NULL DEFAULT 1.0,
|
||||
|
||||
@@ -120,11 +120,7 @@ func (r *SnapshotRepository) UpdateCounts(
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateExtendedStats updates extended statistics for a snapshot. It also
|
||||
// records the total plaintext size of the distinct chunks the snapshot's
|
||||
// files reference. Only the snapshot's own backup can take that total: a
|
||||
// file's row is shared by every snapshot that lists it, and a later
|
||||
// backup of a changed file rewrites the row with the new chunks.
|
||||
// UpdateExtendedStats updates extended statistics for a snapshot
|
||||
func (r *SnapshotRepository) UpdateExtendedStats(
|
||||
ctx context.Context,
|
||||
tx *sql.Tx,
|
||||
@@ -149,17 +145,7 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
||||
compression_level = ?,
|
||||
upload_bytes = ?,
|
||||
upload_duration_ms = ?,
|
||||
new_chunk_size = ?,
|
||||
uncompressed_chunk_size = (
|
||||
SELECT COALESCE(SUM(c.size), 0)
|
||||
FROM (
|
||||
SELECT DISTINCT fc.chunk_hash
|
||||
FROM snapshot_files sf
|
||||
JOIN file_chunks fc ON sf.file_id = fc.file_id
|
||||
WHERE sf.snapshot_id = snapshots.id
|
||||
) sc
|
||||
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
|
||||
)
|
||||
new_chunk_size = ?
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
@@ -255,7 +241,12 @@ func (r *SnapshotRepository) GetOnlySnapshot(ctx context.Context) (*Snapshot, er
|
||||
return nil, fmt.Errorf("querying snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
snapshots, err := r.scanSnapshotRows(rows)
|
||||
if err != nil {
|
||||
@@ -290,7 +281,12 @@ func (r *SnapshotRepository) ListRecent(
|
||||
return nil, fmt.Errorf("querying snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanSnapshotRows(rows)
|
||||
}
|
||||
@@ -505,7 +501,12 @@ func (r *SnapshotRepository) GetBlobHashes(
|
||||
return nil, fmt.Errorf("querying blob hashes: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var blobs []string
|
||||
|
||||
@@ -569,22 +570,30 @@ func (r *SnapshotRepository) GetSnapshotBlobSizes(
|
||||
return compressed, uncompressed, nil
|
||||
}
|
||||
|
||||
// GetSnapshotUncompressedChunkSize returns the total plaintext size of the
|
||||
// distinct chunks the snapshot's files reference, as the snapshot's backup
|
||||
// recorded it (see UpdateExtendedStats).
|
||||
// GetSnapshotUncompressedChunkSize returns the sum of plaintext sizes of all unique
|
||||
// chunks referenced by a snapshot (via snapshot_files → file_chunks → chunks).
|
||||
func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
||||
ctx context.Context, snapshotID string,
|
||||
) (int64, error) {
|
||||
query := `SELECT uncompressed_chunk_size FROM snapshots WHERE id = ?`
|
||||
query := `
|
||||
SELECT COALESCE(SUM(c.size), 0)
|
||||
FROM (
|
||||
SELECT DISTINCT fc.chunk_hash
|
||||
FROM snapshot_files sf
|
||||
JOIN file_chunks fc ON sf.file_id = fc.file_id
|
||||
WHERE sf.snapshot_id = ?
|
||||
) sc
|
||||
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
|
||||
`
|
||||
|
||||
var uncompressedChunkSize int64
|
||||
var totalSize int64
|
||||
|
||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&uncompressedChunkSize)
|
||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&totalSize)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("querying uncompressed chunk size: %w", err)
|
||||
}
|
||||
|
||||
return uncompressedChunkSize, nil
|
||||
return totalSize, nil
|
||||
}
|
||||
|
||||
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
||||
@@ -625,7 +634,12 @@ func (r *SnapshotRepository) GetIncompleteSnapshots(
|
||||
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
return r.scanSnapshotRows(rows)
|
||||
}
|
||||
@@ -648,7 +662,12 @@ func (r *SnapshotRepository) GetIncompleteByHostname(
|
||||
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Same column set as every other multi-row snapshot query, so the
|
||||
// shared scanner applies — including its timestamp normalization.
|
||||
|
||||
@@ -1,62 +0,0 @@
|
||||
//nolint:testpackage // exercises the unexported cleanupBlobTempFile
|
||||
package snapshot
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"sneak.berlin/go/vaultik/internal/blob"
|
||||
)
|
||||
|
||||
// A blob temp file that cannot be closed or removed must make the cleanup
|
||||
// return an error, not end the process.
|
||||
func TestCleanupBlobTempFileReturnsErrors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
s := &Scanner{fs: fs}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
// prepare runs before the cleanup and makes its close or remove
|
||||
// fail.
|
||||
prepare func(tempFile afero.File) error
|
||||
want error
|
||||
}{
|
||||
{
|
||||
name: "close fails",
|
||||
prepare: func(tempFile afero.File) error { return tempFile.Close() },
|
||||
want: os.ErrClosed,
|
||||
},
|
||||
{
|
||||
name: "remove fails",
|
||||
prepare: func(tempFile afero.File) error {
|
||||
return fs.Remove(tempFile.Name())
|
||||
},
|
||||
want: os.ErrNotExist,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tempFile, err := afero.TempFile(fs, t.TempDir(), "blob-*.tmp")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = tt.prepare(tempFile)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = s.cleanupBlobTempFile(&blob.WithReader{TempFile: tempFile})
|
||||
if !errors.Is(err, tt.want) {
|
||||
t.Fatalf("expected an error wrapping %v, got: %v", tt.want, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1517,8 +1517,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
|
||||
// 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, b.Compressed)
|
||||
err := s.repos.Blobs.UpdateUploaded(ctx, tx, b.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marking blob uploaded: %w", err)
|
||||
}
|
||||
@@ -1554,28 +1553,23 @@ func (s *Scanner) handleBlobReady(
|
||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
||||
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
||||
|
||||
storedSize, blobExists, err := s.uploadBlobIfNeeded(
|
||||
blobExists, err := s.uploadBlobIfNeeded(
|
||||
ctx, blobPath, blobWithReader, startTime, result)
|
||||
if err != nil {
|
||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
|
||||
return errors.Join(
|
||||
fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr)
|
||||
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
|
||||
}
|
||||
|
||||
err = s.recordBlobMetadata(
|
||||
ctx, finishedBlob, storedSize, blobExists, startTime)
|
||||
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime)
|
||||
if err != nil {
|
||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
|
||||
return errors.Join(err, cleanupErr)
|
||||
}
|
||||
|
||||
err = s.cleanupBlobTempFile(blobWithReader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
|
||||
// Chunks from this blob are now committed to DB - remove from pending set
|
||||
s.removePendingChunkHashes(blobWithReader.InsertedChunkHashes)
|
||||
|
||||
@@ -1589,22 +1583,20 @@ func (s *Scanner) handleBlobReady(
|
||||
}
|
||||
|
||||
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
||||
// exist. It returns the size of the object in storage, which for an
|
||||
// existing object can differ from the local copy's, and whether it
|
||||
// already existed.
|
||||
// exist, returns whether it existed
|
||||
func (s *Scanner) uploadBlobIfNeeded(
|
||||
ctx context.Context,
|
||||
blobPath string,
|
||||
blobWithReader *blob.WithReader,
|
||||
startTime time.Time,
|
||||
result *ScanResult,
|
||||
) (int64, bool, error) {
|
||||
) (bool, error) {
|
||||
finishedBlob := blobWithReader.FinishedBlob
|
||||
|
||||
// Check if blob already exists (deduplication after restart)
|
||||
destination := s.storage.Info().Location
|
||||
|
||||
stored, err := s.storage.Stat(ctx, blobPath)
|
||||
_, err := s.storage.Stat(ctx, blobPath)
|
||||
if err == nil {
|
||||
log.Info("Blob already exists in storage, skipping upload",
|
||||
"hash", finishedBlob.Hash,
|
||||
@@ -1613,7 +1605,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
||||
s.ui.Path(destination))
|
||||
|
||||
return stored.Size, true, nil
|
||||
return true, nil
|
||||
}
|
||||
|
||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||
@@ -1627,7 +1619,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
if err != nil {
|
||||
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
||||
|
||||
return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||
return false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||
}
|
||||
|
||||
uploadDuration := time.Since(startTime)
|
||||
@@ -1658,7 +1650,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
||||
}
|
||||
|
||||
return finishedBlob.Compressed, false, nil
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
||||
@@ -1728,7 +1720,6 @@ func (s *Scanner) makeUploadProgressCallback(
|
||||
func (s *Scanner) recordBlobMetadata(
|
||||
ctx context.Context,
|
||||
finishedBlob *blob.FinishedBlob,
|
||||
storedSize int64,
|
||||
blobExists bool,
|
||||
startTime time.Time,
|
||||
) error {
|
||||
@@ -1740,8 +1731,7 @@ func (s *Scanner) recordBlobMetadata(
|
||||
uploadDuration := time.Since(startTime)
|
||||
|
||||
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
|
||||
err := s.repos.Blobs.UpdateUploaded(
|
||||
txCtx, tx, finishedBlob.ID, storedSize)
|
||||
err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
||||
}
|
||||
@@ -1772,22 +1762,20 @@ func (s *Scanner) recordBlobMetadata(
|
||||
}
|
||||
|
||||
// cleanupBlobTempFile closes and removes the blob's temporary file
|
||||
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) error {
|
||||
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) {
|
||||
if blobWithReader.TempFile != nil {
|
||||
tempName := blobWithReader.TempFile.Name()
|
||||
|
||||
err := blobWithReader.TempFile.Close()
|
||||
if err != nil {
|
||||
return fmt.Errorf("closing blob temp file: %w", err)
|
||||
log.Fatal("Failed to close temp file", "file", tempName, "error", err)
|
||||
}
|
||||
|
||||
err = s.fs.Remove(tempName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("removing blob temp file: %w", err)
|
||||
log.Fatal("Failed to remove temp file", "file", tempName, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// streamingChunkInfo tracks chunk metadata collected during streaming
|
||||
|
||||
+32
-6
@@ -29,7 +29,10 @@ var (
|
||||
ErrURLCredentials = errors.New(
|
||||
"storage URL must not carry credentials; " +
|
||||
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||
ErrRcloneRemoteParams = errors.New(
|
||||
"rclone URL must not carry parameters in the remote name; " +
|
||||
"set them on the remote with rclone config instead")
|
||||
)
|
||||
|
||||
// URL represents a parsed storage URL.
|
||||
@@ -90,8 +93,9 @@ func parseFileURL(rawPath string) (*URL, error) {
|
||||
}
|
||||
|
||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||
// the userinfo and any query parameter other than endpoint, region and
|
||||
// ssl, so a credential-bearing URL is never stored or echoed.
|
||||
// the userinfo or in the endpoint parameter's userinfo, and any query
|
||||
// parameter other than endpoint, region and ssl, so a credential-bearing
|
||||
// URL is never stored or echoed.
|
||||
func parseS3URL(rawURL string) (*URL, error) {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
@@ -114,6 +118,14 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// A host name never contains "@", so one in an endpoint, with or
|
||||
// without a scheme, can only introduce userinfo.
|
||||
for _, endpoint := range query["endpoint"] {
|
||||
if strings.Contains(endpoint, "@") {
|
||||
return nil, ErrURLCredentials
|
||||
}
|
||||
}
|
||||
|
||||
return &URL{
|
||||
Scheme: schemeS3,
|
||||
Bucket: bucket,
|
||||
@@ -125,9 +137,23 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
}
|
||||
|
||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||
// query parameters, so credentials in the userinfo and any parameter at
|
||||
// all are rejected rather than silently ignored.
|
||||
// query parameters, so credentials in the userinfo, parameters in the
|
||||
// remote name and any query parameter at all are rejected rather than
|
||||
// silently ignored.
|
||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
// The remote name runs up to the first "/", "?" or "#". It never
|
||||
// contains a comma; one starts rclone's remote,param=value form,
|
||||
// whose values can be credentials. This is checked before url.Parse,
|
||||
// whose error for such a name can print part of it.
|
||||
remote := strings.TrimPrefix(rawURL, "rclone://")
|
||||
if end := strings.IndexAny(remote, "/?#"); end >= 0 {
|
||||
remote = remote[:end]
|
||||
}
|
||||
|
||||
if strings.Contains(remote, ",") {
|
||||
return nil, ErrRcloneRemoteParams
|
||||
}
|
||||
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return nil, wrapParseError(err)
|
||||
@@ -137,7 +163,7 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
return nil, ErrURLCredentials
|
||||
}
|
||||
|
||||
remote := u.Host
|
||||
remote = u.Host
|
||||
if remote == "" {
|
||||
return nil, ErrMissingRemote
|
||||
}
|
||||
|
||||
@@ -150,6 +150,19 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret, "mybucket"},
|
||||
},
|
||||
{
|
||||
name: "s3 userinfo in endpoint",
|
||||
raw: "s3://mybucket?endpoint=https://" +
|
||||
key + ":" + secret + "@minio.example.com",
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret},
|
||||
},
|
||||
{
|
||||
name: "s3 userinfo in endpoint without scheme",
|
||||
raw: "s3://mybucket?endpoint=" + key + ":" + secret + "@minio.example.com",
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret},
|
||||
},
|
||||
{
|
||||
name: "s3 unknown query param",
|
||||
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
||||
@@ -203,6 +216,54 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestParseStorageURLRejectsRcloneRemoteParams checks that an rclone
|
||||
// remote name carrying parameters (rclone's "remote,param=value" form,
|
||||
// which can hold a credential) is rejected without echoing the value,
|
||||
// including the URLs that url.Parse itself refuses.
|
||||
func TestParseStorageURLRejectsRcloneRemoteParams(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
raw string
|
||||
secrets []string // must not appear in the error message
|
||||
}{
|
||||
{
|
||||
name: "configured remote",
|
||||
raw: "rclone://gdrive,client_secret=topsecret/backups",
|
||||
secrets: []string{"topsecret"},
|
||||
},
|
||||
{
|
||||
name: "remote defined inline",
|
||||
raw: "rclone://:s3,access_key_id=KEY,secret_access_key=SECRET/bucket",
|
||||
secrets: []string{"KEY", "SECRET"},
|
||||
},
|
||||
{
|
||||
name: "colon in a parameter value",
|
||||
raw: "rclone://gdrive,client_secret=ab:cd/backups",
|
||||
secrets: []string{"cd"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
_, err := storage.ParseStorageURL(tc.raw)
|
||||
if !errors.Is(err, storage.ErrRcloneRemoteParams) {
|
||||
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
|
||||
tc.raw, err, storage.ErrRcloneRemoteParams)
|
||||
}
|
||||
|
||||
for _, secret := range tc.secrets {
|
||||
if strings.Contains(err.Error(), secret) {
|
||||
t.Errorf("error message leaked %q: %v", secret, err.Error())
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
||||
// itself fails, the wrapped error carries only the inner cause, not the
|
||||
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -11,9 +9,7 @@ import (
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"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"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
@@ -81,174 +77,6 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
|
||||
"deep verify should fail on a corrupted blob")
|
||||
}
|
||||
|
||||
// TestDeepVerifyJSONCountsBlobs checks the verified and missing counts
|
||||
// that deep verify reports under --json: on a healthy snapshot, with a
|
||||
// damaged blob, and with missing blobs. Deep verify stops at the first
|
||||
// damaged blob, so the blobs read before it are the ones verified; when
|
||||
// blobs are missing it reads no blob contents, but counts every missing
|
||||
// blob.
|
||||
func TestDeepVerifyJSONCountsBlobs(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||
|
||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||
|
||||
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "ok", healthy.Status)
|
||||
require.Len(t, keys, healthy.BlobCount)
|
||||
require.Equal(t, healthy.BlobCount, healthy.Verified)
|
||||
require.Zero(t, healthy.Missing)
|
||||
|
||||
// Damage the last blob read, keeping its length, so every blob before
|
||||
// it is read and passes first.
|
||||
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
|
||||
blob, err := afero.ReadFile(fs, lastPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
blob[len(blob)/2] ^= 0xff
|
||||
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
|
||||
|
||||
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Equal(t, healthy.BlobCount-1, got.Verified)
|
||||
require.Zero(t, got.Missing)
|
||||
|
||||
// Delete the first blob and the damaged last one: both must be
|
||||
// counted as missing, not only the first one found.
|
||||
var deletedSize int64
|
||||
|
||||
for _, key := range []string{keys[0], keys[len(keys)-1]} {
|
||||
info, statErr := storer.Stat(ctx, key)
|
||||
require.NoError(t, statErr)
|
||||
|
||||
deletedSize += info.Size
|
||||
|
||||
require.NoError(t, storer.Delete(ctx, key))
|
||||
}
|
||||
|
||||
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Equal(t, 2, got.Missing)
|
||||
require.Equal(t, deletedSize, got.MissingSize)
|
||||
require.Zero(t, got.Verified)
|
||||
}
|
||||
|
||||
// TestDeepVerifyJSONCountsWrongSizeBlob checks that under --deep --json a
|
||||
// blob of the wrong size is counted in mismatched and fails the run, and
|
||||
// that the check goes on past it to count a missing blob after it.
|
||||
func TestDeepVerifyJSONCountsWrongSizeBlob(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||
|
||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||
|
||||
// Make the first blob checked longer, and delete the last one.
|
||||
firstPath := filepath.Join(storeDir, keys[0])
|
||||
blob, err := afero.ReadFile(fs, firstPath)
|
||||
require.NoError(t, err)
|
||||
|
||||
blob = append(blob, []byte("extra")...)
|
||||
require.NoError(t, afero.WriteFile(fs, firstPath, blob, 0o600))
|
||||
require.NoError(t, storer.Delete(ctx, keys[len(keys)-1]))
|
||||
|
||||
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Equal(t, 1, got.Mismatched)
|
||||
require.Equal(t, 1, got.Missing)
|
||||
require.Zero(t, got.Verified)
|
||||
}
|
||||
|
||||
// TestDeepVerifyStopsOnBlobThatCannotBeChecked checks that a blob that is
|
||||
// present in storage but cannot be checked is not counted as missing:
|
||||
// deep verify stops and fails with the storage error.
|
||||
func TestDeepVerifyStopsOnBlobThatCannotBeChecked(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||
|
||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||
failing := statFailingStorer{Storer: storer, failKeys: []string{keys[0]}}
|
||||
|
||||
got, err := deepVerifyJSON(ctx, t, cfg, failing, snapshotID)
|
||||
require.Error(t, err)
|
||||
require.Equal(t, "failed", got.Status)
|
||||
require.Zero(t, got.Missing)
|
||||
require.Contains(t, got.ErrorMessage, errStatDenied.Error())
|
||||
}
|
||||
|
||||
// backUpSeveralBlobs backs up random data, which does not compress or
|
||||
// deduplicate, so it fills at least three blobs in a store on disk at
|
||||
// storeDir. It returns the blobs' storage keys sorted, which is hash
|
||||
// order, the order deep verify reads them in.
|
||||
func backUpSeveralBlobs(
|
||||
ctx context.Context, t *testing.T, storeDir string,
|
||||
) (*config.Config, *storage.FileStorer, string, []string) {
|
||||
t.Helper()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
tempDir := t.TempDir()
|
||||
dataDir := filepath.Join(tempDir, "source")
|
||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||
|
||||
chunkSize := int64(32 * 1024)
|
||||
maxBlobSize := int64(128 * 1024)
|
||||
|
||||
data := make([]byte, maxBlobSize*4)
|
||||
_, err := rand.Read(data)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||
require.NoError(t, afero.WriteFile(fs,
|
||||
filepath.Join(dataDir, "data.bin"), data, 0o644))
|
||||
|
||||
cfg, storer, snapshotID := runFileStorageBackup(
|
||||
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
||||
|
||||
keys := storedBlobKeys(t, storeDir)
|
||||
require.GreaterOrEqual(t, len(keys), 3)
|
||||
|
||||
return cfg, storer, snapshotID, keys
|
||||
}
|
||||
|
||||
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
|
||||
// the decoded result and the error RunDeepVerify returned.
|
||||
func deepVerifyJSON(
|
||||
ctx context.Context, t *testing.T,
|
||||
cfg *config.Config, storer storage.Storer, snapshotID string,
|
||||
) (vaultik.VerifyResult, error) {
|
||||
t.Helper()
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
v := &vaultik.Vaultik{
|
||||
Config: cfg,
|
||||
Storage: storer,
|
||||
Fs: afero.NewOsFs(),
|
||||
Stdout: &out,
|
||||
Stderr: io.Discard,
|
||||
UI: ui.NewWithColor(io.Discard, false),
|
||||
}
|
||||
v.SetContext(ctx)
|
||||
|
||||
err := v.RunDeepVerify(snapshotID,
|
||||
&vaultik.VerifyOptions{Deep: true, JSON: true})
|
||||
|
||||
return decodeVerifyResult(t, out.Bytes()), err
|
||||
}
|
||||
|
||||
// corruptOneBlob flips a middle byte of the first blob file found under
|
||||
// blobsDir, leaving the file length unchanged.
|
||||
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||
|
||||
+12
-27
@@ -103,12 +103,11 @@ type RestoreResult struct {
|
||||
BlobsDownloaded int
|
||||
BytesDownloaded int64
|
||||
Duration time.Duration
|
||||
FilesFailed int
|
||||
FailedFiles []string // Paths of files that failed to restore
|
||||
// Verification results (only populated if Verify option is set)
|
||||
FilesVerified int
|
||||
BytesVerified int64
|
||||
VerifyFailedFiles []string // Paths of restored files that failed verification
|
||||
FilesVerified int
|
||||
BytesVerified int64
|
||||
FilesFailed int
|
||||
FailedFiles []string // Paths of files that failed verification
|
||||
}
|
||||
|
||||
// Restore restores files from a snapshot to the target directory
|
||||
@@ -659,34 +658,20 @@ func (v *Vaultik) handleRestoreVerification(
|
||||
opts *RestoreOptions,
|
||||
result *RestoreResult,
|
||||
) error {
|
||||
// A file that failed to restore was not written and is already
|
||||
// reported as such, so it is not verified.
|
||||
failedToRestore := make(map[string]bool, len(result.FailedFiles))
|
||||
for _, path := range result.FailedFiles {
|
||||
failedToRestore[path] = true
|
||||
}
|
||||
|
||||
restored := make([]*database.File, 0, len(files))
|
||||
for _, file := range files {
|
||||
if !failedToRestore[file.Path.String()] {
|
||||
restored = append(restored, file)
|
||||
}
|
||||
}
|
||||
|
||||
err := v.verifyRestoredFiles(v.ctx, repos, restored, opts.TargetDir, result)
|
||||
err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verification failed: %w", err)
|
||||
}
|
||||
|
||||
if len(result.VerifyFailedFiles) > 0 {
|
||||
if result.FilesFailed > 0 {
|
||||
v.UI.Errorf("Verification failed: %s files did not match expected checksums.",
|
||||
v.UI.Count(len(result.VerifyFailedFiles)))
|
||||
v.UI.Count(result.FilesFailed))
|
||||
|
||||
for _, path := range result.VerifyFailedFiles {
|
||||
for _, path := range result.FailedFiles {
|
||||
v.UI.Detailf("%s", v.UI.Path(path))
|
||||
}
|
||||
|
||||
return fmt.Errorf("%d %w", len(result.VerifyFailedFiles), errFilesFailedVerify)
|
||||
return fmt.Errorf("%d %w", result.FilesFailed, errFilesFailedVerify)
|
||||
}
|
||||
|
||||
v.UI.Completef("Verified %s files (%s).",
|
||||
@@ -1448,8 +1433,8 @@ func (v *Vaultik) verifyRestoredFiles(
|
||||
if err != nil {
|
||||
log.Error("File verification failed", "path", file.Path, "error", err)
|
||||
|
||||
result.VerifyFailedFiles = append(
|
||||
result.VerifyFailedFiles, file.Path.String())
|
||||
result.FilesFailed++
|
||||
result.FailedFiles = append(result.FailedFiles, file.Path.String())
|
||||
}
|
||||
|
||||
bytesProcessed += file.Size
|
||||
@@ -1465,7 +1450,7 @@ func (v *Vaultik) verifyRestoredFiles(
|
||||
log.Info("Verification complete",
|
||||
"files_verified", result.FilesVerified,
|
||||
"bytes_verified", ubytes(result.BytesVerified),
|
||||
"files_failed", len(result.VerifyFailedFiles),
|
||||
"files_failed", result.FilesFailed,
|
||||
)
|
||||
|
||||
return nil
|
||||
|
||||
@@ -4,9 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
@@ -106,130 +104,6 @@ func assertRestoreSkipsLostFiles(
|
||||
}
|
||||
}
|
||||
|
||||
// TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles restores with
|
||||
// SkipErrors and Verify after one blob of a two-blob snapshot was
|
||||
// deleted. Verification must check only the files that were written, so
|
||||
// a file of the deleted blob is reported as failed to restore and not
|
||||
// also as failing verification, and Restore must still return an error.
|
||||
func TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
backup := backupThenDeleteOneBlob(ctx, t)
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)
|
||||
v.UI = ui.NewWithColor(&out, false)
|
||||
|
||||
err := v.Restore(&vaultik.RestoreOptions{
|
||||
SnapshotID: backup.snapshotID,
|
||||
TargetDir: backup.restoreDir,
|
||||
Verify: true,
|
||||
SkipErrors: true,
|
||||
})
|
||||
|
||||
require.Error(t, err, "restore must fail when files were skipped")
|
||||
assert.Contains(t, err.Error(),
|
||||
fmt.Sprintf("%d file(s) failed to restore", len(backup.lost)))
|
||||
|
||||
written := len(backup.files) - len(backup.lost)
|
||||
|
||||
assert.Contains(t, out.String(), fmt.Sprintf("Verifying %d files", written))
|
||||
assert.Contains(t, out.String(), fmt.Sprintf("Verified %d files", written))
|
||||
assert.NotContains(t, out.String(), "Verification failed")
|
||||
}
|
||||
|
||||
// TestRestoreSkipErrorsVerifyReportsFailuresApart restores with
|
||||
// SkipErrors and Verify after one blob of a two-blob snapshot was
|
||||
// deleted, while one file that does restore is written one byte too long.
|
||||
// The restore warning must list only the files of the deleted blob, the
|
||||
// verification summary only the long file, each path once, and Restore
|
||||
// must return an error.
|
||||
func TestRestoreSkipErrorsVerifyReportsFailuresApart(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
backup := backupThenDeleteOneBlob(ctx, t)
|
||||
|
||||
var long string
|
||||
|
||||
for path := range backup.files {
|
||||
if !backup.lost[path] {
|
||||
long = path
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
fs := trailingByteFs{Fs: backup.fs, suffix: long}
|
||||
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, fs)
|
||||
v.UI = ui.NewWithColor(&out, false)
|
||||
|
||||
err := v.Restore(&vaultik.RestoreOptions{
|
||||
SnapshotID: backup.snapshotID,
|
||||
TargetDir: backup.restoreDir,
|
||||
Verify: true,
|
||||
SkipErrors: true,
|
||||
})
|
||||
require.Error(t, err, "restore must fail when files failed")
|
||||
|
||||
_, afterWarning, found := strings.Cut(out.String(),
|
||||
fmt.Sprintf("%d file(s) failed to restore:", len(backup.lost)))
|
||||
require.Truef(t, found, "no restore warning in:\n%s", out.String())
|
||||
|
||||
warning, summary, found := strings.Cut(afterWarning,
|
||||
"Verification failed: 1 files did not match")
|
||||
require.Truef(t, found, "no verification summary in:\n%s", out.String())
|
||||
|
||||
for path := range backup.lost {
|
||||
assert.Equalf(t, 1, strings.Count(warning, path),
|
||||
"%s must be listed once as failed to restore", path)
|
||||
assert.NotContainsf(t, summary, path,
|
||||
"%s was not restored and must not fail verification", path)
|
||||
}
|
||||
|
||||
assert.NotContainsf(t, warning, long,
|
||||
"%s was restored and must not be listed as failed to restore", long)
|
||||
assert.Equalf(t, 1, strings.Count(summary, long),
|
||||
"%s must be listed once as failing verification", long)
|
||||
}
|
||||
|
||||
// trailingByteFs appends one byte, on close, to the file it opens whose
|
||||
// path ends in suffix, so that file is restored one byte too long and
|
||||
// fails verification.
|
||||
type trailingByteFs struct {
|
||||
afero.Fs
|
||||
|
||||
suffix string
|
||||
}
|
||||
|
||||
//nolint:ireturn // afero.Fs.OpenFile is defined to return the interface
|
||||
func (fs trailingByteFs) OpenFile(
|
||||
name string, flag int, perm os.FileMode,
|
||||
) (afero.File, error) {
|
||||
f, err := fs.Fs.OpenFile(name, flag, perm)
|
||||
if err != nil || !strings.HasSuffix(name, fs.suffix) {
|
||||
return f, err
|
||||
}
|
||||
|
||||
return trailingByteFile{File: f}, nil
|
||||
}
|
||||
|
||||
type trailingByteFile struct {
|
||||
afero.File
|
||||
}
|
||||
|
||||
func (f trailingByteFile) Close() error {
|
||||
_, _ = f.Write([]byte{'x'})
|
||||
|
||||
return f.File.Close()
|
||||
}
|
||||
|
||||
// TestRestoreMissingBlobAbortsWithoutSkipErrors checks that a deleted blob
|
||||
// still ends the restore with an error when SkipErrors is not set.
|
||||
func TestRestoreMissingBlobAbortsWithoutSkipErrors(t *testing.T) {
|
||||
|
||||
@@ -1,83 +0,0 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"filippo.io/age"
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/ui"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// TestVerifyPassesWhenBackupReusesStoredBlob backs up a file, then backs it
|
||||
// up again from an empty local index, as after `database delete`, with an
|
||||
// extra age recipient. The second backup builds the same blob, finds it
|
||||
// already stored and skips the upload, but its own copy is larger, because
|
||||
// age writes a header entry for each recipient. The second snapshot must
|
||||
// record the size of the stored object, so shallow and deep verify pass.
|
||||
func TestVerifyPassesWhenBackupReusesStoredBlob(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")
|
||||
secondDBPath := filepath.Join(tempDir, "second.sqlite")
|
||||
|
||||
chunkSize := int64(64 * 1024)
|
||||
maxBlobSize := int64(512 * 1024)
|
||||
|
||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||
require.NoError(t, afero.WriteFile(fs,
|
||||
filepath.Join(dataDir, "data.bin"),
|
||||
bytesPattern("reuse-", int(chunkSize*3)), 0o644))
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
runFileStorageBackup(ctx, t, fs, dataDir, storeDir,
|
||||
filepath.Join(tempDir, "first.sqlite"), chunkSize, maxBlobSize)
|
||||
|
||||
env := setupDedupBackupEnv(
|
||||
ctx, t, fs, storeDir, secondDBPath, chunkSize, maxBlobSize)
|
||||
|
||||
defer func() { _ = env.db.Close() }()
|
||||
|
||||
extra, err := age.GenerateX25519Identity()
|
||||
require.NoError(t, err)
|
||||
|
||||
env.cfg.AgeRecipients = append(env.cfg.AgeRecipients,
|
||||
extra.Recipient().String())
|
||||
|
||||
snapshotID, result := runDedupSnapshot(ctx, t, env.sm, env.makeScanner(),
|
||||
env.cfg.Hostname, "second", dataDir, secondDBPath)
|
||||
require.Positive(t, result.BlobsCreated)
|
||||
require.Zero(t, result.BlobsUploaded,
|
||||
"the second backup should find every blob already stored")
|
||||
|
||||
require.NoError(t, env.db.Close())
|
||||
|
||||
verifier := &vaultik.Vaultik{
|
||||
Config: env.cfg,
|
||||
Storage: env.storer,
|
||||
Fs: fs,
|
||||
Stdout: io.Discard,
|
||||
Stderr: io.Discard,
|
||||
UI: ui.NewWithColor(io.Discard, false),
|
||||
}
|
||||
verifier.SetContext(ctx)
|
||||
|
||||
require.NoError(t,
|
||||
verifier.VerifySnapshotWithOptions(snapshotID, &vaultik.VerifyOptions{}),
|
||||
"shallow verify should pass")
|
||||
require.NoError(t,
|
||||
verifier.VerifySnapshotWithOptions(snapshotID,
|
||||
&vaultik.VerifyOptions{Deep: true}),
|
||||
"deep verify should pass")
|
||||
}
|
||||
@@ -627,8 +627,7 @@ func selectSnapshotsToPurge(
|
||||
}
|
||||
|
||||
// confirmAndExecutePurge shows deletion candidates, confirms with the
|
||||
// user, and deletes snapshots. It stops with an error at the first
|
||||
// snapshot it cannot delete.
|
||||
// user, and deletes snapshots.
|
||||
func (v *Vaultik) confirmAndExecutePurge(
|
||||
toDelete []SnapshotInfo, force, quiet bool,
|
||||
) error {
|
||||
@@ -670,11 +669,10 @@ func (v *Vaultik) confirmAndExecutePurge(
|
||||
snapshotID := snap.ID.String()
|
||||
log.Info("Deleting snapshot", "id", snapshotID)
|
||||
|
||||
// Stop before the remote delete: a snapshot still in the local
|
||||
// index must keep its metadata on the destination store.
|
||||
err := v.deleteSnapshotFromLocalDB(snapshotID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("deleting snapshot %s from local index: %w", snapshotID, err)
|
||||
log.Error("Failed to delete from local database",
|
||||
"snapshot_id", snapshotID, "error", err)
|
||||
}
|
||||
|
||||
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
||||
|
||||
@@ -483,14 +483,13 @@ func TestListSnapshots_UnreadableManifestDoesNotHideOthers(t *testing.T) {
|
||||
//
|
||||
//nolint:tagliatelle // snake_case is the established output format
|
||||
type listJSONRow struct {
|
||||
ID string `json:"id"`
|
||||
RemoteKey string `json:"remote_key"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
CompressedSize int64 `json:"compressed_size"`
|
||||
UncompressedSize int64 `json:"uncompressed_size"`
|
||||
NewChunkSize int64 `json:"new_chunk_size"`
|
||||
LocallyTracked bool `json:"locally_tracked"`
|
||||
RemotePresent *bool `json:"remote_present"`
|
||||
ID string `json:"id"`
|
||||
RemoteKey string `json:"remote_key"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
CompressedSize int64 `json:"compressed_size"`
|
||||
NewChunkSize int64 `json:"new_chunk_size"`
|
||||
LocallyTracked bool `json:"locally_tracked"`
|
||||
RemotePresent *bool `json:"remote_present"`
|
||||
}
|
||||
|
||||
// decodeListJSON parses the command's stdout, which must contain
|
||||
|
||||
@@ -1,85 +0,0 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// TestUncompressedSizeOfChangedFileStaysFixed covers
|
||||
// https://git.eeqj.de/sneak/vaultik/issues/360. A backup that changes a
|
||||
// file leaves the earlier snapshot's UNCOMPRESSED SIZE as it was when that
|
||||
// snapshot's backup completed. The file's row in the local index is shared
|
||||
// by both snapshots and rewritten by the second backup.
|
||||
func TestUncompressedSizeOfChangedFileStaysFixed(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
fs := afero.NewOsFs()
|
||||
tempDir := t.TempDir()
|
||||
dataDir := filepath.Join(tempDir, "src")
|
||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||
changedPath := filepath.Join(dataDir, "changed.bin")
|
||||
|
||||
ctx := context.Background()
|
||||
files := writeFaultSourceTree(t, fs, dataDir)
|
||||
cfg := changedFileConfig(dataDir, dbPath)
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, appendedFileSize)
|
||||
|
||||
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
|
||||
require.NoError(t, err)
|
||||
|
||||
db, err := database.New(ctx, dbPath)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||
|
||||
require.NoError(t, backUp(v, "first"))
|
||||
|
||||
// With nothing to compare against, sizeOfNewChunks totals every
|
||||
// distinct chunk of the files.
|
||||
firstID := localSnapshotID(ctx, t, repos, "first")
|
||||
firstSize := sizeOfNewChunks(t, files, nil)
|
||||
require.Equal(t, firstSize, listedUncompressedSizes(t, v)[firstID],
|
||||
"the first snapshot must count the chunks of every file it holds")
|
||||
|
||||
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||
require.NoError(t, backUp(v, "second"))
|
||||
|
||||
secondID := localSnapshotID(ctx, t, repos, "second")
|
||||
sizes := listedUncompressedSizes(t, v)
|
||||
assert.Equal(t, firstSize, sizes[firstID],
|
||||
"a later backup must not change the first snapshot's figure")
|
||||
assert.Equal(t, sizeOfNewChunks(t, files, nil), sizes[secondID],
|
||||
"the second snapshot must count the changed file's chunks")
|
||||
}
|
||||
|
||||
// listedUncompressedSizes runs `snapshot list --json` and returns each
|
||||
// listed snapshot's uncompressed_size by snapshot ID.
|
||||
func listedUncompressedSizes(t *testing.T, v *vaultik.Vaultik) map[string]int64 {
|
||||
t.Helper()
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
v.Stdout = &out
|
||||
require.NoError(t, v.ListSnapshots(true))
|
||||
|
||||
sizes := make(map[string]int64)
|
||||
for _, row := range decodeListJSON(t, out.String()) {
|
||||
sizes[row.ID] = row.UncompressedSize
|
||||
}
|
||||
|
||||
return sizes
|
||||
}
|
||||
+17
-39
@@ -17,7 +17,6 @@ import (
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
)
|
||||
|
||||
// Sentinel errors for snapshot verification failures.
|
||||
@@ -33,9 +32,7 @@ var (
|
||||
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
||||
errManifestMissingBlob = errors.New(
|
||||
"manifest omits blob present in database")
|
||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||
errBlobsNotInStorage = errors.New(
|
||||
"blobs missing from storage or stored with the wrong size")
|
||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||
)
|
||||
|
||||
// verifyStatusFailed is the JSON status value for a failed verification.
|
||||
@@ -152,6 +149,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
||||
}
|
||||
|
||||
result.Status = "ok"
|
||||
result.Verified = len(dbBlobs)
|
||||
|
||||
if opts.JSON {
|
||||
return v.outputVerifyJSON(result)
|
||||
@@ -308,7 +306,7 @@ func (v *Vaultik) runVerificationSteps(
|
||||
v.stdoutf("Checking blob existence in remote storage...\n")
|
||||
}
|
||||
|
||||
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
|
||||
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
||||
if err != nil {
|
||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||
}
|
||||
@@ -319,8 +317,7 @@ func (v *Vaultik) runVerificationSteps(
|
||||
len(dbBlobs), ubytes(totalSize))
|
||||
}
|
||||
|
||||
err = v.performDeepVerificationFromDB(
|
||||
dbBlobs, tdb.db.Conn(), opts, result, identities)
|
||||
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
||||
if err != nil {
|
||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||
}
|
||||
@@ -672,14 +669,8 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
|
||||
return nil
|
||||
}
|
||||
|
||||
// verifyBlobExistenceFromDB checks that every blob from the database is
|
||||
// in storage with the size the database records. It checks them all,
|
||||
// counting the missing and wrong-size blobs in result, and fails if it
|
||||
// found any. Any storage error other than not found stops the check and
|
||||
// fails with that error.
|
||||
func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||
blobs []snapshot.BlobInfo, result *VerifyResult,
|
||||
) error {
|
||||
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
||||
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
||||
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
||||
|
||||
for i, blob := range blobs {
|
||||
@@ -692,23 +683,17 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||
// Construct blob path
|
||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
||||
|
||||
// Check blob exists
|
||||
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||
switch {
|
||||
case errors.Is(err, storage.ErrNotFound):
|
||||
log.Warn("Blob missing from storage", "hash", blob.Hash)
|
||||
if err != nil {
|
||||
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
||||
}
|
||||
|
||||
result.Missing++
|
||||
result.MissingSize += blob.CompressedSize
|
||||
case err != nil:
|
||||
return fmt.Errorf("checking blob %s in storage: %w",
|
||||
blob.Hash, err)
|
||||
case stat.Size != blob.CompressedSize:
|
||||
log.Warn("Blob has the wrong size in storage",
|
||||
"hash", blob.Hash,
|
||||
"storage_size", stat.Size,
|
||||
"database_size", blob.CompressedSize)
|
||||
|
||||
result.Mismatched++
|
||||
// Verify size matches
|
||||
if stat.Size != blob.CompressedSize {
|
||||
return fmt.Errorf(
|
||||
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
||||
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
||||
}
|
||||
|
||||
// Progress update every 100 blobs
|
||||
@@ -722,22 +707,16 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||
}
|
||||
}
|
||||
|
||||
if result.Missing > 0 || result.Mismatched > 0 {
|
||||
return fmt.Errorf("%w: %d missing, %d with the wrong size",
|
||||
errBlobsNotInStorage, result.Missing, result.Mismatched)
|
||||
}
|
||||
|
||||
log.Info("✓ All blobs exist in storage")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// performDeepVerificationFromDB downloads and verifies the content of
|
||||
// each blob using the database as source, stopping at the first blob that
|
||||
// fails. Each blob that passes is counted in result.Verified.
|
||||
// each blob using the database as source.
|
||||
func (v *Vaultik) performDeepVerificationFromDB(
|
||||
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
||||
result *VerifyResult, identities []age.Identity,
|
||||
identities []age.Identity,
|
||||
) error {
|
||||
// Calculate total bytes for ETA
|
||||
var totalBytesExpected int64
|
||||
@@ -760,7 +739,6 @@ func (v *Vaultik) performDeepVerificationFromDB(
|
||||
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
||||
}
|
||||
|
||||
result.Verified++
|
||||
bytesProcessed += blobInfo.CompressedSize
|
||||
elapsed := time.Since(startTime)
|
||||
remaining := len(blobs) - (i + 1)
|
||||
|
||||
Reference in New Issue
Block a user