Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2c4af6d0df | ||
|
|
42eb50b151 | ||
|
|
25ee9211d3 | ||
|
|
3361e9ebd9 | ||
|
|
fb8abcd049 | ||
|
|
6aceaaf086 |
+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
|
||||
|
||||
@@ -294,7 +294,8 @@ 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, recorded when its backup ran).
|
||||
were not stored before). Both chunk sizes are recorded when the
|
||||
snapshot's backup runs.
|
||||
|
||||
Listing the destination store does **not** require the age secret key,
|
||||
so it works in vaultik's intended configuration, where the backed-up
|
||||
@@ -338,7 +339,9 @@ 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).
|
||||
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.
|
||||
* `--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)
|
||||
@@ -375,7 +378,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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
+3
-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.
|
||||
@@ -122,6 +122,7 @@ 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
|
||||
@@ -184,7 +185,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
|
||||
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
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")
|
||||
}
|
||||
@@ -154,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 = ?
|
||||
`
|
||||
|
||||
@@ -168,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) {
|
||||
|
||||
@@ -89,6 +89,7 @@ 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,7 +120,11 @@ func (r *SnapshotRepository) UpdateCounts(
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateExtendedStats updates extended statistics for a snapshot
|
||||
// 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.
|
||||
func (r *SnapshotRepository) UpdateExtendedStats(
|
||||
ctx context.Context,
|
||||
tx *sql.Tx,
|
||||
@@ -145,7 +149,17 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
||||
compression_level = ?,
|
||||
upload_bytes = ?,
|
||||
upload_duration_ms = ?,
|
||||
new_chunk_size = ?
|
||||
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
|
||||
)
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
@@ -555,30 +569,22 @@ func (r *SnapshotRepository) GetSnapshotBlobSizes(
|
||||
return compressed, uncompressed, nil
|
||||
}
|
||||
|
||||
// GetSnapshotUncompressedChunkSize returns the sum of plaintext sizes of all unique
|
||||
// chunks referenced by a snapshot (via snapshot_files → file_chunks → chunks).
|
||||
// GetSnapshotUncompressedChunkSize returns the total plaintext size of the
|
||||
// distinct chunks the snapshot's files reference, as the snapshot's backup
|
||||
// recorded it (see UpdateExtendedStats).
|
||||
func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
||||
ctx context.Context, snapshotID string,
|
||||
) (int64, error) {
|
||||
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
|
||||
`
|
||||
query := `SELECT uncompressed_chunk_size FROM snapshots WHERE id = ?`
|
||||
|
||||
var totalSize int64
|
||||
var uncompressedChunkSize int64
|
||||
|
||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&totalSize)
|
||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&uncompressedChunkSize)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("querying uncompressed chunk size: %w", err)
|
||||
}
|
||||
|
||||
return totalSize, nil
|
||||
return uncompressedChunkSize, nil
|
||||
}
|
||||
|
||||
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
||||
|
||||
@@ -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,7 +1554,7 @@ 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 {
|
||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||
@@ -1562,7 +1563,8 @@ func (s *Scanner) handleBlobReady(
|
||||
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 {
|
||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||
|
||||
@@ -1587,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,
|
||||
@@ -1609,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.",
|
||||
@@ -1623,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)
|
||||
@@ -1654,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.
|
||||
@@ -1724,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 {
|
||||
@@ -1735,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)
|
||||
}
|
||||
|
||||
+27
-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).",
|
||||
@@ -1433,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
|
||||
@@ -1450,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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -627,7 +627,8 @@ func selectSnapshotsToPurge(
|
||||
}
|
||||
|
||||
// confirmAndExecutePurge shows deletion candidates, confirms with the
|
||||
// user, and deletes snapshots.
|
||||
// user, and deletes snapshots. It stops with an error at the first
|
||||
// snapshot it cannot delete.
|
||||
func (v *Vaultik) confirmAndExecutePurge(
|
||||
toDelete []SnapshotInfo, force, quiet bool,
|
||||
) error {
|
||||
@@ -669,10 +670,11 @@ 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 {
|
||||
log.Error("Failed to delete from local database",
|
||||
"snapshot_id", snapshotID, "error", err)
|
||||
return fmt.Errorf("deleting snapshot %s from local index: %w", snapshotID, err)
|
||||
}
|
||||
|
||||
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
||||
|
||||
@@ -483,13 +483,14 @@ 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"`
|
||||
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"`
|
||||
UncompressedSize int64 `json:"uncompressed_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
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user