clawbot
38ebfd843a
An interrupted blob upload left the blob's chunks, blob_chunks, and blobs rows committed before the upload was attempted, so a later run deduplicated against data that never reached storage and produced a snapshot that reported success but could not be restored. Fix (issue option b): a chunk counts as known only when a blob holding it has uploaded_ts set, and each run drops un-uploaded blob rows and the chunks they orphan at startup, so the affected data is re-chunked and re-uploaded. A blob recorded with no remote backend is marked uploaded so the invariant holds uniformly. The reproduction is the interrupted-upload test from #72: its t.Skip is removed and it passes against this fix, and this branch's earlier duplicate copy is dropped. The interrupted metadata-export case is split to #177. Model: opus-4-8
66 lines
1.6 KiB
Go
66 lines
1.6 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
)
|
|
|
|
// List returns every chunk in the index, ordered by chunk hash.
|
|
func (r *ChunkRepository) List(ctx context.Context) ([]*Chunk, error) {
|
|
return r.list(ctx, `
|
|
SELECT chunk_hash, size
|
|
FROM chunks
|
|
ORDER BY chunk_hash
|
|
`)
|
|
}
|
|
|
|
// ListInUploadedBlobs returns the chunks that are stored in a blob whose
|
|
// upload has completed (uploaded_ts set), ordered by chunk hash. These
|
|
// are the only chunks a backup may safely deduplicate against: a chunk
|
|
// recorded solely in a blob that was never uploaded refers to data that
|
|
// is not in remote storage, so trusting it would silently drop that data
|
|
// from later snapshots.
|
|
func (r *ChunkRepository) ListInUploadedBlobs(ctx context.Context) ([]*Chunk, error) {
|
|
return r.list(ctx, `
|
|
SELECT DISTINCT c.chunk_hash, c.size
|
|
FROM chunks c
|
|
JOIN blob_chunks bc ON c.chunk_hash = bc.chunk_hash
|
|
JOIN blobs b ON bc.blob_id = b.id
|
|
WHERE b.uploaded_ts IS NOT NULL
|
|
ORDER BY c.chunk_hash
|
|
`)
|
|
}
|
|
|
|
// list runs a chunk-selecting query and scans the (chunk_hash, size) rows.
|
|
func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, error) {
|
|
rows, err := r.db.conn.QueryContext(ctx, query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("querying chunks: %w", err)
|
|
}
|
|
|
|
defer func() {
|
|
err := rows.Close()
|
|
if err != nil {
|
|
Fatalf("failed to close rows: %v", err)
|
|
}
|
|
}()
|
|
|
|
var chunks []*Chunk
|
|
|
|
for rows.Next() {
|
|
var chunk Chunk
|
|
|
|
err := rows.Scan(
|
|
&chunk.ChunkHash,
|
|
&chunk.Size,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("scanning chunk: %w", err)
|
|
}
|
|
|
|
chunks = append(chunks, &chunk)
|
|
}
|
|
|
|
return chunks, rows.Err()
|
|
}
|