Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
71d5561dfe |
@@ -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 the chunks that snapshot stored that
|
uncompressed chunk size; size of chunks newly referenced by that
|
||||||
were not stored before, recorded when its backup ran).
|
snapshot).
|
||||||
|
|
||||||
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 come only from the
|
the uncompressed and "new chunk" columns, which can only be computed
|
||||||
local index. Their timestamp and compressed size are real,
|
from the 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,7 +119,6 @@ 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
|
||||||
|
|||||||
@@ -86,7 +86,6 @@ 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,7 +129,6 @@ 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,
|
||||||
@@ -144,19 +143,18 @@ 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, newChunkSize, snapshotID)
|
uploadBytes, uploadDurationMs, snapshotID)
|
||||||
} else {
|
} else {
|
||||||
_, err = r.db.ExecWithLog(ctx, query,
|
_, err = r.db.ExecWithLog(ctx, query,
|
||||||
blobUncompressedSize, compressionRatio, compressionLevel,
|
blobUncompressedSize, compressionRatio, compressionLevel,
|
||||||
uploadBytes, uploadDurationMs, newChunkSize, snapshotID)
|
uploadBytes, uploadDurationMs, snapshotID)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -596,24 +594,45 @@ func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
|
|||||||
return totalSize, nil
|
return totalSize, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
|
// GetSnapshotNewChunkSize returns the sum of plaintext sizes of chunks that are
|
||||||
// snapshot's backup stored that were not stored before, as the backup
|
// referenced by this snapshot but not by any earlier completed snapshot known to
|
||||||
// recorded it. It is not derived from the snapshot's files: a file's row is
|
// the local database. The result is the marginal uncompressed data this snapshot
|
||||||
// shared by every snapshot that lists it and is rewritten when the file
|
// added to the dedup pool — i.e., the delta from prior snapshots.
|
||||||
// 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 := `SELECT new_chunk_size FROM snapshots WHERE id = ?`
|
query := `
|
||||||
|
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 newChunkSize int64
|
var totalSize int64
|
||||||
|
|
||||||
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&newChunkSize)
|
err := r.db.conn.QueryRowContext(
|
||||||
|
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 newChunkSize, nil
|
return totalSize, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetIncompleteSnapshots returns all snapshots that haven't been completed
|
// GetIncompleteSnapshots returns all snapshots that haven't been completed
|
||||||
|
|||||||
@@ -135,9 +135,7 @@ 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. ChunksCreated counts the
|
// includes them and BytesScanned does not.
|
||||||
// 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
|
||||||
@@ -147,7 +145,6 @@ 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
|
||||||
@@ -1869,12 +1866,11 @@ func (s *Scanner) processFileStreaming(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// updateChunkStats counts a chunk that was not already stored, and its
|
// updateChunkStats counts a chunk that was not already stored. The scan
|
||||||
// size. The scan result's file counts, BytesScanned and BytesSkipped are
|
// result's file counts, BytesScanned and BytesSkipped are not touched
|
||||||
// not touched here: the scan phase counts each file once.
|
// 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,7 +199,6 @@ func (sm *SnapshotManager) UpdateSnapshotStatsExtended(
|
|||||||
stats.CompressionLevel,
|
stats.CompressionLevel,
|
||||||
stats.BytesUploaded,
|
stats.BytesUploaded,
|
||||||
stats.UploadDurationMs,
|
stats.UploadDurationMs,
|
||||||
stats.NewChunkSize,
|
|
||||||
)
|
)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -907,7 +906,6 @@ 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
|
||||||
|
|||||||
@@ -161,14 +161,13 @@ func TestVerifyManifestBlobsRejectsShortHash(t *testing.T) {
|
|||||||
Blobs: []snapshot.BlobInfo{{Hash: "abc", CompressedSize: 1}},
|
Blobs: []snapshot.BlobInfo{{Hash: "abc", CompressedSize: 1}},
|
||||||
}
|
}
|
||||||
|
|
||||||
verified, missing, mismatched, missingSize, unchecked, err :=
|
verified, missing, mismatched, missingSize, err :=
|
||||||
v.verifyManifestBlobs(manifest, &VerifyOptions{JSON: true})
|
v.verifyManifestBlobs(manifest, &VerifyOptions{JSON: true})
|
||||||
require.ErrorIs(t, err, errInvalidBlobHash)
|
require.ErrorIs(t, err, errInvalidBlobHash)
|
||||||
require.Zero(t, verified)
|
require.Zero(t, verified)
|
||||||
require.Zero(t, missing)
|
require.Zero(t, missing)
|
||||||
require.Zero(t, mismatched)
|
require.Zero(t, mismatched)
|
||||||
require.Zero(t, missingSize)
|
require.Zero(t, missingSize)
|
||||||
require.Empty(t, unchecked)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestVerifyBlobChunksRejectsNegativeLength proves a blob_chunks row with a
|
// TestVerifyBlobChunksRejectsNegativeLength proves a blob_chunks row with a
|
||||||
|
|||||||
@@ -1,133 +0,0 @@
|
|||||||
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
|
|
||||||
}
|
|
||||||
@@ -691,6 +691,8 @@ func (v *Vaultik) downloadSnapshotDB(
|
|||||||
return nil, "", fmt.Errorf("downloading %s: %w", dbKey, err)
|
return nil, "", fmt.Errorf("downloading %s: %w", dbKey, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
reader = newCloseOnCancelReader(v.ctx, reader)
|
||||||
|
|
||||||
defer func() { _ = reader.Close() }()
|
defer func() { _ = reader.Close() }()
|
||||||
|
|
||||||
// Decrypt and decompress straight from the storage stream, then stream
|
// Decrypt and decompress straight from the storage stream, then stream
|
||||||
|
|||||||
@@ -207,30 +207,37 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
|
|||||||
"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
|
// stallingStorer wraps a Storer so that each download of a key chosen by
|
||||||
// over the first half of the blob and then stalls, ignoring the context,
|
// stalls hands over the first half of the object and then stalls,
|
||||||
// as a file:// blob on a hung USB stick or network share does. Only
|
// ignoring the context, as a file:// read on a hung USB stick or network
|
||||||
// closing the returned reader ends the stall.
|
// share does. Only closing the returned reader ends the stall.
|
||||||
type stallingBlobStorer struct {
|
type stallingStorer struct {
|
||||||
storage.Storer
|
storage.Storer
|
||||||
|
|
||||||
|
stalls func(key string) bool
|
||||||
once sync.Once
|
once sync.Once
|
||||||
stalled chan struct{}
|
stalled chan struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
|
func newStallingStorer(
|
||||||
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
|
inner storage.Storer, stalls func(key string) bool,
|
||||||
|
) *stallingStorer {
|
||||||
|
return &stallingStorer{
|
||||||
|
Storer: inner,
|
||||||
|
stalls: stalls,
|
||||||
|
stalled: make(chan struct{}),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *stallingBlobStorer) Get(
|
func (s *stallingStorer) Get(
|
||||||
ctx context.Context, key string,
|
ctx context.Context, key string,
|
||||||
) (io.ReadCloser, error) {
|
) (io.ReadCloser, error) {
|
||||||
rc, err := s.Storer.Get(ctx, key)
|
rc, err := s.Storer.Get(ctx, key)
|
||||||
if err != nil || !strings.HasPrefix(key, "blobs/") {
|
if err != nil || !s.stalls(key) {
|
||||||
return rc, err
|
return rc, err
|
||||||
}
|
}
|
||||||
|
|
||||||
blob, err := io.ReadAll(rc)
|
object, err := io.ReadAll(rc)
|
||||||
_ = rc.Close()
|
_ = rc.Close()
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -242,7 +249,7 @@ func (s *stallingBlobStorer) Get(
|
|||||||
pr, pw := io.Pipe()
|
pr, pw := io.Pipe()
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
_, _ = pw.Write(blob[:len(blob)/2])
|
_, _ = pw.Write(object[:len(object)/2])
|
||||||
|
|
||||||
s.once.Do(func() { close(s.stalled) })
|
s.once.Do(func() { close(s.stalled) })
|
||||||
}()
|
}()
|
||||||
@@ -250,17 +257,26 @@ func (s *stallingBlobStorer) Get(
|
|||||||
return pr, nil
|
return pr, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a
|
// TestInterruptDuringStalledReadCleansTempDir stops a restore and a deep
|
||||||
// deep verify while a blob read is stalled on a source that ignores the
|
// verify while a read of a blob or of the snapshot database is stalled
|
||||||
// context. Stopping must end that read, so the operation unwinds and
|
// on a source that ignores the context. Stopping must end that read, so
|
||||||
// removes its decrypted scratch files well within the shutdown timeout
|
// the operation unwinds and removes its decrypted scratch files well
|
||||||
// (issue #325).
|
// within the shutdown timeout (issues #325 and #356).
|
||||||
//
|
//
|
||||||
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
|
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
|
||||||
// so it can assert on exactly the scratch files the operation created.
|
// so it can assert on exactly the scratch files the operation created.
|
||||||
func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
func TestInterruptDuringStalledReadCleansTempDir(t *testing.T) {
|
||||||
log.Initialize(log.Config{})
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
stalledReads := map[string]func(key string) bool{
|
||||||
|
"blob": func(key string) bool {
|
||||||
|
return strings.HasPrefix(key, "blobs/")
|
||||||
|
},
|
||||||
|
"snapshot database": func(key string) bool {
|
||||||
|
return strings.HasSuffix(key, "/db.zst.age")
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
||||||
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
||||||
return v.Restore(&RestoreOptions{
|
return v.Restore(&RestoreOptions{
|
||||||
@@ -273,52 +289,58 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
for name, op := range operations {
|
for readName, stalls := range stalledReads {
|
||||||
t.Run(name, func(t *testing.T) {
|
for opName, op := range operations {
|
||||||
fs := afero.NewOsFs()
|
t.Run(readName+" read during "+opName, func(t *testing.T) {
|
||||||
root := t.TempDir()
|
fs := afero.NewOsFs()
|
||||||
|
root := t.TempDir()
|
||||||
|
|
||||||
// Random content, so the blob stays large enough to stall
|
// Random content, so the blob stays large enough to stall
|
||||||
// part-way through its compressed, encrypted body.
|
// part-way through its compressed, encrypted body.
|
||||||
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
|
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
|
||||||
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
|
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
|
||||||
|
|
||||||
scratch := filepath.Join(root, "scratch")
|
scratch := filepath.Join(root, "scratch")
|
||||||
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
||||||
t.Setenv("TMPDIR", scratch)
|
t.Setenv("TMPDIR", scratch)
|
||||||
|
|
||||||
stalling := newStallingBlobStorer(storer)
|
stalling := newStallingStorer(storer, stalls)
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
v := newRestoreVaultik(ctx, cfg, stalling, fs)
|
v := newRestoreVaultik(ctx, cfg, stalling, fs)
|
||||||
v.cancel = cancel
|
v.cancel = cancel
|
||||||
|
|
||||||
var opErr error
|
var opErr error
|
||||||
|
|
||||||
stop := v.StartOperation(func() {
|
stop := v.StartOperation(func() {
|
||||||
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
|
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
|
||||||
|
})
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-stalling.stalled:
|
||||||
|
case <-time.After(30 * time.Second):
|
||||||
|
t.Fatal("the operation never reached the stalled read")
|
||||||
|
}
|
||||||
|
|
||||||
|
// The snapshot database's temp directory is created just
|
||||||
|
// after its read starts, so it can appear after the stall.
|
||||||
|
require.Eventually(t, func() bool {
|
||||||
|
return len(scratchEntries(t, scratch)) > 0
|
||||||
|
}, 10*time.Second, 10*time.Millisecond,
|
||||||
|
"expected decrypted scratch files to exist mid-read")
|
||||||
|
|
||||||
|
stopCtx, stopCancel := context.WithTimeout(
|
||||||
|
context.Background(), 10*time.Second)
|
||||||
|
defer stopCancel()
|
||||||
|
|
||||||
|
require.True(t, stop(stopCtx),
|
||||||
|
"stop timed out; the stalled read was not interrupted")
|
||||||
|
require.ErrorIs(t, opErr, context.Canceled)
|
||||||
|
require.Empty(t, scratchEntries(t, scratch),
|
||||||
|
"decrypted scratch files remained after the interrupt")
|
||||||
})
|
})
|
||||||
|
}
|
||||||
select {
|
|
||||||
case <-stalling.stalled:
|
|
||||||
case <-time.After(30 * time.Second):
|
|
||||||
t.Fatal("the operation never reached a blob read")
|
|
||||||
}
|
|
||||||
|
|
||||||
require.NotEmpty(t, scratchEntries(t, scratch),
|
|
||||||
"expected decrypted scratch files to exist mid-read")
|
|
||||||
|
|
||||||
stopCtx, stopCancel := context.WithTimeout(
|
|
||||||
context.Background(), 10*time.Second)
|
|
||||||
defer stopCancel()
|
|
||||||
|
|
||||||
require.True(t, stop(stopCtx),
|
|
||||||
"stop timed out; the stalled blob read was not interrupted")
|
|
||||||
require.ErrorIs(t, opErr, context.Canceled)
|
|
||||||
require.Empty(t, scratchEntries(t, scratch),
|
|
||||||
"decrypted scratch files remained after the interrupt")
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,20 +4,15 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"path"
|
|
||||||
"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/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"
|
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
)
|
)
|
||||||
@@ -156,224 +151,6 @@ func TestShallowVerifyDetectsMissingDatabase(t *testing.T) {
|
|||||||
"shallow verify must fail when db.zst.age is absent")
|
"shallow verify must fail when db.zst.age is absent")
|
||||||
}
|
}
|
||||||
|
|
||||||
// errStatDenied stands in for any storage error other than not-found: a
|
|
||||||
// permission error, an S3 403, a dropped connection.
|
|
||||||
var errStatDenied = errors.New("injected stat failure")
|
|
||||||
|
|
||||||
// statFailingStorer fails Stat with errStatDenied for the keys in
|
|
||||||
// failKeys and passes every other call to the wrapped Storer.
|
|
||||||
type statFailingStorer struct {
|
|
||||||
storage.Storer
|
|
||||||
|
|
||||||
failKeys []string
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s statFailingStorer) Stat(
|
|
||||||
ctx context.Context, key string,
|
|
||||||
) (*storage.ObjectInfo, error) {
|
|
||||||
if slices.Contains(s.failKeys, key) {
|
|
||||||
return nil, fmt.Errorf("stat %q: %w", key, errStatDenied)
|
|
||||||
}
|
|
||||||
|
|
||||||
return s.Storer.Stat(ctx, key)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestShallowVerifyReportsStatErrorsApartFromMissing backs up a real
|
|
||||||
// snapshot, deletes one blob, and makes storage fail Stat with an error
|
|
||||||
// other than not-found for a second blob and for the encrypted database.
|
|
||||||
// Only the deleted blob may be counted as missing. The second blob and
|
|
||||||
// the database must be reported as could not be checked, with the error
|
|
||||||
// storage returned, and verify must fail, in the text output and under
|
|
||||||
// --json.
|
|
||||||
func TestShallowVerifyReportsStatErrorsApartFromMissing(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 content, so the snapshot spans several blobs.
|
|
||||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
|
||||||
appendRandomBytes(t, fs, map[string][]byte{},
|
|
||||||
filepath.Join(dataDir, "data.bin"), maxBlobSize*4)
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
cfg, storer, snapshotID := runFileStorageBackup(
|
|
||||||
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
|
||||||
|
|
||||||
blobKeys := storedBlobKeys(t, storeDir)
|
|
||||||
require.GreaterOrEqual(t, len(blobKeys), 2, "expected at least two blobs")
|
|
||||||
|
|
||||||
missingKey, uncheckedKey := blobKeys[0], blobKeys[1]
|
|
||||||
missingHash, uncheckedHash := path.Base(missingKey), path.Base(uncheckedKey)
|
|
||||||
|
|
||||||
require.NoError(t, os.Remove(filepath.Join(storeDir, missingKey)))
|
|
||||||
|
|
||||||
dbKey := "metadata/" + snapshot.RemoteSnapshotKey(snapshotID) + "/db.zst.age"
|
|
||||||
failing := statFailingStorer{
|
|
||||||
Storer: storer,
|
|
||||||
failKeys: []string{uncheckedKey, dbKey},
|
|
||||||
}
|
|
||||||
|
|
||||||
newVerifier := func(out io.Writer) *vaultik.Vaultik {
|
|
||||||
v := &vaultik.Vaultik{
|
|
||||||
Config: cfg,
|
|
||||||
Storage: failing,
|
|
||||||
Fs: fs,
|
|
||||||
Stdout: out,
|
|
||||||
Stderr: io.Discard,
|
|
||||||
UI: ui.NewWithColor(io.Discard, false),
|
|
||||||
}
|
|
||||||
v.SetContext(ctx)
|
|
||||||
|
|
||||||
return v
|
|
||||||
}
|
|
||||||
|
|
||||||
var out bytes.Buffer
|
|
||||||
|
|
||||||
err := newVerifier(&out).VerifySnapshotWithOptions(
|
|
||||||
snapshotID, &vaultik.VerifyOptions{JSON: true})
|
|
||||||
require.Error(t, err)
|
|
||||||
|
|
||||||
result := decodeVerifyResult(t, out.Bytes())
|
|
||||||
require.Equal(t, "failed", result.Status)
|
|
||||||
require.Equal(t, 1, result.Missing, "only the deleted blob is missing")
|
|
||||||
require.Equal(t, result.BlobCount-2, result.Verified)
|
|
||||||
require.Len(t, result.Unchecked, 1)
|
|
||||||
require.Equal(t, uncheckedHash, result.Unchecked[0].Hash)
|
|
||||||
require.Contains(t, result.Unchecked[0].Error, errStatDenied.Error())
|
|
||||||
require.False(t, result.DatabaseMissing,
|
|
||||||
"a database that could not be checked is not missing")
|
|
||||||
require.Contains(t, result.DatabaseCheckError, errStatDenied.Error())
|
|
||||||
|
|
||||||
out.Reset()
|
|
||||||
err = newVerifier(&out).VerifySnapshotWithOptions(
|
|
||||||
snapshotID, &vaultik.VerifyOptions{})
|
|
||||||
require.Error(t, err)
|
|
||||||
|
|
||||||
text := out.String()
|
|
||||||
require.Contains(t, text, "Missing: "+missingHash)
|
|
||||||
require.NotContains(t, text, "Missing: "+uncheckedHash)
|
|
||||||
require.Contains(t, text, "Could not check: "+uncheckedHash)
|
|
||||||
require.Contains(t, text, "Encrypted database: could not be checked")
|
|
||||||
require.Contains(t, text, errStatDenied.Error())
|
|
||||||
require.Contains(t, text, "FAILED")
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestShallowVerifyQuietFailsOnStatErrorAlone backs up a healthy snapshot,
|
|
||||||
// then makes storage fail Stat, with an error other than not-found, for
|
|
||||||
// one blob alone and then for the encrypted database alone. Each run must
|
|
||||||
// fail verify, and under --quiet, which drops the report, the error
|
|
||||||
// storage returned must still reach stderr.
|
|
||||||
//
|
|
||||||
//nolint:paralleltest // captureProcessStderr replaces os.Stderr
|
|
||||||
func TestShallowVerifyQuietFailsOnStatErrorAlone(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
|
|
||||||
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)
|
|
||||||
|
|
||||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
|
||||||
require.NoError(t, afero.WriteFile(fs,
|
|
||||||
filepath.Join(dataDir, "data.bin"),
|
|
||||||
bytesPattern("shallow-quiet-", int(maxBlobSize*2)), 0o644))
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
cfg, storer, snapshotID := runFileStorageBackup(
|
|
||||||
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
|
||||||
|
|
||||||
newQuietVerifier := func(s storage.Storer) *vaultik.Vaultik {
|
|
||||||
v := &vaultik.Vaultik{
|
|
||||||
Config: cfg,
|
|
||||||
Storage: s,
|
|
||||||
Fs: fs,
|
|
||||||
Stdout: io.Discard,
|
|
||||||
Stderr: io.Discard,
|
|
||||||
UI: ui.NewWithColor(io.Discard, false),
|
|
||||||
}
|
|
||||||
v.SetContext(ctx)
|
|
||||||
v.UI.SetQuiet(true)
|
|
||||||
|
|
||||||
return v
|
|
||||||
}
|
|
||||||
|
|
||||||
require.NoError(t,
|
|
||||||
newQuietVerifier(storer).VerifySnapshotWithOptions(
|
|
||||||
snapshotID, &vaultik.VerifyOptions{}),
|
|
||||||
"shallow verify should pass on a healthy snapshot")
|
|
||||||
|
|
||||||
blobKeys := storedBlobKeys(t, storeDir)
|
|
||||||
require.NotEmpty(t, blobKeys)
|
|
||||||
|
|
||||||
dbKey := "metadata/" + snapshot.RemoteSnapshotKey(snapshotID) + "/db.zst.age"
|
|
||||||
|
|
||||||
for _, failKey := range []string{blobKeys[0], dbKey} {
|
|
||||||
v := newQuietVerifier(statFailingStorer{
|
|
||||||
Storer: storer,
|
|
||||||
failKeys: []string{failKey},
|
|
||||||
})
|
|
||||||
|
|
||||||
var err error
|
|
||||||
|
|
||||||
stderr := captureProcessStderr(t, func() {
|
|
||||||
err = v.VerifySnapshotWithOptions(
|
|
||||||
snapshotID, &vaultik.VerifyOptions{})
|
|
||||||
})
|
|
||||||
|
|
||||||
require.Error(t, err,
|
|
||||||
"verify must fail when %s could not be checked", failKey)
|
|
||||||
require.Contains(t, stderr, errStatDenied.Error(),
|
|
||||||
"under --quiet the error for %s must still reach stderr", failKey)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// storedBlobKeys returns the storage key of every blob file under
|
|
||||||
// storeDir, sorted.
|
|
||||||
func storedBlobKeys(t *testing.T, storeDir string) []string {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
var keys []string
|
|
||||||
|
|
||||||
err := filepath.Walk(filepath.Join(storeDir, "blobs"),
|
|
||||||
func(p string, info os.FileInfo, err error) error {
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
if info.IsDir() {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
rel, err := filepath.Rel(storeDir, p)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
keys = append(keys, filepath.ToSlash(rel))
|
|
||||||
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
return keys
|
|
||||||
}
|
|
||||||
|
|
||||||
// growOneBlob appends bytes to the first blob file found under blobsDir,
|
// growOneBlob appends bytes to the first blob file found under blobsDir,
|
||||||
// changing its on-disk size so it no longer matches the manifest.
|
// changing its on-disk size so it no longer matches the manifest.
|
||||||
func growOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
func growOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||||
|
|||||||
@@ -14,7 +14,6 @@ import (
|
|||||||
|
|
||||||
"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"
|
|
||||||
"sneak.berlin/go/vaultik/internal/types"
|
"sneak.berlin/go/vaultik/internal/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -210,7 +209,6 @@ 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
|
||||||
@@ -343,7 +341,6 @@ 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
|
||||||
@@ -406,7 +403,6 @@ 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)
|
||||||
@@ -778,10 +774,18 @@ func (v *Vaultik) VerifySnapshotWithOptions(
|
|||||||
v.stdoutf("Checking blob presence and sizes...\n")
|
v.stdoutf("Checking blob presence and sizes...\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
v.checkSnapshotDatabase(remoteKey, result)
|
// A snapshot is only restorable if its encrypted database is present
|
||||||
|
// alongside the blobs. Shallow verify checks that the object exists; it
|
||||||
|
// does not decrypt it (that is deep verify's job).
|
||||||
|
dbPath := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey)
|
||||||
|
|
||||||
result.Verified, result.Missing, result.Mismatched, result.MissingSize,
|
_, dbErr := v.Storage.Stat(v.ctx, dbPath)
|
||||||
result.Unchecked, err = v.verifyManifestBlobs(manifest, opts)
|
if dbErr != nil {
|
||||||
|
result.DatabaseMissing = true
|
||||||
|
}
|
||||||
|
|
||||||
|
result.Verified, result.Missing, result.Mismatched, result.MissingSize, err =
|
||||||
|
v.verifyManifestBlobs(manifest, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
result.Status = verifyStatusFailed
|
result.Status = verifyStatusFailed
|
||||||
@@ -818,46 +822,25 @@ func (v *Vaultik) printVerifyHeader(snapshotID string, opts *VerifyOptions) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkSnapshotDatabase records in result whether the snapshot's
|
|
||||||
// encrypted database is missing or could not be checked. A snapshot is
|
|
||||||
// only restorable if its database is present alongside the blobs.
|
|
||||||
// Shallow verify checks that the object exists; it does not decrypt it
|
|
||||||
// (that is deep verify's job).
|
|
||||||
func (v *Vaultik) checkSnapshotDatabase(remoteKey string, result *VerifyResult) {
|
|
||||||
dbPath := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey)
|
|
||||||
|
|
||||||
_, err := v.Storage.Stat(v.ctx, dbPath)
|
|
||||||
|
|
||||||
switch {
|
|
||||||
case errors.Is(err, storage.ErrNotFound):
|
|
||||||
result.DatabaseMissing = true
|
|
||||||
case err != nil:
|
|
||||||
result.DatabaseCheckError = err.Error()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// verifyManifestBlobs checks that each blob in the manifest is present in
|
// verifyManifestBlobs checks that each blob in the manifest is present in
|
||||||
// storage with the size the manifest records, returning the counts of
|
// storage with the size the manifest records, returning the counts of
|
||||||
// blobs that were present with the right size, absent, and present but the
|
// blobs that were present with the right size, absent, and present but the
|
||||||
// wrong size, the total bytes of the absent blobs, and the blobs it could
|
// wrong size, plus the total bytes of the absent blobs. It does not read
|
||||||
// not check because storage returned an error other than not-found. It
|
// blob contents; deep verification (RunDeepVerify) does that. The size
|
||||||
// does not read blob contents; deep verification (RunDeepVerify) does
|
// comparison matches the deep path (see verifyBlobExistenceFromDB).
|
||||||
// that. The size comparison matches the deep path (see
|
|
||||||
// verifyBlobExistenceFromDB).
|
|
||||||
func (v *Vaultik) verifyManifestBlobs(
|
func (v *Vaultik) verifyManifestBlobs(
|
||||||
manifest *snapshot.Manifest, opts *VerifyOptions,
|
manifest *snapshot.Manifest, opts *VerifyOptions,
|
||||||
) (int, int, int, int64, []UncheckedBlob, error) {
|
) (int, int, int, int64, error) {
|
||||||
var (
|
var (
|
||||||
verified, missing, mismatched int
|
verified, missing, mismatched int
|
||||||
missingSize int64
|
missingSize int64
|
||||||
unchecked []UncheckedBlob
|
|
||||||
)
|
)
|
||||||
|
|
||||||
for _, blob := range manifest.Blobs {
|
for _, blob := range manifest.Blobs {
|
||||||
// The manifest is unauthenticated, so its blob hashes are checked
|
// The manifest is unauthenticated, so its blob hashes are checked
|
||||||
// before being spliced into a storage path.
|
// before being spliced into a storage path.
|
||||||
if !isBlobHash(blob.Hash) {
|
if !isBlobHash(blob.Hash) {
|
||||||
return 0, 0, 0, 0, nil, fmt.Errorf("%w: %s",
|
return 0, 0, 0, 0, fmt.Errorf("%w: %s",
|
||||||
errInvalidBlobHash, shortHash(blob.Hash))
|
errInvalidBlobHash, shortHash(blob.Hash))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -866,7 +849,7 @@ func (v *Vaultik) verifyManifestBlobs(
|
|||||||
|
|
||||||
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||||
switch {
|
switch {
|
||||||
case errors.Is(err, storage.ErrNotFound):
|
case err != nil:
|
||||||
if !opts.JSON && !v.UI.Quiet() {
|
if !opts.JSON && !v.UI.Quiet() {
|
||||||
v.stdoutf(" Missing: %s (%s)\n",
|
v.stdoutf(" Missing: %s (%s)\n",
|
||||||
blob.Hash, ubytes(blob.CompressedSize))
|
blob.Hash, ubytes(blob.CompressedSize))
|
||||||
@@ -874,13 +857,6 @@ func (v *Vaultik) verifyManifestBlobs(
|
|||||||
|
|
||||||
missing++
|
missing++
|
||||||
missingSize += blob.CompressedSize
|
missingSize += blob.CompressedSize
|
||||||
case err != nil:
|
|
||||||
if !opts.JSON && !v.UI.Quiet() {
|
|
||||||
v.stdoutf(" Could not check: %s (%v)\n", blob.Hash, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
unchecked = append(unchecked,
|
|
||||||
UncheckedBlob{Hash: blob.Hash, Error: err.Error()})
|
|
||||||
case stat.Size != blob.CompressedSize:
|
case stat.Size != blob.CompressedSize:
|
||||||
if !opts.JSON && !v.UI.Quiet() {
|
if !opts.JSON && !v.UI.Quiet() {
|
||||||
v.stdoutf(" Wrong size: %s (store has %s, manifest lists %s)\n",
|
v.stdoutf(" Wrong size: %s (store has %s, manifest lists %s)\n",
|
||||||
@@ -893,7 +869,7 @@ func (v *Vaultik) verifyManifestBlobs(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return verified, missing, mismatched, missingSize, unchecked, nil
|
return verified, missing, mismatched, missingSize, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// formatVerifyResult outputs the final verification results as JSON or
|
// formatVerifyResult outputs the final verification results as JSON or
|
||||||
@@ -914,19 +890,9 @@ func (v *Vaultik) formatVerifyResult(
|
|||||||
return v.outputVerifyJSON(result)
|
return v.outputVerifyJSON(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
// --quiet drops the report but not the errors storage returned, which
|
// Under --quiet a failure is still returned, and the cli layer
|
||||||
// are logged on stderr. A failure is still returned either way, and
|
// prints it on stderr.
|
||||||
// the cli layer prints it on stderr.
|
if !v.UI.Quiet() {
|
||||||
if v.UI.Quiet() {
|
|
||||||
for _, blob := range result.Unchecked {
|
|
||||||
log.Warn("Could not check blob", "blob", blob.Hash, "error", blob.Error)
|
|
||||||
}
|
|
||||||
|
|
||||||
if result.DatabaseCheckError != "" {
|
|
||||||
log.Warn("Could not check the encrypted database",
|
|
||||||
"snapshot", result.SnapshotID, "error", result.DatabaseCheckError)
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
v.printVerifySummary(result, failure)
|
v.printVerifySummary(result, failure)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -952,19 +918,10 @@ func (v *Vaultik) printVerifySummary(result *VerifyResult, failure string) {
|
|||||||
v.stdoutf(" Wrong size: %d blobs\n", result.Mismatched)
|
v.stdoutf(" Wrong size: %d blobs\n", result.Mismatched)
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(result.Unchecked) > 0 {
|
|
||||||
v.stdoutf(" Could not check: %d blobs\n", len(result.Unchecked))
|
|
||||||
}
|
|
||||||
|
|
||||||
if result.DatabaseMissing {
|
if result.DatabaseMissing {
|
||||||
v.stdoutf(" Encrypted database: missing\n")
|
v.stdoutf(" Encrypted database: missing\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
if result.DatabaseCheckError != "" {
|
|
||||||
v.stdoutf(" Encrypted database: could not be checked (%s)\n",
|
|
||||||
result.DatabaseCheckError)
|
|
||||||
}
|
|
||||||
|
|
||||||
v.stdoutf(" Status: ")
|
v.stdoutf(" Status: ")
|
||||||
|
|
||||||
if failure != "" {
|
if failure != "" {
|
||||||
@@ -992,19 +949,10 @@ func shallowVerifyFailure(result *VerifyResult) string {
|
|||||||
fmt.Sprintf("%d blobs have the wrong size", result.Mismatched))
|
fmt.Sprintf("%d blobs have the wrong size", result.Mismatched))
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(result.Unchecked) > 0 {
|
|
||||||
parts = append(parts, fmt.Sprintf("%d blobs could not be checked",
|
|
||||||
len(result.Unchecked)))
|
|
||||||
}
|
|
||||||
|
|
||||||
if result.DatabaseMissing {
|
if result.DatabaseMissing {
|
||||||
parts = append(parts, "the encrypted database is missing")
|
parts = append(parts, "the encrypted database is missing")
|
||||||
}
|
}
|
||||||
|
|
||||||
if result.DatabaseCheckError != "" {
|
|
||||||
parts = append(parts, "the encrypted database could not be checked")
|
|
||||||
}
|
|
||||||
|
|
||||||
return strings.Join(parts, "; ")
|
return strings.Join(parts, "; ")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -487,7 +487,6 @@ 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"`
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -57,25 +57,11 @@ type VerifyResult struct {
|
|||||||
Missing int `json:"missing"`
|
Missing int `json:"missing"`
|
||||||
MissingSize int64 `json:"missing_size,omitempty"`
|
MissingSize int64 `json:"missing_size,omitempty"`
|
||||||
Mismatched int `json:"mismatched,omitempty"`
|
Mismatched int `json:"mismatched,omitempty"`
|
||||||
// Unchecked lists the blobs shallow verify could not check, because
|
|
||||||
// storage answered with an error other than not-found, so whether
|
|
||||||
// they are present is unknown.
|
|
||||||
Unchecked []UncheckedBlob `json:"unchecked,omitempty"`
|
|
||||||
// DatabaseMissing is set by shallow verify when the snapshot's
|
// DatabaseMissing is set by shallow verify when the snapshot's
|
||||||
// encrypted database (metadata/<key>/db.zst.age) is absent, which
|
// encrypted database (metadata/<key>/db.zst.age) is absent, which
|
||||||
// makes the snapshot unrestorable regardless of the blobs.
|
// makes the snapshot unrestorable regardless of the blobs.
|
||||||
DatabaseMissing bool `json:"database_missing,omitempty"`
|
DatabaseMissing bool `json:"database_missing,omitempty"`
|
||||||
// DatabaseCheckError is set by shallow verify to the error storage
|
ErrorMessage string `json:"error,omitempty"`
|
||||||
// returned when the encrypted database could not be checked.
|
|
||||||
DatabaseCheckError string `json:"database_check_error,omitempty"`
|
|
||||||
ErrorMessage string `json:"error,omitempty"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// UncheckedBlob is a blob shallow verify could not check, with the error
|
|
||||||
// storage returned for it.
|
|
||||||
type UncheckedBlob struct {
|
|
||||||
Hash string `json:"hash"`
|
|
||||||
Error string `json:"error"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// deepVerifyFailure records a failure in the result and returns it appropriately
|
// deepVerifyFailure records a failure in the result and returns it appropriately
|
||||||
@@ -260,6 +246,8 @@ func (v *Vaultik) downloadVerifiedSnapshotDB(
|
|||||||
fmt.Errorf("failed to download database: %w", err))
|
fmt.Errorf("failed to download database: %w", err))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
dbReader = newCloseOnCancelReader(v.ctx, dbReader)
|
||||||
|
|
||||||
defer func() { _ = dbReader.Close() }()
|
defer func() { _ = dbReader.Close() }()
|
||||||
|
|
||||||
tdb, err := v.decryptAndLoadDatabase(dbReader, identities)
|
tdb, err := v.decryptAndLoadDatabase(dbReader, identities)
|
||||||
|
|||||||
Reference in New Issue
Block a user