Compare commits
6
Commits
4b391a4c98
...
2660f7320d
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2660f7320d | ||
|
|
e48bafccef | ||
|
|
bffd4578fc | ||
|
|
f4ab1ba5ae | ||
|
|
005927a7fc | ||
|
|
09a0c01c43 |
@@ -293,8 +293,8 @@ needed.
|
|||||||
**`snapshot list`**: Show every snapshot known to this host — the union
|
**`snapshot list`**: Show every snapshot known to this host — the union
|
||||||
of the local index and the backup destination store — with timestamps
|
of the local index and the backup destination store — with timestamps
|
||||||
and three sizes per snapshot (compressed remote size; total
|
and three sizes per snapshot (compressed remote size; total
|
||||||
uncompressed chunk size; size of chunks newly referenced by that
|
uncompressed chunk size; size of the chunks that snapshot stored that
|
||||||
snapshot).
|
were not stored before, recorded when its backup ran).
|
||||||
|
|
||||||
Listing the destination store does **not** require the age secret key,
|
Listing the destination store does **not** require the age secret key,
|
||||||
so it works in vaultik's intended configuration, where the backed-up
|
so it works in vaultik's intended configuration, where the backed-up
|
||||||
@@ -307,8 +307,8 @@ only the hash, so hostname and snapshot name exist solely in the local
|
|||||||
index and in the encrypted per-snapshot database. Snapshots found only
|
index and in the encrypted per-snapshot database. Snapshots found only
|
||||||
on the destination store are therefore listed as
|
on the destination store are therefore listed as
|
||||||
`<remote only:<abbreviated remote key>>` and show `<remote only>` in
|
`<remote only:<abbreviated remote key>>` and show `<remote only>` in
|
||||||
the uncompressed and "new chunk" columns, which can only be computed
|
the uncompressed and "new chunk" columns, which come only from the
|
||||||
from the local index. Their timestamp and compressed size are real,
|
local index. Their timestamp and compressed size are real,
|
||||||
read from the manifest.
|
read from the manifest.
|
||||||
|
|
||||||
Snapshots in the local index with no counterpart on the destination
|
Snapshots in the local index with no counterpart on the destination
|
||||||
|
|||||||
@@ -119,6 +119,7 @@ Tracks backup snapshots.
|
|||||||
- `completed_at` (INTEGER) - Completion timestamp (NULL if in progress)
|
- `completed_at` (INTEGER) - Completion timestamp (NULL if in progress)
|
||||||
- `file_count` (INTEGER) - Number of files in snapshot
|
- `file_count` (INTEGER) - Number of files in snapshot
|
||||||
- `chunk_count` (INTEGER) - Number of chunks this snapshot stored that were not stored before
|
- `chunk_count` (INTEGER) - Number of chunks this snapshot stored that were not stored before
|
||||||
|
- `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
|
- `blob_count` (INTEGER) - Number of blobs this snapshot created
|
||||||
- `total_size` (INTEGER) - Total size of all files
|
- `total_size` (INTEGER) - Total size of all files
|
||||||
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs
|
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package database
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -60,22 +61,24 @@ func (r *Repositories) WithTx(ctx context.Context, fn TxFunc) error {
|
|||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if p := recover(); p != nil {
|
if p := recover(); p != nil {
|
||||||
rollbackErr := tx.Rollback()
|
// The panic is the failure to report, so a rollback error
|
||||||
if rollbackErr != nil {
|
// is dropped.
|
||||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
_ = tx.Rollback()
|
||||||
}
|
|
||||||
|
|
||||||
panic(p)
|
panic(p)
|
||||||
} else if err != nil {
|
|
||||||
rollbackErr := tx.Rollback()
|
|
||||||
if rollbackErr != nil {
|
|
||||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
err = fn(ctx, tx)
|
err = fn(ctx, tx)
|
||||||
if err != nil {
|
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
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
"errors"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -235,6 +236,125 @@ func TestRepositoriesTransactionRollback(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// When the context is cancelled during the body, database/sql rolls the
|
||||||
|
// transaction back itself, so WithTx's own rollback finds it already
|
||||||
|
// done. WithTx must return the body's error alone, without that rollback.
|
||||||
|
func TestRepositoriesTransactionContextCancelled(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.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
file := &database.File{
|
||||||
|
Path: "/test/cancelled_file.txt",
|
||||||
|
MTime: time.Now().Truncate(time.Second),
|
||||||
|
Size: 1024,
|
||||||
|
Mode: 0644,
|
||||||
|
UID: 1000,
|
||||||
|
GID: 1000,
|
||||||
|
}
|
||||||
|
|
||||||
|
err := repos.Files.Create(ctx, tx, file)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
file, err := repos.Files.GetByPath(
|
||||||
|
context.Background(), "/test/cancelled_file.txt",
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("error checking for file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if file != nil {
|
||||||
|
t.Error("file should not exist after the cancelled transaction")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A body that ends the transaction itself with ROLLBACK makes WithTx's
|
||||||
|
// own rollback fail. WithTx must return that failure together with the
|
||||||
|
// body's error.
|
||||||
|
func TestRepositoriesTransactionRollbackFails(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
err := repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
_, err := tx.ExecContext(ctx, "ROLLBACK")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return errIntentionalRollback
|
||||||
|
})
|
||||||
|
if !errors.Is(err, errIntentionalRollback) {
|
||||||
|
t.Fatalf("expected the body's error, got: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !strings.Contains(err.Error(), "rolling back transaction") {
|
||||||
|
t.Errorf("expected the failed rollback in the error, got: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// When the body ends the transaction itself with ROLLBACK and then
|
||||||
|
// panics, WithTx's own rollback fails; the panic must still reach the
|
||||||
|
// caller.
|
||||||
|
func TestRepositoriesTransactionPanicRollbackFails(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
if recover() == nil {
|
||||||
|
t.Error("expected the body's panic to reach the caller")
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
_ = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
_, err := tx.ExecContext(ctx, "ROLLBACK")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
panic(errIntentionalRollback)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestRepositoriesReadTransaction(t *testing.T) {
|
func TestRepositoriesReadTransaction(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -86,6 +86,7 @@ CREATE TABLE IF NOT EXISTS snapshots (
|
|||||||
completed_at INTEGER,
|
completed_at INTEGER,
|
||||||
file_count INTEGER NOT NULL DEFAULT 0,
|
file_count INTEGER NOT NULL DEFAULT 0,
|
||||||
chunk_count INTEGER NOT NULL DEFAULT 0,
|
chunk_count INTEGER NOT NULL DEFAULT 0,
|
||||||
|
new_chunk_size INTEGER NOT NULL DEFAULT 0,
|
||||||
blob_count INTEGER NOT NULL DEFAULT 0,
|
blob_count INTEGER NOT NULL DEFAULT 0,
|
||||||
total_size INTEGER NOT NULL DEFAULT 0,
|
total_size INTEGER NOT NULL DEFAULT 0,
|
||||||
blob_size INTEGER NOT NULL DEFAULT 0,
|
blob_size INTEGER NOT NULL DEFAULT 0,
|
||||||
|
|||||||
@@ -129,6 +129,7 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
|||||||
compressionLevel int,
|
compressionLevel int,
|
||||||
uploadBytes int64,
|
uploadBytes int64,
|
||||||
uploadDurationMs int64,
|
uploadDurationMs int64,
|
||||||
|
newChunkSize int64,
|
||||||
) error {
|
) error {
|
||||||
compressionRatio, err := r.extendedCompressionRatio(
|
compressionRatio, err := r.extendedCompressionRatio(
|
||||||
ctx, tx, snapshotID, blobUncompressedSize,
|
ctx, tx, snapshotID, blobUncompressedSize,
|
||||||
@@ -143,18 +144,19 @@ func (r *SnapshotRepository) UpdateExtendedStats(
|
|||||||
compression_ratio = ?,
|
compression_ratio = ?,
|
||||||
compression_level = ?,
|
compression_level = ?,
|
||||||
upload_bytes = ?,
|
upload_bytes = ?,
|
||||||
upload_duration_ms = ?
|
upload_duration_ms = ?,
|
||||||
|
new_chunk_size = ?
|
||||||
WHERE id = ?
|
WHERE id = ?
|
||||||
`
|
`
|
||||||
|
|
||||||
if tx != nil {
|
if tx != nil {
|
||||||
_, err = tx.ExecContext(ctx, query,
|
_, err = tx.ExecContext(ctx, query,
|
||||||
blobUncompressedSize, compressionRatio, compressionLevel,
|
blobUncompressedSize, compressionRatio, compressionLevel,
|
||||||
uploadBytes, uploadDurationMs, snapshotID)
|
uploadBytes, uploadDurationMs, newChunkSize, snapshotID)
|
||||||
} else {
|
} else {
|
||||||
_, err = r.db.ExecWithLog(ctx, query,
|
_, err = r.db.ExecWithLog(ctx, query,
|
||||||
blobUncompressedSize, compressionRatio, compressionLevel,
|
blobUncompressedSize, compressionRatio, compressionLevel,
|
||||||
uploadBytes, uploadDurationMs, snapshotID)
|
uploadBytes, uploadDurationMs, newChunkSize, snapshotID)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -594,45 +596,24 @@ func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
|||||||
return totalSize, nil
|
return totalSize, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetSnapshotNewChunkSize returns the sum of plaintext sizes of chunks that are
|
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
||||||
// referenced by this snapshot but not by any earlier completed snapshot known to
|
// snapshot's backup stored that were not stored before, as the backup
|
||||||
// the local database. The result is the marginal uncompressed data this snapshot
|
// recorded it. It is not derived from the snapshot's files: a file's row is
|
||||||
// added to the dedup pool — i.e., the delta from prior snapshots.
|
// shared by every snapshot that lists it and is rewritten when the file
|
||||||
|
// changes, so it shows only the file's latest chunks.
|
||||||
func (r *SnapshotRepository) GetSnapshotNewChunkSize(
|
func (r *SnapshotRepository) GetSnapshotNewChunkSize(
|
||||||
ctx context.Context, snapshotID string,
|
ctx context.Context, snapshotID string,
|
||||||
) (int64, error) {
|
) (int64, error) {
|
||||||
query := `
|
query := `SELECT new_chunk_size FROM snapshots WHERE id = ?`
|
||||||
WITH this_snap_chunks AS (
|
|
||||||
SELECT DISTINCT fc.chunk_hash
|
|
||||||
FROM snapshot_files sf
|
|
||||||
JOIN file_chunks fc ON sf.file_id = fc.file_id
|
|
||||||
WHERE sf.snapshot_id = ?
|
|
||||||
),
|
|
||||||
prior_chunks AS (
|
|
||||||
SELECT DISTINCT fc.chunk_hash
|
|
||||||
FROM snapshots s
|
|
||||||
JOIN snapshot_files sf ON sf.snapshot_id = s.id
|
|
||||||
JOIN file_chunks fc ON fc.file_id = sf.file_id
|
|
||||||
WHERE s.completed_at IS NOT NULL
|
|
||||||
AND s.id != ?
|
|
||||||
AND s.started_at < (SELECT started_at FROM snapshots WHERE id = ?)
|
|
||||||
)
|
|
||||||
SELECT COALESCE(SUM(c.size), 0)
|
|
||||||
FROM chunks c
|
|
||||||
JOIN this_snap_chunks t ON c.chunk_hash = t.chunk_hash
|
|
||||||
WHERE c.chunk_hash NOT IN (SELECT chunk_hash FROM prior_chunks)
|
|
||||||
`
|
|
||||||
|
|
||||||
var totalSize int64
|
var newChunkSize int64
|
||||||
|
|
||||||
err := r.db.conn.QueryRowContext(
|
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&newChunkSize)
|
||||||
ctx, query, snapshotID, snapshotID, snapshotID,
|
|
||||||
).Scan(&totalSize)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, fmt.Errorf("querying new chunk size: %w", err)
|
return 0, fmt.Errorf("querying new chunk size: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return totalSize, nil
|
return newChunkSize, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetIncompleteSnapshots returns all snapshots that haven't been completed
|
// GetIncompleteSnapshots returns all snapshots that haven't been completed
|
||||||
|
|||||||
@@ -10,6 +10,8 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
"unicode"
|
"unicode"
|
||||||
|
|
||||||
|
"github.com/dustin/go-humanize"
|
||||||
)
|
)
|
||||||
|
|
||||||
// groupSeparator joins an open group path to an attribute key. This
|
// groupSeparator joins an open group path to an attribute key. This
|
||||||
@@ -304,18 +306,13 @@ func formatDuration(d time.Duration) string {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// formatBytes formats bytes in a human-readable way
|
// formatBytes formats a byte count with humanize.Bytes, as every command
|
||||||
|
// prints one. humanize.Bytes takes no negative count, so one is printed
|
||||||
|
// as it is.
|
||||||
func formatBytes(b int64) string {
|
func formatBytes(b int64) string {
|
||||||
const unit = 1024
|
if b < 0 {
|
||||||
if b < unit {
|
|
||||||
return fmt.Sprintf("%d B", b)
|
return fmt.Sprintf("%d B", b)
|
||||||
}
|
}
|
||||||
|
|
||||||
div, exp := int64(unit), 0
|
return humanize.Bytes(uint64(b))
|
||||||
for n := b / unit; n >= unit; n /= unit {
|
|
||||||
div *= unit
|
|
||||||
exp++
|
|
||||||
}
|
|
||||||
|
|
||||||
return fmt.Sprintf("%.1f %cB", float64(b)/float64(div), "KMGTPE"[exp])
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -200,7 +200,8 @@ func TestTTYHandlerWithGroupQualifiesKeys(t *testing.T) {
|
|||||||
func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
|
func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
const oneAndAHalfKiB = 1536
|
// Printed as "1.0 MB", in powers of 1000 like every other byte count.
|
||||||
|
const oneMegabyte = 1_000_000
|
||||||
|
|
||||||
for name, testCase := range map[string]struct {
|
for name, testCase := range map[string]struct {
|
||||||
derive func(*slog.Logger) *slog.Logger
|
derive func(*slog.Logger) *slog.Logger
|
||||||
@@ -223,13 +224,13 @@ func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
|
|||||||
var buf bytes.Buffer
|
var buf bytes.Buffer
|
||||||
|
|
||||||
logger := slog.New(log.NewTTYHandler(&buf, debugHandlerOptions()))
|
logger := slog.New(log.NewTTYHandler(&buf, debugHandlerOptions()))
|
||||||
testCase.derive(logger).Info("uploaded", "bytes", oneAndAHalfKiB)
|
testCase.derive(logger).Info("uploaded", "bytes", oneMegabyte)
|
||||||
|
|
||||||
line := ansiEscape.ReplaceAllString(buf.String(), "")
|
line := ansiEscape.ReplaceAllString(buf.String(), "")
|
||||||
|
|
||||||
assert.Contains(t, line, testCase.key+"=1.5 KB",
|
assert.Contains(t, line, testCase.key+"=1.0 MB",
|
||||||
"a byte count must be human-readable however it is qualified")
|
"a byte count must be human-readable however it is qualified")
|
||||||
assert.NotContains(t, line, strconv.Itoa(oneAndAHalfKiB),
|
assert.NotContains(t, line, strconv.Itoa(oneMegabyte),
|
||||||
"the raw number must not survive the formatting")
|
"the raw number must not survive the formatting")
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -135,7 +135,9 @@ type ScannerConfig struct {
|
|||||||
// are counted per file: BytesScanned is the size of the new and changed
|
// are counted per file: BytesScanned is the size of the new and changed
|
||||||
// files, BytesSkipped that of the unchanged ones. FilesFailed counts the
|
// files, BytesSkipped that of the unchanged ones. FilesFailed counts the
|
||||||
// new and changed files that phase 2 could not store; FilesScanned
|
// new and changed files that phase 2 could not store; FilesScanned
|
||||||
// includes them and BytesScanned does not.
|
// includes them and BytesScanned does not. ChunksCreated counts the
|
||||||
|
// chunks that were not already stored, and NewChunkSize is their total
|
||||||
|
// size.
|
||||||
type ScanResult struct {
|
type ScanResult struct {
|
||||||
FilesScanned int
|
FilesScanned int
|
||||||
FilesSkipped int
|
FilesSkipped int
|
||||||
@@ -145,6 +147,7 @@ type ScanResult struct {
|
|||||||
BytesSkipped int64
|
BytesSkipped int64
|
||||||
BytesDeleted int64
|
BytesDeleted int64
|
||||||
ChunksCreated int
|
ChunksCreated int
|
||||||
|
NewChunkSize int64
|
||||||
BlobsCreated int
|
BlobsCreated int
|
||||||
BlobsUploaded int
|
BlobsUploaded int
|
||||||
BytesUploaded int64
|
BytesUploaded int64
|
||||||
@@ -1866,11 +1869,12 @@ func (s *Scanner) processFileStreaming(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// updateChunkStats counts a chunk that was not already stored. The scan
|
// updateChunkStats counts a chunk that was not already stored, and its
|
||||||
// result's file counts, BytesScanned and BytesSkipped are not touched
|
// size. The scan result's file counts, BytesScanned and BytesSkipped are
|
||||||
// here: the scan phase counts each file once.
|
// not touched here: the scan phase counts each file once.
|
||||||
func (s *Scanner) updateChunkStats(chunkSize int64, result *ScanResult) {
|
func (s *Scanner) updateChunkStats(chunkSize int64, result *ScanResult) {
|
||||||
result.ChunksCreated++
|
result.ChunksCreated++
|
||||||
|
result.NewChunkSize += chunkSize
|
||||||
|
|
||||||
if s.progress != nil {
|
if s.progress != nil {
|
||||||
s.progress.GetStats().ChunksCreated.Add(1)
|
s.progress.GetStats().ChunksCreated.Add(1)
|
||||||
|
|||||||
@@ -199,6 +199,7 @@ func (sm *SnapshotManager) UpdateSnapshotStatsExtended(
|
|||||||
stats.CompressionLevel,
|
stats.CompressionLevel,
|
||||||
stats.BytesUploaded,
|
stats.BytesUploaded,
|
||||||
stats.UploadDurationMs,
|
stats.UploadDurationMs,
|
||||||
|
stats.NewChunkSize,
|
||||||
)
|
)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -906,6 +907,7 @@ type ExtendedBackupStats struct {
|
|||||||
BlobUncompressedSize int64 // Total uncompressed size of all referenced blobs
|
BlobUncompressedSize int64 // Total uncompressed size of all referenced blobs
|
||||||
CompressionLevel int // Compression level used for this snapshot
|
CompressionLevel int // Compression level used for this snapshot
|
||||||
UploadDurationMs int64 // Total milliseconds spent uploading to S3
|
UploadDurationMs int64 // Total milliseconds spent uploading to S3
|
||||||
|
NewChunkSize int64 // Total size of the chunks counted in ChunksCreated
|
||||||
}
|
}
|
||||||
|
|
||||||
// deleteOtherSnapshots deletes all snapshots except the current one
|
// deleteOtherSnapshots deletes all snapshots except the current one
|
||||||
|
|||||||
@@ -103,6 +103,8 @@ func (v *Vaultik) FetchAndDecryptBlob(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// FetchBlob downloads a blob and returns a reader for the encrypted data.
|
// FetchBlob downloads a blob and returns a reader for the encrypted data.
|
||||||
|
// The reader is closed as soon as ctx is cancelled, which ends a read
|
||||||
|
// blocked inside it.
|
||||||
func (v *Vaultik) FetchBlob(
|
func (v *Vaultik) FetchBlob(
|
||||||
ctx context.Context, blobHash string,
|
ctx context.Context, blobHash string,
|
||||||
) (io.ReadCloser, error) {
|
) (io.ReadCloser, error) {
|
||||||
@@ -120,5 +122,47 @@ func (v *Vaultik) FetchBlob(
|
|||||||
return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err)
|
return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return rc, nil
|
return newCloseOnCancelReader(ctx, rc), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// closeOnCancelReader closes the reader it wraps when ctx is cancelled.
|
||||||
|
// Not every backend's reader stops on cancel by itself (the file
|
||||||
|
// backend's does not), and a read stalled on a slow or hung source would
|
||||||
|
// otherwise outlast the shutdown timeout, so the process would exit
|
||||||
|
// before restore or deep verify removed its decrypted temporary files.
|
||||||
|
type closeOnCancelReader struct {
|
||||||
|
io.ReadCloser
|
||||||
|
|
||||||
|
ctx context.Context //nolint:containedctx // Read reports its error
|
||||||
|
stop func() bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newCloseOnCancelReader(
|
||||||
|
ctx context.Context, rc io.ReadCloser,
|
||||||
|
) *closeOnCancelReader {
|
||||||
|
return &closeOnCancelReader{
|
||||||
|
ReadCloser: rc,
|
||||||
|
ctx: ctx,
|
||||||
|
stop: context.AfterFunc(ctx, func() { _ = rc.Close() }),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read returns the context's error, not the closed reader's, once ctx
|
||||||
|
// has been cancelled.
|
||||||
|
func (r *closeOnCancelReader) Read(p []byte) (int, error) {
|
||||||
|
n, err := r.ReadCloser.Read(p)
|
||||||
|
if err != nil && r.ctx.Err() != nil {
|
||||||
|
return n, r.ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close closes the wrapped reader, unless the cancel already closed it.
|
||||||
|
func (r *closeOnCancelReader) Close() error {
|
||||||
|
if !r.stop() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return r.ReadCloser.Close()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,15 +1,20 @@
|
|||||||
package vaultik_test
|
package vaultik_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
|
"crypto/rand"
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/spf13/afero"
|
"github.com/spf13/afero"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
)
|
)
|
||||||
@@ -77,6 +82,161 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
|
|||||||
"deep verify should fail on a corrupted blob")
|
"deep verify should fail on a corrupted blob")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestDeepVerifyJSONCountsBlobs checks the counts that deep verify
|
||||||
|
// reports under --json: on a healthy snapshot, with a blob that cannot
|
||||||
|
// be checked, with a damaged blob, with missing blobs, and with a blob
|
||||||
|
// of the wrong size. Deep verify stops at the first damaged blob, so the
|
||||||
|
// blobs read before it are the ones verified; when blobs are missing or
|
||||||
|
// the wrong size it reads no blob contents, but counts every such blob.
|
||||||
|
func TestDeepVerifyJSONCountsBlobs(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")
|
||||||
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||||
|
|
||||||
|
chunkSize := int64(32 * 1024)
|
||||||
|
maxBlobSize := int64(128 * 1024)
|
||||||
|
|
||||||
|
// Random data does not compress or deduplicate, so it fills several
|
||||||
|
// blobs.
|
||||||
|
data := make([]byte, maxBlobSize*4)
|
||||||
|
_, err := rand.Read(data)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(dataDir, "data.bin"), data, 0o644))
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
cfg, storer, snapshotID := runFileStorageBackup(
|
||||||
|
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
||||||
|
|
||||||
|
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, "ok", healthy.Status)
|
||||||
|
require.GreaterOrEqual(t, healthy.BlobCount, 3)
|
||||||
|
require.Equal(t, healthy.BlobCount, healthy.Verified)
|
||||||
|
require.Zero(t, healthy.Missing)
|
||||||
|
|
||||||
|
// Deep verify reads blobs in hash order, which is the sorted order of
|
||||||
|
// their storage keys.
|
||||||
|
keys, err := storer.List(ctx, "blobs/")
|
||||||
|
require.NoError(t, err)
|
||||||
|
slices.Sort(keys)
|
||||||
|
require.Len(t, keys, healthy.BlobCount)
|
||||||
|
|
||||||
|
// A blob that is present but cannot be checked is not missing: the
|
||||||
|
// check stops with the storage error.
|
||||||
|
denied := &statDeniedStorer{Storer: storer, deniedKey: keys[0]}
|
||||||
|
|
||||||
|
got, err := deepVerifyJSON(ctx, t, cfg, denied, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Zero(t, got.Missing)
|
||||||
|
require.Contains(t, got.ErrorMessage, os.ErrPermission.Error())
|
||||||
|
|
||||||
|
// Damage the last blob read, keeping its length, so every blob before
|
||||||
|
// it is read and passes first.
|
||||||
|
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
|
||||||
|
blob, err := afero.ReadFile(fs, lastPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
blob[len(blob)/2] ^= 0xff
|
||||||
|
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
|
||||||
|
|
||||||
|
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Equal(t, healthy.BlobCount-1, got.Verified)
|
||||||
|
require.Zero(t, got.Missing)
|
||||||
|
|
||||||
|
// Delete the first blob and the damaged last one: both must be
|
||||||
|
// counted as missing, not only the first one found.
|
||||||
|
var deletedSize int64
|
||||||
|
|
||||||
|
for _, key := range []string{keys[0], keys[len(keys)-1]} {
|
||||||
|
info, statErr := storer.Stat(ctx, key)
|
||||||
|
require.NoError(t, statErr)
|
||||||
|
|
||||||
|
deletedSize += info.Size
|
||||||
|
|
||||||
|
require.NoError(t, storer.Delete(ctx, key))
|
||||||
|
}
|
||||||
|
|
||||||
|
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Equal(t, 2, got.Missing)
|
||||||
|
require.Equal(t, deletedSize, got.MissingSize)
|
||||||
|
require.Zero(t, got.Verified)
|
||||||
|
|
||||||
|
// Make a blob between the two deleted ones longer: it is counted as
|
||||||
|
// the wrong size, and both missing blobs are still counted.
|
||||||
|
grownPath := filepath.Join(storeDir, keys[1])
|
||||||
|
blob, err = afero.ReadFile(fs, grownPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
blob = append(blob, []byte("extra")...)
|
||||||
|
require.NoError(t, afero.WriteFile(fs, grownPath, blob, 0o600))
|
||||||
|
|
||||||
|
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Equal(t, 1, got.Mismatched)
|
||||||
|
require.Equal(t, 2, got.Missing)
|
||||||
|
require.Zero(t, got.Verified)
|
||||||
|
}
|
||||||
|
|
||||||
|
// statDeniedStorer passes every call through to the wrapped Storer,
|
||||||
|
// except that Stat of deniedKey fails with a permission error, as it
|
||||||
|
// does for a blob that is present but cannot be read.
|
||||||
|
type statDeniedStorer struct {
|
||||||
|
storage.Storer
|
||||||
|
|
||||||
|
deniedKey string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *statDeniedStorer) Stat(
|
||||||
|
ctx context.Context, key string,
|
||||||
|
) (*storage.ObjectInfo, error) {
|
||||||
|
if key == s.deniedKey {
|
||||||
|
return nil, os.ErrPermission
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.Storer.Stat(ctx, key)
|
||||||
|
}
|
||||||
|
|
||||||
|
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
|
||||||
|
// the decoded result and the error RunDeepVerify returned.
|
||||||
|
func deepVerifyJSON(
|
||||||
|
ctx context.Context, t *testing.T,
|
||||||
|
cfg *config.Config, storer storage.Storer, snapshotID string,
|
||||||
|
) (vaultik.VerifyResult, error) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
var out bytes.Buffer
|
||||||
|
|
||||||
|
v := &vaultik.Vaultik{
|
||||||
|
Config: cfg,
|
||||||
|
Storage: storer,
|
||||||
|
Fs: afero.NewOsFs(),
|
||||||
|
Stdout: &out,
|
||||||
|
Stderr: io.Discard,
|
||||||
|
UI: ui.NewWithColor(io.Discard, false),
|
||||||
|
}
|
||||||
|
v.SetContext(ctx)
|
||||||
|
|
||||||
|
err := v.RunDeepVerify(snapshotID,
|
||||||
|
&vaultik.VerifyOptions{Deep: true, JSON: true})
|
||||||
|
|
||||||
|
return decodeVerifyResult(t, out.Bytes()), err
|
||||||
|
}
|
||||||
|
|
||||||
// corruptOneBlob flips a middle byte of the first blob file found under
|
// corruptOneBlob flips a middle byte of the first blob file found under
|
||||||
// blobsDir, leaving the file length unchanged.
|
// blobsDir, leaving the file length unchanged.
|
||||||
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||||
|
|||||||
@@ -89,22 +89,6 @@ type SnapshotInfo struct {
|
|||||||
RemotePresent *bool `json:"remote_present"`
|
RemotePresent *bool `json:"remote_present"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// formatBytes formats bytes in a human-readable format
|
|
||||||
func formatBytes(bytes int64) string {
|
|
||||||
const unit = 1024
|
|
||||||
if bytes < unit {
|
|
||||||
return fmt.Sprintf("%d B", bytes)
|
|
||||||
}
|
|
||||||
|
|
||||||
div, exp := int64(unit), 0
|
|
||||||
for n := bytes / unit; n >= unit; n /= unit {
|
|
||||||
div *= unit
|
|
||||||
exp++
|
|
||||||
}
|
|
||||||
|
|
||||||
return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp])
|
|
||||||
}
|
|
||||||
|
|
||||||
// parseSnapshotTimestamp extracts the timestamp from a snapshot ID
|
// parseSnapshotTimestamp extracts the timestamp from a snapshot ID
|
||||||
// Format: hostname_snapshotname_2026-01-12T14:41:15Z
|
// Format: hostname_snapshotname_2026-01-12T14:41:15Z
|
||||||
func parseSnapshotTimestamp(snapshotID string) (time.Time, error) {
|
func parseSnapshotTimestamp(snapshotID string) (time.Time, error) {
|
||||||
|
|||||||
@@ -0,0 +1,133 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"maps"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/chunker"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestNewChunkSizeOfChangedFileStaysFixed covers
|
||||||
|
// https://git.eeqj.de/sneak/vaultik/issues/311. A backup whose only change
|
||||||
|
// is a modified file lists the size of the chunks that change added as its
|
||||||
|
// NEW CHUNK SIZE, and a later backup that changes the file again leaves
|
||||||
|
// every earlier snapshot's figure as it was. The file's row in the local
|
||||||
|
// index is shared by all three snapshots and rewritten by each backup.
|
||||||
|
func TestNewChunkSizeOfChangedFileStaysFixed(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"))
|
||||||
|
|
||||||
|
firstFiles := maps.Clone(files)
|
||||||
|
firstID := localSnapshotID(ctx, t, repos, "first")
|
||||||
|
|
||||||
|
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||||
|
require.NoError(t, backUp(v, "second"))
|
||||||
|
|
||||||
|
secondID := localSnapshotID(ctx, t, repos, "second")
|
||||||
|
secondAdded := sizeOfNewChunks(t, files, firstFiles)
|
||||||
|
require.Positive(t, secondAdded)
|
||||||
|
assert.Equal(t, secondAdded, listedNewChunkSizes(t, v)[secondID],
|
||||||
|
"the second snapshot must count the chunks the change added")
|
||||||
|
|
||||||
|
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
|
||||||
|
require.NoError(t, backUp(v, "third"))
|
||||||
|
|
||||||
|
sizes := listedNewChunkSizes(t, v)
|
||||||
|
assert.Equal(t, sizeOfNewChunks(t, firstFiles, nil), sizes[firstID],
|
||||||
|
"a later backup must not change the first snapshot's figure")
|
||||||
|
assert.Equal(t, secondAdded, sizes[secondID],
|
||||||
|
"a later backup must not change the second snapshot's figure")
|
||||||
|
}
|
||||||
|
|
||||||
|
// sizeOfNewChunks returns the total size of the distinct chunks of the
|
||||||
|
// files in after that no file in before has, cut the way the backup cuts
|
||||||
|
// them.
|
||||||
|
func sizeOfNewChunks(t *testing.T, after, before map[string][]byte) int64 {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
known := make(map[string]bool)
|
||||||
|
for _, chunk := range chunksOf(t, before) {
|
||||||
|
known[chunk.Hash] = true
|
||||||
|
}
|
||||||
|
|
||||||
|
var total int64
|
||||||
|
|
||||||
|
for _, chunk := range chunksOf(t, after) {
|
||||||
|
if !known[chunk.Hash] {
|
||||||
|
known[chunk.Hash] = true
|
||||||
|
total += chunk.Size
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return total
|
||||||
|
}
|
||||||
|
|
||||||
|
// chunksOf returns the chunks of every file in files.
|
||||||
|
func chunksOf(t *testing.T, files map[string][]byte) []chunker.Chunk {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
c := chunker.NewChunker(faultChunkSize)
|
||||||
|
|
||||||
|
var all []chunker.Chunk
|
||||||
|
|
||||||
|
for _, content := range files {
|
||||||
|
chunks, err := c.ChunkReader(bytes.NewReader(content))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
all = append(all, chunks...)
|
||||||
|
}
|
||||||
|
|
||||||
|
return all
|
||||||
|
}
|
||||||
|
|
||||||
|
// listedNewChunkSizes runs `snapshot list --json` and returns each listed
|
||||||
|
// snapshot's new_chunk_size by snapshot ID.
|
||||||
|
func listedNewChunkSizes(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.NewChunkSize
|
||||||
|
}
|
||||||
|
|
||||||
|
return sizes
|
||||||
|
}
|
||||||
@@ -309,11 +309,22 @@ func (v *Vaultik) restoreAllFiles(
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
plan, err := newRestorePlan(v.ctx, repos, files, chunkToBlobMap, blobIDToHash)
|
plan, unplanned, err := newRestorePlan(
|
||||||
|
v.ctx, repos, files, chunkToBlobMap, blobIDToHash)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("building restore plan: %w", err)
|
return nil, fmt.Errorf("building restore plan: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Ranging over files rather than the map reports them in a stable order.
|
||||||
|
for _, file := range files {
|
||||||
|
if planErr, ok := unplanned[file.ID]; ok {
|
||||||
|
err = v.handleRestoreFileError(plan, opts, result, file, file.ID, planErr)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
filesByID, totalBytesExpected := indexRestoreFiles(files)
|
filesByID, totalBytesExpected := indexRestoreFiles(files)
|
||||||
|
|
||||||
v.UI.Beginf("Restoring %s files (%s) to %s.",
|
v.UI.Beginf("Restoring %s files (%s) to %s.",
|
||||||
|
|||||||
@@ -142,14 +142,14 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// scratchEntries returns the vaultik blob-cache and snapshot-database
|
// scratchEntries returns the vaultik blob-cache and snapshot-database
|
||||||
// scratch entries currently present in dir.
|
// scratch entries (restore's and deep verify's) currently present in dir.
|
||||||
func scratchEntries(t *testing.T, dir string) []string {
|
func scratchEntries(t *testing.T, dir string) []string {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
var matches []string
|
var matches []string
|
||||||
|
|
||||||
for _, pattern := range []string{
|
for _, pattern := range []string{
|
||||||
"vaultik-blobcache-*", "vaultik-restore-*",
|
"vaultik-blobcache-*", "vaultik-restore-*", "vaultik-verify-*",
|
||||||
} {
|
} {
|
||||||
found, err := filepath.Glob(filepath.Join(dir, pattern))
|
found, err := filepath.Glob(filepath.Join(dir, pattern))
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -206,3 +206,119 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
|
|||||||
assert.NotContains(t, out.String(), srcPath,
|
assert.NotContains(t, out.String(), srcPath,
|
||||||
"the cancel was reported as a failed file")
|
"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 {
|
||||||
|
storage.Storer
|
||||||
|
|
||||||
|
once sync.Once
|
||||||
|
stalled chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
|
||||||
|
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *stallingBlobStorer) Get(
|
||||||
|
ctx context.Context, key string,
|
||||||
|
) (io.ReadCloser, error) {
|
||||||
|
rc, err := s.Storer.Get(ctx, key)
|
||||||
|
if err != nil || !strings.HasPrefix(key, "blobs/") {
|
||||||
|
return rc, err
|
||||||
|
}
|
||||||
|
|
||||||
|
blob, err := io.ReadAll(rc)
|
||||||
|
_ = rc.Close()
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// The writer is never closed, so once the first half has been read
|
||||||
|
// the next Read blocks until the reader is closed.
|
||||||
|
pr, pw := io.Pipe()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
_, _ = pw.Write(blob[:len(blob)/2])
|
||||||
|
|
||||||
|
s.once.Do(func() { close(s.stalled) })
|
||||||
|
}()
|
||||||
|
|
||||||
|
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).
|
||||||
|
//
|
||||||
|
// 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) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
||||||
|
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
||||||
|
return v.Restore(&RestoreOptions{
|
||||||
|
SnapshotID: snapshotID,
|
||||||
|
TargetDir: targetDir,
|
||||||
|
})
|
||||||
|
},
|
||||||
|
"deep verify": func(v *Vaultik, snapshotID, _ string) error {
|
||||||
|
return v.RunDeepVerify(snapshotID, &VerifyOptions{Deep: true})
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for name, op := range operations {
|
||||||
|
t.Run(name, 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)
|
||||||
|
|
||||||
|
scratch := filepath.Join(root, "scratch")
|
||||||
|
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
||||||
|
t.Setenv("TMPDIR", scratch)
|
||||||
|
|
||||||
|
stalling := newStallingBlobStorer(storer)
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
v := newRestoreVaultik(ctx, cfg, stalling, fs)
|
||||||
|
v.cancel = cancel
|
||||||
|
|
||||||
|
var opErr error
|
||||||
|
|
||||||
|
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 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")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -61,8 +61,8 @@ func craftedFile(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestRestoreShortChunkHashInNoBlob proves a file whose short chunk hash
|
// TestRestoreShortChunkHashInNoBlob proves a file whose short chunk hash
|
||||||
// has no blob_chunks row fails restore planning and the chunk write with
|
// has no blob_chunks row is left out of the restore plan with an error,
|
||||||
// an error.
|
// and fails the chunk write with an error.
|
||||||
func TestRestoreShortChunkHashInNoBlob(t *testing.T) {
|
func TestRestoreShortChunkHashInNoBlob(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -77,9 +77,11 @@ func TestRestoreShortChunkHashInNoBlob(t *testing.T) {
|
|||||||
chunkToBlobMap, err := v.buildChunkToBlobMap(ctx, repos)
|
chunkToBlobMap, err := v.buildChunkToBlobMap(ctx, repos)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
_, err = newRestorePlan(ctx, repos, []*database.File{file},
|
plan, unplanned, err := newRestorePlan(ctx, repos, []*database.File{file},
|
||||||
chunkToBlobMap, map[string]string{})
|
chunkToBlobMap, map[string]string{})
|
||||||
require.ErrorIs(t, err, errPlanChunkMissing)
|
require.NoError(t, err)
|
||||||
|
require.ErrorIs(t, unplanned[file.ID], errPlanChunkMissing)
|
||||||
|
require.False(t, plan.hasPending())
|
||||||
|
|
||||||
fileChunks, err := repos.FileChunks.GetByFileID(ctx, file.ID)
|
fileChunks, err := repos.FileChunks.GetByFileID(ctx, file.ID)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|||||||
@@ -41,20 +41,25 @@ type restorePlan struct {
|
|||||||
|
|
||||||
// newRestorePlan builds the file→blob index for the given files. Files
|
// newRestorePlan builds the file→blob index for the given files. Files
|
||||||
// whose chunks reference no blobs (symlinks, directories) start in the
|
// whose chunks reference no blobs (symlinks, directories) start in the
|
||||||
// ready queue immediately.
|
// ready queue immediately. A file with a chunk that no blob holds cannot
|
||||||
|
// be restored: it is left out of the plan and returned, by ID, with the
|
||||||
|
// reason.
|
||||||
func newRestorePlan(
|
func newRestorePlan(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
repos *database.Repositories,
|
repos *database.Repositories,
|
||||||
files []*database.File,
|
files []*database.File,
|
||||||
chunkToBlobMap map[string]*database.BlobChunk,
|
chunkToBlobMap map[string]*database.BlobChunk,
|
||||||
blobIDToHash map[string]string,
|
blobIDToHash map[string]string,
|
||||||
) (*restorePlan, error) {
|
) (*restorePlan, map[types.FileID]error, error) {
|
||||||
p := &restorePlan{
|
p := &restorePlan{
|
||||||
fileBlobs: make(map[types.FileID]map[string]struct{}, len(files)),
|
fileBlobs: make(map[types.FileID]map[string]struct{}, len(files)),
|
||||||
blobFiles: make(map[string]map[types.FileID]struct{}),
|
blobFiles: make(map[string]map[types.FileID]struct{}),
|
||||||
ready: make([]types.FileID, 0, len(files)),
|
ready: make([]types.FileID, 0, len(files)),
|
||||||
cached: make(map[string]struct{}),
|
cached: make(map[string]struct{}),
|
||||||
}
|
}
|
||||||
|
unplanned := make(map[types.FileID]error)
|
||||||
|
|
||||||
|
nextFile:
|
||||||
for _, f := range files {
|
for _, f := range files {
|
||||||
if f.IsSymlink() || f.Mode&uint32(os.ModeDir) != 0 {
|
if f.IsSymlink() || f.Mode&uint32(os.ModeDir) != 0 {
|
||||||
// No chunks to fetch — restore can run immediately.
|
// No chunks to fetch — restore can run immediately.
|
||||||
@@ -66,7 +71,7 @@ func newRestorePlan(
|
|||||||
|
|
||||||
fileChunks, err := repos.FileChunks.GetByFileID(ctx, f.ID)
|
fileChunks, err := repos.FileChunks.GetByFileID(ctx, f.ID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("planning %s: %w", f.Path, err)
|
return nil, nil, fmt.Errorf("planning %s: %w", f.Path, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
blobs := make(map[string]struct{})
|
blobs := make(map[string]struct{})
|
||||||
@@ -74,14 +79,18 @@ func newRestorePlan(
|
|||||||
for _, fc := range fileChunks {
|
for _, fc := range fileChunks {
|
||||||
bc, ok := chunkToBlobMap[fc.ChunkHash.String()]
|
bc, ok := chunkToBlobMap[fc.ChunkHash.String()]
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, fmt.Errorf("planning %s: %w: %s",
|
unplanned[f.ID] = fmt.Errorf("%w: %s",
|
||||||
f.Path, errPlanChunkMissing, shortHash(fc.ChunkHash.String()))
|
errPlanChunkMissing, shortHash(fc.ChunkHash.String()))
|
||||||
|
|
||||||
|
continue nextFile
|
||||||
}
|
}
|
||||||
|
|
||||||
hash, ok := blobIDToHash[bc.BlobID.String()]
|
hash, ok := blobIDToHash[bc.BlobID.String()]
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, fmt.Errorf("planning %s: %w: %s",
|
unplanned[f.ID] = fmt.Errorf("%w: %s",
|
||||||
f.Path, errPlanBlobIDMissing, bc.BlobID)
|
errPlanBlobIDMissing, bc.BlobID)
|
||||||
|
|
||||||
|
continue nextFile
|
||||||
}
|
}
|
||||||
|
|
||||||
blobs[hash] = struct{}{}
|
blobs[hash] = struct{}{}
|
||||||
@@ -103,7 +112,7 @@ func newRestorePlan(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return p, nil
|
return p, unplanned, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// markBlobCached records that the named blob is now resident in the
|
// markBlobCached records that the named blob is now resident in the
|
||||||
|
|||||||
@@ -28,7 +28,8 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// missingBlobBackup is a snapshot of single-chunk files spread over two
|
// missingBlobBackup is a snapshot of single-chunk files spread over two
|
||||||
// blobs, with one of those blobs deleted from the store.
|
// blobs, with one of those blobs lost: deleted from the store, or missing
|
||||||
|
// from the snapshot metadata.
|
||||||
type missingBlobBackup struct {
|
type missingBlobBackup struct {
|
||||||
fs afero.Fs
|
fs afero.Fs
|
||||||
cfg *config.Config
|
cfg *config.Config
|
||||||
@@ -37,20 +38,39 @@ type missingBlobBackup struct {
|
|||||||
restoreDir string
|
restoreDir string
|
||||||
// files holds the original content by source path.
|
// files holds the original content by source path.
|
||||||
files map[string][]byte
|
files map[string][]byte
|
||||||
// lost holds the source paths whose chunk is in the deleted blob.
|
// lost holds the source paths whose chunk is in the lost blob.
|
||||||
lost map[string]bool
|
lost map[string]bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestRestoreSkipErrorsSkipsFilesOfMissingBlob restores with SkipErrors
|
// TestRestoreSkipErrorsSkipsFilesOfMissingBlob restores with SkipErrors
|
||||||
// after one blob of a two-blob snapshot was deleted. Every file stored in
|
// after one blob of a two-blob snapshot was deleted from the store.
|
||||||
// that blob must be reported as failed and left absent, every other file
|
|
||||||
// must be restored intact, and Restore must still return an error.
|
|
||||||
func TestRestoreSkipErrorsSkipsFilesOfMissingBlob(t *testing.T) {
|
func TestRestoreSkipErrorsSkipsFilesOfMissingBlob(t *testing.T) {
|
||||||
log.Initialize(log.Config{})
|
log.Initialize(log.Config{})
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
backup := backupThenDeleteOneBlob(ctx, t)
|
assertRestoreSkipsLostFiles(ctx, t, backupThenDeleteOneBlob(ctx, t))
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRestoreSkipErrorsSkipsFilesWithChunkInNoBlob restores with
|
||||||
|
// SkipErrors a snapshot whose metadata lists chunks that no blob holds, as
|
||||||
|
// earlier backup defects left behind.
|
||||||
|
func TestRestoreSkipErrorsSkipsFilesWithChunkInNoBlob(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
assertRestoreSkipsLostFiles(ctx, t, backupThenDropOneBlobFromMetadata(ctx, t))
|
||||||
|
}
|
||||||
|
|
||||||
|
// assertRestoreSkipsLostFiles restores backup with SkipErrors. Every file
|
||||||
|
// stored in the lost blob must be reported as failed and left absent,
|
||||||
|
// every other file must be restored intact, and Restore must still return
|
||||||
|
// an error.
|
||||||
|
func assertRestoreSkipsLostFiles(
|
||||||
|
ctx context.Context, t *testing.T, backup *missingBlobBackup,
|
||||||
|
) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
var out bytes.Buffer
|
var out bytes.Buffer
|
||||||
|
|
||||||
@@ -72,14 +92,14 @@ func TestRestoreSkipErrorsSkipsFilesOfMissingBlob(t *testing.T) {
|
|||||||
|
|
||||||
if backup.lost[path] {
|
if backup.lost[path] {
|
||||||
assert.Containsf(t, out.String(), path,
|
assert.Containsf(t, out.String(), path,
|
||||||
"%s needs the deleted blob and must be reported", path)
|
"%s needs the lost blob and must be reported", path)
|
||||||
assert.NoFileExists(t, restored)
|
assert.NoFileExists(t, restored)
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
got, err := afero.ReadFile(backup.fs, restored)
|
got, err := afero.ReadFile(backup.fs, restored)
|
||||||
require.NoErrorf(t, err, "%s does not need the deleted blob", path)
|
require.NoErrorf(t, err, "%s does not need the lost blob", path)
|
||||||
assert.Equalf(t, content, got, "%s restored with wrong content", path)
|
assert.Equalf(t, content, got, "%s restored with wrong content", path)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -103,6 +123,30 @@ func TestRestoreMissingBlobAbortsWithoutSkipErrors(t *testing.T) {
|
|||||||
require.ErrorIs(t, err, storage.ErrNotFound)
|
require.ErrorIs(t, err, storage.ErrNotFound)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRestoreChunkInNoBlobAbortsWithoutSkipErrors checks that a chunk held
|
||||||
|
// by no blob still ends the restore, before any file is written, when
|
||||||
|
// SkipErrors is not set.
|
||||||
|
func TestRestoreChunkInNoBlobAbortsWithoutSkipErrors(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
backup := backupThenDropOneBlobFromMetadata(ctx, t)
|
||||||
|
|
||||||
|
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)
|
||||||
|
|
||||||
|
err := v.Restore(&vaultik.RestoreOptions{
|
||||||
|
SnapshotID: backup.snapshotID,
|
||||||
|
TargetDir: backup.restoreDir,
|
||||||
|
})
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "chunk missing from blob map")
|
||||||
|
|
||||||
|
for path := range backup.files {
|
||||||
|
assert.NoFileExists(t, filepath.Join(backup.restoreDir, path))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// backupThenDeleteOneBlob backs up missingBlobFileCount single-chunk
|
// backupThenDeleteOneBlob backs up missingBlobFileCount single-chunk
|
||||||
// files, then deletes from the store the blob holding the first of them.
|
// files, then deletes from the store the blob holding the first of them.
|
||||||
func backupThenDeleteOneBlob(
|
func backupThenDeleteOneBlob(
|
||||||
@@ -110,6 +154,54 @@ func backupThenDeleteOneBlob(
|
|||||||
) *missingBlobBackup {
|
) *missingBlobBackup {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
|
backup, db, lostBlob := backupSingleChunkFiles(ctx, t)
|
||||||
|
require.NoError(t, db.Close())
|
||||||
|
|
||||||
|
require.NoError(t, backup.storer.Delete(ctx, fmt.Sprintf(
|
||||||
|
"blobs/%s/%s/%s", lostBlob[:2], lostBlob[2:4], lostBlob)))
|
||||||
|
|
||||||
|
return backup
|
||||||
|
}
|
||||||
|
|
||||||
|
// backupThenDropOneBlobFromMetadata backs up missingBlobFileCount
|
||||||
|
// single-chunk files, then removes the blob holding the first of them from
|
||||||
|
// the snapshot and exports the snapshot metadata again. The exported
|
||||||
|
// metadata then lists chunks that no blob holds.
|
||||||
|
func backupThenDropOneBlobFromMetadata(
|
||||||
|
ctx context.Context, t *testing.T,
|
||||||
|
) *missingBlobBackup {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
backup, db, lostBlob := backupSingleChunkFiles(ctx, t)
|
||||||
|
|
||||||
|
// The uploads row refers to the blob row, so it would keep the export
|
||||||
|
// from deleting it.
|
||||||
|
for _, query := range []string{
|
||||||
|
"DELETE FROM snapshot_blobs WHERE blob_hash = ?",
|
||||||
|
"DELETE FROM uploads WHERE blob_hash = ?",
|
||||||
|
} {
|
||||||
|
_, err := db.Conn().ExecContext(ctx, query, lostBlob)
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
sm := newFaultSnapshotManager(
|
||||||
|
backup.fs, backup.storer, backup.cfg, database.NewRepositories(db))
|
||||||
|
require.NoError(t,
|
||||||
|
sm.ExportSnapshotMetadata(ctx, db.Path(), backup.snapshotID))
|
||||||
|
require.NoError(t, db.Close())
|
||||||
|
|
||||||
|
return backup
|
||||||
|
}
|
||||||
|
|
||||||
|
// backupSingleChunkFiles backs up missingBlobFileCount single-chunk files.
|
||||||
|
// It returns the backup, with lost naming the files stored in the same
|
||||||
|
// blob as the first of them, the still open local index, and the hash of
|
||||||
|
// that blob.
|
||||||
|
func backupSingleChunkFiles(
|
||||||
|
ctx context.Context, t *testing.T,
|
||||||
|
) (*missingBlobBackup, *database.DB, string) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
fs := afero.NewOsFs()
|
fs := afero.NewOsFs()
|
||||||
tempDir := t.TempDir()
|
tempDir := t.TempDir()
|
||||||
dataDir := filepath.Join(tempDir, "src")
|
dataDir := filepath.Join(tempDir, "src")
|
||||||
@@ -143,16 +235,11 @@ func backupThenDeleteOneBlob(
|
|||||||
blobOfFile[path] = blobHashOfSingleChunkFile(ctx, t, repos, path)
|
blobOfFile[path] = blobHashOfSingleChunkFile(ctx, t, repos, path)
|
||||||
}
|
}
|
||||||
|
|
||||||
require.NoError(t, db.Close())
|
lostBlob := blobOfFile[filepath.Join(dataDir, "file-00.bin")]
|
||||||
|
|
||||||
deleted := blobOfFile[filepath.Join(dataDir, "file-00.bin")]
|
|
||||||
require.NoError(t, storer.Delete(ctx, fmt.Sprintf(
|
|
||||||
"blobs/%s/%s/%s", deleted[:2], deleted[2:4], deleted)))
|
|
||||||
|
|
||||||
lost := make(map[string]bool)
|
lost := make(map[string]bool)
|
||||||
|
|
||||||
for path, hash := range blobOfFile {
|
for path, hash := range blobOfFile {
|
||||||
if hash == deleted {
|
if hash == lostBlob {
|
||||||
lost[path] = true
|
lost[path] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -168,7 +255,7 @@ func backupThenDeleteOneBlob(
|
|||||||
restoreDir: filepath.Join(tempDir, "restored"),
|
restoreDir: filepath.Join(tempDir, "restored"),
|
||||||
files: files,
|
files: files,
|
||||||
lost: lost,
|
lost: lost,
|
||||||
}
|
}, db, lostBlob
|
||||||
}
|
}
|
||||||
|
|
||||||
// blobHashOfSingleChunkFile returns the hash of the blob holding the one
|
// blobHashOfSingleChunkFile returns the hash of the blob holding the one
|
||||||
|
|||||||
@@ -209,6 +209,7 @@ type snapshotStats struct {
|
|||||||
totalFiles int
|
totalFiles int
|
||||||
totalBytes int64
|
totalBytes int64
|
||||||
totalChunks int
|
totalChunks int
|
||||||
|
newChunkSize int64
|
||||||
totalBlobs int
|
totalBlobs int
|
||||||
totalBytesSkipped int64
|
totalBytesSkipped int64
|
||||||
totalFilesSkipped int
|
totalFilesSkipped int
|
||||||
@@ -341,6 +342,7 @@ func (v *Vaultik) scanAllDirectories(
|
|||||||
stats.totalFiles += result.FilesScanned
|
stats.totalFiles += result.FilesScanned
|
||||||
stats.totalBytes += result.BytesScanned
|
stats.totalBytes += result.BytesScanned
|
||||||
stats.totalChunks += result.ChunksCreated
|
stats.totalChunks += result.ChunksCreated
|
||||||
|
stats.newChunkSize += result.NewChunkSize
|
||||||
stats.totalBlobs += result.BlobsCreated
|
stats.totalBlobs += result.BlobsCreated
|
||||||
stats.totalFilesSkipped += result.FilesSkipped
|
stats.totalFilesSkipped += result.FilesSkipped
|
||||||
stats.totalFilesFailed += result.FilesFailed
|
stats.totalFilesFailed += result.FilesFailed
|
||||||
@@ -403,6 +405,7 @@ func (v *Vaultik) finalizeSnapshotMetadata(
|
|||||||
BlobUncompressedSize: stats.blobUncompressedSize,
|
BlobUncompressedSize: stats.blobUncompressedSize,
|
||||||
CompressionLevel: v.Config.CompressionLevel,
|
CompressionLevel: v.Config.CompressionLevel,
|
||||||
UploadDurationMs: stats.uploadDuration.Milliseconds(),
|
UploadDurationMs: stats.uploadDuration.Milliseconds(),
|
||||||
|
NewChunkSize: stats.newChunkSize,
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.SnapshotManager.UpdateSnapshotStatsExtended(v.ctx, snapshotID, extStats)
|
err = v.SnapshotManager.UpdateSnapshotStatsExtended(v.ctx, snapshotID, extStats)
|
||||||
@@ -633,7 +636,7 @@ func (v *Vaultik) confirmAndExecutePurge(
|
|||||||
v.stdoutf(" %s (%s, %s)\n",
|
v.stdoutf(" %s (%s, %s)\n",
|
||||||
snap.ID,
|
snap.ID,
|
||||||
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
||||||
formatBytes(snap.CompressedSize))
|
ubytes(snap.CompressedSize))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -553,8 +553,8 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error {
|
|||||||
|
|
||||||
if snap.LocallyTracked {
|
if snap.LocallyTracked {
|
||||||
id = snap.ID.String()
|
id = snap.ID.String()
|
||||||
uncompressed = formatBytes(snap.UncompressedSize)
|
uncompressed = ubytes(snap.UncompressedSize)
|
||||||
newChunks = formatBytes(snap.NewChunkSize)
|
newChunks = ubytes(snap.NewChunkSize)
|
||||||
} else {
|
} else {
|
||||||
id = formatRemoteOnlyID(snap.RemoteKey)
|
id = formatRemoteOnlyID(snap.RemoteKey)
|
||||||
uncompressed = remoteOnlyCell
|
uncompressed = remoteOnlyCell
|
||||||
@@ -564,7 +564,7 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error {
|
|||||||
_, err = fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
|
_, err = fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
|
||||||
id,
|
id,
|
||||||
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
snap.Timestamp.Format("2006-01-02 15:04:05"),
|
||||||
formatBytes(snap.CompressedSize),
|
ubytes(snap.CompressedSize),
|
||||||
uncompressed,
|
uncompressed,
|
||||||
newChunks)
|
newChunks)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -212,8 +212,8 @@ const (
|
|||||||
listConfiguredName = "home"
|
listConfiguredName = "home"
|
||||||
listLocalID = "testhost_home_2026-03-01T10:00:00Z"
|
listLocalID = "testhost_home_2026-03-01T10:00:00Z"
|
||||||
listRemoteID = "otherhost_media_2026-03-02T11:22:33Z"
|
listRemoteID = "otherhost_media_2026-03-02T11:22:33Z"
|
||||||
// fiveMegabytes formats as "5.0 MB" through formatBytes.
|
// fiveMegabytes is printed as "5.0 MB" by every command.
|
||||||
fiveMegabytes = 5 * 1024 * 1024
|
fiveMegabytes = 5_000_000
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestListSnapshots_RemoteWithoutSecretKey is the regression guard for
|
// TestListSnapshots_RemoteWithoutSecretKey is the regression guard for
|
||||||
@@ -306,6 +306,48 @@ func TestListSnapshots_RemoteOnlyRowRendering(t *testing.T) {
|
|||||||
"are not in the local index")
|
"are not in the local index")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestSnapshotListAndPurgePrintSizesLikeRemoteInfo checks that the
|
||||||
|
// `snapshot list` table and the purge listing print a byte count the way
|
||||||
|
// `remote info` does: in powers of 1000, as the MB label says.
|
||||||
|
func TestSnapshotListAndPurgePrintSizesLikeRemoteInfo(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const wantSize = "5.0 MB" // fiveMegabytes
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
|
||||||
|
// A remote-only snapshot: `snapshot list` and `remote info` both take
|
||||||
|
// its size from the manifest.
|
||||||
|
remoteKey := env.addRemote(t, listRemoteID,
|
||||||
|
time.Date(2026, 3, 2, 11, 22, 33, 0, time.UTC))
|
||||||
|
|
||||||
|
// A locally tracked snapshot of the same size for the purge to list.
|
||||||
|
// The purge prints the snapshot's blob size, the last argument here.
|
||||||
|
start := time.Date(2026, 3, 1, 10, 0, 0, 0, time.UTC)
|
||||||
|
env.addLocal(t, listLocalID, start)
|
||||||
|
env.addRemote(t, listLocalID, start)
|
||||||
|
require.NoError(t, env.v.Repositories.Snapshots.UpdateCounts(
|
||||||
|
context.Background(), nil, listLocalID, 0, 0, 0, 0, fiveMegabytes))
|
||||||
|
|
||||||
|
require.NoError(t, env.v.RemoteInfo(false))
|
||||||
|
assert.Contains(t, findTableRow(t, env.stdout.String(), remoteKey[:16]),
|
||||||
|
wantSize, "remote info")
|
||||||
|
|
||||||
|
env.stdout.Reset()
|
||||||
|
require.NoError(t, env.v.ListSnapshots(false))
|
||||||
|
assert.Contains(t, findTableRow(t, env.stdout.String(),
|
||||||
|
"<remote only:"+remoteKey[:12]+">"), wantSize, "snapshot list")
|
||||||
|
|
||||||
|
// Without --force and with nothing on stdin the purge lists what it
|
||||||
|
// would delete, reads no confirmation, and deletes nothing.
|
||||||
|
env.stdout.Reset()
|
||||||
|
require.NoError(t, env.v.PurgeSnapshotsWithOptions(
|
||||||
|
&vaultik.SnapshotPurgeOptions{OlderThan: "1d"}))
|
||||||
|
assert.Contains(t, findTableRow(t, env.stdout.String(), listLocalID),
|
||||||
|
wantSize, "snapshot purge")
|
||||||
|
}
|
||||||
|
|
||||||
// findTableRow returns the single output line containing needle.
|
// findTableRow returns the single output line containing needle.
|
||||||
func findTableRow(t *testing.T, out, needle string) string {
|
func findTableRow(t *testing.T, out, needle string) string {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
@@ -445,6 +487,7 @@ type listJSONRow struct {
|
|||||||
RemoteKey string `json:"remote_key"`
|
RemoteKey string `json:"remote_key"`
|
||||||
Timestamp string `json:"timestamp"`
|
Timestamp string `json:"timestamp"`
|
||||||
CompressedSize int64 `json:"compressed_size"`
|
CompressedSize int64 `json:"compressed_size"`
|
||||||
|
NewChunkSize int64 `json:"new_chunk_size"`
|
||||||
LocallyTracked bool `json:"locally_tracked"`
|
LocallyTracked bool `json:"locally_tracked"`
|
||||||
RemotePresent *bool `json:"remote_present"`
|
RemotePresent *bool `json:"remote_present"`
|
||||||
}
|
}
|
||||||
|
|||||||
+39
-17
@@ -17,6 +17,7 @@ import (
|
|||||||
"sneak.berlin/go/vaultik/internal/database"
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Sentinel errors for snapshot verification failures.
|
// Sentinel errors for snapshot verification failures.
|
||||||
@@ -32,7 +33,9 @@ var (
|
|||||||
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
||||||
errManifestMissingBlob = errors.New(
|
errManifestMissingBlob = errors.New(
|
||||||
"manifest omits blob present in database")
|
"manifest omits blob present in database")
|
||||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||||
|
errBlobsNotInStorage = errors.New(
|
||||||
|
"blobs missing from storage or stored with the wrong size")
|
||||||
)
|
)
|
||||||
|
|
||||||
// verifyStatusFailed is the JSON status value for a failed verification.
|
// verifyStatusFailed is the JSON status value for a failed verification.
|
||||||
@@ -135,7 +138,6 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
result.Status = "ok"
|
result.Status = "ok"
|
||||||
result.Verified = len(dbBlobs)
|
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return v.outputVerifyJSON(result)
|
return v.outputVerifyJSON(result)
|
||||||
@@ -290,7 +292,7 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
v.stdoutf("Checking blob existence in remote storage...\n")
|
v.stdoutf("Checking blob existence in remote storage...\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -301,7 +303,8 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
len(dbBlobs), ubytes(totalSize))
|
len(dbBlobs), ubytes(totalSize))
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
err = v.performDeepVerificationFromDB(
|
||||||
|
dbBlobs, tdb.db.Conn(), opts, result, identities)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -653,8 +656,14 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
// verifyBlobExistenceFromDB checks that every blob from the database is
|
||||||
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
// in storage with the size the database records. It checks them all,
|
||||||
|
// counting the missing and wrong-size blobs in result, and fails if it
|
||||||
|
// found any. Any storage error other than not found stops the check and
|
||||||
|
// fails with that error.
|
||||||
|
func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||||
|
blobs []snapshot.BlobInfo, result *VerifyResult,
|
||||||
|
) error {
|
||||||
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
||||||
|
|
||||||
for i, blob := range blobs {
|
for i, blob := range blobs {
|
||||||
@@ -667,17 +676,23 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|||||||
// Construct blob path
|
// Construct blob path
|
||||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
||||||
|
|
||||||
// Check blob exists
|
|
||||||
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||||
if err != nil {
|
switch {
|
||||||
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
case errors.Is(err, storage.ErrNotFound):
|
||||||
}
|
log.Warn("Blob missing from storage", "hash", blob.Hash)
|
||||||
|
|
||||||
// Verify size matches
|
result.Missing++
|
||||||
if stat.Size != blob.CompressedSize {
|
result.MissingSize += blob.CompressedSize
|
||||||
return fmt.Errorf(
|
case err != nil:
|
||||||
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
return fmt.Errorf("checking blob %s in storage: %w",
|
||||||
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
blob.Hash, err)
|
||||||
|
case stat.Size != blob.CompressedSize:
|
||||||
|
log.Warn("Blob has the wrong size in storage",
|
||||||
|
"hash", blob.Hash,
|
||||||
|
"storage_size", stat.Size,
|
||||||
|
"database_size", blob.CompressedSize)
|
||||||
|
|
||||||
|
result.Mismatched++
|
||||||
}
|
}
|
||||||
|
|
||||||
// Progress update every 100 blobs
|
// Progress update every 100 blobs
|
||||||
@@ -691,16 +706,22 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if result.Missing > 0 || result.Mismatched > 0 {
|
||||||
|
return fmt.Errorf("%w: %d missing, %d with the wrong size",
|
||||||
|
errBlobsNotInStorage, result.Missing, result.Mismatched)
|
||||||
|
}
|
||||||
|
|
||||||
log.Info("✓ All blobs exist in storage")
|
log.Info("✓ All blobs exist in storage")
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// performDeepVerificationFromDB downloads and verifies the content of
|
// performDeepVerificationFromDB downloads and verifies the content of
|
||||||
// each blob using the database as source.
|
// each blob using the database as source, stopping at the first blob that
|
||||||
|
// fails. Each blob that passes is counted in result.Verified.
|
||||||
func (v *Vaultik) performDeepVerificationFromDB(
|
func (v *Vaultik) performDeepVerificationFromDB(
|
||||||
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
||||||
identities []age.Identity,
|
result *VerifyResult, identities []age.Identity,
|
||||||
) error {
|
) error {
|
||||||
// Calculate total bytes for ETA
|
// Calculate total bytes for ETA
|
||||||
var totalBytesExpected int64
|
var totalBytesExpected int64
|
||||||
@@ -723,6 +744,7 @@ func (v *Vaultik) performDeepVerificationFromDB(
|
|||||||
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
result.Verified++
|
||||||
bytesProcessed += blobInfo.CompressedSize
|
bytesProcessed += blobInfo.CompressedSize
|
||||||
elapsed := time.Since(startTime)
|
elapsed := time.Since(startTime)
|
||||||
remaining := len(blobs) - (i + 1)
|
remaining := len(blobs) - (i + 1)
|
||||||
|
|||||||
Reference in New Issue
Block a user