From 82eb352eb53e5a4874d6a255c4500e8a61e33a93 Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 7 Aug 2026 16:53:23 +0000 Subject: [PATCH] Apply linter autofixes: internal/vaultik (refs #61) --- internal/vaultik/blob_fetch.go | 10 +- internal/vaultik/blob_fetch_hash_test.go | 12 + internal/vaultik/blobcache.go | 39 ++- internal/vaultik/blobcache_test.go | 20 +- internal/vaultik/helpers.go | 10 +- internal/vaultik/helpers_test.go | 3 + internal/vaultik/info.go | 49 +++- internal/vaultik/integration_test.go | 77 ++++-- internal/vaultik/prune.go | 51 +++- internal/vaultik/purge_per_name_test.go | 3 + internal/vaultik/remove_snapshot_test.go | 14 +- internal/vaultik/restore.go | 140 +++++++++-- internal/vaultik/restore_locality_test.go | 32 ++- internal/vaultik/restore_plan.go | 24 ++ internal/vaultik/restore_sweeper.go | 11 +- .../restore_sweeper_integration_test.go | 24 +- internal/vaultik/snapshot.go | 229 +++++++++++++++--- internal/vaultik/storage_bind.go | 5 +- internal/vaultik/vaultik.go | 9 +- internal/vaultik/verify.go | 73 +++++- internal/vaultik/verify_test.go | 3 + 21 files changed, 730 insertions(+), 108 deletions(-) diff --git a/internal/vaultik/blob_fetch.go b/internal/vaultik/blob_fetch.go index dde5f34..1bd5931 100644 --- a/internal/vaultik/blob_fetch.go +++ b/internal/vaultik/blob_fetch.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" "time" @@ -26,9 +27,10 @@ type hashVerifyReader struct { func (h *hashVerifyReader) Read(p []byte) (int, error) { n, err := h.reader.Read(p) - if err == io.EOF { + if errors.Is(err, io.EOF) { h.done = true } + return n, err } @@ -41,6 +43,7 @@ func (h *hashVerifyReader) Close() error { firstHash := h.reader.Sum256() secondHasher := sha256.New() secondHasher.Write(firstHash) + actualHashHex := hex.EncodeToString(secondHasher.Sum(nil)) if actualHashHex != h.blobHash { return fmt.Errorf("blob hash mismatch: expected %s, got %s", h.blobHash[:16], actualHashHex[:16]) @@ -50,6 +53,7 @@ func (h *hashVerifyReader) Close() error { if readerErr != nil { return readerErr } + return fetcherErr } @@ -66,6 +70,7 @@ func (v *Vaultik) FetchAndDecryptBlob(ctx context.Context, blobHash string, expe reader, err := blobgen.NewReader(rc, identity) if err != nil { _ = rc.Close() + return nil, fmt.Errorf("creating blob reader: %w", err) } @@ -86,6 +91,7 @@ func (v *Vaultik) FetchBlob(ctx context.Context, blobHash string, expectedSize i t0 := time.Now() rc, err := v.Storage.Get(ctx, blobPath) getDur := time.Since(t0) + if err != nil { return nil, 0, fmt.Errorf("downloading blob %s: %w", blobHash[:16], err) } @@ -93,8 +99,10 @@ func (v *Vaultik) FetchBlob(ctx context.Context, blobHash string, expectedSize i t0 = time.Now() info, err := v.Storage.Stat(ctx, blobPath) statDur := time.Since(t0) + if err != nil { _ = rc.Close() + return nil, 0, fmt.Errorf("stat blob %s: %w", blobHash[:16], err) } diff --git a/internal/vaultik/blob_fetch_hash_test.go b/internal/vaultik/blob_fetch_hash_test.go index c8f6099..6dee0b6 100644 --- a/internal/vaultik/blob_fetch_hash_test.go +++ b/internal/vaultik/blob_fetch_hash_test.go @@ -24,17 +24,22 @@ func TestFetchAndDecryptBlobVerifiesHash(t *testing.T) { // Create test data and encrypt it using blobgen.Writer plaintext := []byte("hello world test data for blob hash verification") + var encBuf bytes.Buffer + writer, err := blobgen.NewWriter(&encBuf, 1, []string{identity.Recipient().String()}) if err != nil { t.Fatalf("creating blobgen writer: %v", err) } + if _, err := writer.Write(plaintext); err != nil { t.Fatalf("writing plaintext: %v", err) } + if err := writer.Close(); err != nil { t.Fatalf("closing writer: %v", err) } + encryptedData := encBuf.Bytes() // Compute correct double-SHA-256 hash of the plaintext (matches blobgen.Writer.Sum256) @@ -51,6 +56,7 @@ func TestFetchAndDecryptBlobVerifiesHash(t *testing.T) { // Set up mock storage with the blob at the correct path mockStorage := NewMockStorer() blobPath := "blobs/" + correctHash[:2] + "/" + correctHash[2:4] + "/" + correctHash + mockStorage.mu.Lock() mockStorage.data[blobPath] = encryptedData mockStorage.mu.Unlock() @@ -63,13 +69,16 @@ func TestFetchAndDecryptBlobVerifiesHash(t *testing.T) { if err != nil { t.Fatalf("expected success, got error: %v", err) } + data, err := io.ReadAll(rc) if err != nil { t.Fatalf("reading stream: %v", err) } + if err := rc.Close(); err != nil { t.Fatalf("close (hash verification) failed: %v", err) } + if !bytes.Equal(data, plaintext) { t.Fatalf("decrypted data mismatch: got %q, want %q", data, plaintext) } @@ -79,6 +88,7 @@ func TestFetchAndDecryptBlobVerifiesHash(t *testing.T) { // Use a fake hash that doesn't match the actual plaintext fakeHash := strings.Repeat("ab", 32) // 64 hex chars fakePath := "blobs/" + fakeHash[:2] + "/" + fakeHash[2:4] + "/" + fakeHash + mockStorage.mu.Lock() mockStorage.data[fakePath] = encryptedData mockStorage.mu.Unlock() @@ -89,10 +99,12 @@ func TestFetchAndDecryptBlobVerifiesHash(t *testing.T) { } // Read all data — hash is verified on Close _, _ = io.ReadAll(rc) + err = rc.Close() if err == nil { t.Fatal("expected error for mismatched hash, got nil") } + if !strings.Contains(err.Error(), "hash mismatch") { t.Fatalf("expected hash mismatch error, got: %v", err) } diff --git a/internal/vaultik/blobcache.go b/internal/vaultik/blobcache.go index 7ebbefe..2a28f18 100644 --- a/internal/vaultik/blobcache.go +++ b/internal/vaultik/blobcache.go @@ -53,6 +53,7 @@ func newBlobDiskCache(maxBytes int64) (*blobDiskCache, error) { if err != nil { return nil, fmt.Errorf("creating blob cache dir: %w", err) } + return &blobDiskCache{ dir: dir, maxBytes: maxBytes, @@ -70,21 +71,25 @@ func (c *blobDiskCache) unlink(e *blobDiskCacheEntry) { } else { c.head = e.next } + if e.next != nil { e.next.prev = e.prev } else { c.tail = e.prev } + e.prev = nil e.next = nil } func (c *blobDiskCache) pushFront(e *blobDiskCacheEntry) { e.prev = nil + e.next = c.head if c.head != nil { c.head.prev = e } + c.head = e if c.tail == nil { c.tail = e @@ -95,6 +100,7 @@ func (c *blobDiskCache) evictLRU() { if c.tail == nil { return } + victim := c.tail c.unlink(victim) delete(c.items, victim.key) @@ -121,7 +127,8 @@ func (c *blobDiskCache) Put(key string, data []byte) error { delete(c.items, key) } - if err := os.WriteFile(c.path(key), data, 0600); err != nil { + err := os.WriteFile(c.path(key), data, 0600) + if err != nil { return fmt.Errorf("writing blob to cache: %w", err) } @@ -163,14 +170,19 @@ func (c *blobDiskCache) PutFromReader(key string, r io.Reader) (int64, error) { if err != nil { return 0, fmt.Errorf("creating cache file: %w", err) } + written, copyErr := io.Copy(f, r) closeErr := f.Close() + if copyErr != nil { _ = os.Remove(c.path(key)) + return written, fmt.Errorf("streaming to cache file: %w", copyErr) } + if closeErr != nil { _ = os.Remove(c.path(key)) + return written, fmt.Errorf("closing cache file: %w", closeErr) } @@ -182,6 +194,7 @@ func (c *blobDiskCache) PutFromReader(key string, r io.Reader) (int64, error) { // so this branch is effectively unreachable there. if written > c.maxBytes { _ = os.Remove(c.path(key)) + return written, nil } @@ -205,11 +218,14 @@ func (c *blobDiskCache) PutFromReader(key string, r io.Reader) (int64, error) { func (c *blobDiskCache) Get(key string) ([]byte, bool) { c.mu.Lock() c.getCalls++ + e, ok := c.items[key] if !ok { c.mu.Unlock() + return nil, false } + c.unlink(e) c.pushFront(e) c.mu.Unlock() @@ -223,8 +239,10 @@ func (c *blobDiskCache) Get(key string) ([]byte, bool) { c.curBytes -= e.size } c.mu.Unlock() + return nil, false } + return data, true } @@ -232,15 +250,20 @@ func (c *blobDiskCache) Get(key string) ([]byte, bool) { func (c *blobDiskCache) ReadAt(key string, offset, length int64) ([]byte, error) { c.mu.Lock() c.readAtCalls++ + e, ok := c.items[key] if !ok { c.mu.Unlock() + return nil, fmt.Errorf("key %q not in cache", key) } + if offset+length > e.size { c.mu.Unlock() + return nil, fmt.Errorf("read beyond blob size: offset=%d length=%d size=%d", offset, length, e.size) } + c.unlink(e) c.pushFront(e) c.mu.Unlock() @@ -255,6 +278,7 @@ func (c *blobDiskCache) ReadAt(key string, offset, length int64) ([]byte, error) if _, err := f.ReadAt(buf, offset); err != nil { return nil, err } + return buf, nil } @@ -262,7 +286,9 @@ func (c *blobDiskCache) ReadAt(key string, offset, length int64) ([]byte, error) func (c *blobDiskCache) Has(key string) bool { c.mu.Lock() defer c.mu.Unlock() + _, ok := c.items[key] + return ok } @@ -272,10 +298,12 @@ func (c *blobDiskCache) Has(key string) bool { func (c *blobDiskCache) Delete(key string) { c.mu.Lock() defer c.mu.Unlock() + e, ok := c.items[key] if !ok { return } + c.unlink(e) delete(c.items, key) c.curBytes -= e.size @@ -287,10 +315,12 @@ func (c *blobDiskCache) Delete(key string) { func (c *blobDiskCache) Keys() []string { c.mu.Lock() defer c.mu.Unlock() + keys := make([]string, 0, len(c.items)) for k := range c.items { keys = append(keys, k) } + return keys } @@ -298,6 +328,7 @@ func (c *blobDiskCache) Keys() []string { func (c *blobDiskCache) Size() int64 { c.mu.Lock() defer c.mu.Unlock() + return c.curBytes } @@ -305,6 +336,7 @@ func (c *blobDiskCache) Size() int64 { func (c *blobDiskCache) Len() int { c.mu.Lock() defer c.mu.Unlock() + return len(c.items) } @@ -312,6 +344,7 @@ func (c *blobDiskCache) Len() int { func (c *blobDiskCache) GetCalls() int { c.mu.Lock() defer c.mu.Unlock() + return c.getCalls } @@ -319,6 +352,7 @@ func (c *blobDiskCache) GetCalls() int { func (c *blobDiskCache) ReadAtCalls() int { c.mu.Lock() defer c.mu.Unlock() + return c.readAtCalls } @@ -327,6 +361,7 @@ func (c *blobDiskCache) ReadAtCalls() int { func (c *blobDiskCache) PeakLen() int { c.mu.Lock() defer c.mu.Unlock() + return c.peakLen } @@ -334,9 +369,11 @@ func (c *blobDiskCache) PeakLen() int { func (c *blobDiskCache) Close() error { c.mu.Lock() defer c.mu.Unlock() + c.items = nil c.head = nil c.tail = nil c.curBytes = 0 + return os.RemoveAll(c.dir) } diff --git a/internal/vaultik/blobcache_test.go b/internal/vaultik/blobcache_test.go index 778aadd..7467eff 100644 --- a/internal/vaultik/blobcache_test.go +++ b/internal/vaultik/blobcache_test.go @@ -23,6 +23,7 @@ func TestBlobDiskCache_BasicGetPut(t *testing.T) { if !ok { t.Fatal("expected cache hit") } + if !bytes.Equal(got, data) { t.Fatalf("got %q, want %q", got, data) } @@ -35,15 +36,19 @@ func TestBlobDiskCache_BasicGetPut(t *testing.T) { func TestBlobDiskCache_EvictionUnderPressure(t *testing.T) { maxBytes := int64(1000) + cache, err := newBlobDiskCache(maxBytes) if err != nil { t.Fatal(err) } + defer func() { _ = cache.Close() }() - for i := 0; i < 5; i++ { + for i := range 5 { data := make([]byte, 300) - if err := cache.Put(fmt.Sprintf("key%d", i), data); err != nil { + + err := cache.Put(fmt.Sprintf("key%d", i), data) + if err != nil { t.Fatal(err) } } @@ -55,6 +60,7 @@ func TestBlobDiskCache_EvictionUnderPressure(t *testing.T) { if !cache.Has("key4") { t.Fatal("expected key4 to be cached") } + if cache.Has("key0") { t.Fatal("expected key0 to be evicted") } @@ -87,6 +93,7 @@ func TestBlobDiskCache_UpdateInPlace(t *testing.T) { if err := cache.Put("key1", []byte("v1")); err != nil { t.Fatal(err) } + if err := cache.Put("key1", []byte("version2")); err != nil { t.Fatal(err) } @@ -95,12 +102,15 @@ func TestBlobDiskCache_UpdateInPlace(t *testing.T) { if !ok { t.Fatal("expected hit") } + if string(got) != "version2" { t.Fatalf("got %q, want %q", got, "version2") } + if cache.Len() != 1 { t.Fatalf("expected 1 entry, got %d", cache.Len()) } + if cache.Size() != int64(len("version2")) { t.Fatalf("expected size %d, got %d", len("version2"), cache.Size()) } @@ -117,6 +127,7 @@ func TestBlobDiskCache_ReadAt(t *testing.T) { if _, err := rand.Read(data); err != nil { t.Fatal(err) } + if err := cache.Put("blob1", data); err != nil { t.Fatal(err) } @@ -125,6 +136,7 @@ func TestBlobDiskCache_ReadAt(t *testing.T) { if err != nil { t.Fatal(err) } + if !bytes.Equal(chunk, data[100:300]) { t.Fatal("ReadAt returned wrong data") } @@ -149,6 +161,7 @@ func TestBlobDiskCache_Close(t *testing.T) { if err := cache.Put("key1", []byte("data")); err != nil { t.Fatal(err) } + if err := cache.Close(); err != nil { t.Fatal(err) } @@ -165,6 +178,7 @@ func TestBlobDiskCache_LRUOrder(t *testing.T) { if err := cache.Put("a", d); err != nil { t.Fatal(err) } + if err := cache.Put("b", d); err != nil { t.Fatal(err) } @@ -180,9 +194,11 @@ func TestBlobDiskCache_LRUOrder(t *testing.T) { if !cache.Has("a") { t.Fatal("expected 'a' to survive") } + if !cache.Has("c") { t.Fatal("expected 'c' to be present") } + if cache.Has("b") { t.Fatal("expected 'b' to be evicted") } diff --git a/internal/vaultik/helpers.go b/internal/vaultik/helpers.go index deb0d3b..b10e161 100644 --- a/internal/vaultik/helpers.go +++ b/internal/vaultik/helpers.go @@ -1,6 +1,7 @@ package vaultik import ( + "errors" "fmt" "regexp" "strconv" @@ -29,11 +30,13 @@ func formatBytes(bytes int64) string { 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]) } @@ -42,11 +45,12 @@ func formatBytes(bytes int64) string { func parseSnapshotTimestamp(snapshotID string) (time.Time, error) { parts := strings.Split(snapshotID, "_") if len(parts) < 2 { - return time.Time{}, fmt.Errorf("invalid snapshot ID format: expected hostname_snapshotname_timestamp") + return time.Time{}, errors.New("invalid snapshot ID format: expected hostname_snapshotname_timestamp") } // Last part is the RFC3339 timestamp timestampStr := parts[len(parts)-1] + timestamp, err := time.Parse(time.RFC3339, timestampStr) if err != nil { return time.Time{}, fmt.Errorf("invalid timestamp: %w", err) @@ -80,17 +84,20 @@ func parseDuration(s string) (time.Duration, error) { } re := regexp.MustCompile(`(\d+)\s*([a-zA-Z]+)`) + matches := re.FindAllStringSubmatch(s, -1) if len(matches) == 0 { return 0, fmt.Errorf("invalid duration: %q", s) } var total time.Duration + for _, match := range matches { n, err := strconv.Atoi(match[1]) if err != nil { return 0, fmt.Errorf("invalid number %q: %w", match[1], err) } + unit := strings.ToLower(match[2]) switch unit { case "d", "day", "days": @@ -105,5 +112,6 @@ func parseDuration(s string) (time.Duration, error) { return 0, fmt.Errorf("unknown time unit %q", unit) } } + return total, nil } diff --git a/internal/vaultik/helpers_test.go b/internal/vaultik/helpers_test.go index 76a3ea3..e648bec 100644 --- a/internal/vaultik/helpers_test.go +++ b/internal/vaultik/helpers_test.go @@ -61,11 +61,14 @@ func TestParseDuration(t *testing.T) { if err == nil { t.Fatalf("expected error for %q, got %v", tt.input, got) } + return } + if err != nil { t.Fatalf("unexpected error for %q: %v", tt.input, err) } + if got != tt.want { t.Errorf("parseDuration(%q) = %v, want %v", tt.input, got, tt.want) } diff --git a/internal/vaultik/info.go b/internal/vaultik/info.go index ac5f76f..11d427a 100644 --- a/internal/vaultik/info.go +++ b/internal/vaultik/info.go @@ -30,21 +30,27 @@ func (v *Vaultik) ShowInfo() error { storageInfo := v.Storage.Info() v.printfStdout("Type: %s\n", storageInfo.Type) v.printfStdout("Location: %s\n", storageInfo.Location) + if v.Config.StorageURL != "" { v.printfStdout("Storage URL: %s\n", v.Config.StorageURL) } + if v.Config.S3.Bucket != "" { v.printfStdout("S3 Bucket: %s\n", v.Config.S3.Bucket) } + if v.Config.S3.Prefix != "" { v.printfStdout("S3 Prefix: %s\n", v.Config.S3.Prefix) } + if v.Config.S3.Endpoint != "" { v.printfStdout("S3 Endpoint: %s\n", v.Config.S3.Endpoint) } + if v.Config.S3.Region != "" { v.printfStdout("S3 Region: %s\n", v.Config.S3.Region) } + v.printlnStdout() // Backup Settings @@ -52,12 +58,15 @@ func (v *Vaultik) ShowInfo() error { // Show configured snapshots v.printfStdout("Snapshots:\n") + for _, name := range v.Config.SnapshotNames() { snap := v.Config.Snapshots[name] v.printfStdout(" %s:\n", name) + for _, path := range snap.Paths { v.printfStdout(" - %s\n", path) } + if len(snap.Exclude) > 0 { v.printfStdout(" exclude: %s\n", strings.Join(snap.Exclude, ", ")) } @@ -76,9 +85,11 @@ func (v *Vaultik) ShowInfo() error { // Encryption Configuration v.printfStdout("=== Encryption Configuration ===\n") v.printfStdout("Recipients:\n") + for _, recipient := range v.Config.AgeRecipients { v.printfStdout(" - %s\n", recipient) } + v.printlnStdout() // Local Database @@ -91,22 +102,31 @@ func (v *Vaultik) ShowInfo() error { // Get snapshot count from database query := `SELECT COUNT(*) FROM snapshots WHERE completed_at IS NOT NULL` + var snapshotCount int - if err := v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&snapshotCount); err == nil { + + err := v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&snapshotCount) + if err == nil { v.printfStdout("Snapshots: %d\n", snapshotCount) } // Get blob count from database query = `SELECT COUNT(*) FROM blobs` + var blobCount int - if err := v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&blobCount); err == nil { + + err = v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&blobCount) + if err == nil { v.printfStdout("Blobs: %d\n", blobCount) } // Get file count from database query = `SELECT COUNT(*) FROM files` + var fileCount int - if err := v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&fileCount); err == nil { + + err = v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&fileCount) + if err == nil { v.printfStdout("Files: %d\n", fileCount) } } else { @@ -153,6 +173,7 @@ type RemoteInfoResult struct { // RemoteInfo displays information about remote storage func (v *Vaultik) RemoteInfo(jsonOutput bool) error { log.Info("Starting remote storage info gathering") + result := &RemoteInfoResult{} storageInfo := v.Storage.Info() @@ -193,10 +214,12 @@ func (v *Vaultik) RemoteInfo(jsonOutput bool) error { if jsonOutput { enc := json.NewEncoder(v.Stdout) enc.SetIndent("", " ") + return enc.Encode(result) } v.printRemoteInfoTable(result) + return nil } @@ -214,6 +237,7 @@ func (v *Vaultik) collectSnapshotMetadata() (map[string]*SnapshotMetadataInfo, [ if len(parts) < 3 { continue } + snapshotID := parts[1] if _, exists := snapshotMetadata[snapshotID]; !exists { @@ -221,12 +245,14 @@ func (v *Vaultik) collectSnapshotMetadata() (map[string]*SnapshotMetadataInfo, [ } info := snapshotMetadata[snapshotID] + filename := parts[2] if strings.HasPrefix(filename, "manifest") { info.ManifestSize = obj.Size } else if strings.HasPrefix(filename, "db") { info.DatabaseSize = obj.Size } + info.TotalSize = info.ManifestSize + info.DatabaseSize } @@ -234,6 +260,7 @@ func (v *Vaultik) collectSnapshotMetadata() (map[string]*SnapshotMetadataInfo, [ for id := range snapshotMetadata { snapshotIDs = append(snapshotIDs, id) } + sort.Strings(snapshotIDs) return snapshotMetadata, snapshotIDs, nil @@ -245,26 +272,33 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(snapshotIDs []string, snap for _, snapshotID := range snapshotIDs { manifestKey := fmt.Sprintf("metadata/%s/manifest.json.zst", snapshotID) + reader, err := v.Storage.Get(v.ctx, manifestKey) if err != nil { log.Warn("Failed to get manifest", "snapshot", snapshotID, "error", err) + continue } manifest, err := snapshot.DecodeManifest(reader) _ = reader.Close() + if err != nil { log.Warn("Failed to decode manifest", "snapshot", snapshotID, "error", err) + continue } info := snapshotMetadata[snapshotID] info.BlobCount = manifest.BlobCount + var blobsSize int64 + for _, blob := range manifest.Blobs { referencedBlobs[blob.Hash] = blob.CompressedSize blobsSize += blob.CompressedSize } + info.BlobsSize = blobsSize } @@ -274,11 +308,13 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(snapshotIDs []string, snap // populateRemoteInfoResult fills in the result's snapshot and referenced blob stats func (v *Vaultik) populateRemoteInfoResult(result *RemoteInfoResult, snapshotMetadata map[string]*SnapshotMetadataInfo, snapshotIDs []string, referencedBlobs map[string]int64) { var totalMetadataSize int64 + for _, id := range snapshotIDs { info := snapshotMetadata[id] result.Snapshots = append(result.Snapshots, *info) totalMetadataSize += info.TotalSize } + result.TotalMetadataSize = totalMetadataSize result.TotalMetadataCount = len(snapshotIDs) @@ -301,10 +337,12 @@ func (v *Vaultik) scanRemoteBlobStorage(result *RemoteInfoResult, referencedBlob if obj.Err != nil { return fmt.Errorf("listing blobs: %w", obj.Err) } + parts := strings.Split(obj.Key, "/") if len(parts) < 4 { continue } + hash := parts[3] allBlobs[hash] = obj.Size result.TotalBlobCount++ @@ -324,11 +362,13 @@ func (v *Vaultik) scanRemoteBlobStorage(result *RemoteInfoResult, referencedBlob // printRemoteInfoTable renders the human-readable remote info output func (v *Vaultik) printRemoteInfoTable(result *RemoteInfoResult) { v.printfStdout("\n=== Snapshot Metadata ===\n") + if len(result.Snapshots) == 0 { v.printfStdout("No snapshots found\n") } else { v.printfStdout("%-45s %12s %12s %12s %10s %12s\n", "SNAPSHOT", "MANIFEST", "DATABASE", "TOTAL", "BLOBS", "BLOB SIZE") v.printfStdout("%-45s %12s %12s %12s %10s %12s\n", strings.Repeat("-", 45), strings.Repeat("-", 12), strings.Repeat("-", 12), strings.Repeat("-", 12), strings.Repeat("-", 10), strings.Repeat("-", 12)) + for _, info := range result.Snapshots { v.printfStdout("%-45s %12s %12s %12s %10s %12s\n", truncateString(info.SnapshotID, 45), @@ -339,6 +379,7 @@ func (v *Vaultik) printRemoteInfoTable(result *RemoteInfoResult) { humanize.Bytes(uint64(info.BlobsSize)), ) } + v.printfStdout("%-45s %12s %12s %12s %10s %12s\n", strings.Repeat("-", 45), strings.Repeat("-", 12), strings.Repeat("-", 12), strings.Repeat("-", 12), strings.Repeat("-", 10), strings.Repeat("-", 12)) v.printfStdout("%-45s %12s %12s %12s\n", fmt.Sprintf("Total (%d snapshots)", result.TotalMetadataCount), "", "", humanize.Bytes(uint64(result.TotalMetadataSize))) } @@ -361,8 +402,10 @@ func truncateString(s string, maxLen int) string { if len(s) <= maxLen { return s } + if maxLen <= 3 { return s[:maxLen] } + return s[:maxLen-3] + "..." } diff --git a/internal/vaultik/integration_test.go b/internal/vaultik/integration_test.go index 945d448..78a1c18 100644 --- a/internal/vaultik/integration_test.go +++ b/internal/vaultik/integration_test.go @@ -43,11 +43,14 @@ func (m *MockStorer) Put(ctx context.Context, key string, reader io.Reader) erro defer m.mu.Unlock() m.calls = append(m.calls, "Put:"+key) + data, err := io.ReadAll(reader) if err != nil { return err } + m.data[key] = data + return nil } @@ -60,10 +63,12 @@ func (m *MockStorer) Get(ctx context.Context, key string) (io.ReadCloser, error) defer m.mu.Unlock() m.calls = append(m.calls, "Get:"+key) + data, exists := m.data[key] if !exists { return nil, storage.ErrNotFound } + return io.NopCloser(bytes.NewReader(data)), nil } @@ -72,10 +77,12 @@ func (m *MockStorer) Stat(ctx context.Context, key string) (*storage.ObjectInfo, defer m.mu.Unlock() m.calls = append(m.calls, "Stat:"+key) + data, exists := m.data[key] if !exists { return nil, storage.ErrNotFound } + return &storage.ObjectInfo{ Key: key, Size: int64(len(data)), @@ -88,6 +95,7 @@ func (m *MockStorer) Delete(ctx context.Context, key string) error { m.calls = append(m.calls, "Delete:"+key) delete(m.data, key) + return nil } @@ -96,12 +104,15 @@ func (m *MockStorer) List(ctx context.Context, prefix string) ([]string, error) defer m.mu.Unlock() m.calls = append(m.calls, "List:"+prefix) + var keys []string + for key := range m.data { if len(prefix) == 0 || (len(key) >= len(prefix) && key[:len(prefix)] == prefix) { keys = append(keys, key) } } + return keys, nil } @@ -109,6 +120,7 @@ func (m *MockStorer) ListStream(ctx context.Context, prefix string) <-chan stora ch := make(chan storage.ObjectInfo) go func() { defer close(ch) + m.mu.Lock() defer m.mu.Unlock() @@ -121,6 +133,7 @@ func (m *MockStorer) ListStream(ctx context.Context, prefix string) <-chan stora } } }() + return ch } @@ -138,6 +151,7 @@ func (m *MockStorer) GetCalls() []string { calls := make([]string, len(m.calls)) copy(calls, m.calls) + return calls } @@ -172,14 +186,16 @@ func TestEndToEndBackup(t *testing.T) { "/home/user/code", } for _, dir := range dirs { - if err := fs.MkdirAll(dir, 0755); err != nil { + err := fs.MkdirAll(dir, 0755) + if err != nil { t.Fatalf("failed to create directory %s: %v", dir, err) } } // Create test files for path, content := range testFiles { - if err := afero.WriteFile(fs, path, []byte(content), 0644); err != nil { + err := afero.WriteFile(fs, path, []byte(content), 0644) + if err != nil { t.Fatalf("failed to create test file %s: %v", path, err) } } @@ -216,9 +232,11 @@ func TestEndToEndBackup(t *testing.T) { // Create in-memory database db, err := database.New(ctx, ":memory:") + require.NoError(t, err) defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -246,6 +264,7 @@ func TestEndToEndBackup(t *testing.T) { VaultikVersion: "test-version", StartedAt: time.Now(), } + return repos.Snapshots.Create(ctx, tx, snapshot) }) require.NoError(t, err) @@ -258,9 +277,9 @@ func TestEndToEndBackup(t *testing.T) { // The scanner counts both files and directories, so we have: // 4 files + 4 directories (/home, /home/user, /home/user/documents, /home/user/pictures, /home/user/code) assert.GreaterOrEqual(t, result.FilesScanned, 4, "Should scan at least 4 files") - assert.Greater(t, result.BytesScanned, int64(0), "Should scan some bytes") - assert.Greater(t, result.ChunksCreated, 0, "Should create chunks") - assert.Greater(t, result.BlobsCreated, 0, "Should create blobs") + assert.Positive(t, result.BytesScanned, "Should scan some bytes") + assert.Positive(t, result.ChunksCreated, "Should create chunks") + assert.Positive(t, result.BlobsCreated, "Should create blobs") // Verify storage operations calls := mockStorage.GetCalls() @@ -268,6 +287,7 @@ func TestEndToEndBackup(t *testing.T) { // Should have uploaded at least one blob blobUploads := 0 + for _, call := range calls { if len(call) > 4 && call[:4] == "Put:" { if len(call) > 10 && call[4:10] == "blobs/" { @@ -275,27 +295,30 @@ func TestEndToEndBackup(t *testing.T) { } } } - assert.Greater(t, blobUploads, 0, "Should upload at least one blob") + + assert.Positive(t, blobUploads, "Should upload at least one blob") // Verify files in database files, err := repos.Files.ListByPrefix(ctx, "/home/user") require.NoError(t, err) // Count only regular files (not directories) regularFiles := 0 + for _, f := range files { if f.Mode&0x80000000 == 0 { // Check if regular file (not directory) regularFiles++ } } + assert.Equal(t, 4, regularFiles, "Should have 4 regular files in database") // Verify chunks were created by checking a specific file fileChunks, err := repos.FileChunks.GetByPath(ctx, "/home/user/documents/file1.txt") require.NoError(t, err) - assert.Greater(t, len(fileChunks), 0, "Should have chunks for file1.txt") + assert.NotEmpty(t, fileChunks, "Should have chunks for file1.txt") // Verify blobs were uploaded to storage - assert.Greater(t, mockStorage.GetStorageSize(), 0, "Should have blobs in storage") + assert.Positive(t, mockStorage.GetStorageSize(), "Should have blobs in storage") // Complete the snapshot - just verify we got results // In a real integration test, we'd update the snapshot record @@ -337,9 +360,11 @@ func TestBackupAndVerify(t *testing.T) { // Create test database ctx := context.Background() db, err := database.New(ctx, ":memory:") + require.NoError(t, err) defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -366,6 +391,7 @@ func TestBackupAndVerify(t *testing.T) { VaultikVersion: "test-version", StartedAt: time.Now(), } + return repos.Snapshots.Create(ctx, tx, snapshot) }) require.NoError(t, err) @@ -375,7 +401,7 @@ func TestBackupAndVerify(t *testing.T) { require.NoError(t, err) // Verify backup created blobs - assert.Greater(t, result.BlobsCreated, 0, "Should create at least one blob") + assert.Positive(t, result.BlobsCreated, "Should create at least one blob") assert.Equal(t, mockStorage.GetStorageSize(), result.BlobsCreated, "Storage should have the blobs") // Verify we can retrieve the blob from storage @@ -391,18 +417,19 @@ func TestBackupAndVerify(t *testing.T) { // Get blob info blobInfo, err := mockStorage.Stat(ctx, blobKey) require.NoError(t, err) - assert.Greater(t, blobInfo.Size, int64(0), "Blob should have content") + assert.Positive(t, blobInfo.Size, "Blob should have content") // Get blob content reader, err := mockStorage.Get(ctx, blobKey) require.NoError(t, err) + defer func() { _ = reader.Close() }() // Verify blob data is encrypted (should not contain plaintext) blobData, err := io.ReadAll(reader) require.NoError(t, err) assert.NotContains(t, string(blobData), testContent, "Blob should be encrypted") - assert.Greater(t, len(blobData), 0, "Blob should have data") + assert.NotEmpty(t, blobData, "Blob should have data") } t.Logf("Backup and verify test completed successfully") @@ -418,6 +445,7 @@ func TestBackupAndRestore(t *testing.T) { // Create real temp directory for the database (SQLite needs real filesystem) realTempDir, err := os.MkdirTemp("", "vaultik-test-") require.NoError(t, err) + defer func() { _ = os.RemoveAll(realTempDir) }() // Use real OS filesystem for this test @@ -434,10 +462,14 @@ func TestBackupAndRestore(t *testing.T) { // Create directories and files for path, content := range testFiles { dir := filepath.Dir(path) - if err := fs.MkdirAll(dir, 0755); err != nil { + + err := fs.MkdirAll(dir, 0755) + if err != nil { t.Fatalf("failed to create directory %s: %v", dir, err) } - if err := afero.WriteFile(fs, path, []byte(content), 0644); err != nil { + + err = afero.WriteFile(fs, path, []byte(content), 0644) + if err != nil { t.Fatalf("failed to create test file %s: %v", path, err) } } @@ -455,6 +487,7 @@ func TestBackupAndRestore(t *testing.T) { dbPath := filepath.Join(realTempDir, "test.db") db, err := database.New(ctx, dbPath) require.NoError(t, err) + defer func() { _ = db.Close() }() repos := database.NewRepositories(db) @@ -558,6 +591,7 @@ func TestEndToEndFileStorage(t *testing.T) { fs := afero.NewOsFs() tempDir, err := os.MkdirTemp("", "vaultik-e2e-") require.NoError(t, err) + defer func() { _ = os.RemoveAll(tempDir) }() dataDir := filepath.Join(tempDir, "source") @@ -618,6 +652,7 @@ func TestEndToEndFileStorage(t *testing.T) { db, err := database.New(ctx, dbPath) require.NoError(t, err) + defer func() { _ = db.Close() }() repos := database.NewRepositories(db) @@ -644,8 +679,8 @@ func TestEndToEndFileStorage(t *testing.T) { scanResult, err := scanner.Scan(ctx, dataDir, snapshotID) require.NoError(t, err) - require.Greater(t, scanResult.FilesScanned, 0) - require.Greater(t, scanResult.BlobsCreated, 0) + require.Positive(t, scanResult.FilesScanned) + require.Positive(t, scanResult.BlobsCreated) require.NoError(t, sm.CompleteSnapshot(ctx, snapshotID)) require.NoError(t, sm.ExportSnapshotMetadata(ctx, dbPath, snapshotID)) @@ -656,6 +691,7 @@ func TestEndToEndFileStorage(t *testing.T) { blobInfo, err := os.Stat(filepath.Join(storeDir, "blobs")) require.NoError(t, err) require.True(t, blobInfo.IsDir()) + metaInfo, err := os.Stat(filepath.Join(storeDir, "metadata", snapshot.RemoteSnapshotKey(snapshotID))) require.NoError(t, err) require.True(t, metaInfo.IsDir()) @@ -721,6 +757,7 @@ func TestDedupOnlySnapshotRestores(t *testing.T) { fs := afero.NewOsFs() tempDir, err := os.MkdirTemp("", "vaultik-dedup-") require.NoError(t, err) + defer func() { _ = os.RemoveAll(tempDir) }() dataDir := filepath.Join(tempDir, "source") @@ -756,7 +793,9 @@ func TestDedupOnlySnapshotRestores(t *testing.T) { ctx := context.Background() db, err := database.New(ctx, dbPath) require.NoError(t, err) + defer func() { _ = db.Close() }() + repos := database.NewRepositories(db) makeScanner := func() *snapshot.Scanner { @@ -780,13 +819,14 @@ func TestDedupOnlySnapshotRestores(t *testing.T) { require.NoError(t, err) r1, err := makeScanner().Scan(ctx, dataDir, id1) require.NoError(t, err) - require.Greater(t, r1.BlobsCreated, 0, "first snapshot should upload at least one blob") + require.Positive(t, r1.BlobsCreated, "first snapshot should upload at least one blob") require.NoError(t, sm.CompleteSnapshot(ctx, id1)) require.NoError(t, sm.ExportSnapshotMetadata(ctx, dbPath, id1)) // Second snapshot — same data, every chunk dedups. Sleep past the // second-precision timestamp so the snapshot IDs differ. time.Sleep(1100 * time.Millisecond) + id2, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "dedup", "v", "g") require.NoError(t, err) r2, err := makeScanner().Scan(ctx, dataDir, id2) @@ -833,5 +873,6 @@ func bytesPattern(tag string, n int) []byte { for i := range out { out[i] = byte(tag[i%len(tag)] ^ byte(i&0xff)) } + return out } diff --git a/internal/vaultik/prune.go b/internal/vaultik/prune.go index 8b3cfac..e358c07 100644 --- a/internal/vaultik/prune.go +++ b/internal/vaultik/prune.go @@ -2,6 +2,7 @@ package vaultik import ( "encoding/json" + "errors" "fmt" "strings" @@ -23,20 +24,24 @@ type PruneOptions struct { // confirming with the user. func (v *Vaultik) NukeRemote(force bool) error { if !force { - return fmt.Errorf("nuke requires --force (this deletes ALL remote snapshots and blobs)") + return errors.New("nuke requires --force (this deletes ALL remote snapshots and blobs)") } v.UI.Begin("Removing all snapshot metadata from backup destination store.") + if _, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true}); err != nil { return fmt.Errorf("removing all snapshots: %w", err) } v.UI.Begin("Removing any blobs still present in backup destination store.") - if err := v.PruneBlobs(&PruneOptions{Force: true}); err != nil { + + err := v.PruneBlobs(&PruneOptions{Force: true}) + if err != nil { return fmt.Errorf("pruning blobs: %w", err) } v.UI.Complete("Backup destination store is now empty.") + return nil } @@ -55,7 +60,8 @@ type PruneBlobsResult struct { // prefer this method over PruneDatabase or PruneBlobs individually // unless it specifically wants one half. func (v *Vaultik) Prune(opts *PruneOptions) error { - if err := v.EnsureStorageBinding(); err != nil { + err := v.EnsureStorageBinding() + if err != nil { return err } // First reconcile local snapshot records against remote metadata: @@ -63,12 +69,15 @@ func (v *Vaultik) Prune(opts *PruneOptions) error { // store is treated as gone. This used to be the separate 'snapshot // cleanup' command and is now folded in so a single 'vaultik prune' // gets the local index fully back in sync with the destination. - if err := v.CleanupLocalSnapshots(); err != nil { + err = v.CleanupLocalSnapshots() + if err != nil { return fmt.Errorf("reconciling local snapshots with remote: %w", err) } + if _, err := v.PruneDatabase(); err != nil { return fmt.Errorf("pruning local database: %w", err) } + return v.PruneBlobs(opts) } @@ -92,27 +101,35 @@ func (v *Vaultik) PruneBlobs(opts *PruneOptions) error { if len(unreferencedBlobs) == 0 { log.Info("No unreferenced blobs found") + if opts.JSON { return v.outputPruneBlobsJSON(result) } + v.printlnStdout("No unreferenced blobs to remove.") + return nil } log.Info("Found unreferenced blobs", "count", len(unreferencedBlobs), "total_size", humanize.Bytes(uint64(totalSize))) + if !opts.JSON { v.printfStdout("Found %d unreferenced blob(s) totaling %s\n", len(unreferencedBlobs), humanize.Bytes(uint64(totalSize))) } if !opts.Force && !opts.JSON { v.printfStdout("\nDelete %d unreferenced blob(s)? [y/N] ", len(unreferencedBlobs)) + var confirm string if _, err := v.scanStdin(&confirm); err != nil { v.printlnStdout("Cancelled") + return nil } + if strings.ToLower(confirm) != "y" { v.printlnStdout("Cancelled") + return nil } } @@ -124,6 +141,7 @@ func (v *Vaultik) PruneBlobs(opts *PruneOptions) error { } v.printfStdout("\nDeleted %d blob(s) totaling %s\n", result.BlobsDeleted, humanize.Bytes(uint64(result.BytesFreed))) + if result.BlobsFailed > 0 { v.printfStdout("Failed to delete %d blob(s)\n", result.BlobsFailed) } @@ -140,6 +158,7 @@ func (v *Vaultik) collectReferencedBlobs() (map[string]bool, error) { if err != nil { return nil, fmt.Errorf("listing snapshot keys: %w", err) } + log.Info("Found manifests in remote storage", "count", len(remoteKeys)) allBlobsReferenced := make(map[string]bool) @@ -147,18 +166,23 @@ func (v *Vaultik) collectReferencedBlobs() (map[string]bool, error) { for _, remoteKey := range remoteKeys { log.Debug("Processing manifest", "remote_key", remoteKey) + manifest, err := v.downloadManifestByKey(remoteKey) if err != nil { log.Error("Failed to download manifest", "remote_key", remoteKey, "error", err) + continue } + for _, blob := range manifest.Blobs { allBlobsReferenced[blob.Hash] = true } + manifestCount++ } log.Info("Processed manifests", "count", manifestCount, "unique_blobs_referenced", len(allBlobsReferenced)) + return allBlobsReferenced, nil } @@ -166,12 +190,14 @@ func (v *Vaultik) collectReferencedBlobs() (map[string]bool, error) { func (v *Vaultik) listUniqueSnapshotIDs() ([]string, error) { objectCh := v.Storage.ListStream(v.ctx, "metadata/") seen := make(map[string]bool) + var snapshotIDs []string for object := range objectCh { if object.Err != nil { return nil, fmt.Errorf("listing metadata objects: %w", object.Err) } + parts := strings.Split(object.Key, "/") if len(parts) >= 2 && parts[0] == "metadata" && parts[1] != "" { if strings.HasSuffix(object.Key, "/") || strings.Contains(object.Key, "/manifest.json.zst") { @@ -183,12 +209,14 @@ func (v *Vaultik) listUniqueSnapshotIDs() ([]string, error) { } } } + return snapshotIDs, nil } // listAllRemoteBlobs returns a map of all blob hashes to their sizes in remote storage func (v *Vaultik) listAllRemoteBlobs() (map[string]int64, error) { log.Info("Listing all blobs in storage") + allBlobs := make(map[string]int64) blobObjectCh := v.Storage.ListStream(v.ctx, "blobs/") @@ -196,6 +224,7 @@ func (v *Vaultik) listAllRemoteBlobs() (map[string]int64, error) { if object.Err != nil { return nil, fmt.Errorf("listing blobs: %w", object.Err) } + parts := strings.Split(object.Key, "/") if len(parts) == 4 && parts[0] == "blobs" { allBlobs[parts[3]] = object.Size @@ -203,19 +232,24 @@ func (v *Vaultik) listAllRemoteBlobs() (map[string]int64, error) { } log.Info("Found blobs in storage", "count", len(allBlobs)) + return allBlobs, nil } // findUnreferencedBlobs returns blob hashes not referenced by any manifest and their total size func (v *Vaultik) findUnreferencedBlobs(allBlobs map[string]int64, referenced map[string]bool) ([]string, int64) { - var unreferenced []string - var totalSize int64 + var ( + unreferenced []string + totalSize int64 + ) + for hash, size := range allBlobs { if !referenced[hash] { unreferenced = append(unreferenced, hash) totalSize += size } } + return unreferenced, totalSize } @@ -226,8 +260,10 @@ func (v *Vaultik) deleteUnreferencedBlobs(unreferencedBlobs []string, allBlobs m for i, hash := range unreferencedBlobs { blobPath := fmt.Sprintf("blobs/%s/%s/%s", hash[:2], hash[2:4], hash) - if err := v.Storage.Delete(v.ctx, blobPath); err != nil { + err := v.Storage.Delete(v.ctx, blobPath) + if err != nil { log.Error("Failed to delete blob", "hash", hash, "error", err) + continue } @@ -256,5 +292,6 @@ func (v *Vaultik) deleteUnreferencedBlobs(unreferencedBlobs []string, allBlobs m func (v *Vaultik) outputPruneBlobsJSON(result *PruneBlobsResult) error { encoder := json.NewEncoder(v.Stdout) encoder.SetIndent("", " ") + return encoder.Encode(result) } diff --git a/internal/vaultik/purge_per_name_test.go b/internal/vaultik/purge_per_name_test.go index ff86be3..b8c4782 100644 --- a/internal/vaultik/purge_per_name_test.go +++ b/internal/vaultik/purge_per_name_test.go @@ -79,16 +79,19 @@ func setupPurgeTest(t *testing.T, snapshotIDs []string) *vaultik.Vaultik { // listRemainingSnapshots returns IDs of all completed snapshots in the database. func listRemainingSnapshots(t *testing.T, v *vaultik.Vaultik) []string { t.Helper() + ctx := context.Background() dbSnaps, err := v.Repositories.Snapshots.ListRecent(ctx, 10000) require.NoError(t, err) var ids []string + for _, s := range dbSnaps { if s.CompletedAt != nil { ids = append(ids, s.ID.String()) } } + return ids } diff --git a/internal/vaultik/remove_snapshot_test.go b/internal/vaultik/remove_snapshot_test.go index 5dd03ec..2bbf33f 100644 --- a/internal/vaultik/remove_snapshot_test.go +++ b/internal/vaultik/remove_snapshot_test.go @@ -37,7 +37,9 @@ func (s *testStorer) Put(ctx context.Context, key string, reader io.Reader) erro if err != nil { return err } + s.data[key] = data + return nil } @@ -53,6 +55,7 @@ func (s *testStorer) Get(ctx context.Context, key string) (io.ReadCloser, error) if !exists { return nil, storage.ErrNotFound } + return io.NopCloser(bytes.NewReader(data)), nil } @@ -64,6 +67,7 @@ func (s *testStorer) Stat(ctx context.Context, key string) (*storage.ObjectInfo, if !exists { return nil, storage.ErrNotFound } + return &storage.ObjectInfo{ Key: key, Size: int64(len(data)), @@ -75,6 +79,7 @@ func (s *testStorer) Delete(ctx context.Context, key string) error { defer s.mu.Unlock() delete(s.data, key) + return nil } @@ -83,11 +88,13 @@ func (s *testStorer) List(ctx context.Context, prefix string) ([]string, error) defer s.mu.Unlock() var keys []string + for key := range s.data { if prefix == "" || strings.HasPrefix(key, prefix) { keys = append(keys, key) } } + return keys, nil } @@ -96,6 +103,7 @@ func (s *testStorer) ListStream(ctx context.Context, prefix string) <-chan stora go func() { defer close(ch) + s.mu.Lock() defer s.mu.Unlock() @@ -115,13 +123,16 @@ func (s *testStorer) ListStream(ctx context.Context, prefix string) <-chan stora func (s *testStorer) hasKey(key string) bool { s.mu.Lock() defer s.mu.Unlock() + _, exists := s.data[key] + return exists } func (s *testStorer) keyCount() int { s.mu.Lock() defer s.mu.Unlock() + return len(s.data) } @@ -175,6 +186,7 @@ func addBlob(t *testing.T, store *testStorer, hash string) { // Create zstd compressed data var buf bytes.Buffer + writer, _ := zstd.NewWriter(&buf) _, _ = writer.Write([]byte("blob data")) _ = writer.Close() @@ -366,7 +378,7 @@ func TestRemoveAllSnapshots_NoSnapshots(t *testing.T) { result, err := tv.RemoveAllSnapshots(opts) require.NoError(t, err) - assert.Len(t, result.SnapshotsRemoved, 0) + assert.Empty(t, result.SnapshotsRemoved) // Verify output assert.Contains(t, tv.Stdout.String(), "No snapshots found") diff --git a/internal/vaultik/restore.go b/internal/vaultik/restore.go index bb7715c..eea762c 100644 --- a/internal/vaultik/restore.go +++ b/internal/vaultik/restore.go @@ -5,6 +5,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" "math" @@ -62,16 +63,20 @@ func (v *Vaultik) Restore(opts *RestoreOptions) error { // Step 1: Download and decrypt the snapshot metadata database log.Info("Downloading snapshot metadata...") + tempDB, err := v.downloadSnapshotDB(opts.SnapshotID, identity) if err != nil { return fmt.Errorf("downloading snapshot database: %w", err) } + defer func() { - if err := tempDB.Close(); err != nil { + err := tempDB.Close() + if err != nil { log.Debug("Failed to close temp database", "error", err) } // Clean up temp file - if err := v.Fs.Remove(tempDB.Path()); err != nil { + err = v.Fs.Remove(tempDB.Path()) + if err != nil { log.Debug("Failed to remove temp database", "error", err) } }() @@ -87,6 +92,7 @@ func (v *Vaultik) Restore(opts *RestoreOptions) error { if len(files) == 0 { log.Warn("No files found to restore") v.UI.Warning("No files found to restore.") + return nil } @@ -132,6 +138,7 @@ func (v *Vaultik) Restore(opts *RestoreOptions) error { if result.FilesFailed > 0 { v.UI.Warning("%d file(s) failed to restore:", result.FilesFailed) + for _, path := range result.FailedFiles { v.UI.Detail("%s", v.UI.Path(path)) } @@ -139,7 +146,8 @@ func (v *Vaultik) Restore(opts *RestoreOptions) error { // Run verification if requested if opts.Verify { - if err := v.handleRestoreVerification(repos, files, opts, result); err != nil { + err := v.handleRestoreVerification(repos, files, opts, result) + if err != nil { return err } } @@ -154,13 +162,14 @@ func (v *Vaultik) Restore(opts *RestoreOptions) error { // prepareRestoreIdentity validates that an age secret key is configured and parses it func (v *Vaultik) prepareRestoreIdentity() (age.Identity, error) { if v.Config.AgeSecretKey == "" { - return nil, fmt.Errorf("decryption key required for restore\n\nSet the VAULTIK_AGE_SECRET_KEY environment variable to your age private key:\n export VAULTIK_AGE_SECRET_KEY='AGE-SECRET-KEY-...'") + return nil, errors.New("decryption key required for restore\n\nSet the VAULTIK_AGE_SECRET_KEY environment variable to your age private key:\n export VAULTIK_AGE_SECRET_KEY='AGE-SECRET-KEY-...'") } identity, err := age.ParseX25519Identity(v.Config.AgeSecretKey) if err != nil { return nil, fmt.Errorf("parsing age secret key: %w", err) } + return identity, nil } @@ -187,13 +196,16 @@ func (v *Vaultik) restoreAllFiles( if err != nil { return nil, fmt.Errorf("creating blob cache: %w", err) } + if v.restoreCacheObserver != nil { v.restoreCacheObserver(blobCache) } + defer func() { if v.restoreCacheObserver != nil { v.restoreCacheObserver(blobCache) } + _ = blobCache.Close() }() @@ -208,7 +220,9 @@ func (v *Vaultik) restoreAllFiles( if err != nil { return nil, fmt.Errorf("fetching blob index: %w", err) } + blobIDToHash := make(map[string]string, len(blobsByID)) + blobByHash := make(map[string]*database.Blob, len(blobsByID)) for id, blob := range blobsByID { hash := blob.Hash.String() @@ -257,9 +271,11 @@ func (v *Vaultik) restoreAllFiles( // Periodic progress output, matching the snapshot create cadence. startTime := time.Now() lastStatusTime := startTime + const statusInterval = 15 * time.Second processed := 0 + for plan.hasPending() { if v.ctx.Err() != nil { return nil, v.ctx.Err() @@ -282,31 +298,44 @@ func (v *Vaultik) restoreAllFiles( if next.IsZero() { break } + for _, hash := range plan.blobsNeeded(next) { blob, ok := blobByHash[hash] if !ok { return nil, fmt.Errorf("blob hash %s missing from blob index", hash[:16]) } - if err := session.downloadBlobToCache(hash, blob.CompressedSize); err != nil { + + err := session.downloadBlobToCache(hash, blob.CompressedSize) + if err != nil { return nil, fmt.Errorf("downloading blob %s: %w", hash[:16], err) } + result.BlobsDownloaded++ result.BytesDownloaded += blob.CompressedSize + plan.markBlobCached(hash) } + continue } file := filesByID[fileID] - if err := session.restoreFile(file); err != nil { + + err := session.restoreFile(file) + if err != nil { log.Error("Failed to restore file", "path", file.Path, "error", err) + if !opts.SkipErrors { return nil, fmt.Errorf("restoring %s: %w (pass --skip-errors to continue past restore failures)", file.Path, err) } + v.UI.Error("Failed to restore %s: %v. Skipping (--skip-errors).", v.UI.Path(file.Path.String()), err) + result.FilesFailed++ result.FailedFiles = append(result.FailedFiles, file.Path.String()) + plan.finishFile(fileID) + continue } @@ -315,10 +344,12 @@ func (v *Vaultik) restoreAllFiles( // plan's indexes so future picks ignore it. sweeper.fileRestored(fileID.String()) plan.finishFile(fileID) + processed++ if time.Since(lastStatusTime) >= statusInterval { v.printRestoreProgress(processed, len(files), result.BytesRestored, totalBytesExpected, startTime) + lastStatusTime = time.Now() } @@ -344,6 +375,7 @@ func (v *Vaultik) printRestoreProgress(filesDone, totalFiles int, bytesDone, tot fileRate := float64(filesDone) / elapsed.Seconds() remainingBytes := totalBytes - bytesDone + var eta time.Duration if byteRate > 0 && remainingBytes > 0 { eta = time.Duration(float64(remainingBytes)/byteRate) * time.Second @@ -361,8 +393,10 @@ func (v *Vaultik) printRestoreProgress(filesDone, totalFiles int, bytesDone, tot v.UI.Duration(elapsed), v.UI.Time(time.Now().Add(eta)), v.UI.Duration(eta)) + return } + v.UI.Progress("Restore: %s/%s files (%s), %s/%s, %s, %.0f files/sec, restore elapsed: %s.", v.UI.Count(filesDone), v.UI.Count(totalFiles), @@ -381,22 +415,26 @@ func (v *Vaultik) handleRestoreVerification( opts *RestoreOptions, result *RestoreResult, ) error { - if err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result); err != nil { + err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result) + if err != nil { return fmt.Errorf("verification failed: %w", err) } if result.FilesFailed > 0 { v.UI.Error("Verification failed: %s files did not match expected checksums.", v.UI.Count(result.FilesFailed)) + for _, path := range result.FailedFiles { v.UI.Detail("%s", v.UI.Path(path)) } + return fmt.Errorf("%d files failed verification", result.FilesFailed) } v.UI.Complete("Verified %s files (%s).", v.UI.Count(result.FilesVerified), v.UI.Size(result.BytesVerified)) + return nil } @@ -418,6 +456,7 @@ func (v *Vaultik) downloadSnapshotDB(snapshotID string, identity age.Identity) ( if err != nil { return nil, fmt.Errorf("reading encrypted data: %w", err) } + log.Debug("Downloaded encrypted database", "size", humanize.Bytes(uint64(len(encryptedData)))) // Decrypt and decompress using blobgen.Reader @@ -432,6 +471,7 @@ func (v *Vaultik) downloadSnapshotDB(snapshotID string, identity age.Identity) ( if err != nil { return nil, fmt.Errorf("decrypting and decompressing: %w", err) } + log.Debug("Decrypted database", "size", humanize.Bytes(uint64(len(dbData)))) // Create a temporary database file and write the binary SQLite data directly @@ -439,18 +479,23 @@ func (v *Vaultik) downloadSnapshotDB(snapshotID string, identity age.Identity) ( if err != nil { return nil, fmt.Errorf("creating temp file: %w", err) } + tempPath := tempFile.Name() // Write the binary SQLite database directly if _, err := tempFile.Write(dbData); err != nil { _ = tempFile.Close() _ = v.Fs.Remove(tempPath) + return nil, fmt.Errorf("writing database file: %w", err) } + if err := tempFile.Close(); err != nil { _ = v.Fs.Remove(tempPath) + return nil, fmt.Errorf("closing temp file: %w", err) } + log.Debug("Created restore database", "path", tempPath) // Open the database @@ -471,6 +516,7 @@ func (v *Vaultik) getFilesToRestore(ctx context.Context, repos *database.Reposit // Get files matching the path filters var result []*database.File + seen := make(map[string]bool) for _, filter := range pathFilters { @@ -498,23 +544,30 @@ func (v *Vaultik) getFilesToRestore(ctx context.Context, repos *database.Reposit func (v *Vaultik) buildChunkToBlobMap(ctx context.Context, repos *database.Repositories) (map[string]*database.BlobChunk, error) { // Query all blob_chunks query := `SELECT blob_id, chunk_hash, offset, length FROM blob_chunks` + rows, err := repos.DB().Conn().QueryContext(ctx, query) if err != nil { return nil, fmt.Errorf("querying blob_chunks: %w", err) } + defer func() { _ = rows.Close() }() result := make(map[string]*database.BlobChunk) + for rows.Next() { - var bc database.BlobChunk - var blobIDStr, chunkHashStr string + var ( + bc database.BlobChunk + blobIDStr, chunkHashStr string + ) if err := rows.Scan(&blobIDStr, &chunkHashStr, &bc.Offset, &bc.Length); err != nil { return nil, fmt.Errorf("scanning blob_chunk: %w", err) } + blobID, err := types.ParseBlobID(blobIDStr) if err != nil { return nil, fmt.Errorf("parsing blob ID: %w", err) } + bc.BlobID = blobID bc.ChunkHash = types.ChunkHash(chunkHashStr) result[chunkHashStr] = &bc @@ -553,16 +606,22 @@ type restoreSession struct { // restoreFile dispatches to the right per-kind restorer. func (s *restoreSession) restoreFile(file *database.File) error { targetPath := filepath.Join(s.opts.TargetDir, file.Path.String()) + parentDir := filepath.Dir(targetPath) - if err := s.v.Fs.MkdirAll(parentDir, 0755); err != nil { + + err := s.v.Fs.MkdirAll(parentDir, 0755) + if err != nil { return fmt.Errorf("creating parent directory: %w", err) } + if file.IsSymlink() { return s.restoreSymlink(file, targetPath) } + if file.Mode&uint32(os.ModeDir) != 0 { return s.restoreDirectory(file, targetPath) } + return s.restoreRegularFile(file, targetPath) } @@ -572,37 +631,50 @@ func (s *restoreSession) restoreSymlink(file *database.File, targetPath string) // afero.MemMapFs doesn't support symlinks, so route real-FS // symlinks through os. if _, ok := s.v.Fs.(*afero.OsFs); ok { - if err := os.Symlink(file.LinkTarget.String(), targetPath); err != nil { + err := os.Symlink(file.LinkTarget.String(), targetPath) + if err != nil { return fmt.Errorf("creating symlink: %w", err) } } else { log.Debug("Symlink creation not supported on this filesystem", "path", file.Path, "target", file.LinkTarget) } + s.result.FilesRestored++ + log.Debug("Restored symlink", "path", file.Path, "target", file.LinkTarget) + return nil } // restoreDirectory restores a directory with its permissions, mtime, // and (on real filesystems, with sufficient privileges) ownership. func (s *restoreSession) restoreDirectory(file *database.File, targetPath string) error { - if err := s.v.Fs.MkdirAll(targetPath, os.FileMode(file.Mode)); err != nil { + err := s.v.Fs.MkdirAll(targetPath, os.FileMode(file.Mode)) + if err != nil { return fmt.Errorf("creating directory: %w", err) } - if err := s.v.Fs.Chmod(targetPath, os.FileMode(file.Mode)); err != nil { + + err = s.v.Fs.Chmod(targetPath, os.FileMode(file.Mode)) + if err != nil { log.Debug("Failed to set directory permissions", "path", targetPath, "error", err) } + if s.runningAsRoot { if _, ok := s.v.Fs.(*afero.OsFs); ok { - if err := os.Chown(targetPath, int(file.UID), int(file.GID)); err != nil { + err := os.Chown(targetPath, int(file.UID), int(file.GID)) + if err != nil { log.Debug("Failed to set directory ownership", "path", targetPath, "error", err) } } } - if err := s.v.Fs.Chtimes(targetPath, file.MTime, file.MTime); err != nil { + + err = s.v.Fs.Chtimes(targetPath, file.MTime, file.MTime) + if err != nil { log.Debug("Failed to set directory mtime", "path", targetPath, "error", err) } + s.result.FilesRestored++ + return nil } @@ -617,16 +689,20 @@ func (s *restoreSession) restoreRegularFile(file *database.File, targetPath stri t0 := time.Now() fileChunks, err := s.repos.FileChunks.GetByFileID(s.ctx, file.ID) fileChunksQueryDur := time.Since(t0) + if err != nil { return fmt.Errorf("getting file chunks: %w", err) } t0 = time.Now() + outFile, err := s.v.Fs.Create(targetPath) createDur := time.Since(t0) + if err != nil { return fmt.Errorf("creating output file: %w", err) } + defer func() { _ = outFile.Close() }() var ( @@ -638,10 +714,12 @@ func (s *restoreSession) restoreRegularFile(file *database.File, targetPath stri for _, fc := range fileChunks { chunkHashStr := fc.ChunkHash.String() + blobChunk, ok := s.chunkToBlobMap[chunkHashStr] if !ok { return fmt.Errorf("chunk %s not found in any blob", chunkHashStr[:16]) } + blobHash, ok := s.blobIDToHash[blobChunk.BlobID.String()] if !ok { return fmt.Errorf("blob id %s missing from hash index", blobChunk.BlobID) @@ -650,6 +728,7 @@ func (s *restoreSession) restoreRegularFile(file *database.File, targetPath stri t0 = time.Now() chunkData, err := s.blobCache.ReadAt(blobHash, blobChunk.Offset, blobChunk.Length) readAtDur += time.Since(t0) + if err != nil { return fmt.Errorf("reading chunk %s from cached blob %s: %w", fc.ChunkHash[:16], blobHash[:16], err) } @@ -657,13 +736,17 @@ func (s *restoreSession) restoreRegularFile(file *database.File, targetPath stri t0 = time.Now() n, err := outFile.Write(chunkData) writeDur += time.Since(t0) + if err != nil { return fmt.Errorf("writing chunk: %w", err) } + bytesWritten += int64(n) t0 = time.Now() + s.sweeper.chunkRestored(int64(n)) + sweeperDur += time.Since(t0) } @@ -682,16 +765,20 @@ func (s *restoreSession) restoreRegularFile(file *database.File, targetPath stri if err := outFile.Close(); err != nil { return fmt.Errorf("closing output file: %w", err) } + if err := s.v.Fs.Chmod(targetPath, os.FileMode(file.Mode)); err != nil { log.Debug("Failed to set file permissions", "path", targetPath, "error", err) } + if s.runningAsRoot { if _, ok := s.v.Fs.(*afero.OsFs); ok { - if err := os.Chown(targetPath, int(file.UID), int(file.GID)); err != nil { + err := os.Chown(targetPath, int(file.UID), int(file.GID)) + if err != nil { log.Debug("Failed to set file ownership", "path", targetPath, "error", err) } } } + if err := s.v.Fs.Chtimes(targetPath, file.MTime, file.MTime); err != nil { log.Debug("Failed to set file mtime", "path", targetPath, "error", err) } @@ -700,6 +787,7 @@ func (s *restoreSession) restoreRegularFile(file *database.File, targetPath stri s.result.BytesRestored += bytesWritten log.Debug("Restored file", "path", file.Path, "size", humanize.Bytes(uint64(bytesWritten))) + return nil } @@ -715,6 +803,7 @@ func (s *restoreSession) downloadBlobToCache(blobHash string, expectedSize int64 t0 := time.Now() rc, err := s.v.FetchAndDecryptBlob(s.ctx, blobHash, expectedSize, s.identity) fetchSetupDur := time.Since(t0) + if err != nil { return err } @@ -723,9 +812,11 @@ func (s *restoreSession) downloadBlobToCache(blobHash string, expectedSize int64 written, copyErr := s.blobCache.PutFromReader(blobHash, rc) streamDur := time.Since(t0) closeErr := rc.Close() + if copyErr != nil { return copyErr } + if closeErr != nil { return closeErr } @@ -738,6 +829,7 @@ func (s *restoreSession) downloadBlobToCache(blobHash string, expectedSize int64 "ms_fetch_setup", fetchSetupDur.Milliseconds(), "ms_stream_decrypt_decompress", streamDur.Milliseconds(), ) + return nil } @@ -751,18 +843,21 @@ func (v *Vaultik) verifyRestoredFiles( ) error { // Calculate total bytes to verify for progress bar var totalBytes int64 + regularFiles := make([]*database.File, 0, len(files)) for _, file := range files { // Skip symlinks and directories - only verify regular files if file.IsSymlink() || file.Mode&uint32(os.ModeDir) != 0 { continue } + regularFiles = append(regularFiles, file) totalBytes += file.Size } if len(regularFiles) == 0 { log.Info("No regular files to verify") + return nil } @@ -776,28 +871,34 @@ func (v *Vaultik) verifyRestoredFiles( startTime := time.Now() lastStatusTime := startTime + const statusInterval = 15 * time.Second var bytesProcessed int64 + for i, file := range regularFiles { if ctx.Err() != nil { return ctx.Err() } targetPath := filepath.Join(targetDir, file.Path.String()) + bytesVerified, err := v.verifyFile(ctx, repos, file, targetPath) if err != nil { log.Error("File verification failed", "path", file.Path, "error", err) + result.FilesFailed++ result.FailedFiles = append(result.FailedFiles, file.Path.String()) } else { result.FilesVerified++ result.BytesVerified += bytesVerified } + bytesProcessed += file.Size if time.Since(lastStatusTime) >= statusInterval { v.printVerifyProgress(i+1, len(regularFiles), bytesProcessed, totalBytes, startTime) + lastStatusTime = time.Now() } } @@ -821,6 +922,7 @@ func (v *Vaultik) printVerifyProgress(filesDone, totalFiles int, bytesDone, tota fileRate := float64(filesDone) / elapsed.Seconds() remainingBytes := totalBytes - bytesDone + var eta time.Duration if byteRate > 0 && remainingBytes > 0 { eta = time.Duration(float64(remainingBytes)/byteRate) * time.Second @@ -838,8 +940,10 @@ func (v *Vaultik) printVerifyProgress(filesDone, totalFiles int, bytesDone, tota v.UI.Duration(elapsed), v.UI.Time(time.Now().Add(eta)), v.UI.Duration(eta)) + return } + v.UI.Progress("Verify: %s/%s files (%s), %s/%s, %s, %.0f files/sec, verify elapsed: %s.", v.UI.Count(filesDone), v.UI.Count(totalFiles), @@ -873,6 +977,7 @@ func (v *Vaultik) verifyFile( // Verify each chunk var bytesVerified int64 + for _, fc := range fileChunks { // Get chunk size from database chunk, err := repos.Chunks.GetByHash(ctx, fc.ChunkHash.String()) @@ -882,10 +987,12 @@ func (v *Vaultik) verifyFile( // Read chunk data from file chunkData := make([]byte, chunk.Size) + n, err := io.ReadFull(f, chunkData) if err != nil { return bytesVerified, fmt.Errorf("reading chunk data: %w", err) } + if int64(n) != chunk.Size { return bytesVerified, fmt.Errorf("short read: expected %d bytes, got %d", chunk.Size, n) } @@ -904,5 +1011,6 @@ func (v *Vaultik) verifyFile( } log.Debug("File verified", "path", file.Path, "bytes", bytesVerified, "chunks", len(fileChunks)) + return bytesVerified, nil } diff --git a/internal/vaultik/restore_locality_test.go b/internal/vaultik/restore_locality_test.go index 4a29bc2..0e7c52d 100644 --- a/internal/vaultik/restore_locality_test.go +++ b/internal/vaultik/restore_locality_test.go @@ -6,6 +6,7 @@ import ( "crypto/rand" "fmt" "io" + "maps" "os" "path/filepath" "sort" @@ -55,6 +56,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { fs := afero.NewOsFs() tempDir, err := os.MkdirTemp("", "vaultik-locality-") require.NoError(t, err) + defer func() { _ = os.RemoveAll(tempDir) }() dataDir := filepath.Join(tempDir, "source") @@ -86,8 +88,9 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { path string data []byte } + sources := make([]*source, srcCount) - for i := 0; i < srcCount; i++ { + for i := range srcCount { s := &source{ path: fmt.Sprintf("src-%03d.bin", i+1), data: randomBytes(t, srcBytes), @@ -105,11 +108,14 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { sourceBlob int // 0, 1, or 2 sourceIndex int // index into sources slice } + groupReps := []int{0, perBlob, 2 * perBlob} // 0, 5, 10 letters := []byte{'A', 'B', 'C'} + var copies []copyFile - for i := 0; i < 3; i++ { - for j := 0; j < blobsCount; j++ { + + for i := range 3 { + for j := range blobsCount { seq := i*blobsCount + j + 1 name := fmt.Sprintf("cp-%03d-%c.bin", seq, letters[j]) path := filepath.Join(dataDir, name) @@ -143,6 +149,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { db, err := database.New(ctx, dbPath) require.NoError(t, err) + defer func() { _ = db.Close() }() repos := database.NewRepositories(db) @@ -187,6 +194,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { // immediately before close) so we read PeakLen and call counters // from the same instance the production code used. var cacheRef *blobDiskCache + v := &Vaultik{ Config: cfg, Storage: counter, @@ -214,6 +222,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { require.NoErrorf(t, err, "source missing after restore: %s", s.path) require.Truef(t, bytes.Equal(got, s.data), "byte mismatch for source %s", s.path) } + for _, c := range copies { restored := filepath.Join(restoreDir, c.path) got, err := afero.ReadFile(fs, restored) @@ -226,6 +235,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { if !filterBlobKey(key) { continue } + assert.Equalf(t, 1, n, "blob %s fetched %d times, want exactly 1", key, n) } @@ -248,9 +258,11 @@ func TestRestoreLocalityAndReadAt(t *testing.T) { // chunker picks non-degenerate FastCDC boundaries. func randomBytes(t *testing.T, n int) []byte { t.Helper() + b := make([]byte, n) _, err := rand.Read(b) require.NoError(t, err) + return b } @@ -258,21 +270,27 @@ func randomBytes(t *testing.T, n int) []byte { // relative keys for every blob file present. func listBlobKeys(t *testing.T, storeDir string) []string { t.Helper() + var keys []string + root := filepath.Join(storeDir, "blobs") err := filepath.Walk(root, func(p string, info os.FileInfo, err error) error { if err != nil { return err } + if info.IsDir() { return nil } + rel, _ := filepath.Rel(storeDir, p) keys = append(keys, rel) + return nil }) require.NoError(t, err) sort.Strings(keys) + return keys } @@ -289,6 +307,7 @@ func filterBlobKey(key string) bool { // vaultik for access to unexported cache internals. type countingStorerInternal struct { storage.Storer + mu sync.Mutex counts map[string]int } @@ -301,15 +320,16 @@ func (c *countingStorerInternal) Get(ctx context.Context, key string) (io.ReadCl c.mu.Lock() c.counts[key]++ c.mu.Unlock() + return c.Storer.Get(ctx, key) } func (c *countingStorerInternal) snapshot() map[string]int { c.mu.Lock() defer c.mu.Unlock() + out := make(map[string]int, len(c.counts)) - for k, v := range c.counts { - out[k] = v - } + maps.Copy(out, c.counts) + return out } diff --git a/internal/vaultik/restore_plan.go b/internal/vaultik/restore_plan.go index 5b53680..c6cf5cb 100644 --- a/internal/vaultik/restore_plan.go +++ b/internal/vaultik/restore_plan.go @@ -53,26 +53,33 @@ func newRestorePlan( // No chunks to fetch — restore can run immediately. p.fileBlobs[f.ID] = nil p.ready = append(p.ready, f.ID) + continue } + fileChunks, err := repos.FileChunks.GetByFileID(ctx, f.ID) if err != nil { return nil, fmt.Errorf("planning %s: %w", f.Path, err) } + blobs := make(map[string]struct{}) + for _, fc := range fileChunks { bc, ok := chunkToBlobMap[fc.ChunkHash.String()] if !ok { return nil, fmt.Errorf("planning %s: chunk %s missing from blob map", f.Path, fc.ChunkHash.String()[:16]) } + hash, ok := blobIDToHash[bc.BlobID.String()] if !ok { return nil, fmt.Errorf("planning %s: blob id %s missing from id-to-hash map", f.Path, bc.BlobID) } + blobs[hash] = struct{}{} } + p.fileBlobs[f.ID] = blobs for hash := range blobs { set, ok := p.blobFiles[hash] @@ -80,12 +87,15 @@ func newRestorePlan( set = make(map[types.FileID]struct{}) p.blobFiles[hash] = set } + set[f.ID] = struct{}{} } + if len(blobs) == 0 { p.ready = append(p.ready, f.ID) } } + return p, nil } @@ -96,10 +106,12 @@ func (p *restorePlan) markBlobCached(blobHash string) { if _, already := p.cached[blobHash]; already { return } + p.cached[blobHash] = struct{}{} for fileID := range p.blobFiles[blobHash] { blobs := p.fileBlobs[fileID] delete(blobs, blobHash) + if len(blobs) == 0 { p.ready = append(p.ready, fileID) } @@ -112,8 +124,10 @@ func (p *restorePlan) popReady() (types.FileID, bool) { if len(p.ready) == 0 { return types.FileID{}, false } + id := p.ready[0] p.ready = p.ready[1:] + return id, true } @@ -123,17 +137,20 @@ func (p *restorePlan) finishFile(fileID types.FileID) { for hash := range p.fileBlobs[fileID] { if set, ok := p.blobFiles[hash]; ok { delete(set, fileID) + if len(set) == 0 { delete(p.blobFiles, hash) } } } + delete(p.fileBlobs, fileID) // Also scrub the file from any blobFiles entries where it might // still appear even after its uncached-blob set was emptied. for hash, set := range p.blobFiles { if _, ok := set[fileID]; ok { delete(set, fileID) + if len(set) == 0 { delete(p.blobFiles, hash) } @@ -150,8 +167,11 @@ func (p *restorePlan) finishFile(fileID types.FileID) { // The zero FileID return means nothing is pending. func (p *restorePlan) pickNextDownload() types.FileID { var best types.FileID + bestCount := math.MaxInt + var bestID string + for id, blobs := range p.fileBlobs { n := len(blobs) if n == 0 { @@ -159,6 +179,7 @@ func (p *restorePlan) pickNextDownload() types.FileID { // popReady; ignore here just in case. continue } + idStr := id.String() if n < bestCount || (n == bestCount && (best.IsZero() || idStr < bestID)) { best = id @@ -166,16 +187,19 @@ func (p *restorePlan) pickNextDownload() types.FileID { bestID = idStr } } + return best } // blobsNeeded returns the uncached blob hashes for fileID in any order. func (p *restorePlan) blobsNeeded(fileID types.FileID) []string { blobs := p.fileBlobs[fileID] + out := make([]string, 0, len(blobs)) for h := range blobs { out = append(out, h) } + return out } diff --git a/internal/vaultik/restore_sweeper.go b/internal/vaultik/restore_sweeper.go index 2594844..f3a47f0 100644 --- a/internal/vaultik/restore_sweeper.go +++ b/internal/vaultik/restore_sweeper.go @@ -42,6 +42,7 @@ func newRestoreSweeper(ctx context.Context, repos *database.Repositories, cache if threshold <= 0 { threshold = 1 } + return &restoreSweeper{ ctx: ctx, repos: repos, @@ -65,6 +66,7 @@ func (s *restoreSweeper) chunkRestored(n int64) { if s.bytesAccum < s.threshold { return } + s.bytesAccum = 0 s.sweep() } @@ -77,8 +79,10 @@ func (s *restoreSweeper) sweep() { needed, err := s.blobStillNeeded(blobHash) if err != nil { log.Debug("sweeper referencing-files query failed", "blob_hash", blobHash[:16], "error", err) + continue } + if !needed { s.cache.Delete(blobHash) } @@ -104,15 +108,20 @@ func (s *restoreSweeper) blobStillNeeded(blobHash string) (bool, error) { for rows.Next() { var fileID string - if err := rows.Scan(&fileID); err != nil { + + err := rows.Scan(&fileID) + if err != nil { return true, fmt.Errorf("scanning file_id: %w", err) } + if _, ok := s.restored[fileID]; !ok { return true, nil } } + if err := rows.Err(); err != nil { return true, err } + return false, nil } diff --git a/internal/vaultik/restore_sweeper_integration_test.go b/internal/vaultik/restore_sweeper_integration_test.go index 044f3a2..d5b21e0 100644 --- a/internal/vaultik/restore_sweeper_integration_test.go +++ b/internal/vaultik/restore_sweeper_integration_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "maps" "math/rand" "os" "path/filepath" @@ -45,6 +46,7 @@ func TestRestoreSweeperEvictsBlobs(t *testing.T) { fs := afero.NewOsFs() tempDir, err := os.MkdirTemp("", "vaultik-sweeper-") require.NoError(t, err) + defer func() { _ = os.RemoveAll(tempDir) }() dataDir := filepath.Join(tempDir, "source") @@ -63,19 +65,22 @@ func TestRestoreSweeperEvictsBlobs(t *testing.T) { duplicateFiles = 10 fileSize = 1 * 1024 * 1024 ) + rng := rand.New(rand.NewSource(42)) type sourceFile struct { path string data []byte } + uniques := make([]sourceFile, 0, uniqueFiles) expected := make(map[string][]byte, uniqueFiles+duplicateFiles) - for i := 0; i < uniqueFiles; i++ { + for i := range uniqueFiles { data := make([]byte, fileSize) _, err := rng.Read(data) require.NoError(t, err) + path := filepath.Join(dataDir, fmt.Sprintf("unique-%02d.bin", i)) require.NoError(t, afero.WriteFile(fs, path, data, 0o644)) uniques = append(uniques, sourceFile{path: path, data: data}) @@ -112,6 +117,7 @@ func TestRestoreSweeperEvictsBlobs(t *testing.T) { db, err := database.New(ctx, dbPath) require.NoError(t, err) + defer func() { _ = db.Close() }() repos := database.NewRepositories(db) @@ -184,14 +190,18 @@ func TestRestoreSweeperEvictsBlobs(t *testing.T) { // sweeper evicted a still-needed blob; 0 means the cache silently // stopped being consulted. blobDownloads := 0 + for key, count := range counter.snapshot() { if !strings.HasPrefix(key, "blobs/") { continue } + assert.Equalf(t, 1, count, "blob %s should have been downloaded exactly once during restore, got %d", key, count) + blobDownloads++ } + assert.Equal(t, blobCount, blobDownloads, "every blob on disk should have been fetched exactly once during restore") t.Logf("restore downloaded %d blobs, each exactly once", blobDownloads) @@ -202,6 +212,7 @@ func TestRestoreSweeperEvictsBlobs(t *testing.T) { // re-downloading blobs that are evicted while still needed. type countingStorer struct { storage.Storer + mu sync.Mutex counts map[string]int } @@ -214,16 +225,17 @@ func (c *countingStorer) Get(ctx context.Context, key string) (io.ReadCloser, er c.mu.Lock() c.counts[key]++ c.mu.Unlock() + return c.Storer.Get(ctx, key) } func (c *countingStorer) snapshot() map[string]int { c.mu.Lock() defer c.mu.Unlock() + out := make(map[string]int, len(c.counts)) - for k, v := range c.counts { - out[k] = v - } + maps.Copy(out, c.counts) + return out } @@ -232,17 +244,21 @@ func (c *countingStorer) snapshot() map[string]int { // expected number of restore-time downloads. func countBlobsOnDisk(t *testing.T, storeDir string) int { t.Helper() + count := 0 root := filepath.Join(storeDir, "blobs") err := filepath.Walk(root, func(_ string, info os.FileInfo, err error) error { if err != nil { return err } + if !info.IsDir() { count++ } + return nil }) require.NoError(t, err) + return count } diff --git a/internal/vaultik/snapshot.go b/internal/vaultik/snapshot.go index ba5c017..2cf437a 100644 --- a/internal/vaultik/snapshot.go +++ b/internal/vaultik/snapshot.go @@ -2,6 +2,7 @@ package vaultik import ( "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -36,7 +37,8 @@ func (v *Vaultik) CreateSnapshot(opts *SnapshotCreateOptions) error { "index_path", v.Config.IndexPath, ) - if err := v.EnsureStorageBinding(); err != nil { + err := v.EnsureStorageBinding() + if err != nil { return err } @@ -72,12 +74,13 @@ func (v *Vaultik) CreateSnapshot(opts *SnapshotCreateOptions) error { } if len(snapshotNames) == 0 { - return fmt.Errorf("no snapshots configured") + return errors.New("no snapshots configured") } // Process each named snapshot for snapIdx, snapName := range snapshotNames { - if err := v.createNamedSnapshot(opts, hostname, snapName, snapIdx+1, len(snapshotNames)); err != nil { + err := v.createNamedSnapshot(opts, hostname, snapName, snapIdx+1, len(snapshotNames)) + if err != nil { return err } } @@ -88,7 +91,8 @@ func (v *Vaultik) CreateSnapshot(opts *SnapshotCreateOptions) error { } if opts.Prune { - if err := v.runPostBackupPrune(snapshotNames, opts.KeepNewerThan); err != nil { + err := v.runPostBackupPrune(snapshotNames, opts.KeepNewerThan) + if err != nil { return fmt.Errorf("post-backup prune: %w", err) } } @@ -127,11 +131,13 @@ func (v *Vaultik) runPostBackupPrune(snapshotNames []string, keepNewerThan strin purgeOpts.KeepLatest = true } - if err := v.PurgeSnapshotsWithOptions(purgeOpts); err != nil { + err := v.PurgeSnapshotsWithOptions(purgeOpts) + if err != nil { return fmt.Errorf("purging old snapshots: %w", err) } - if err := v.PruneBlobs(&PruneOptions{Force: true}); err != nil { + err = v.PruneBlobs(&PruneOptions{Force: true}) + if err != nil { return fmt.Errorf("pruning orphaned blobs: %w", err) } @@ -178,6 +184,7 @@ func (v *Vaultik) createNamedSnapshot(opts *SnapshotCreateOptions, hostname, sna if err != nil { return fmt.Errorf("creating snapshot: %w", err) } + log.Info("Beginning snapshot", "snapshot_id", snapshotID, "name", snapName) v.UI.Begin("Creating snapshot %s.", v.UI.Snapshot(snapshotID)) @@ -201,6 +208,7 @@ func (v *Vaultik) createNamedSnapshot(opts *SnapshotCreateOptions, hostname, sna "duration", time.Since(snapshotStartTime)) v.printSnapshotSummary(snapshotID, snapshotStartTime, stats) + return nil } @@ -238,12 +246,14 @@ func (v *Vaultik) scanAllDirectories(scanner *snapshot.Scanner, resolvedDirs []s select { case <-v.ctx.Done(): log.Info("Snapshot creation cancelled") + return nil, v.ctx.Err() default: } log.Info("Scanning directory", "path", dir) v.UI.Begin("Enumerating snapshot source files in %s (%d of %d).", v.UI.Path(dir), i+1, len(resolvedDirs)) + result, err := scanner.Scan(v.ctx, dir, snapshotID) if err != nil { return nil, fmt.Errorf("failed to scan %s: %w", dir, err) @@ -297,15 +307,18 @@ func (v *Vaultik) finalizeSnapshotMetadata(snapshotID string, stats *snapshotSta UploadDurationMs: stats.uploadDuration.Milliseconds(), } - if err := v.SnapshotManager.UpdateSnapshotStatsExtended(v.ctx, snapshotID, extStats); err != nil { + err := v.SnapshotManager.UpdateSnapshotStatsExtended(v.ctx, snapshotID, extStats) + if err != nil { return fmt.Errorf("updating snapshot stats: %w", err) } - if err := v.SnapshotManager.CompleteSnapshot(v.ctx, snapshotID); err != nil { + err = v.SnapshotManager.CompleteSnapshot(v.ctx, snapshotID) + if err != nil { return fmt.Errorf("completing snapshot: %w", err) } - if err := v.SnapshotManager.ExportSnapshotMetadata(v.ctx, v.Config.IndexPath, snapshotID); err != nil { + err = v.SnapshotManager.ExportSnapshotMetadata(v.ctx, v.Config.IndexPath, snapshotID) + if err != nil { return fmt.Errorf("exporting snapshot metadata: %w", err) } @@ -318,6 +331,7 @@ func (v *Vaultik) uploadSpeed(bytesUploaded int64, duration time.Duration) strin if bytesUploaded <= 0 || duration <= 0 { return v.UI.Speed(0) } + return v.UI.Speed(float64(bytesUploaded) / duration.Seconds()) } @@ -338,6 +352,7 @@ func (v *Vaultik) printSnapshotSummary(snapshotID string, startTime time.Time, s } v.UI.Complete("Created snapshot %s.", v.UI.Snapshot(snapshotID)) + filesMsg := fmt.Sprintf("Files: %s examined, %s backed up, %s unchanged", v.UI.Count(stats.totalFiles), v.UI.Count(totalFilesChanged), @@ -345,6 +360,7 @@ func (v *Vaultik) printSnapshotSummary(snapshotID string, startTime time.Time, s if stats.totalFilesDeleted > 0 { filesMsg += fmt.Sprintf(", %s deleted", v.UI.Count(stats.totalFilesDeleted)) } + v.UI.Detail("%s.", filesMsg) dataMsg := fmt.Sprintf("Data: %s total (%s backed up)", @@ -353,6 +369,7 @@ func (v *Vaultik) printSnapshotSummary(snapshotID string, startTime time.Time, s if stats.totalBytesDeleted > 0 { dataMsg += fmt.Sprintf(", %s deleted", v.UI.Size(stats.totalBytesDeleted)) } + v.UI.Detail("%s.", dataMsg) if stats.totalBlobsUploaded > 0 { @@ -366,6 +383,7 @@ func (v *Vaultik) printSnapshotSummary(snapshotID string, startTime time.Time, s v.UI.Duration(stats.uploadDuration), v.uploadSpeed(stats.totalBytesUploaded, stats.uploadDuration)) } + v.UI.Detail("Snapshot create duration: %s.", v.UI.Duration(snapshotDuration)) } @@ -375,12 +393,14 @@ func (v *Vaultik) getSnapshotBlobSizes(snapshotID string) (compressed int64, unc if err != nil { return 0, 0 } + for _, hash := range blobHashes { if blob, err := v.Repositories.Blobs.GetByHash(v.ctx, hash); err == nil && blob != nil { compressed += blob.CompressedSize uncompressed += blob.UncompressedSize } } + return compressed, uncompressed } @@ -418,6 +438,7 @@ func (v *Vaultik) ListSnapshots(jsonOutput bool) error { if ls.CompletedAt == nil { continue } + snapshots = append(snapshots, v.snapshotInfoFromLocal(ls)) } @@ -428,6 +449,7 @@ func (v *Vaultik) ListSnapshots(jsonOutput bool) error { if jsonOutput { encoder := json.NewEncoder(v.Stdout) encoder.SetIndent("", " ") + return encoder.Encode(snapshots) } @@ -442,6 +464,7 @@ func (v *Vaultik) ListSnapshots(jsonOutput bool) error { remoteKeys, err := v.listAllRemoteSnapshotKeys() if err != nil { v.UI.Warning("Could not list backup destination store: %v.", err) + return nil } @@ -450,20 +473,25 @@ func (v *Vaultik) ListSnapshots(jsonOutput bool) error { if ls.CompletedAt == nil { continue } + localKeys[snapshot.RemoteSnapshotKey(ls.ID.String())] = ls.ID.String() } + remoteSet := make(map[string]bool, len(remoteKeys)) for _, k := range remoteKeys { remoteSet[k] = true } var localOnly []string + for key, humanID := range localKeys { if !remoteSet[key] { localOnly = append(localOnly, humanID) } } + var remoteOnlyCount int + for key := range remoteSet { if _, ok := localKeys[key]; !ok { remoteOnlyCount++ @@ -472,11 +500,14 @@ func (v *Vaultik) ListSnapshots(jsonOutput bool) error { if len(localOnly) > 0 { v.UI.Warning("%d local snapshot record(s) not found in backup destination store:", len(localOnly)) + for _, id := range localOnly { v.UI.Info("%s", v.UI.Snapshot(id)) } + v.UI.Info("Run 'vaultik snapshot cleanup' to remove stale local records.") } + if remoteOnlyCount > 0 { v.UI.Notice("NOTE: %d remote snapshot(s) found in backup destination store but not in local database.", remoteOnlyCount) } @@ -493,6 +524,7 @@ func (v *Vaultik) snapshotInfoFromLocal(ls *database.Snapshot) SnapshotInfo { totalSize, err := v.Repositories.Snapshots.GetSnapshotTotalCompressedSize(v.ctx, idStr) if err != nil { log.Warn("Failed to get total compressed size", "id", idStr, "error", err) + totalSize = ls.BlobSize } @@ -523,19 +555,24 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error { if _, err := fmt.Fprintln(w, "CONFIGURED SNAPSHOTS:"); err != nil { return err } + if _, err := fmt.Fprintln(w, "NAME\tPATHS"); err != nil { return err } + if _, err := fmt.Fprintln(w, "────\t─────"); err != nil { return err } + for _, name := range v.Config.SnapshotNames() { snap := v.Config.Snapshots[name] + paths := strings.Join(snap.Paths, ", ") if _, err := fmt.Fprintf(w, "%s\t%s\n", name, paths); err != nil { return err } } + if _, err := fmt.Fprintln(w); err != nil { return err } @@ -543,9 +580,11 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error { if _, err := fmt.Fprintln(w, "REMOTE SNAPSHOTS:"); err != nil { return err } + if _, err := fmt.Fprintln(w, "SNAPSHOT ID\tTIMESTAMP\tCOMPRESSED SIZE\tUNCOMPRESSED SIZE\tNEW CHUNK SIZE"); err != nil { return err } + if _, err := fmt.Fprintln(w, "───────────\t─────────\t───────────────\t─────────────────\t──────────────"); err != nil { return err } @@ -554,10 +593,12 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error { for _, snap := range snapshots { uncompressed := remoteOnlyCell newChunks := remoteOnlyCell + if snap.LocallyTracked { uncompressed = formatBytes(snap.UncompressedSize) newChunks = formatBytes(snap.NewChunkSize) } + if _, err := fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", snap.ID, snap.Timestamp.Format("2006-01-02 15:04:05"), @@ -611,11 +652,13 @@ func (v *Vaultik) PurgeSnapshotsWithOptions(opts *SnapshotPurgeOptions) error { if s.CompletedAt == nil { continue } + if len(nameFilter) > 0 { if _, ok := nameFilter[parseSnapshotName(s.ID.String())]; !ok { continue } } + snapshots = append(snapshots, SnapshotInfo{ ID: s.ID, Timestamp: s.StartedAt, @@ -634,12 +677,15 @@ func (v *Vaultik) PurgeSnapshotsWithOptions(opts *SnapshotPurgeOptions) error { // Keep the latest snapshot per snapshot name. Snapshots are sorted // newest-first, so the first occurrence of each name is kept. seen := make(map[string]bool) + for _, snap := range snapshots { name := parseSnapshotName(snap.ID.String()) if seen[name] { toDelete = append(toDelete, snap) + continue } + seen[name] = true } } else if opts.OlderThan != "" { @@ -660,6 +706,7 @@ func (v *Vaultik) PurgeSnapshotsWithOptions(opts *SnapshotPurgeOptions) error { if !opts.Quiet { v.printlnStdout("No snapshots to delete") } + return nil } @@ -670,6 +717,7 @@ func (v *Vaultik) PurgeSnapshotsWithOptions(opts *SnapshotPurgeOptions) error { func (v *Vaultik) confirmAndExecutePurge(toDelete []SnapshotInfo, force, quiet bool) error { if !quiet { v.printfStdout("The following snapshots will be deleted:\n\n") + for _, snap := range toDelete { v.printfStdout(" %s (%s, %s)\n", snap.ID, @@ -681,14 +729,18 @@ func (v *Vaultik) confirmAndExecutePurge(toDelete []SnapshotInfo, force, quiet b // Confirm unless --force is used if !force { v.printfStdout("\nDelete %d snapshot(s)? [y/N] ", len(toDelete)) + var confirm string if _, err := v.scanStdin(&confirm); err != nil { // Treat EOF or error as "no" v.printlnStdout("Cancelled") + return nil } + if strings.ToLower(confirm) != "y" { v.printlnStdout("Cancelled") + return nil } } else if !quiet { @@ -699,10 +751,14 @@ func (v *Vaultik) confirmAndExecutePurge(toDelete []SnapshotInfo, force, quiet b for _, snap := range toDelete { snapshotID := snap.ID.String() log.Info("Deleting snapshot", "id", snapshotID) - if err := v.deleteSnapshotFromLocalDB(snapshotID); err != nil { + + err := v.deleteSnapshotFromLocalDB(snapshotID) + if err != nil { log.Error("Failed to delete from local database", "snapshot_id", snapshotID, "error", err) } - if err := v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID)); err != nil { + + err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID)) + if err != nil { return fmt.Errorf("deleting snapshot %s from remote: %w", snapshotID, err) } } @@ -711,7 +767,8 @@ func (v *Vaultik) confirmAndExecutePurge(toDelete []SnapshotInfo, force, quiet b // separate command after a purge. Guarded against nil for tests // that don't wire up a SnapshotManager. if v.SnapshotManager != nil { - if err := v.SnapshotManager.CleanupOrphanedData(v.ctx); err != nil { + err := v.SnapshotManager.CleanupOrphanedData(v.ctx) + if err != nil { log.Warn("Failed to clean up orphaned local data after purge", "error", err) } } @@ -730,6 +787,7 @@ func (v *Vaultik) VerifySnapshot(snapshotID string, deep bool) error { if deep { return v.RunDeepVerify(snapshotID, opts) } + return v.VerifySnapshotWithOptions(snapshotID, opts) } @@ -740,6 +798,7 @@ func (v *Vaultik) VerifySnapshotWithOptions(snapshotID string, opts *VerifyOptio if opts.Deep { return v.RunDeepVerify(snapshotID, opts) } + result := &VerifyResult{ SnapshotID: snapshotID, Mode: "shallow", @@ -754,8 +813,10 @@ func (v *Vaultik) VerifySnapshotWithOptions(snapshotID string, opts *VerifyOptio if opts.JSON { result.Status = "failed" result.ErrorMessage = fmt.Sprintf("downloading manifest: %v", err) + return v.outputVerifyJSON(result) } + return fmt.Errorf("downloading manifest: %w", err) } @@ -766,11 +827,13 @@ func (v *Vaultik) VerifySnapshotWithOptions(snapshotID string, opts *VerifyOptio v.printfStdout("Snapshot information:\n") v.printfStdout(" Blob count: %d\n", manifest.BlobCount) v.printfStdout(" Total size: %s\n", humanize.Bytes(uint64(manifest.TotalCompressedSize))) + if manifest.Timestamp != "" { if t, err := time.Parse(time.RFC3339, manifest.Timestamp); err == nil { v.printfStdout(" Created: %s\n", t.Format("2006-01-02 15:04:05 MST")) } } + v.printlnStdout() // Check each blob exists @@ -792,9 +855,11 @@ func (v *Vaultik) printVerifyHeader(snapshotID string, opts *VerifyOptions) { if !opts.JSON { v.printfStdout("Verifying snapshot %s\n", snapshotID) + if !snapshotTime.IsZero() { v.printfStdout("Snapshot time: %s\n", snapshotTime.Format("2006-01-02 15:04:05 MST")) } + v.printlnStdout() } } @@ -810,12 +875,14 @@ func (v *Vaultik) verifyManifestBlobsExist(manifest *snapshot.Manifest, opts *Ve if !opts.JSON { v.printfStdout(" Missing: %s (%s)\n", blob.Hash, humanize.Bytes(uint64(blob.CompressedSize))) } + missing++ missingSize += blob.CompressedSize } else { verified++ } } + return verified, missing, missingSize } @@ -828,22 +895,28 @@ func (v *Vaultik) formatVerifyResult(result *VerifyResult, manifest *snapshot.Ma } else { result.Status = "ok" } + return v.outputVerifyJSON(result) } v.printfStdout("\nVerification complete:\n") v.printfStdout(" Verified: %d blobs (%s)\n", result.Verified, humanize.Bytes(uint64(manifest.TotalCompressedSize-result.MissingSize))) + if result.Missing > 0 { v.printfStdout(" Missing: %d blobs (%s)\n", result.Missing, humanize.Bytes(uint64(result.MissingSize))) } else { v.printfStdout(" Missing: 0 blobs\n") } + v.printfStdout(" Status: ") + if result.Missing > 0 { v.printfStdout("FAILED - %d blobs are missing\n", result.Missing) + return fmt.Errorf("%d blobs are missing", result.Missing) } + v.printfStdout("OK - All blobs verified\n") return nil @@ -853,12 +926,16 @@ func (v *Vaultik) formatVerifyResult(result *VerifyResult, manifest *snapshot.Ma func (v *Vaultik) outputVerifyJSON(result *VerifyResult) error { encoder := json.NewEncoder(v.Stdout) encoder.SetIndent("", " ") - if err := encoder.Encode(result); err != nil { + + err := encoder.Encode(result) + if err != nil { return fmt.Errorf("encoding JSON: %w", err) } + if result.Status == "failed" { return fmt.Errorf("verification failed: %s", result.ErrorMessage) } + return nil } @@ -871,10 +948,12 @@ func (v *Vaultik) CleanupLocalSnapshots() error { if err := v.EnsureStorageBinding(); err != nil { return err } + remoteKeys, err := v.listAllRemoteSnapshotKeys() if err != nil { return err } + remoteSet := make(map[string]bool, len(remoteKeys)) for _, k := range remoteKeys { remoteSet[k] = true @@ -886,14 +965,19 @@ func (v *Vaultik) CleanupLocalSnapshots() error { } var removed int + for _, snap := range localSnapshots { id := snap.ID.String() if !remoteSet[snapshot.RemoteSnapshotKey(id)] { v.printfStdout("Removing stale local record: %s\n", id) - if err := v.deleteSnapshotFromLocalDB(id); err != nil { + + err := v.deleteSnapshotFromLocalDB(id) + if err != nil { log.Error("Failed to delete local snapshot", "snapshot_id", id, "error", err) + continue } + removed++ } } @@ -903,6 +987,7 @@ func (v *Vaultik) CleanupLocalSnapshots() error { } else { v.printfStdout("Removed %d stale local snapshot record(s).\n", removed) } + return nil } @@ -948,6 +1033,7 @@ func (v *Vaultik) syncWithRemote() error { if strings.HasPrefix(parts[1], ".") { continue } + remoteSnapshots[parts[1]] = true } } @@ -962,11 +1048,14 @@ func (v *Vaultik) syncWithRemote() error { // Remove local snapshots that don't exist remotely removedCount := 0 + for _, snapshot := range localSnapshots { snapshotIDStr := snapshot.ID.String() if !remoteSnapshots[snapshotIDStr] { log.Info("Removing local snapshot not found in remote", "snapshot_id", snapshot.ID) - if err := v.deleteSnapshotFromLocalDB(snapshotIDStr); err != nil { + + err := v.deleteSnapshotFromLocalDB(snapshotIDStr) + if err != nil { log.Error("Failed to delete local snapshot", "snapshot_id", snapshot.ID, "error", err) } else { removedCount++ @@ -1015,22 +1104,28 @@ func (v *Vaultik) RemoveSnapshot(snapshotID string, opts *RemoveOptions) (*Remov SnapshotID: snapshotID, } - if err := v.EnsureStorageBinding(); err != nil { + err := v.EnsureStorageBinding() + if err != nil { return result, err } if opts.DryRun { result.DryRun = true + if !opts.JSON { v.printfStdout("Would remove snapshot: %s\n", snapshotID) + if !opts.LocalOnly { v.printlnStdout("Would also remove snapshot metadata from remote storage (blobs untouched)") } + v.printlnStdout("[Dry run - no changes made]") } + if opts.JSON { return result, v.outputRemoveJSON(result) } + return result, nil } @@ -1040,32 +1135,41 @@ func (v *Vaultik) RemoveSnapshot(snapshotID string, opts *RemoveOptions) (*Remov } else { v.printfStdout("Remove snapshot '%s' from local database AND its metadata from remote storage? [y/N] ", snapshotID) } + var confirm string if _, err := v.scanStdin(&confirm); err != nil { v.printlnStdout("Cancelled") + return result, nil } + if strings.ToLower(confirm) != "y" { v.printlnStdout("Cancelled") + return result, nil } } log.Info("Removing snapshot from local database", "snapshot_id", snapshotID) - if err := v.deleteSnapshotFromLocalDB(snapshotID); err != nil { + err = v.deleteSnapshotFromLocalDB(snapshotID) + if err != nil { return result, fmt.Errorf("removing from local database: %w", err) } if !opts.LocalOnly { log.Info("Removing snapshot metadata from remote storage", "snapshot_id", snapshotID) + remoteKey := snapshot.RemoteSnapshotKey(snapshotID) - if err := v.deleteRemoteSnapshotByKey(remoteKey); err != nil { + + err := v.deleteRemoteSnapshotByKey(remoteKey) + if err != nil { // Warn-and-proceed: the local-DB removal has already // happened, so let the user know the remote half didn't // finish and they can retry with `vaultik prune` once the // destination store is reachable. log.Warn("Could not remove snapshot metadata from remote storage", "error", err) + if v.UI != nil { v.UI.Warning("Could not remove snapshot metadata from remote: %v. Run '%s' once the remote is reachable to finish cleanup.", err, pruneCommandHint) } @@ -1075,7 +1179,8 @@ func (v *Vaultik) RemoveSnapshot(snapshotID string, opts *RemoveOptions) (*Remov } if v.SnapshotManager != nil { - if err := v.SnapshotManager.CleanupOrphanedData(v.ctx); err != nil { + err := v.SnapshotManager.CleanupOrphanedData(v.ctx) + if err != nil { log.Warn("Failed to clean up orphaned local data after removal", "error", err) } } @@ -1085,6 +1190,7 @@ func (v *Vaultik) RemoveSnapshot(snapshotID string, opts *RemoveOptions) (*Remov } v.printfStdout("Removed snapshot '%s' from local database\n", snapshotID) + if !opts.LocalOnly && result.RemoteRemoved { v.printlnStdout("Removed snapshot metadata from remote storage") v.printfStdout("\nNote: The removed snapshot's blobs remain on the remote. Run '%s' to delete any blobs no longer referenced by any remaining remote snapshot.\n", pruneCommandHint) @@ -1102,6 +1208,7 @@ func (v *Vaultik) RemoveAllSnapshots(opts *RemoveOptions) (*RemoveResult, error) if err := v.EnsureStorageBinding(); err != nil { return nil, err } + localSnaps, err := v.localSnapshotIDs() if err != nil { return nil, fmt.Errorf("listing local snapshots: %w", err) @@ -1124,7 +1231,9 @@ func (v *Vaultik) RemoveAllSnapshots(opts *RemoveOptions) (*RemoveResult, error) for _, id := range localSnaps { knownLocalKeys[snapshot.RemoteSnapshotKey(id)] = id } + var orphanRemoteKeys []string + for _, key := range remoteKeys { if _, known := knownLocalKeys[key]; !known { orphanRemoteKeys = append(orphanRemoteKeys, key) @@ -1135,6 +1244,7 @@ func (v *Vaultik) RemoveAllSnapshots(opts *RemoveOptions) (*RemoveResult, error) if !opts.JSON { v.printlnStdout("No snapshots found") } + return &RemoveResult{}, nil } @@ -1152,15 +1262,19 @@ func (v *Vaultik) localSnapshotIDs() ([]string, error) { if v.Repositories == nil { return nil, nil } + snaps, err := v.Repositories.Snapshots.ListRecent(v.ctx, 100000) if err != nil { return nil, err } + ids := make([]string, 0, len(snaps)) for _, s := range snaps { ids = append(ids, s.ID.String()) } + sort.Strings(ids) + return ids, nil } @@ -1171,10 +1285,13 @@ func (v *Vaultik) localSnapshotIDs() ([]string, error) { // fatal. func (v *Vaultik) listAllRemoteSnapshotKeys() ([]string, error) { log.Info("Listing all remote snapshots") + objectCh := v.Storage.ListStream(v.ctx, "metadata/") seen := make(map[string]bool) + var keys []string + for object := range objectCh { if object.Err != nil { return nil, fmt.Errorf("listing remote snapshots: %w", object.Err) @@ -1186,6 +1303,7 @@ func (v *Vaultik) listAllRemoteSnapshotKeys() ([]string, error) { if strings.HasPrefix(parts[1], ".") { continue } + if strings.HasSuffix(object.Key, "/") || strings.Contains(object.Key, "/manifest.json.zst") { key := parts[1] if !seen[key] { @@ -1202,18 +1320,23 @@ func (v *Vaultik) listAllRemoteSnapshotKeys() ([]string, error) { // handleRemoveAllDryRun handles the dry-run mode for removing all snapshots func (v *Vaultik) handleRemoveAllDryRun(localSnaps, orphanRemoteKeys []string, opts *RemoveOptions) (*RemoveResult, error) { result := &RemoveResult{DryRun: true} + result.SnapshotsRemoved = append(result.SnapshotsRemoved, localSnaps...) if !opts.LocalOnly { result.SnapshotsRemoved = append(result.SnapshotsRemoved, orphanRemoteKeys...) } + if !opts.JSON { v.printfStdout("Would remove %d local snapshot(s):\n", len(localSnaps)) + for _, id := range localSnaps { v.printfStdout(" %s\n", id) } + if !opts.LocalOnly { if len(orphanRemoteKeys) > 0 { v.printfStdout("Would also remove %d orphan remote snapshot key(s):\n", len(orphanRemoteKeys)) + for _, key := range orphanRemoteKeys { v.printfStdout(" %s\n", key) } @@ -1221,11 +1344,14 @@ func (v *Vaultik) handleRemoveAllDryRun(localSnaps, orphanRemoteKeys []string, o v.printlnStdout("Would also remove snapshot metadata from remote storage (blobs untouched)") } } + v.printlnStdout("[Dry run - no changes made]") } + if opts.JSON { return result, v.outputRemoveJSON(result) } + return result, nil } @@ -1235,24 +1361,29 @@ func (v *Vaultik) handleRemoveAllDryRun(localSnaps, orphanRemoteKeys []string, o // hint is the next step. func (v *Vaultik) executeRemoveAll(localSnaps, orphanRemoteKeys []string, opts *RemoveOptions) (*RemoveResult, error) { if !opts.Force { - return nil, fmt.Errorf("--all requires --force") + return nil, errors.New("--all requires --force") } log.Info("Removing all snapshots", "local_count", len(localSnaps), "orphan_remote_count", len(orphanRemoteKeys)) result := &RemoveResult{} remoteErrors := 0 + for _, snapshotID := range localSnaps { log.Info("Removing snapshot", "snapshot_id", snapshotID) - if err := v.deleteSnapshotFromLocalDB(snapshotID); err != nil { + err := v.deleteSnapshotFromLocalDB(snapshotID) + if err != nil { log.Error("Failed to remove from local database", "snapshot_id", snapshotID, "error", err) + continue } if !opts.LocalOnly { - if err := v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID)); err != nil { + err := v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID)) + if err != nil { log.Warn("Failed to remove snapshot metadata from remote", "snapshot_id", snapshotID, "error", err) + remoteErrors++ } } @@ -1263,11 +1394,16 @@ func (v *Vaultik) executeRemoveAll(localSnaps, orphanRemoteKeys []string, opts * if !opts.LocalOnly { for _, key := range orphanRemoteKeys { log.Info("Removing orphan remote snapshot", "remote_key", key) - if err := v.deleteRemoteSnapshotByKey(key); err != nil { + + err := v.deleteRemoteSnapshotByKey(key) + if err != nil { log.Warn("Failed to remove orphan from remote", "remote_key", key, "error", err) + remoteErrors++ + continue } + result.SnapshotsRemoved = append(result.SnapshotsRemoved, key) } @@ -1279,7 +1415,8 @@ func (v *Vaultik) executeRemoveAll(localSnaps, orphanRemoteKeys []string, opts * } if v.SnapshotManager != nil { - if err := v.SnapshotManager.CleanupOrphanedData(v.ctx); err != nil { + err := v.SnapshotManager.CleanupOrphanedData(v.ctx) + if err != nil { log.Warn("Failed to clean up orphaned local data after bulk removal", "error", err) } } @@ -1289,6 +1426,7 @@ func (v *Vaultik) executeRemoveAll(localSnaps, orphanRemoteKeys []string, opts * } v.printfStdout("Removed %d snapshot(s)\n", len(result.SnapshotsRemoved)) + if !opts.LocalOnly && result.RemoteRemoved { v.printlnStdout("Removed snapshot metadata from remote storage") v.printfStdout("\nNote: Removed snapshots' blobs remain on the remote. Run '%s' to delete any blobs no longer referenced by any remaining remote snapshot.\n", pruneCommandHint) @@ -1304,16 +1442,23 @@ func (v *Vaultik) deleteSnapshotFromLocalDB(snapshotID string) error { } // Delete related records first to avoid foreign key constraints - if err := v.Repositories.Snapshots.DeleteSnapshotFiles(v.ctx, snapshotID); err != nil { + err := v.Repositories.Snapshots.DeleteSnapshotFiles(v.ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot files for %s: %w", snapshotID, err) } - if err := v.Repositories.Snapshots.DeleteSnapshotBlobs(v.ctx, snapshotID); err != nil { + + err = v.Repositories.Snapshots.DeleteSnapshotBlobs(v.ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot blobs for %s: %w", snapshotID, err) } - if err := v.Repositories.Snapshots.DeleteSnapshotUploads(v.ctx, snapshotID); err != nil { + + err = v.Repositories.Snapshots.DeleteSnapshotUploads(v.ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot uploads for %s: %w", snapshotID, err) } - if err := v.Repositories.Snapshots.Delete(v.ctx, snapshotID); err != nil { + + err = v.Repositories.Snapshots.Delete(v.ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot record %s: %w", snapshotID, err) } @@ -1330,17 +1475,21 @@ func (v *Vaultik) deleteRemoteSnapshotByKey(remoteKey string) error { objectCh := v.Storage.ListStream(v.ctx, prefix) var objectsToDelete []string + for object := range objectCh { if object.Err != nil { return fmt.Errorf("listing objects: %w", object.Err) } + objectsToDelete = append(objectsToDelete, object.Key) } for _, key := range objectsToDelete { - if err := v.Storage.Delete(v.ctx, key); err != nil { + err := v.Storage.Delete(v.ctx, key) + if err != nil { return fmt.Errorf("removing %s: %w", key, err) } + log.Debug("Deleted remote object", "key", key) } @@ -1351,6 +1500,7 @@ func (v *Vaultik) deleteRemoteSnapshotByKey(remoteKey string) error { func (v *Vaultik) outputRemoveJSON(result *RemoveResult) error { encoder := json.NewEncoder(v.Stdout) encoder.SetIndent("", " ") + return encoder.Encode(result) } @@ -1384,16 +1534,23 @@ func (v *Vaultik) PruneDatabase() (*PruneResult, error) { snapshotIDStr := snapshot.ID.String() log.Info("Deleting incomplete snapshot", "snapshot_id", snapshot.ID) // Delete related records first - if err := v.Repositories.Snapshots.DeleteSnapshotFiles(v.ctx, snapshotIDStr); err != nil { + err := v.Repositories.Snapshots.DeleteSnapshotFiles(v.ctx, snapshotIDStr) + if err != nil { log.Error("Failed to delete snapshot files", "snapshot_id", snapshot.ID, "error", err) } - if err := v.Repositories.Snapshots.DeleteSnapshotBlobs(v.ctx, snapshotIDStr); err != nil { + + err = v.Repositories.Snapshots.DeleteSnapshotBlobs(v.ctx, snapshotIDStr) + if err != nil { log.Error("Failed to delete snapshot blobs", "snapshot_id", snapshot.ID, "error", err) } - if err := v.Repositories.Snapshots.DeleteSnapshotUploads(v.ctx, snapshotIDStr); err != nil { + + err = v.Repositories.Snapshots.DeleteSnapshotUploads(v.ctx, snapshotIDStr) + if err != nil { log.Error("Failed to delete snapshot uploads", "snapshot_id", snapshot.ID, "error", err) } - if err := v.Repositories.Snapshots.Delete(v.ctx, snapshotIDStr); err != nil { + + err = v.Repositories.Snapshots.Delete(v.ctx, snapshotIDStr) + if err != nil { log.Error("Failed to delete snapshot", "snapshot_id", snapshot.ID, "error", err) } else { result.SnapshotsDeleted++ @@ -1427,6 +1584,7 @@ func (v *Vaultik) PruneDatabase() (*PruneResult, error) { ) snapshotCountAfter := snapshotCountBefore - result.SnapshotsDeleted + v.UI.Complete("Pruned local index database.") v.UI.Detail("Incomplete snapshots: %d removed (%d remain).", result.SnapshotsDeleted, snapshotCountAfter) v.UI.Detail("Orphaned files: %d removed (%d remain).", result.FilesDeleted, fileCountAfter) @@ -1451,10 +1609,13 @@ func (v *Vaultik) getTableCount(tableName string) (int64, error) { } var count int64 - query := fmt.Sprintf("SELECT COUNT(*) FROM %s", tableName) + + query := "SELECT COUNT(*) FROM " + tableName + err := v.DB.Conn().QueryRowContext(v.ctx, query).Scan(&count) if err != nil { return 0, err } + return count, nil } diff --git a/internal/vaultik/storage_bind.go b/internal/vaultik/storage_bind.go index 8c1083d..27f79c2 100644 --- a/internal/vaultik/storage_bind.go +++ b/internal/vaultik/storage_bind.go @@ -54,10 +54,13 @@ func (v *Vaultik) EnsureStorageBinding() error { } if stored == "" { - if err := v.Repositories.LocalMeta.Set(v.ctx, database.LocalMetaKeyStorageURL, configured); err != nil { + err := v.Repositories.LocalMeta.Set(v.ctx, database.LocalMetaKeyStorageURL, configured) + if err != nil { return fmt.Errorf("recording local storage binding: %w", err) } + log.Info("Bound local index to storage destination", "storage_url", configured) + return nil } diff --git a/internal/vaultik/vaultik.go b/internal/vaultik/vaultik.go index 91c93c1..29f4274 100644 --- a/internal/vaultik/vaultik.go +++ b/internal/vaultik/vaultik.go @@ -3,6 +3,7 @@ package vaultik import ( "bytes" "context" + "errors" "fmt" "io" "os" @@ -125,8 +126,9 @@ func (v *Vaultik) CanDecrypt() bool { // Returns an error if no recipients are configured func (v *Vaultik) GetEncryptor() (*crypto.Encryptor, error) { if len(v.Config.AgeRecipients) == 0 { - return nil, fmt.Errorf("no age recipients configured") + return nil, errors.New("no age recipients configured") } + return crypto.NewEncryptor(v.Config.AgeRecipients) } @@ -134,8 +136,9 @@ func (v *Vaultik) GetEncryptor() (*crypto.Encryptor, error) { // Returns an error if no secret key is configured func (v *Vaultik) GetDecryptor() (*crypto.Decryptor, error) { if v.Config.AgeSecretKey == "" { - return nil, fmt.Errorf("no age secret key configured") + return nil, errors.New("no age secret key configured") } + return crypto.NewDecryptor(v.Config.AgeSecretKey) } @@ -162,6 +165,7 @@ func (v *Vaultik) scanStdin(a ...any) (int, error) { // TestVaultik wraps a Vaultik with captured stdout/stderr for testing type TestVaultik struct { *Vaultik + Stdout *bytes.Buffer Stderr *bytes.Buffer Stdin *bytes.Buffer @@ -175,6 +179,7 @@ func NewForTesting(storage storage.Storer) *TestVaultik { stdout := &bytes.Buffer{} stderr := &bytes.Buffer{} stdin := &bytes.Buffer{} + return &TestVaultik{ Vaultik: &Vaultik{ Storage: storage, diff --git a/internal/vaultik/verify.go b/internal/vaultik/verify.go index 9ad2f42..a37688a 100644 --- a/internal/vaultik/verify.go +++ b/internal/vaultik/verify.go @@ -39,13 +39,16 @@ type VerifyResult struct { // deepVerifyFailure records a failure in the result and returns it appropriately func (v *Vaultik) deepVerifyFailure(result *VerifyResult, opts *VerifyOptions, msg string, err error) error { result.Status = "failed" + result.ErrorMessage = msg if opts.JSON { return v.outputVerifyJSON(result) } + if err != nil { return err } + return fmt.Errorf("%s", msg) } @@ -58,10 +61,12 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error { if !v.CanDecrypt() { msg := "VAULTIK_AGE_SECRET_KEY not set; required for deep verification" + return v.deepVerifyFailure(result, opts, msg, fmt.Errorf("%s", msg)) } log.Info("Starting snapshot verification", "snapshot_id", snapshotID, "mode", "deep") + if !opts.JSON { v.printfStdout("Deep verification of snapshot: %s\n\n", snapshotID) } @@ -77,10 +82,12 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error { }() result.BlobCount = len(dbBlobs) + var totalSize int64 for _, blob := range dbBlobs { totalSize += blob.CompressedSize } + result.TotalSize = totalSize if err := v.runVerificationSteps(manifest, dbBlobs, tempDB, opts, result, totalSize); err != nil { @@ -112,15 +119,18 @@ func (v *Vaultik) loadVerificationData(snapshotID string, opts *VerifyOptions, r // Download manifest manifestPath := fmt.Sprintf("metadata/%s/manifest.json.zst", remoteKey) log.Info("Downloading manifest", "path", manifestPath) + if !opts.JSON { v.printfStdout("Downloading manifest...\n") } + manifestReader, err := v.Storage.Get(v.ctx, manifestPath) if err != nil { return nil, nil, nil, v.deepVerifyFailure(result, opts, fmt.Sprintf("failed to download manifest: %v", err), fmt.Errorf("failed to download manifest: %w", err)) } + defer func() { _ = manifestReader.Close() }() manifest, err := snapshot.DecodeManifest(manifestReader) @@ -133,6 +143,7 @@ func (v *Vaultik) loadVerificationData(snapshotID string, opts *VerifyOptions, r log.Info("Manifest loaded", "manifest_blob_count", manifest.BlobCount, "manifest_total_size", humanize.Bytes(uint64(manifest.TotalCompressedSize))) + if !opts.JSON { v.printfStdout("Manifest loaded: %d blobs (%s)\n", manifest.BlobCount, humanize.Bytes(uint64(manifest.TotalCompressedSize))) v.printfStdout("Downloading and decrypting database...\n") @@ -141,12 +152,14 @@ func (v *Vaultik) loadVerificationData(snapshotID string, opts *VerifyOptions, r // Download and decrypt database dbPath := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey) log.Info("Downloading encrypted database", "path", dbPath) + dbReader, err := v.Storage.Get(v.ctx, dbPath) if err != nil { return nil, nil, nil, v.deepVerifyFailure(result, opts, fmt.Sprintf("failed to download database: %v", err), fmt.Errorf("failed to download database: %w", err)) } + defer func() { _ = dbReader.Close() }() tdb, err := v.decryptAndLoadDatabase(dbReader, v.Config.AgeSecretKey) @@ -159,6 +172,7 @@ func (v *Vaultik) loadVerificationData(snapshotID string, opts *VerifyOptions, r dbBlobs, err := v.getBlobsFromDatabase(snapshotID, tdb.DB) if err != nil { _ = tdb.Close() + return nil, nil, nil, v.deepVerifyFailure(result, opts, fmt.Sprintf("failed to get blobs from database: %v", err), fmt.Errorf("failed to get blobs from database: %w", err)) @@ -172,6 +186,7 @@ func (v *Vaultik) loadVerificationData(snapshotID string, opts *VerifyOptions, r log.Info("Database loaded", "db_blob_count", len(dbBlobs), "db_total_size", humanize.Bytes(uint64(dbTotalSize))) + if !opts.JSON { v.printfStdout("Database loaded: %d blobs (%s)\n", len(dbBlobs), humanize.Bytes(uint64(dbTotalSize))) } @@ -184,7 +199,9 @@ func (v *Vaultik) runVerificationSteps(manifest *snapshot.Manifest, dbBlobs []sn if !opts.JSON { v.printfStdout("Verifying manifest against database...\n") } - if err := v.verifyManifestAgainstDatabase(manifest, dbBlobs); err != nil { + + err := v.verifyManifestAgainstDatabase(manifest, dbBlobs) + if err != nil { return v.deepVerifyFailure(result, opts, err.Error(), err) } @@ -192,7 +209,9 @@ func (v *Vaultik) runVerificationSteps(manifest *snapshot.Manifest, dbBlobs []sn v.printfStdout("Manifest verified.\n") v.printfStdout("Checking blob existence in remote storage...\n") } - if err := v.verifyBlobExistenceFromDB(dbBlobs); err != nil { + + err = v.verifyBlobExistenceFromDB(dbBlobs) + if err != nil { return v.deepVerifyFailure(result, opts, err.Error(), err) } @@ -200,7 +219,9 @@ func (v *Vaultik) runVerificationSteps(manifest *snapshot.Manifest, dbBlobs []sn v.printfStdout("All blobs exist.\n") v.printfStdout("Downloading and verifying blob contents (%d blobs, %s)...\n", len(dbBlobs), humanize.Bytes(uint64(totalSize))) } - if err := v.performDeepVerificationFromDB(dbBlobs, tdb.DB, opts); err != nil { + + err = v.performDeepVerificationFromDB(dbBlobs, tdb.DB, opts) + if err != nil { return v.deepVerifyFailure(result, opts, err.Error(), err) } @@ -210,12 +231,14 @@ func (v *Vaultik) runVerificationSteps(manifest *snapshot.Manifest, dbBlobs []sn // tempDB wraps sql.DB with cleanup type tempDB struct { *sql.DB + tempPath string } func (t *tempDB) Close() error { err := t.DB.Close() _ = os.Remove(t.tempPath) + return err } @@ -245,16 +268,20 @@ func (v *Vaultik) decryptAndLoadDatabase(reader io.ReadCloser, secretKey string) if err != nil { return nil, fmt.Errorf("failed to create temp file: %w", err) } + tempPath := tempFile.Name() // Stream decompress directly to file log.Info("Decompressing database...") + written, err := io.Copy(tempFile, decompressor) if err != nil { _ = tempFile.Close() _ = os.Remove(tempPath) + return nil, fmt.Errorf("failed to decompress database: %w", err) } + _ = tempFile.Close() log.Info("Database decompressed", "size", humanize.Bytes(uint64(written))) @@ -263,6 +290,7 @@ func (v *Vaultik) decryptAndLoadDatabase(reader io.ReadCloser, secretKey string) db, err := sql.Open("sqlite", tempPath) if err != nil { _ = os.Remove(tempPath) + return nil, fmt.Errorf("failed to open database: %w", err) } @@ -332,21 +360,28 @@ func (v *Vaultik) verifyBlobChunks(db *sql.DB, blobHash string, decompressor io. WHERE b.blob_hash = ? ORDER BY bc.offset ` + rows, err := db.QueryContext(v.ctx, query, blobHash) if err != nil { return 0, fmt.Errorf("failed to query blob chunks: %w", err) } + defer func() { _ = rows.Close() }() var lastOffset int64 = -1 + chunkCount := 0 totalRead := int64(0) // Verify each chunk in the blob for rows.Next() { - var chunkHash string - var offset, length int64 - if err := rows.Scan(&chunkHash, &offset, &length); err != nil { + var ( + chunkHash string + offset, length int64 + ) + + err := rows.Scan(&chunkHash, &offset, &length) + if err != nil { return 0, fmt.Errorf("failed to scan chunk row: %w", err) } @@ -354,6 +389,7 @@ func (v *Vaultik) verifyBlobChunks(db *sql.DB, blobHash string, decompressor io. if offset <= lastOffset { return 0, fmt.Errorf("chunks out of order: offset %d after %d", offset, lastOffset) } + lastOffset = offset // Read chunk data from decompressed stream @@ -363,6 +399,7 @@ func (v *Vaultik) verifyBlobChunks(db *sql.DB, blobHash string, decompressor io. if _, err := io.CopyN(io.Discard, decompressor, skipBytes); err != nil { return 0, fmt.Errorf("failed to skip to offset %d: %w", offset, err) } + totalRead = offset } @@ -371,6 +408,7 @@ func (v *Vaultik) verifyBlobChunks(db *sql.DB, blobHash string, decompressor io. if _, err := io.ReadFull(decompressor, chunkData); err != nil { return 0, fmt.Errorf("failed to read chunk at offset %d: %w", offset, err) } + totalRead += length // Verify chunk hash @@ -401,6 +439,7 @@ func (v *Vaultik) verifyBlobFinalIntegrity(decompressor io.Reader, blobHasher ha if err != nil { return fmt.Errorf("failed to check for remaining blob data: %w", err) } + if remaining > 0 { return fmt.Errorf("blob has %d unexpected trailing bytes not covered by chunk list", remaining) } @@ -424,19 +463,27 @@ func (v *Vaultik) getBlobsFromDatabase(snapshotID string, db *sql.DB) ([]snapsho WHERE sb.snapshot_id = ? ORDER BY b.blob_hash ` + rows, err := db.QueryContext(v.ctx, query, snapshotID) if err != nil { return nil, fmt.Errorf("failed to query snapshot blobs: %w", err) } + defer func() { _ = rows.Close() }() var blobs []snapshot.BlobInfo + for rows.Next() { - var hash string - var size int64 - if err := rows.Scan(&hash, &size); err != nil { + var ( + hash string + size int64 + ) + + err := rows.Scan(&hash, &size) + if err != nil { return nil, fmt.Errorf("failed to scan blob row: %w", err) } + blobs = append(blobs, snapshot.BlobInfo{ Hash: hash, CompressedSize: size, @@ -481,6 +528,7 @@ func (v *Vaultik) verifyManifestAgainstDatabase(manifest *snapshot.Manifest, dbB if !exists { return fmt.Errorf("manifest contains blob %s not in database", hash) } + if dbSize != manifestSize { return fmt.Errorf("blob %s size mismatch: database has %d bytes, manifest has %d bytes", hash, dbSize, manifestSize) @@ -491,6 +539,7 @@ func (v *Vaultik) verifyManifestAgainstDatabase(manifest *snapshot.Manifest, dbB "manifest_blobs", len(manifestBlobMap), "database_blobs", len(dbBlobMap), ) + return nil } @@ -525,6 +574,7 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error { } log.Info("✓ All blobs exist in storage") + return nil } @@ -546,7 +596,8 @@ func (v *Vaultik) performDeepVerificationFromDB(blobs []snapshot.BlobInfo, db *s for i, blobInfo := range blobs { // Verify individual blob - if err := v.verifyBlob(blobInfo, db); err != nil { + err := v.verifyBlob(blobInfo, db) + if err != nil { return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err) } @@ -556,8 +607,10 @@ func (v *Vaultik) performDeepVerificationFromDB(blobs []snapshot.BlobInfo, db *s // Calculate ETA based on bytes processed var eta time.Duration + if bytesProcessed > 0 { bytesPerSec := float64(bytesProcessed) / elapsed.Seconds() + bytesRemaining := totalBytesExpected - bytesProcessed if bytesPerSec > 0 { eta = time.Duration(float64(bytesRemaining)/bytesPerSec) * time.Second diff --git a/internal/vaultik/verify_test.go b/internal/vaultik/verify_test.go index 6ff6adc..c1fc1c3 100644 --- a/internal/vaultik/verify_test.go +++ b/internal/vaultik/verify_test.go @@ -25,6 +25,7 @@ func TestTeeReaderWithDecryption(t *testing.T) { // Compress the data var compressedBuf bytes.Buffer + compressor, err := zstd.NewWriter(&compressedBuf, zstd.WithEncoderLevel(zstd.SpeedDefault)) require.NoError(t, err) _, err = compressor.Write(testData) @@ -40,6 +41,7 @@ func TestTeeReaderWithDecryption(t *testing.T) { require.NoError(t, err) var encryptedBuf bytes.Buffer + err = encryptor.EncryptStream(&encryptedBuf, bytes.NewReader(compressedBuf.Bytes())) require.NoError(t, err) @@ -68,6 +70,7 @@ func TestTeeReaderWithDecryption(t *testing.T) { // Decompress decompressor, err := zstd.NewReader(decryptedReader) require.NoError(t, err) + defer decompressor.Close() // Read all decompressed data (simulating chunk verification)