check / check (push) Waiting to run
WithReadTx ended the process when its rollback failed, including after Ctrl-C, when database/sql has already rolled the transaction back and Rollback returns sql.ErrTxDone. It now handles the rollback as WithTx does. The query functions ended the process when closing their rows failed; they now ignore the deferred close, because database/sql reports a close failure through rows.Err(), which each returns, once the rows are read to the end. The unused database.Fatalf is removed. cleanupBlobTempFile returns its close or remove failure and the upload step returns it, so such a failure now fails the backup. No test makes a rows close fail: SQLite fails it only with the error the last read already returned. Model: opus-5-5
197 lines
4.0 KiB
Go
197 lines
4.0 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"sneak.berlin/go/vaultik/internal/log"
|
|
)
|
|
|
|
// ChunkRepository provides access to the chunks table, which tracks
|
|
// content-defined chunks by hash and size.
|
|
type ChunkRepository struct {
|
|
db *DB
|
|
}
|
|
|
|
// NewChunkRepository creates a ChunkRepository backed by db.
|
|
func NewChunkRepository(db *DB) *ChunkRepository {
|
|
return &ChunkRepository{db: db}
|
|
}
|
|
|
|
// Create inserts a chunk row (idempotently), using tx when non-nil.
|
|
func (r *ChunkRepository) Create(ctx context.Context, tx *sql.Tx, chunk *Chunk) error {
|
|
query := `
|
|
INSERT INTO chunks (chunk_hash, size)
|
|
VALUES (?, ?)
|
|
ON CONFLICT(chunk_hash) DO NOTHING
|
|
`
|
|
|
|
var err error
|
|
if tx != nil {
|
|
_, err = tx.ExecContext(ctx, query, chunk.ChunkHash, chunk.Size)
|
|
} else {
|
|
_, err = r.db.ExecWithLog(ctx, query, chunk.ChunkHash, chunk.Size)
|
|
}
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("inserting chunk: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetByHash returns the chunk with the given hash, or nil if it is not
|
|
// known to the index.
|
|
func (r *ChunkRepository) GetByHash(ctx context.Context, hash string) (*Chunk, error) {
|
|
query := `
|
|
SELECT chunk_hash, size
|
|
FROM chunks
|
|
WHERE chunk_hash = ?
|
|
`
|
|
|
|
var chunk Chunk
|
|
|
|
err := r.db.conn.QueryRowContext(ctx, query, hash).Scan(
|
|
&chunk.ChunkHash,
|
|
&chunk.Size,
|
|
)
|
|
|
|
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 chunk: %w", err)
|
|
}
|
|
|
|
return &chunk, nil
|
|
}
|
|
|
|
// GetByHashes returns the chunks whose hashes appear in hashes, ordered by
|
|
// chunk hash. Unknown hashes are silently omitted from the result.
|
|
func (r *ChunkRepository) GetByHashes(
|
|
ctx context.Context, hashes []string,
|
|
) ([]*Chunk, error) {
|
|
if len(hashes) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
query := `
|
|
SELECT chunk_hash, size
|
|
FROM chunks
|
|
WHERE chunk_hash IN (`
|
|
|
|
args := make([]any, len(hashes))
|
|
|
|
var querySb75 strings.Builder
|
|
|
|
for i, hash := range hashes {
|
|
if i > 0 {
|
|
querySb75.WriteString(", ")
|
|
}
|
|
|
|
querySb75.WriteString("?")
|
|
|
|
args[i] = hash
|
|
}
|
|
|
|
query += querySb75.String()
|
|
|
|
query += ") ORDER BY chunk_hash"
|
|
|
|
rows, err := r.db.conn.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("querying chunks: %w", err)
|
|
}
|
|
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
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()
|
|
}
|
|
|
|
// ListUnpacked returns up to limit chunks that are not yet stored in any
|
|
// blob, ordered by chunk hash.
|
|
func (r *ChunkRepository) ListUnpacked(
|
|
ctx context.Context, limit int,
|
|
) ([]*Chunk, error) {
|
|
query := `
|
|
SELECT c.chunk_hash, c.size
|
|
FROM chunks c
|
|
LEFT JOIN blob_chunks bc ON c.chunk_hash = bc.chunk_hash
|
|
WHERE bc.chunk_hash IS NULL
|
|
ORDER BY c.chunk_hash
|
|
LIMIT ?
|
|
`
|
|
|
|
rows, err := r.db.conn.QueryContext(ctx, query, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("querying unpacked chunks: %w", err)
|
|
}
|
|
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
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()
|
|
}
|
|
|
|
// DeleteOrphaned deletes chunks that are not referenced by any file or blob
|
|
func (r *ChunkRepository) DeleteOrphaned(ctx context.Context) error {
|
|
query := `
|
|
DELETE FROM chunks
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM file_chunks
|
|
WHERE file_chunks.chunk_hash = chunks.chunk_hash
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM blob_chunks
|
|
WHERE blob_chunks.chunk_hash = chunks.chunk_hash
|
|
)
|
|
`
|
|
|
|
result, err := r.db.ExecWithLog(ctx, query)
|
|
if err != nil {
|
|
return fmt.Errorf("deleting orphaned chunks: %w", err)
|
|
}
|
|
|
|
rowsAffected, _ := result.RowsAffected()
|
|
if rowsAffected > 0 {
|
|
log.Debug("Deleted orphaned chunks", "count", rowsAffected)
|
|
}
|
|
|
|
return nil
|
|
}
|