Files
vaultik/internal/database/snapshots.go
T
clawbot 49eed7a3e5
check / check (push) Waiting to run
Count each file, byte and upload once in backup statistics (closes #225)
The scanner added a changed file's bytes again for each new chunk and
counted a file as unchanged for each chunk already stored. It now
counts files and bytes once, in the scan phase, and counts its own
uploads, so a --cron run, which has no progress reporter, records
them. The blob count no longer adds earlier paths' blobs again.

The snapshots row now stores the size of all files in total_size and
the referenced blobs' sizes in blob_size, blob_uncompressed_size and
compression_ratio, as docs/DATAMODEL.md says. Those sizes come from
one query, and a failed query fails the snapshot. DATAMODEL.md now
says chunk_count and blob_count count what the run added.

Removed UpdateSnapshotStats and GetCountBySnapshot, which nothing
calls any more.

Model: opus-5-5
2026-10-07 02:59:27 +02:00

830 lines
21 KiB
Go

package database
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"time"
"sneak.berlin/go/vaultik/internal/types"
)
// Sentinel errors for the single-snapshot invariant that an exported
// per-snapshot metadata database must satisfy.
var (
// ErrNoSnapshotInDatabase means the metadata database has no snapshot
// row at all.
ErrNoSnapshotInDatabase = errors.New("database contains no snapshot")
// ErrMultipleSnapshotsInDatabase means the metadata database holds
// more than the single snapshot an export is supposed to contain.
ErrMultipleSnapshotsInDatabase = errors.New(
"database contains more than one snapshot")
)
// SnapshotRepository provides access to the snapshots table and its
// snapshot_files / snapshot_blobs association tables.
type SnapshotRepository struct {
db *DB
}
// NewSnapshotRepository creates a SnapshotRepository backed by db.
func NewSnapshotRepository(db *DB) *SnapshotRepository {
return &SnapshotRepository{db: db}
}
// Create inserts a snapshot row, using tx when non-nil.
func (r *SnapshotRepository) Create(
ctx context.Context, tx *sql.Tx, snapshot *Snapshot,
) error {
query := `
INSERT INTO snapshots (id, hostname, vaultik_version,
vaultik_git_revision, started_at, completed_at,
file_count, chunk_count, blob_count, total_size, blob_size,
blob_uncompressed_size, compression_ratio, compression_level,
upload_bytes, upload_duration_ms)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
var completedAt *int64
if snapshot.CompletedAt != nil {
ts := snapshot.CompletedAt.Unix()
completedAt = &ts
}
args := []any{
snapshot.ID, snapshot.Hostname, snapshot.VaultikVersion,
snapshot.VaultikGitRevision, snapshot.StartedAt.Unix(),
completedAt, snapshot.FileCount, snapshot.ChunkCount,
snapshot.BlobCount, snapshot.TotalSize, snapshot.BlobSize,
snapshot.BlobUncompressedSize, snapshot.CompressionRatio,
snapshot.CompressionLevel, snapshot.UploadBytes,
snapshot.UploadDurationMs,
}
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query, args...)
} else {
_, err = r.db.ExecWithLog(ctx, query, args...)
}
if err != nil {
return fmt.Errorf("inserting snapshot: %w", err)
}
return nil
}
// UpdateCounts updates a snapshot's file/chunk/blob counters and sizes,
// recomputing the compression ratio, using tx when non-nil.
func (r *SnapshotRepository) UpdateCounts(
ctx context.Context,
tx *sql.Tx,
snapshotID string,
fileCount, chunkCount, blobCount, totalSize, blobSize int64,
) error {
compressionRatio := 1.0
if totalSize > 0 {
compressionRatio = float64(blobSize) / float64(totalSize)
}
query := `
UPDATE snapshots
SET file_count = ?,
chunk_count = ?,
blob_count = ?,
total_size = ?,
blob_size = ?,
compression_ratio = ?
WHERE id = ?
`
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query,
fileCount, chunkCount, blobCount, totalSize, blobSize,
compressionRatio, snapshotID)
} else {
_, err = r.db.ExecWithLog(ctx, query,
fileCount, chunkCount, blobCount, totalSize, blobSize,
compressionRatio, snapshotID)
}
if err != nil {
return fmt.Errorf("updating snapshot: %w", err)
}
return nil
}
// UpdateExtendedStats updates extended statistics for a snapshot
func (r *SnapshotRepository) UpdateExtendedStats(
ctx context.Context,
tx *sql.Tx,
snapshotID string,
blobUncompressedSize int64,
compressionLevel int,
uploadBytes int64,
uploadDurationMs int64,
) error {
compressionRatio, err := r.extendedCompressionRatio(
ctx, tx, snapshotID, blobUncompressedSize,
)
if err != nil {
return err
}
query := `
UPDATE snapshots
SET blob_uncompressed_size = ?,
compression_ratio = ?,
compression_level = ?,
upload_bytes = ?,
upload_duration_ms = ?
WHERE id = ?
`
if tx != nil {
_, err = tx.ExecContext(ctx, query,
blobUncompressedSize, compressionRatio, compressionLevel,
uploadBytes, uploadDurationMs, snapshotID)
} else {
_, err = r.db.ExecWithLog(ctx, query,
blobUncompressedSize, compressionRatio, compressionLevel,
uploadBytes, uploadDurationMs, snapshotID)
}
if err != nil {
return fmt.Errorf("updating extended stats: %w", err)
}
return nil
}
// GetByID returns the snapshot with the given ID, or nil if no such
// snapshot exists.
func (r *SnapshotRepository) GetByID(
ctx context.Context, snapshotID string,
) (*Snapshot, error) {
query := `
SELECT id, hostname, vaultik_version, vaultik_git_revision, started_at, completed_at,
file_count, chunk_count, blob_count, total_size, blob_size, blob_uncompressed_size,
compression_ratio, compression_level, upload_bytes, upload_duration_ms
FROM snapshots
WHERE id = ?
`
var (
snapshot Snapshot
startedAtUnix int64
completedAtUnix *int64
)
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(
&snapshot.ID,
&snapshot.Hostname,
&snapshot.VaultikVersion,
&snapshot.VaultikGitRevision,
&startedAtUnix,
&completedAtUnix,
&snapshot.FileCount,
&snapshot.ChunkCount,
&snapshot.BlobCount,
&snapshot.TotalSize,
&snapshot.BlobSize,
&snapshot.BlobUncompressedSize,
&snapshot.CompressionRatio,
&snapshot.CompressionLevel,
&snapshot.UploadBytes,
&snapshot.UploadDurationMs,
)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil //nolint:nilnil // nil,nil signals not-found; callers check nil
}
if err != nil {
return nil, fmt.Errorf("querying snapshot: %w", err)
}
snapshot.StartedAt = time.Unix(startedAtUnix, 0).UTC()
if completedAtUnix != nil {
t := time.Unix(*completedAtUnix, 0).UTC()
snapshot.CompletedAt = &t
}
return &snapshot, nil
}
// GetOnlySnapshot returns the sole snapshot in an exported per-snapshot
// metadata database. The backup path writes each snapshot's database with
// exactly one snapshot row (see cleanSnapshotDB), so restore and deep
// verify expect exactly one. Zero rows return ErrNoSnapshotInDatabase and
// more than one returns ErrMultipleSnapshotsInDatabase; callers treat
// either as a failed identity check on the downloaded database.
func (r *SnapshotRepository) GetOnlySnapshot(ctx context.Context) (*Snapshot, error) {
query := `
SELECT id, hostname, vaultik_version, vaultik_git_revision,
started_at, completed_at, file_count, chunk_count, blob_count,
total_size, blob_size, compression_ratio
FROM snapshots
LIMIT 2
`
rows, err := r.db.conn.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("querying snapshots: %w", err)
}
defer func() {
err := rows.Close()
if err != nil {
Fatalf("failed to close rows: %v", err)
}
}()
snapshots, err := r.scanSnapshotRows(rows)
if err != nil {
return nil, err
}
switch len(snapshots) {
case 1:
return snapshots[0], nil
case 0:
return nil, ErrNoSnapshotInDatabase
default:
return nil, ErrMultipleSnapshotsInDatabase
}
}
// ListRecent returns up to limit snapshots, most recently started first.
func (r *SnapshotRepository) ListRecent(
ctx context.Context, limit int,
) ([]*Snapshot, error) {
query := `
SELECT id, hostname, vaultik_version, vaultik_git_revision,
started_at, completed_at, file_count, chunk_count, blob_count,
total_size, blob_size, compression_ratio
FROM snapshots
ORDER BY started_at DESC
LIMIT ?
`
rows, err := r.db.conn.QueryContext(ctx, query, limit)
if err != nil {
return nil, fmt.Errorf("querying snapshots: %w", err)
}
defer func() {
err := rows.Close()
if err != nil {
Fatalf("failed to close rows: %v", err)
}
}()
return r.scanSnapshotRows(rows)
}
// MarkComplete marks a snapshot as completed with the current timestamp
func (r *SnapshotRepository) MarkComplete(
ctx context.Context, tx *sql.Tx, snapshotID string,
) error {
query := `
UPDATE snapshots
SET completed_at = ?
WHERE id = ?
`
completedAt := time.Now().UTC().Unix()
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query, completedAt, snapshotID)
} else {
_, err = r.db.ExecWithLog(ctx, query, completedAt, snapshotID)
}
if err != nil {
return fmt.Errorf("marking snapshot complete: %w", err)
}
return nil
}
// AddFile adds a file to a snapshot
func (r *SnapshotRepository) AddFile(
ctx context.Context, tx *sql.Tx, snapshotID string, filePath string,
) error {
query := `
INSERT OR IGNORE INTO snapshot_files (snapshot_id, file_id)
SELECT ?, id FROM files WHERE path = ?
`
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query, snapshotID, filePath)
} else {
_, err = r.db.ExecWithLog(ctx, query, snapshotID, filePath)
}
if err != nil {
return fmt.Errorf("adding file to snapshot: %w", err)
}
return nil
}
// AddFileByID adds a file to a snapshot by file ID
func (r *SnapshotRepository) AddFileByID(
ctx context.Context, tx *sql.Tx, snapshotID string, fileID types.FileID,
) error {
query := `
INSERT OR IGNORE INTO snapshot_files (snapshot_id, file_id)
VALUES (?, ?)
`
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query, snapshotID, fileID.String())
} else {
_, err = r.db.ExecWithLog(ctx, query, snapshotID, fileID.String())
}
if err != nil {
return fmt.Errorf("adding file to snapshot: %w", err)
}
return nil
}
// AddFilesByIDBatch adds multiple files to a snapshot in batched inserts
func (r *SnapshotRepository) AddFilesByIDBatch(
ctx context.Context, tx *sql.Tx, snapshotID string, fileIDs []types.FileID,
) error {
if len(fileIDs) == 0 {
return nil
}
// Each snapshot_files row binds this many SQL variables.
const snapshotFileCols = 2
// Batch at 400 rows to be safe with SQLite's variable limit.
const batchSize = 400
for i := 0; i < len(fileIDs); i += batchSize {
end := min(i+batchSize, len(fileIDs))
batch := fileIDs[i:end]
query := "INSERT OR IGNORE INTO snapshot_files (snapshot_id, file_id) VALUES "
args := make([]any, 0, len(batch)*snapshotFileCols)
var querySb312 strings.Builder
for j, fileID := range batch {
if j > 0 {
querySb312.WriteString(", ")
}
querySb312.WriteString("(?, ?)")
args = append(args, snapshotID, fileID.String())
}
query += querySb312.String()
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query, args...)
} else {
_, err = r.db.ExecWithLog(ctx, query, args...)
}
if err != nil {
return fmt.Errorf("batch adding files to snapshot: %w", err)
}
}
return nil
}
// PopulateReferencedBlobs ensures snapshot_blobs contains an entry for
// every blob that holds a chunk referenced by any file in the snapshot.
// This is necessary because the AddBlob hook only runs when a blob is
// newly uploaded during a snapshot — fully-deduplicated snapshots (where
// every chunk already exists in storage from a prior run) would otherwise
// have an empty snapshot_blobs set and be impossible to restore.
//
// Returns the number of rows inserted (i.e. blobs that were previously
// referenced indirectly via file_chunks but not yet recorded in
// snapshot_blobs for this snapshot).
func (r *SnapshotRepository) PopulateReferencedBlobs(
ctx context.Context, tx *sql.Tx, snapshotID string,
) (int64, error) {
query := `
INSERT OR IGNORE INTO snapshot_blobs (snapshot_id, blob_id, blob_hash)
SELECT DISTINCT ?, blobs.id, blobs.blob_hash
FROM blobs
JOIN blob_chunks ON blob_chunks.blob_id = blobs.id
JOIN file_chunks ON file_chunks.chunk_hash = blob_chunks.chunk_hash
JOIN snapshot_files ON snapshot_files.file_id = file_chunks.file_id
WHERE snapshot_files.snapshot_id = ?
AND blobs.blob_hash IS NOT NULL
`
var (
result sql.Result
err error
)
if tx != nil {
result, err = tx.ExecContext(ctx, query, snapshotID, snapshotID)
} else {
result, err = r.db.ExecWithLog(ctx, query, snapshotID, snapshotID)
}
if err != nil {
return 0, fmt.Errorf("populating referenced blobs: %w", err)
}
n, _ := result.RowsAffected()
return n, nil
}
// AddBlob adds a blob to a snapshot
func (r *SnapshotRepository) AddBlob(
ctx context.Context,
tx *sql.Tx,
snapshotID string,
blobID types.BlobID,
blobHash types.BlobHash,
) error {
query := `
INSERT OR IGNORE INTO snapshot_blobs (snapshot_id, blob_id, blob_hash)
VALUES (?, ?, ?)
`
var err error
if tx != nil {
_, err = tx.ExecContext(ctx, query, snapshotID, blobID.String(), blobHash.String())
} else {
_, err = r.db.ExecWithLog(ctx, query, snapshotID, blobID.String(), blobHash.String())
}
if err != nil {
return fmt.Errorf("adding blob to snapshot: %w", err)
}
return nil
}
// GetBlobHashes returns all blob hashes for a snapshot
func (r *SnapshotRepository) GetBlobHashes(
ctx context.Context, snapshotID string,
) ([]string, error) {
query := `
SELECT sb.blob_hash
FROM snapshot_blobs sb
WHERE sb.snapshot_id = ?
ORDER BY sb.blob_hash
`
rows, err := r.db.conn.QueryContext(ctx, query, snapshotID)
if err != nil {
return nil, fmt.Errorf("querying blob hashes: %w", err)
}
defer func() {
err := rows.Close()
if err != nil {
Fatalf("failed to close rows: %v", err)
}
}()
var blobs []string
for rows.Next() {
var blobHash string
err := rows.Scan(&blobHash)
if err != nil {
return nil, fmt.Errorf("scanning blob hash: %w", err)
}
blobs = append(blobs, blobHash)
}
return blobs, rows.Err()
}
// GetSnapshotTotalCompressedSize returns the total compressed size of all
// blobs referenced by a snapshot.
func (r *SnapshotRepository) GetSnapshotTotalCompressedSize(
ctx context.Context, snapshotID string,
) (int64, error) {
query := `
SELECT COALESCE(SUM(b.compressed_size), 0)
FROM snapshot_blobs sb
JOIN blobs b ON sb.blob_hash = b.blob_hash
WHERE sb.snapshot_id = ?
`
var totalSize int64
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&totalSize)
if err != nil {
return 0, fmt.Errorf("querying total compressed size: %w", err)
}
return totalSize, nil
}
// GetSnapshotBlobSizes returns the total compressed and uncompressed sizes
// of all blobs referenced by a snapshot.
func (r *SnapshotRepository) GetSnapshotBlobSizes(
ctx context.Context, snapshotID string,
) (int64, int64, error) {
query := `
SELECT COALESCE(SUM(b.compressed_size), 0),
COALESCE(SUM(b.uncompressed_size), 0)
FROM snapshot_blobs sb
JOIN blobs b ON sb.blob_hash = b.blob_hash
WHERE sb.snapshot_id = ?
`
var compressed, uncompressed int64
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(
&compressed, &uncompressed)
if err != nil {
return 0, 0, fmt.Errorf("querying snapshot blob sizes: %w", err)
}
return compressed, uncompressed, nil
}
// GetSnapshotUncompressedChunkSize returns the sum of plaintext sizes of all unique
// chunks referenced by a snapshot (via snapshot_files → file_chunks → chunks).
func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
ctx context.Context, snapshotID string,
) (int64, error) {
query := `
SELECT COALESCE(SUM(c.size), 0)
FROM (
SELECT DISTINCT fc.chunk_hash
FROM snapshot_files sf
JOIN file_chunks fc ON sf.file_id = fc.file_id
WHERE sf.snapshot_id = ?
) sc
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
`
var totalSize int64
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&totalSize)
if err != nil {
return 0, fmt.Errorf("querying uncompressed chunk size: %w", err)
}
return totalSize, nil
}
// GetSnapshotNewChunkSize returns the sum of plaintext sizes of chunks that are
// referenced by this snapshot but not by any earlier completed snapshot known to
// the local database. The result is the marginal uncompressed data this snapshot
// added to the dedup pool — i.e., the delta from prior snapshots.
func (r *SnapshotRepository) GetSnapshotNewChunkSize(
ctx context.Context, snapshotID string,
) (int64, error) {
query := `
WITH this_snap_chunks AS (
SELECT DISTINCT fc.chunk_hash
FROM snapshot_files sf
JOIN file_chunks fc ON sf.file_id = fc.file_id
WHERE sf.snapshot_id = ?
),
prior_chunks AS (
SELECT DISTINCT fc.chunk_hash
FROM snapshots s
JOIN snapshot_files sf ON sf.snapshot_id = s.id
JOIN file_chunks fc ON fc.file_id = sf.file_id
WHERE s.completed_at IS NOT NULL
AND s.id != ?
AND s.started_at < (SELECT started_at FROM snapshots WHERE id = ?)
)
SELECT COALESCE(SUM(c.size), 0)
FROM chunks c
JOIN this_snap_chunks t ON c.chunk_hash = t.chunk_hash
WHERE c.chunk_hash NOT IN (SELECT chunk_hash FROM prior_chunks)
`
var totalSize int64
err := r.db.conn.QueryRowContext(
ctx, query, snapshotID, snapshotID, snapshotID,
).Scan(&totalSize)
if err != nil {
return 0, fmt.Errorf("querying new chunk size: %w", err)
}
return totalSize, nil
}
// GetIncompleteSnapshots returns all snapshots that haven't been completed
func (r *SnapshotRepository) GetIncompleteSnapshots(
ctx context.Context,
) ([]*Snapshot, error) {
query := `
SELECT id, hostname, vaultik_version, vaultik_git_revision,
started_at, completed_at, file_count, chunk_count, blob_count,
total_size, blob_size, compression_ratio
FROM snapshots
WHERE completed_at IS NULL
ORDER BY started_at DESC
`
rows, err := r.db.conn.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
}
defer func() {
err := rows.Close()
if err != nil {
Fatalf("failed to close rows: %v", err)
}
}()
return r.scanSnapshotRows(rows)
}
// GetIncompleteByHostname returns all incomplete snapshots for a specific hostname
func (r *SnapshotRepository) GetIncompleteByHostname(
ctx context.Context, hostname string,
) ([]*Snapshot, error) {
query := `
SELECT id, hostname, vaultik_version, vaultik_git_revision,
started_at, completed_at, file_count, chunk_count, blob_count,
total_size, blob_size, compression_ratio
FROM snapshots
WHERE completed_at IS NULL AND hostname = ?
ORDER BY started_at DESC
`
rows, err := r.db.conn.QueryContext(ctx, query, hostname)
if err != nil {
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
}
defer func() {
err := rows.Close()
if err != nil {
Fatalf("failed to close rows: %v", err)
}
}()
// Same column set as every other multi-row snapshot query, so the
// shared scanner applies — including its timestamp normalization.
return r.scanSnapshotRows(rows)
}
// Delete removes a snapshot record
func (r *SnapshotRepository) Delete(ctx context.Context, snapshotID string) error {
query := `DELETE FROM snapshots WHERE id = ?`
_, err := r.db.ExecWithLog(ctx, query, snapshotID)
if err != nil {
return fmt.Errorf("deleting snapshot: %w", err)
}
return nil
}
// DeleteSnapshotFiles removes all snapshot_files entries for a snapshot
func (r *SnapshotRepository) DeleteSnapshotFiles(
ctx context.Context, snapshotID string,
) error {
query := `DELETE FROM snapshot_files WHERE snapshot_id = ?`
_, err := r.db.ExecWithLog(ctx, query, snapshotID)
if err != nil {
return fmt.Errorf("deleting snapshot files: %w", err)
}
return nil
}
// DeleteSnapshotBlobs removes all snapshot_blobs entries for a snapshot
func (r *SnapshotRepository) DeleteSnapshotBlobs(
ctx context.Context, snapshotID string,
) error {
query := `DELETE FROM snapshot_blobs WHERE snapshot_id = ?`
_, err := r.db.ExecWithLog(ctx, query, snapshotID)
if err != nil {
return fmt.Errorf("deleting snapshot blobs: %w", err)
}
return nil
}
// DeleteSnapshotUploads removes all uploads entries for a snapshot
func (r *SnapshotRepository) DeleteSnapshotUploads(
ctx context.Context, snapshotID string,
) error {
query := `DELETE FROM uploads WHERE snapshot_id = ?`
_, err := r.db.ExecWithLog(ctx, query, snapshotID)
if err != nil {
return fmt.Errorf("deleting snapshot uploads: %w", err)
}
return nil
}
// extendedCompressionRatio computes the compression ratio for a snapshot
// from its stored blob_size and the given uncompressed size. Returns 1.0
// when the uncompressed size is zero.
func (r *SnapshotRepository) extendedCompressionRatio(
ctx context.Context,
tx *sql.Tx,
snapshotID string,
blobUncompressedSize int64,
) (float64, error) {
if blobUncompressedSize <= 0 {
return 1.0, nil
}
// Get current blob_size from DB to calculate ratio
var blobSize int64
queryGet := `SELECT blob_size FROM snapshots WHERE id = ?`
var err error
if tx != nil {
err = tx.QueryRowContext(ctx, queryGet, snapshotID).Scan(&blobSize)
} else {
err = r.db.conn.QueryRowContext(ctx, queryGet, snapshotID).Scan(&blobSize)
}
if err != nil {
return 0, fmt.Errorf("getting blob size: %w", err)
}
return float64(blobSize) / float64(blobUncompressedSize), nil
}
// scanSnapshotRows scans the standard snapshot column set from a rows
// iterator into Snapshot records.
func (r *SnapshotRepository) scanSnapshotRows(rows *sql.Rows) ([]*Snapshot, error) {
var snapshots []*Snapshot
for rows.Next() {
var (
snapshot Snapshot
startedAtUnix int64
completedAtUnix *int64
)
err := rows.Scan(
&snapshot.ID,
&snapshot.Hostname,
&snapshot.VaultikVersion,
&snapshot.VaultikGitRevision,
&startedAtUnix,
&completedAtUnix,
&snapshot.FileCount,
&snapshot.ChunkCount,
&snapshot.BlobCount,
&snapshot.TotalSize,
&snapshot.BlobSize,
&snapshot.CompressionRatio,
)
if err != nil {
return nil, fmt.Errorf("scanning snapshot: %w", err)
}
// UTC, matching every other snapshot scanner in this file. The
// column holds a bare Unix second, so the zone is a decode
// choice rather than stored data, and callers render these
// timestamps through zone-less format strings alongside
// timestamps read from remote manifests. Decoding in the host's
// local zone here would put two different wall clocks in one
// column.
snapshot.StartedAt = time.Unix(startedAtUnix, 0).UTC()
if completedAtUnix != nil {
t := time.Unix(*completedAtUnix, 0).UTC()
snapshot.CompletedAt = &t
}
snapshots = append(snapshots, &snapshot)
}
return snapshots, rows.Err()
}