Leave remote info orphan figures unknown when a manifest is unreadable (closes #228)

When a manifest could not be read, remote info skipped it, counted that
snapshot's blobs as orphaned and advised running prune. The orphan
figures are now unknown in that case, with no prune advice; --json gives
them as null and lists the unreadable remote keys in
unreadable_manifests. Only a listed manifest.json.zst is read, so a
directory without one, as an interrupted backup leaves, keeps the
figures known.

Names under metadata/ were used unchecked and printed raw. A name that
is not a remote key is now skipped with a warning. A manifest under it
is not read either, so it also leaves the figures unknown; --json counts
such manifests in skipped_manifest_count.

Model: opus-5-5

internal/vaultik/info.go

@@ -183,6 +183,11 @@ type SnapshotMetadataInfo struct {
 	TotalSize    int64  `json:"total_size"`
 	BlobCount    int    `json:"blob_count"`
 	BlobsSize    int64  `json:"blobs_size"`
+
+	// Set when the listing holds this snapshot's manifest.json.zst. A
+	// backup interrupted before its manifest upload leaves a directory
+	// without one, which prune does not treat as a snapshot.
+	hasManifest bool
 }
 
 // RemoteInfoResult contains all remote storage information
@@ -206,9 +211,20 @@ type RemoteInfoResult struct {
 	ReferencedBlobCount int   `json:"referenced_blob_count"`
 	ReferencedBlobSize  int64 `json:"referenced_blob_size"`
 
-	// Orphaned blobs
-	OrphanedBlobCount int   `json:"orphaned_blob_count"`
-	OrphanedBlobSize  int64 `json:"orphaned_blob_size"`
+	// Orphaned blobs. Both stay nil (null in the JSON) when a manifest
+	// was listed but not read, since that snapshot's blobs would be
+	// counted as orphaned.
+	OrphanedBlobCount *int   `json:"orphaned_blob_count"`
+	OrphanedBlobSize  *int64 `json:"orphaned_blob_size"`
+
+	// Remote key of each snapshot whose manifest could not be read
+	UnreadableManifests []string `json:"unreadable_manifests,omitempty"`
+
+	// Number of manifests not read because the name above them under
+	// metadata/ is not a remote key. The names themselves are not
+	// reported: they come from the destination store and may hold
+	// control characters.
+	SkippedManifestCount int `json:"skipped_manifest_count,omitempty"`
 }
 
 // RemoteInfo displays information about remote storage
@@ -234,16 +250,28 @@ func (v *Vaultik) RemoteInfo(jsonOutput bool) error {
 		v.stdoutf("Scanning snapshot metadata...\n")
 	}
 
-	snapshotMetadata, snapshotIDs, err := v.collectSnapshotMetadata()
+	snapshotMetadata, snapshotIDs, skippedManifestCount, err := v.collectSnapshotMetadata()
 	if err != nil {
 		return err
 	}
 
+	result.SkippedManifestCount = skippedManifestCount
+
 	if showText {
-		v.stdoutf("Downloading %d manifest(s)...\n", len(snapshotIDs))
+		manifestCount := 0
+
+		for _, info := range snapshotMetadata {
+			if info.hasManifest {
+				manifestCount++
+			}
+		}
+
+		v.stdoutf("Downloading %d manifest(s)...\n", manifestCount)
 	}
 
-	referencedBlobs := v.collectReferencedBlobsFromManifests(snapshotIDs, snapshotMetadata)
+	referencedBlobs, unreadableManifests := v.collectReferencedBlobsFromManifests(
+		snapshotIDs, snapshotMetadata)
+	result.UnreadableManifests = unreadableManifests
 
 	v.populateRemoteInfoResult(result, snapshotMetadata, snapshotIDs, referencedBlobs)
 
@@ -256,7 +284,7 @@ func (v *Vaultik) RemoteInfo(jsonOutput bool) error {
 		"snapshots", result.TotalMetadataCount,
 		"total_blobs", result.TotalBlobCount,
 		"referenced_blobs", result.ReferencedBlobCount,
-		"orphaned_blobs", result.OrphanedBlobCount)
+		"unreadable_manifests", len(result.UnreadableManifests))
 
 	if jsonOutput {
 		enc := json.NewEncoder(v.Stdout)
@@ -273,16 +301,18 @@ func (v *Vaultik) RemoteInfo(jsonOutput bool) error {
 }
 
 // collectSnapshotMetadata scans remote metadata and returns
-// per-snapshot info and sorted IDs.
+// per-snapshot info, sorted IDs and the number of manifests it skipped
+// because the name above them is not a remote key.
 func (v *Vaultik) collectSnapshotMetadata() (
-	map[string]*SnapshotMetadataInfo, []string, error,
+	map[string]*SnapshotMetadataInfo, []string, int, error,
 ) {
 	snapshotMetadata := make(map[string]*SnapshotMetadataInfo)
+	skippedManifestCount := 0
 
 	metadataCh := v.Storage.ListStream(v.ctx, "metadata/")
 	for obj := range metadataCh {
 		if obj.Err != nil {
-			return nil, nil, fmt.Errorf("listing metadata: %w", obj.Err)
+			return nil, nil, 0, fmt.Errorf("listing metadata: %w", obj.Err)
 		}
 
 		parts := strings.Split(obj.Key, "/")
@@ -291,6 +321,22 @@ func (v *Vaultik) collectSnapshotMetadata() (
 		}
 
 		snapshotID := parts[1]
+		filename := parts[2]
+		isManifest := filename == "manifest.json.zst"
+
+		// The name comes from the destination store, which is not
+		// trusted, and is printed in the report. Accept it only in the
+		// form of a remote key.
+		if !isBlobHash(snapshotID) {
+			log.Warn("Skipping non-conforming key under metadata/",
+				"key", obj.Key)
+
+			if isManifest {
+				skippedManifestCount++
+			}
+
+			continue
+		}
 
 		if _, exists := snapshotMetadata[snapshotID]; !exists {
 			snapshotMetadata[snapshotID] = &SnapshotMetadataInfo{SnapshotID: snapshotID}
@@ -298,7 +344,10 @@ func (v *Vaultik) collectSnapshotMetadata() (
 
 		info := snapshotMetadata[snapshotID]
 
-		filename := parts[2]
+		if isManifest {
+			info.hasManifest = true
+		}
+
 		if strings.HasPrefix(filename, "manifest") {
 			info.ManifestSize = obj.Size
 		} else if strings.HasPrefix(filename, "db") {
@@ -315,17 +364,25 @@ func (v *Vaultik) collectSnapshotMetadata() (
 
 	sort.Strings(snapshotIDs)
 
-	return snapshotMetadata, snapshotIDs, nil
+	return snapshotMetadata, snapshotIDs, skippedManifestCount, nil
 }
 
-// collectReferencedBlobsFromManifests downloads manifests and returns
-// referenced blob hashes with sizes.
+// collectReferencedBlobsFromManifests downloads the listed manifests
+// and returns referenced blob hashes with sizes, and the remote keys
+// of the manifests it could not read.
 func (v *Vaultik) collectReferencedBlobsFromManifests(
 	snapshotIDs []string, snapshotMetadata map[string]*SnapshotMetadataInfo,
-) map[string]int64 {
+) (map[string]int64, []string) {
 	referencedBlobs := make(map[string]int64)
 
+	var unreadable []string
+
 	for _, snapshotID := range snapshotIDs {
+		info := snapshotMetadata[snapshotID]
+		if !info.hasManifest {
+			continue
+		}
+
 		// snapshotIDs here are remote keys, taken straight from the
 		// metadata/ listing. downloadManifestByKey is the single reader
 		// for remote manifests; see its doc comment.
@@ -333,10 +390,11 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(
 		if err != nil {
 			log.Warn("Failed to read manifest", "snapshot", snapshotID, "error", err)
 
+			unreadable = append(unreadable, snapshotID)
+
 			continue
 		}
 
-		info := snapshotMetadata[snapshotID]
 		info.BlobCount = manifest.BlobCount
 
 		var blobsSize int64
@@ -349,7 +407,7 @@ func (v *Vaultik) collectReferencedBlobsFromManifests(
 		info.BlobsSize = blobsSize
 	}
 
-	return referencedBlobs
+	return referencedBlobs, unreadable
 }
 
 // populateRemoteInfoResult fills in the result's snapshot and
@@ -378,8 +436,9 @@ func (v *Vaultik) populateRemoteInfoResult(
 }
 
 // scanRemoteBlobStorage lists all blobs on remote and computes orphan
-// stats. showText is true only when the human report is being printed
-// (not --json, not --quiet), gating the progress line.
+// stats when every listed manifest was read. showText is true only
+// when the human report is being printed (not --json, not --quiet),
+// gating the progress line.
 func (v *Vaultik) scanRemoteBlobStorage(
 	result *RemoteInfoResult, referencedBlobs map[string]int64, showText bool,
 ) error {
@@ -406,13 +465,28 @@ func (v *Vaultik) scanRemoteBlobStorage(
 		result.TotalBlobSize += obj.Size
 	}
 
+	// A blob named only by a manifest that could not be read, or by one
+	// under a skipped name, would be counted as orphaned, so the orphan
+	// figures stay unknown.
+	if len(result.UnreadableManifests) > 0 || result.SkippedManifestCount > 0 {
+		return nil
+	}
+
+	var (
+		orphanedCount int
+		orphanedSize  int64
+	)
+
 	for hash, size := range allBlobs {
 		if _, referenced := referencedBlobs[hash]; !referenced {
-			result.OrphanedBlobCount++
-			result.OrphanedBlobSize += size
+			orphanedCount++
+			orphanedSize += size
 		}
 	}
 
+	result.OrphanedBlobCount = &orphanedCount
+	result.OrphanedBlobSize = &orphanedSize
+
 	return nil
 }
 
@@ -465,11 +539,21 @@ func (v *Vaultik) printRemoteInfoTable(result *RemoteInfoResult) {
 	v.stdoutf("Referenced by snapshots:    %s (%s)\n",
 		humanize.Comma(int64(result.ReferencedBlobCount)),
 		ubytes(result.ReferencedBlobSize))
-	v.stdoutf("Orphaned (unreferenced):    %s (%s)\n",
-		humanize.Comma(int64(result.OrphanedBlobCount)),
-		ubytes(result.OrphanedBlobSize))
 
-	if result.OrphanedBlobCount > 0 {
+	if result.OrphanedBlobCount == nil {
+		v.stdoutf("Orphaned (unreferenced):    unknown "+
+			"(%d manifest(s) could not be read, "+
+			"%d manifest(s) under a non-conforming name skipped)\n",
+			len(result.UnreadableManifests), result.SkippedManifestCount)
+
+		return
+	}
+
+	v.stdoutf("Orphaned (unreferenced):    %s (%s)\n",
+		humanize.Comma(int64(*result.OrphanedBlobCount)),
+		ubytes(*result.OrphanedBlobSize))
+
+	if *result.OrphanedBlobCount > 0 {
 		v.stdoutf("\nRun 'vaultik prune' to remove orphaned blobs.\n")
 	}
 }