Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d2108fbac4 | ||
|
|
fb8abcd049 | ||
|
|
6aceaaf086 | ||
|
|
3b2b9cec84 | ||
|
|
dfa3bbc9e1 | ||
|
|
740f474c59 | ||
|
|
27e9798da5 |
+24
-3
@@ -5,7 +5,16 @@
|
||||
FROM golangci/golangci-lint:v2.14.0@sha256:ad862ba6b3798cbe0fd9fd7408d498fd74fbd2623a92406b2fd3898faf0bf98f AS lint
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
# 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
|
||||
COPY . .
|
||||
# `golangci-lint run` silently ignores an unknown top-level key in
|
||||
# .golangci.yml, such as a misspelt `linters:`; `config verify` fails on it.
|
||||
@@ -18,7 +27,13 @@ RUN golangci-lint run --config .golangci.yml ./...
|
||||
FROM golang:1.26.1@sha256:cd78d88e00afadbedd272f977d375a6247455f3a4b1178f8ae8bbcb201743a8a AS test
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
# 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
|
||||
COPY . .
|
||||
RUN go test -timeout 90s -race -cover ./... || \
|
||||
{ echo "--- Rerunning with -v for details ---"; \
|
||||
@@ -36,7 +51,13 @@ RUN apk add --no-cache git
|
||||
RUN git config --system --add safe.directory /src
|
||||
WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
# 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
|
||||
COPY . .
|
||||
|
||||
# The VERSION build arg when one is given, otherwise
|
||||
|
||||
@@ -375,7 +375,9 @@ 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 file's chunk hashes match
|
||||
* `--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.
|
||||
|
||||
**`prune`**: Tidy up everything that isn't needed. Runs three passes:
|
||||
(1) reconcile the local index against the destination store — any
|
||||
@@ -386,10 +388,12 @@ manifest on the destination store to compute the still-referenced blob
|
||||
set and delete any blob not in that set. Step (3) reads all remote
|
||||
manifests — network cost scales with the number of snapshots. `snapshot
|
||||
create --prune` runs the same cleanup automatically; this is the
|
||||
manual entry point for the same work.
|
||||
manual entry point for the same work. If a blob cannot be deleted, the
|
||||
command (and `snapshot create --prune`) exits non-zero.
|
||||
* `--force`: Skip confirmation prompt
|
||||
* `--json`: Output stats as JSON. Also skips the confirmation prompt, as
|
||||
`--force` does.
|
||||
`--force` does. A blob that cannot be deleted is counted in
|
||||
`blobs_failed`, and the document is printed before the non-zero exit.
|
||||
|
||||
**`info`**: Display system configuration, storage settings, encryption
|
||||
recipients, and local database statistics.
|
||||
@@ -411,7 +415,8 @@ since `prune` never deletes it.
|
||||
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
||||
backup destination store, leaving the bucket prefix empty. Destructive and
|
||||
irreversible. This is the single supported way to wipe the entire
|
||||
destination store.
|
||||
destination store. If anything cannot be deleted, the command exits
|
||||
non-zero; running it again deletes what is left.
|
||||
* `--force`: Required to confirm destruction.
|
||||
|
||||
**`database delete`**: Delete the local SQLite state database file
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
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
-2
@@ -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
|
||||
- `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
|
||||
- `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.
|
||||
@@ -184,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 = ?` - Mark blob as uploaded
|
||||
- `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)
|
||||
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
||||
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
||||
|
||||
|
||||
@@ -58,12 +58,7 @@ func (r *BlobChunkRepository) GetByBlobID(
|
||||
return nil, fmt.Errorf("querying blob chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var blobChunks []*BlobChunk
|
||||
|
||||
|
||||
+10
-12
@@ -83,12 +83,7 @@ func (r *BlobRepository) GetAll(ctx context.Context) (map[string]*Blob, error) {
|
||||
return nil, fmt.Errorf("querying blobs: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
out := make(map[string]*Blob)
|
||||
|
||||
@@ -159,13 +154,16 @@ func (r *BlobRepository) UpdateFinished(
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateUploaded marks a blob as uploaded
|
||||
// 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.
|
||||
func (r *BlobRepository) UpdateUploaded(
|
||||
ctx context.Context, tx *sql.Tx, id string,
|
||||
ctx context.Context, tx *sql.Tx, id string, storedSize int64,
|
||||
) error {
|
||||
query := `
|
||||
UPDATE blobs
|
||||
SET uploaded_ts = ?
|
||||
UPDATE blobs
|
||||
SET uploaded_ts = ?, compressed_size = ?
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
@@ -173,9 +171,9 @@ func (r *BlobRepository) UpdateUploaded(
|
||||
|
||||
var err error
|
||||
if tx != nil {
|
||||
_, err = tx.ExecContext(ctx, query, now, id)
|
||||
_, err = tx.ExecContext(ctx, query, now, storedSize, id)
|
||||
} else {
|
||||
_, err = r.db.ExecWithLog(ctx, query, now, id)
|
||||
_, err = r.db.ExecWithLog(ctx, query, now, storedSize, 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
|
||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String())
|
||||
// Test UpdateUploaded, with a stored object larger than the local copy
|
||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to update blob as uploaded: %v", err)
|
||||
}
|
||||
@@ -141,6 +141,10 @@ 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,12 +61,7 @@ func (r *ChunkFileRepository) GetByChunkHash(
|
||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanChunkFiles(rows)
|
||||
}
|
||||
@@ -87,12 +82,7 @@ func (r *ChunkFileRepository) GetByFilePath(
|
||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanChunkFiles(rows)
|
||||
}
|
||||
@@ -112,12 +102,7 @@ func (r *ChunkFileRepository) GetByFileID(
|
||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanChunkFiles(rows)
|
||||
}
|
||||
|
||||
@@ -107,12 +107,7 @@ func (r *ChunkRepository) GetByHashes(
|
||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var chunks []*Chunk
|
||||
|
||||
@@ -152,12 +147,7 @@ func (r *ChunkRepository) ListUnpacked(
|
||||
return nil, fmt.Errorf("querying unpacked chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var chunks []*Chunk
|
||||
|
||||
|
||||
@@ -38,12 +38,7 @@ func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, err
|
||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var chunks []*Chunk
|
||||
|
||||
|
||||
@@ -1,12 +0,0 @@
|
||||
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,12 +62,7 @@ func (r *FileChunkRepository) GetByPath(
|
||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanFileChunks(rows)
|
||||
}
|
||||
@@ -88,12 +83,7 @@ func (r *FileChunkRepository) GetByFileID(
|
||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanFileChunks(rows)
|
||||
}
|
||||
@@ -117,12 +107,7 @@ func (r *FileChunkRepository) GetByPathTx(
|
||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
fileChunks, err := r.scanFileChunks(rows)
|
||||
LogSQL("GetByPathTx", "Complete", path, "count", len(fileChunks))
|
||||
|
||||
@@ -175,12 +175,7 @@ func (r *FileRepository) ListModifiedSince(
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var files []*File
|
||||
|
||||
@@ -256,12 +251,7 @@ func (r *FileRepository) ListUnderPath(
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var files []*File
|
||||
|
||||
@@ -304,12 +294,7 @@ func (r *FileRepository) ListIDsWithChunksNotInUploadedBlobs(
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var ids []types.FileID
|
||||
|
||||
@@ -340,12 +325,7 @@ func (r *FileRepository) ListAll(ctx context.Context) ([]*File, error) {
|
||||
return nil, fmt.Errorf("querying files: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var files []*File
|
||||
|
||||
|
||||
@@ -106,22 +106,24 @@ func (r *Repositories) WithReadTx(ctx context.Context, fn TxFunc) error {
|
||||
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
rollbackErr := tx.Rollback()
|
||||
if rollbackErr != nil {
|
||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
||||
}
|
||||
// The panic is the failure to report, so a rollback error
|
||||
// is dropped.
|
||||
_ = tx.Rollback()
|
||||
|
||||
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,3 +411,41 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -241,12 +241,7 @@ func (r *SnapshotRepository) GetOnlySnapshot(ctx context.Context) (*Snapshot, er
|
||||
return nil, fmt.Errorf("querying snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
snapshots, err := r.scanSnapshotRows(rows)
|
||||
if err != nil {
|
||||
@@ -281,12 +276,7 @@ func (r *SnapshotRepository) ListRecent(
|
||||
return nil, fmt.Errorf("querying snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanSnapshotRows(rows)
|
||||
}
|
||||
@@ -501,12 +491,7 @@ func (r *SnapshotRepository) GetBlobHashes(
|
||||
return nil, fmt.Errorf("querying blob hashes: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var blobs []string
|
||||
|
||||
@@ -634,12 +619,7 @@ func (r *SnapshotRepository) GetIncompleteSnapshots(
|
||||
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
return r.scanSnapshotRows(rows)
|
||||
}
|
||||
@@ -662,12 +642,7 @@ func (r *SnapshotRepository) GetIncompleteByHostname(
|
||||
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := rows.Close()
|
||||
if err != nil {
|
||||
Fatalf("failed to close rows: %v", err)
|
||||
}
|
||||
}()
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
// Same column set as every other multi-row snapshot query, so the
|
||||
// shared scanner applies — including its timestamp normalization.
|
||||
|
||||
+29
-2
@@ -18,6 +18,13 @@ import (
|
||||
"github.com/aws/smithy-go/logging"
|
||||
)
|
||||
|
||||
// ErrNoContinuationToken is returned by ListObjects and ListObjectsStream
|
||||
// when the server marks a page of a listing truncated but sends no
|
||||
// continuation token to fetch the rest with, as a server answering V2
|
||||
// listings V1-style does.
|
||||
var ErrNoContinuationToken = errors.New(
|
||||
"server marked the listing truncated but sent no continuation token")
|
||||
|
||||
// Client wraps the AWS S3 client for vaultik operations.
|
||||
// It provides a simplified interface for S3 operations with automatic
|
||||
// prefix handling and connection management. All operations are performed
|
||||
@@ -214,7 +221,7 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
||||
})
|
||||
|
||||
for paginator.HasMorePages() {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
page, err := nextListPage(ctx, paginator)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -235,6 +242,26 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
// nextListPage fetches the next page of a listing. The SDK paginator takes
|
||||
// a page marked truncated without a continuation token for the last page
|
||||
// and stops without an error, so such a page is ErrNoContinuationToken
|
||||
// here instead.
|
||||
func nextListPage(
|
||||
ctx context.Context, paginator *s3.ListObjectsV2Paginator,
|
||||
) (*s3.ListObjectsV2Output, error) {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
truncated := aws.ToBool(page.IsTruncated)
|
||||
if truncated && aws.ToString(page.NextContinuationToken) == "" {
|
||||
return nil, ErrNoContinuationToken
|
||||
}
|
||||
|
||||
return page, nil
|
||||
}
|
||||
|
||||
// HeadObject checks if an object exists in S3.
|
||||
// Returns true if the object exists, false otherwise.
|
||||
// The key is automatically prefixed with the configured prefix.
|
||||
@@ -300,7 +327,7 @@ func (c *Client) ListObjectsStream(
|
||||
})
|
||||
|
||||
for paginator.HasMorePages() {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
page, err := nextListPage(ctx, paginator)
|
||||
if err != nil {
|
||||
ch <- ObjectInfo{Err: err}
|
||||
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
//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,7 +1517,8 @@ 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)
|
||||
err := s.repos.Blobs.UpdateUploaded(
|
||||
ctx, tx, b.ID, b.Compressed)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marking blob uploaded: %w", err)
|
||||
}
|
||||
@@ -1553,23 +1554,28 @@ func (s *Scanner) handleBlobReady(
|
||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
||||
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
||||
|
||||
blobExists, err := s.uploadBlobIfNeeded(
|
||||
storedSize, blobExists, err := s.uploadBlobIfNeeded(
|
||||
ctx, blobPath, blobWithReader, startTime, result)
|
||||
if err != nil {
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||
|
||||
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
|
||||
return errors.Join(
|
||||
fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr)
|
||||
}
|
||||
|
||||
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime)
|
||||
err = s.recordBlobMetadata(
|
||||
ctx, finishedBlob, storedSize, blobExists, startTime)
|
||||
if err != nil {
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
cleanupErr := 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)
|
||||
|
||||
@@ -1583,20 +1589,22 @@ func (s *Scanner) handleBlobReady(
|
||||
}
|
||||
|
||||
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
||||
// exist, returns whether it existed
|
||||
// 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.
|
||||
func (s *Scanner) uploadBlobIfNeeded(
|
||||
ctx context.Context,
|
||||
blobPath string,
|
||||
blobWithReader *blob.WithReader,
|
||||
startTime time.Time,
|
||||
result *ScanResult,
|
||||
) (bool, error) {
|
||||
) (int64, bool, error) {
|
||||
finishedBlob := blobWithReader.FinishedBlob
|
||||
|
||||
// Check if blob already exists (deduplication after restart)
|
||||
destination := s.storage.Info().Location
|
||||
|
||||
_, err := s.storage.Stat(ctx, blobPath)
|
||||
stored, err := s.storage.Stat(ctx, blobPath)
|
||||
if err == nil {
|
||||
log.Info("Blob already exists in storage, skipping upload",
|
||||
"hash", finishedBlob.Hash,
|
||||
@@ -1605,7 +1613,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
||||
s.ui.Path(destination))
|
||||
|
||||
return true, nil
|
||||
return stored.Size, true, nil
|
||||
}
|
||||
|
||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||
@@ -1619,7 +1627,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
if err != nil {
|
||||
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
||||
|
||||
return false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||
return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||
}
|
||||
|
||||
uploadDuration := time.Since(startTime)
|
||||
@@ -1650,7 +1658,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
||||
}
|
||||
|
||||
return false, nil
|
||||
return finishedBlob.Compressed, false, nil
|
||||
}
|
||||
|
||||
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
||||
@@ -1720,6 +1728,7 @@ func (s *Scanner) makeUploadProgressCallback(
|
||||
func (s *Scanner) recordBlobMetadata(
|
||||
ctx context.Context,
|
||||
finishedBlob *blob.FinishedBlob,
|
||||
storedSize int64,
|
||||
blobExists bool,
|
||||
startTime time.Time,
|
||||
) error {
|
||||
@@ -1731,7 +1740,8 @@ 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)
|
||||
err := s.repos.Blobs.UpdateUploaded(
|
||||
txCtx, tx, finishedBlob.ID, storedSize)
|
||||
if err != nil {
|
||||
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
||||
}
|
||||
@@ -1762,20 +1772,22 @@ func (s *Scanner) recordBlobMetadata(
|
||||
}
|
||||
|
||||
// cleanupBlobTempFile closes and removes the blob's temporary file
|
||||
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) {
|
||||
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) error {
|
||||
if blobWithReader.TempFile != nil {
|
||||
tempName := blobWithReader.TempFile.Name()
|
||||
|
||||
err := blobWithReader.TempFile.Close()
|
||||
if err != nil {
|
||||
log.Fatal("Failed to close temp file", "file", tempName, "error", err)
|
||||
return fmt.Errorf("closing blob temp file: %w", err)
|
||||
}
|
||||
|
||||
err = s.fs.Remove(tempName)
|
||||
if err != nil {
|
||||
log.Fatal("Failed to remove temp file", "file", tempName, "error", err)
|
||||
return fmt.Errorf("removing blob temp file: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// streamingChunkInfo tracks chunk metadata collected during streaming
|
||||
|
||||
@@ -1,15 +1,16 @@
|
||||
// Package faultstore provides a storage.Storer wrapper that injects
|
||||
// faults on demand, so tests can reproduce the failure modes a real
|
||||
// backend exhibits: an upload that fails partway, a backend that reports
|
||||
// success while storing nothing, and reads that return corrupt or
|
||||
// truncated bytes. It is the seam called for by the fault-injection
|
||||
// tests (sneak/vaultik issue 72) and is meant to be reused by future
|
||||
// tests rather than re-implemented per case.
|
||||
// success while storing nothing, reads that return corrupt or truncated
|
||||
// bytes, and a delete the backend refuses. It is the seam called for by
|
||||
// the fault-injection tests (sneak/vaultik issue 72) and is meant to be
|
||||
// reused by future tests rather than re-implemented per case.
|
||||
//
|
||||
// The wrapper delegates every method to the inner Storer. Two hooks
|
||||
// change that: OnPut decides the fate of each write, and OnGet decides
|
||||
// how each read's bytes are returned. Both are keyed by the object key,
|
||||
// so a test can fault only blobs, only metadata, or a single object.
|
||||
// The wrapper delegates every method to the inner Storer. Three hooks
|
||||
// change that: OnPut decides the fate of each write, OnGet decides how
|
||||
// each read's bytes are returned, and OnDelete decides whether each
|
||||
// delete fails. All are keyed by the object key, so a test can fault
|
||||
// only blobs, only metadata, or a single object.
|
||||
package faultstore
|
||||
|
||||
import (
|
||||
@@ -25,6 +26,10 @@ import (
|
||||
// ErrInjectedUpload is returned by a Put the OnPut hook chose to fail.
|
||||
var ErrInjectedUpload = errors.New("faultstore: injected upload failure")
|
||||
|
||||
// ErrInjectedDelete is returned by a Delete the OnDelete hook chose to
|
||||
// fail.
|
||||
var ErrInjectedDelete = errors.New("faultstore: injected delete failure")
|
||||
|
||||
// PutAction is the disposition OnPut assigns to a write.
|
||||
type PutAction int
|
||||
|
||||
@@ -66,6 +71,11 @@ type Storer struct {
|
||||
// OnGet, when set, is consulted for every Get with the object key
|
||||
// and damages the returned bytes accordingly.
|
||||
OnGet func(key string) GetFault
|
||||
|
||||
// OnDelete, when set, is consulted before every Delete with the
|
||||
// object key. Returning true fails the delete and leaves the object
|
||||
// in place.
|
||||
OnDelete func(key string) bool
|
||||
}
|
||||
|
||||
// New wraps inner. inner must be non-nil.
|
||||
@@ -154,8 +164,12 @@ func (f *Storer) Stat(ctx context.Context, key string) (*storage.ObjectInfo, err
|
||||
return f.inner.Stat(ctx, key)
|
||||
}
|
||||
|
||||
// Delete delegates unchanged.
|
||||
// Delete removes the object unless OnDelete fails the delete.
|
||||
func (f *Storer) Delete(ctx context.Context, key string) error {
|
||||
if f.OnDelete != nil && f.OnDelete(key) {
|
||||
return fmt.Errorf("%w for %q", ErrInjectedDelete, key)
|
||||
}
|
||||
|
||||
return f.inner.Delete(ctx, key)
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
@@ -22,13 +23,16 @@ import (
|
||||
// s3TestBucket is the bucket created for each in-process S3 server.
|
||||
const s3TestBucket = "test-bucket"
|
||||
|
||||
// Credentials for the tests that build a storer from a config.Config. The
|
||||
// in-process S3 server accepts any.
|
||||
// Credentials for the tests' S3 clients. The test servers accept any.
|
||||
const (
|
||||
s3TestAccessKeyID = "key"
|
||||
s3TestSecretAccessKey = "secret"
|
||||
)
|
||||
|
||||
// s3TestRegion is the region of the S3 clients built here with
|
||||
// s3.NewClient. The test servers ignore it.
|
||||
const s3TestRegion = "us-east-1"
|
||||
|
||||
// newS3Storer builds an s3:// backend backed by a fresh in-process
|
||||
// S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem
|
||||
// over httptest) that internal/s3 and the not-found regression test use,
|
||||
@@ -55,7 +59,7 @@ func newS3Storer(t *testing.T) storage.Storer {
|
||||
Bucket: s3TestBucket,
|
||||
AccessKeyID: "test",
|
||||
SecretAccessKey: "test",
|
||||
Region: "us-east-1",
|
||||
Region: s3TestRegion,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("new client: %v", err)
|
||||
@@ -93,6 +97,68 @@ func TestS3StorerMissingKeyMapsToErrNotFound(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3ListingCutShortIsAnError checks that List and ListStream fail on a
|
||||
// listing page marked truncated that carries no continuation token, as a
|
||||
// server answering V2 listings V1-style sends it. Taken as the last page,
|
||||
// it would hide every later key, and prune would treat the blobs of the
|
||||
// snapshots listed there as unreferenced.
|
||||
func TestS3ListingCutShortIsAnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
prefix = "metadata/"
|
||||
truncatedPage = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||
<Name>test-bucket</Name>
|
||||
<Prefix>metadata/</Prefix>
|
||||
<KeyCount>1</KeyCount>
|
||||
<MaxKeys>1</MaxKeys>
|
||||
<IsTruncated>true</IsTruncated>
|
||||
<Contents>
|
||||
<Key>metadata/snap/manifest.json.zst</Key>
|
||||
<Size>8</Size>
|
||||
</Contents>
|
||||
</ListBucketResult>`
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = io.WriteString(w, truncatedPage)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
client, err := s3.NewClient(context.Background(), s3.Config{
|
||||
Endpoint: srv.URL,
|
||||
Bucket: s3TestBucket,
|
||||
AccessKeyID: s3TestAccessKeyID,
|
||||
SecretAccessKey: s3TestSecretAccessKey,
|
||||
Region: s3TestRegion,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("new client: %v", err)
|
||||
}
|
||||
|
||||
storer := storage.NewS3Storer(client)
|
||||
ctx := context.Background()
|
||||
|
||||
keys, err := storer.List(ctx, prefix)
|
||||
if !errors.Is(err, s3.ErrNoContinuationToken) {
|
||||
t.Errorf("List = %q, %v; want ErrNoContinuationToken", keys, err)
|
||||
}
|
||||
|
||||
var streamErr error
|
||||
|
||||
for obj := range storer.ListStream(ctx, prefix) {
|
||||
if obj.Err != nil {
|
||||
streamErr = obj.Err
|
||||
}
|
||||
}
|
||||
|
||||
if !errors.Is(streamErr, s3.ErrNoContinuationToken) {
|
||||
t.Errorf("ListStream error = %v, want ErrNoContinuationToken", streamErr)
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
|
||||
// writes: the README's remote storage layout, with the prefix joined to
|
||||
// each key by one "/". s3://b/p and s3://b/p/ must be the same
|
||||
|
||||
@@ -5,12 +5,14 @@ import (
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
||||
)
|
||||
|
||||
// TestNukeRemoteLeavesNoFiles checks that remote nuke leaves no file
|
||||
@@ -23,7 +25,7 @@ func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||
v, repos, out := backUpToFileDestination(ctx, t, storeDir)
|
||||
|
||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||
require.NoError(t, err)
|
||||
@@ -61,4 +63,82 @@ func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, files)
|
||||
|
||||
// Nuke deletes the blobs itself, so advice to run prune is wrong.
|
||||
assert.Contains(t, out.String(), "Backup destination store is now empty.")
|
||||
assert.NotContains(t, out.String(), "vaultik prune")
|
||||
}
|
||||
|
||||
// TestNukeRemoteFailsWhenADeleteFails checks that remote nuke returns an
|
||||
// error when a snapshot's metadata or a blob cannot be deleted, does not
|
||||
// report the store empty, and advises running nuke again: prune deletes
|
||||
// no metadata.
|
||||
func TestNukeRemoteFailsWhenADeleteFails(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
for _, refusedPrefix := range []string{"metadata/", "blobs/"} {
|
||||
t.Run(refusedPrefix, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||
|
||||
store := faultstore.New(v.Storage)
|
||||
store.OnDelete = func(key string) bool {
|
||||
return strings.HasPrefix(key, refusedPrefix)
|
||||
}
|
||||
v.Storage = store
|
||||
|
||||
require.Error(t, v.NukeRemote(true))
|
||||
|
||||
assert.NotContains(t, out.String(), "now empty")
|
||||
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||
assert.NotContains(t, out.String(), "vaultik prune")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice checks that
|
||||
// remote nuke, when the local index belongs to a different destination
|
||||
// than the configured one, fails without advising to run nuke again:
|
||||
// that run would fail the same way.
|
||||
func TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||
|
||||
v.Config.StorageURL = "file://" + filepath.Join(t.TempDir(), "elsewhere")
|
||||
|
||||
err := v.NukeRemote(true)
|
||||
|
||||
require.ErrorContains(t, err, "bound to a different backup destination")
|
||||
assert.NotContains(t, out.String(), "vaultik remote nuke --force")
|
||||
}
|
||||
|
||||
// TestNukeRemoteFailsWhenALocalSnapshotCannotBeRemoved checks that a
|
||||
// snapshot remote nuke cannot remove from the local index fails the
|
||||
// run: nuke then leaves that snapshot's metadata on the destination
|
||||
// store.
|
||||
func TestNukeRemoteFailsWhenALocalSnapshotCannotBeRemoved(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||
|
||||
_, err := v.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.Error(t, v.NukeRemote(true))
|
||||
|
||||
assert.NotContains(t, out.String(), "now empty")
|
||||
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||
}
|
||||
|
||||
@@ -19,6 +19,14 @@ type PruneOptions struct {
|
||||
var errNukeRequiresForce = errors.New(
|
||||
"nuke requires --force (this deletes ALL remote snapshots and blobs)")
|
||||
|
||||
// errBlobDeleteFailed reports unreferenced blobs prune could not delete.
|
||||
var errBlobDeleteFailed = errors.New("blob(s) could not be deleted")
|
||||
|
||||
// nukeCommandHint is the command suggested when a remote nuke could not
|
||||
// delete everything: running it again deletes what is left, which
|
||||
// `vaultik prune` does not do for snapshot metadata.
|
||||
const nukeCommandHint = "vaultik remote nuke --force"
|
||||
|
||||
// metadataDirName is the top-level remote directory holding snapshot
|
||||
// metadata.
|
||||
const metadataDirName = "metadata"
|
||||
@@ -39,6 +47,29 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
||||
return errNukeRequiresForce
|
||||
}
|
||||
|
||||
// Checked before anything is deleted: running nuke again would fail
|
||||
// the same way, so the advice below must not follow this error.
|
||||
err := v.EnsureStorageBinding()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = v.emptyDestinationStore()
|
||||
if err != nil {
|
||||
v.UI.Warningf("Run '%s' again to delete what is left on the "+
|
||||
"backup destination store.", nukeCommandHint)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
v.UI.Completef("Backup destination store is now empty.")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// emptyDestinationStore does the deleting for NukeRemote and stops at
|
||||
// the first step that fails.
|
||||
func (v *Vaultik) emptyDestinationStore() error {
|
||||
v.UI.Beginf("Removing all snapshot metadata from backup destination store.")
|
||||
|
||||
_, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true})
|
||||
@@ -62,8 +93,6 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
||||
}
|
||||
}
|
||||
|
||||
v.UI.Completef("Backup destination store is now empty.")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -166,14 +195,17 @@ func (v *Vaultik) PruneBlobs(opts *PruneOptions) error {
|
||||
v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)
|
||||
|
||||
if opts.JSON {
|
||||
return v.outputPruneBlobsJSON(result)
|
||||
err = v.outputPruneBlobsJSON(result)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
v.stdoutf("\nDeleted %d blob(s) totaling %s\n",
|
||||
result.BlobsDeleted, ubytes(result.BytesFreed))
|
||||
}
|
||||
|
||||
v.stdoutf("\nDeleted %d blob(s) totaling %s\n",
|
||||
result.BlobsDeleted, ubytes(result.BytesFreed))
|
||||
|
||||
if result.BlobsFailed > 0 {
|
||||
v.stdoutf("Failed to delete %d blob(s)\n", result.BlobsFailed)
|
||||
return fmt.Errorf("%d %w", result.BlobsFailed, errBlobDeleteFailed)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// refuseBlobDeletes is an OnDelete hook that fails every blob delete.
|
||||
func refuseBlobDeletes(key string) bool {
|
||||
return strings.HasPrefix(key, "blobs/")
|
||||
}
|
||||
|
||||
// TestPruneBlobs_FailedDeleteReturnsError checks that prune returns an
|
||||
// error, so the command exits non-zero, when an unreferenced blob cannot
|
||||
// be deleted.
|
||||
func TestPruneBlobs_FailedDeleteReturnsError(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
env := newListEnv(t)
|
||||
|
||||
// The store holds no manifest, so nothing references this blob.
|
||||
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||
|
||||
store := faultstore.New(env.store)
|
||||
store.OnDelete = refuseBlobDeletes
|
||||
env.v.Storage = store
|
||||
|
||||
err := env.v.PruneBlobs(&vaultik.PruneOptions{Force: true})
|
||||
|
||||
require.ErrorContains(t, err, "1 blob(s) could not be deleted")
|
||||
}
|
||||
|
||||
// TestPruneBlobs_JSONFailedDeleteReturnsError checks that prune --json
|
||||
// prints its document and then returns an error when an unreferenced
|
||||
// blob cannot be deleted.
|
||||
func TestPruneBlobs_JSONFailedDeleteReturnsError(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
env := newListEnv(t)
|
||||
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||
|
||||
store := faultstore.New(env.store)
|
||||
store.OnDelete = refuseBlobDeletes
|
||||
env.v.Storage = store
|
||||
|
||||
err := env.v.PruneBlobs(&vaultik.PruneOptions{JSON: true})
|
||||
|
||||
require.Error(t, err, "prune must fail when a blob cannot be deleted")
|
||||
|
||||
var result vaultik.PruneBlobsResult
|
||||
|
||||
require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &result),
|
||||
"stdout must hold the JSON document, got:\n%s", env.stdout.String())
|
||||
assert.Equal(t, 1, result.BlobsFailed)
|
||||
assert.Equal(t, 0, result.BlobsDeleted)
|
||||
}
|
||||
|
||||
// TestCreateSnapshotPrune_FailedBlobDeleteReturnsError checks that the
|
||||
// prune step of snapshot create --prune fails the run when an
|
||||
// unreferenced blob cannot be deleted.
|
||||
func TestCreateSnapshotPrune_FailedBlobDeleteReturnsError(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
storeDir := filepath.Join(t.TempDir(), "store")
|
||||
v, _, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||
|
||||
// No snapshot references this blob, so the prune step deletes it.
|
||||
orphan := filepath.Join(storeDir, "blobs", testBlobHashA[:2],
|
||||
testBlobHashA[2:4], testBlobHashA)
|
||||
require.NoError(t, os.MkdirAll(filepath.Dir(orphan), 0o750))
|
||||
require.NoError(t, os.WriteFile(orphan, []byte("orphan"), 0o600))
|
||||
|
||||
store := faultstore.New(v.Storage)
|
||||
store.OnDelete = refuseBlobDeletes
|
||||
v.Storage = store
|
||||
|
||||
err := v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||
Cron: true,
|
||||
Prune: true,
|
||||
})
|
||||
|
||||
require.ErrorContains(t, err, "1 blob(s) could not be deleted")
|
||||
assert.FileExists(t, orphan)
|
||||
}
|
||||
@@ -476,7 +476,8 @@ func TestRemoveAllSnapshots_WithForce(t *testing.T) {
|
||||
out := tv.Stdout.String()
|
||||
assert.Contains(t, out, "Removed 2 snapshot(s)")
|
||||
assert.Contains(t, out, "Removed snapshot metadata from remote storage")
|
||||
assert.Contains(t, out, "vaultik prune")
|
||||
// Remote nuke, the only caller, deletes the blobs next itself.
|
||||
assert.NotContains(t, out, "vaultik prune")
|
||||
}
|
||||
|
||||
func TestRemoveAllSnapshots_DryRun(t *testing.T) {
|
||||
|
||||
+29
-12
@@ -103,11 +103,12 @@ 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
|
||||
FilesFailed int
|
||||
FailedFiles []string // Paths of files that failed verification
|
||||
FilesVerified int
|
||||
BytesVerified int64
|
||||
VerifyFailedFiles []string // Paths of restored files that failed verification
|
||||
}
|
||||
|
||||
// Restore restores files from a snapshot to the target directory
|
||||
@@ -658,20 +659,34 @@ func (v *Vaultik) handleRestoreVerification(
|
||||
opts *RestoreOptions,
|
||||
result *RestoreResult,
|
||||
) error {
|
||||
err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result)
|
||||
// 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)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verification failed: %w", err)
|
||||
}
|
||||
|
||||
if result.FilesFailed > 0 {
|
||||
if len(result.VerifyFailedFiles) > 0 {
|
||||
v.UI.Errorf("Verification failed: %s files did not match expected checksums.",
|
||||
v.UI.Count(result.FilesFailed))
|
||||
v.UI.Count(len(result.VerifyFailedFiles)))
|
||||
|
||||
for _, path := range result.FailedFiles {
|
||||
for _, path := range result.VerifyFailedFiles {
|
||||
v.UI.Detailf("%s", v.UI.Path(path))
|
||||
}
|
||||
|
||||
return fmt.Errorf("%d %w", result.FilesFailed, errFilesFailedVerify)
|
||||
return fmt.Errorf("%d %w", len(result.VerifyFailedFiles), errFilesFailedVerify)
|
||||
}
|
||||
|
||||
v.UI.Completef("Verified %s files (%s).",
|
||||
@@ -702,6 +717,8 @@ func (v *Vaultik) downloadSnapshotDB(
|
||||
return nil, "", fmt.Errorf("downloading %s: %w", dbKey, err)
|
||||
}
|
||||
|
||||
reader = newCloseOnCancelReader(v.ctx, reader)
|
||||
|
||||
defer func() { _ = reader.Close() }()
|
||||
|
||||
// Decrypt and decompress straight from the storage stream, then stream
|
||||
@@ -1431,8 +1448,8 @@ func (v *Vaultik) verifyRestoredFiles(
|
||||
if err != nil {
|
||||
log.Error("File verification failed", "path", file.Path, "error", err)
|
||||
|
||||
result.FilesFailed++
|
||||
result.FailedFiles = append(result.FailedFiles, file.Path.String())
|
||||
result.VerifyFailedFiles = append(
|
||||
result.VerifyFailedFiles, file.Path.String())
|
||||
}
|
||||
|
||||
bytesProcessed += file.Size
|
||||
@@ -1448,7 +1465,7 @@ func (v *Vaultik) verifyRestoredFiles(
|
||||
log.Info("Verification complete",
|
||||
"files_verified", result.FilesVerified,
|
||||
"bytes_verified", ubytes(result.BytesVerified),
|
||||
"files_failed", result.FilesFailed,
|
||||
"files_failed", len(result.VerifyFailedFiles),
|
||||
)
|
||||
|
||||
return nil
|
||||
|
||||
@@ -207,30 +207,37 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
|
||||
"the cancel was reported as a failed file")
|
||||
}
|
||||
|
||||
// stallingBlobStorer wraps a Storer so that every blob download hands
|
||||
// over the first half of the blob and then stalls, ignoring the context,
|
||||
// as a file:// blob on a hung USB stick or network share does. Only
|
||||
// closing the returned reader ends the stall.
|
||||
type stallingBlobStorer struct {
|
||||
// stallingStorer wraps a Storer so that each download of a key chosen by
|
||||
// stalls hands over the first half of the object and then stalls,
|
||||
// ignoring the context, as a file:// read on a hung USB stick or network
|
||||
// share does. Only closing the returned reader ends the stall.
|
||||
type stallingStorer struct {
|
||||
storage.Storer
|
||||
|
||||
stalls func(key string) bool
|
||||
once sync.Once
|
||||
stalled chan struct{}
|
||||
}
|
||||
|
||||
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
|
||||
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
|
||||
func newStallingStorer(
|
||||
inner storage.Storer, stalls func(key string) bool,
|
||||
) *stallingStorer {
|
||||
return &stallingStorer{
|
||||
Storer: inner,
|
||||
stalls: stalls,
|
||||
stalled: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *stallingBlobStorer) Get(
|
||||
func (s *stallingStorer) Get(
|
||||
ctx context.Context, key string,
|
||||
) (io.ReadCloser, error) {
|
||||
rc, err := s.Storer.Get(ctx, key)
|
||||
if err != nil || !strings.HasPrefix(key, "blobs/") {
|
||||
if err != nil || !s.stalls(key) {
|
||||
return rc, err
|
||||
}
|
||||
|
||||
blob, err := io.ReadAll(rc)
|
||||
object, err := io.ReadAll(rc)
|
||||
_ = rc.Close()
|
||||
|
||||
if err != nil {
|
||||
@@ -242,7 +249,7 @@ func (s *stallingBlobStorer) Get(
|
||||
pr, pw := io.Pipe()
|
||||
|
||||
go func() {
|
||||
_, _ = pw.Write(blob[:len(blob)/2])
|
||||
_, _ = pw.Write(object[:len(object)/2])
|
||||
|
||||
s.once.Do(func() { close(s.stalled) })
|
||||
}()
|
||||
@@ -250,17 +257,26 @@ func (s *stallingBlobStorer) Get(
|
||||
return pr, nil
|
||||
}
|
||||
|
||||
// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a
|
||||
// deep verify while a blob read is stalled on a source that ignores the
|
||||
// context. Stopping must end that read, so the operation unwinds and
|
||||
// removes its decrypted scratch files well within the shutdown timeout
|
||||
// (issue #325).
|
||||
// TestInterruptDuringStalledReadCleansTempDir stops a restore and a deep
|
||||
// verify while a read of a blob or of the snapshot database is stalled
|
||||
// on a source that ignores the context. Stopping must end that read, so
|
||||
// the operation unwinds and removes its decrypted scratch files well
|
||||
// within the shutdown timeout (issues #325 and #356).
|
||||
//
|
||||
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
|
||||
// so it can assert on exactly the scratch files the operation created.
|
||||
func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
||||
func TestInterruptDuringStalledReadCleansTempDir(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
|
||||
stalledReads := map[string]func(key string) bool{
|
||||
"blob": func(key string) bool {
|
||||
return strings.HasPrefix(key, "blobs/")
|
||||
},
|
||||
"snapshot database": func(key string) bool {
|
||||
return strings.HasSuffix(key, "/db.zst.age")
|
||||
},
|
||||
}
|
||||
|
||||
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
||||
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
||||
return v.Restore(&RestoreOptions{
|
||||
@@ -273,52 +289,58 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
for name, op := range operations {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
fs := afero.NewOsFs()
|
||||
root := t.TempDir()
|
||||
for readName, stalls := range stalledReads {
|
||||
for opName, op := range operations {
|
||||
t.Run(readName+" read during "+opName, func(t *testing.T) {
|
||||
fs := afero.NewOsFs()
|
||||
root := t.TempDir()
|
||||
|
||||
// Random content, so the blob stays large enough to stall
|
||||
// part-way through its compressed, encrypted body.
|
||||
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
|
||||
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
|
||||
// Random content, so the blob stays large enough to stall
|
||||
// part-way through its compressed, encrypted body.
|
||||
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
|
||||
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
|
||||
|
||||
scratch := filepath.Join(root, "scratch")
|
||||
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
||||
t.Setenv("TMPDIR", scratch)
|
||||
scratch := filepath.Join(root, "scratch")
|
||||
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
||||
t.Setenv("TMPDIR", scratch)
|
||||
|
||||
stalling := newStallingBlobStorer(storer)
|
||||
stalling := newStallingStorer(storer, stalls)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
v := newRestoreVaultik(ctx, cfg, stalling, fs)
|
||||
v.cancel = cancel
|
||||
v := newRestoreVaultik(ctx, cfg, stalling, fs)
|
||||
v.cancel = cancel
|
||||
|
||||
var opErr error
|
||||
var opErr error
|
||||
|
||||
stop := v.StartOperation(func() {
|
||||
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
|
||||
stop := v.StartOperation(func() {
|
||||
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
|
||||
})
|
||||
|
||||
select {
|
||||
case <-stalling.stalled:
|
||||
case <-time.After(30 * time.Second):
|
||||
t.Fatal("the operation never reached the stalled read")
|
||||
}
|
||||
|
||||
// The snapshot database's temp directory is created just
|
||||
// after its read starts, so it can appear after the stall.
|
||||
require.Eventually(t, func() bool {
|
||||
return len(scratchEntries(t, scratch)) > 0
|
||||
}, 10*time.Second, 10*time.Millisecond,
|
||||
"expected decrypted scratch files to exist mid-read")
|
||||
|
||||
stopCtx, stopCancel := context.WithTimeout(
|
||||
context.Background(), 10*time.Second)
|
||||
defer stopCancel()
|
||||
|
||||
require.True(t, stop(stopCtx),
|
||||
"stop timed out; the stalled read was not interrupted")
|
||||
require.ErrorIs(t, opErr, context.Canceled)
|
||||
require.Empty(t, scratchEntries(t, scratch),
|
||||
"decrypted scratch files remained after the interrupt")
|
||||
})
|
||||
|
||||
select {
|
||||
case <-stalling.stalled:
|
||||
case <-time.After(30 * time.Second):
|
||||
t.Fatal("the operation never reached a blob read")
|
||||
}
|
||||
|
||||
require.NotEmpty(t, scratchEntries(t, scratch),
|
||||
"expected decrypted scratch files to exist mid-read")
|
||||
|
||||
stopCtx, stopCancel := context.WithTimeout(
|
||||
context.Background(), 10*time.Second)
|
||||
defer stopCancel()
|
||||
|
||||
require.True(t, stop(stopCtx),
|
||||
"stop timed out; the stalled blob read was not interrupted")
|
||||
require.ErrorIs(t, opErr, context.Canceled)
|
||||
require.Empty(t, scratchEntries(t, scratch),
|
||||
"decrypted scratch files remained after the interrupt")
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,9 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
@@ -104,6 +106,130 @@ 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) {
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
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")
|
||||
}
|
||||
@@ -25,6 +25,7 @@ var (
|
||||
errSnapshotVerifyFailed = errors.New("verification failed")
|
||||
errRemoveAllNeedsForce = errors.New("--all requires --force")
|
||||
errSnapshotNotFound = errors.New("snapshot not found")
|
||||
errSnapshotRemoveFailed = errors.New("snapshot(s) could not be removed")
|
||||
errInvalidTableName = errors.New("invalid table name")
|
||||
errNoAgeRecipients = errors.New(
|
||||
"creating a snapshot needs at least one public key in " +
|
||||
@@ -1591,8 +1592,8 @@ func (v *Vaultik) handleRemoveAllDryRun(
|
||||
|
||||
// executeRemoveAll deletes every local snapshot and, unless LocalOnly
|
||||
// is set, every corresponding remote metadata directory plus any
|
||||
// orphan remote keys. Blobs are not touched; the printed prune-command
|
||||
// hint is the next step.
|
||||
// orphan remote keys. Blobs are not touched. It returns an error when
|
||||
// any snapshot could not be removed.
|
||||
func (v *Vaultik) executeRemoveAll(
|
||||
localSnaps, orphanRemoteKeys []string, opts *RemoveOptions,
|
||||
) (*RemoveResult, error) {
|
||||
@@ -1606,18 +1607,11 @@ func (v *Vaultik) executeRemoveAll(
|
||||
|
||||
result := &RemoveResult{}
|
||||
|
||||
remoteErrors := v.removeAllLocalSnapshots(localSnaps, opts, result)
|
||||
failures := v.removeAllLocalSnapshots(localSnaps, opts, result)
|
||||
|
||||
if !opts.LocalOnly {
|
||||
remoteErrors += v.removeAllOrphanRemotes(orphanRemoteKeys, result)
|
||||
|
||||
if remoteErrors == 0 {
|
||||
result.RemoteRemoved = true
|
||||
} else if v.UI != nil {
|
||||
v.UI.Warningf("Some remote metadata deletions failed. Run '%s' "+
|
||||
"once the remote is healthy to clean up unreferenced blobs.",
|
||||
pruneCommandHint)
|
||||
}
|
||||
failures += v.removeAllOrphanRemotes(orphanRemoteKeys, result)
|
||||
result.RemoteRemoved = failures == 0
|
||||
}
|
||||
|
||||
if v.SnapshotManager != nil {
|
||||
@@ -1627,6 +1621,10 @@ func (v *Vaultik) executeRemoveAll(
|
||||
}
|
||||
}
|
||||
|
||||
if failures > 0 {
|
||||
return result, fmt.Errorf("%d %w", failures, errSnapshotRemoveFailed)
|
||||
}
|
||||
|
||||
if opts.JSON {
|
||||
return result, v.outputRemoveJSON(result)
|
||||
}
|
||||
@@ -1635,9 +1633,6 @@ func (v *Vaultik) executeRemoveAll(
|
||||
|
||||
if !opts.LocalOnly && result.RemoteRemoved {
|
||||
v.printlnStdout("Removed snapshot metadata from remote storage")
|
||||
v.stdoutf("\nNote: Removed snapshots' blobs remain on the remote. "+
|
||||
"Run '%s' to delete any blobs no longer referenced by any "+
|
||||
"remaining remote snapshot.\n", pruneCommandHint)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
@@ -1680,12 +1675,13 @@ func (v *Vaultik) deleteIncompleteSnapshot(snapshotIDStr string, id any) bool {
|
||||
}
|
||||
|
||||
// removeAllLocalSnapshots removes each known local snapshot (and, unless
|
||||
// --local-only, its remote metadata), returning the number of remote
|
||||
// deletion failures.
|
||||
// --local-only, its remote metadata), returning the number of snapshots
|
||||
// it could not remove. A snapshot left in the local index keeps its
|
||||
// remote metadata too, since it is not an orphan remote key.
|
||||
func (v *Vaultik) removeAllLocalSnapshots(
|
||||
localSnaps []string, opts *RemoveOptions, result *RemoveResult,
|
||||
) int {
|
||||
remoteErrors := 0
|
||||
failures := 0
|
||||
|
||||
for _, snapshotID := range localSnaps {
|
||||
log.Info("Removing snapshot", "snapshot_id", snapshotID)
|
||||
@@ -1695,6 +1691,8 @@ func (v *Vaultik) removeAllLocalSnapshots(
|
||||
log.Error("Failed to remove from local database",
|
||||
"snapshot_id", snapshotID, "error", err)
|
||||
|
||||
failures++
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -1705,14 +1703,14 @@ func (v *Vaultik) removeAllLocalSnapshots(
|
||||
log.Warn("Failed to remove snapshot metadata from remote",
|
||||
"snapshot_id", snapshotID, "error", err)
|
||||
|
||||
remoteErrors++
|
||||
failures++
|
||||
}
|
||||
}
|
||||
|
||||
result.SnapshotsRemoved = append(result.SnapshotsRemoved, snapshotID)
|
||||
}
|
||||
|
||||
return remoteErrors
|
||||
return failures
|
||||
}
|
||||
|
||||
// removeAllOrphanRemotes deletes remote snapshot metadata that has no
|
||||
|
||||
@@ -260,6 +260,8 @@ func (v *Vaultik) downloadVerifiedSnapshotDB(
|
||||
fmt.Errorf("failed to download database: %w", err))
|
||||
}
|
||||
|
||||
dbReader = newCloseOnCancelReader(v.ctx, dbReader)
|
||||
|
||||
defer func() { _ = dbReader.Close() }()
|
||||
|
||||
tdb, err := v.decryptAndLoadDatabase(dbReader, identities)
|
||||
|
||||
Reference in New Issue
Block a user