Implement persistent SQLite scan database

scan now synchronizes a database that survives between runs (SFDUPES_DATABASE, default /var/lib/sfdupes/db.sqlite) instead of emitting a stream: operands are resolved to absolute paths, unchanged files (same size, mtime not newer than recorded) are never re-read, new and changed files are hashed, and records under the scanned operands that were not verified this run are deleted; records outside the operands are untouched. All changes commit in a single transaction, and WAL journaling with a busy timeout keeps a report run during a cron scan safe. report and trees read the database (no positional arguments); the NUL-terminated stream format, its parser, and the malformed-record handling are gone. The driver is modernc.org/sqlite (pure Go), so builds keep cgo disabled.
2026-07-24 03:08:49 +07:00
parent e0d578a707
commit d8fbcb32c2
12 changed files with 1229 additions and 379 deletions
--- a/db.go
+++ b/db.go
@@ -0,0 +1,319 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	// The pure-Go SQLite driver, registered as "sqlite"; keeps cgo
+	// disabled.
+	_ "modernc.org/sqlite"
+)
+
+// defaultDatabasePath is where the persistent scan database lives when
+// SFDUPES_DATABASE is not set.
+const defaultDatabasePath = "/var/lib/sfdupes/db.sqlite"
+
+// databaseEnv is the environment variable that overrides the database
+// path.
+const databaseEnv = "SFDUPES_DATABASE"
+
+// schemaVersion is the database schema version this build reads and
+// writes, stored in PRAGMA user_version.
+const schemaVersion = 1
+
+// dbDirPerm is the mode for a database parent directory created by
+// scan.
+const dbDirPerm = 0o755
+
+// createTableSQL is the schema applied to a fresh database. Paths are
+// BLOBs because Unix paths are raw bytes, not guaranteed UTF-8.
+const createTableSQL = `
+CREATE TABLE files (
+    path  BLOB PRIMARY KEY,
+    size  INTEGER NOT NULL,
+    mtime INTEGER NOT NULL,
+    head  TEXT NOT NULL,
+    tail  TEXT NOT NULL
+) WITHOUT ROWID
+`
+
+// upsertSQL inserts one file record, replacing any existing record for
+// the same path.
+const upsertSQL = `
+INSERT INTO files (path, size, mtime, head, tail)
+VALUES (?, ?, ?, ?, ?)
+ON CONFLICT (path) DO UPDATE SET
+    size = excluded.size, mtime = excluded.mtime,
+    head = excluded.head, tail = excluded.tail
+`
+
+// errNoDatabase reports a missing database file for report/trees.
+var errNoDatabase = errors.New(
+	"no database (run \"sfdupes scan\" first, or set " + databaseEnv + ")")
+
+// errSchemaVersion reports a database whose schema version this build
+// does not understand.
+var errSchemaVersion = errors.New("unsupported database schema version")
+
+// databasePath resolves the database location: SFDUPES_DATABASE when
+// set and non-empty, the compiled-in default otherwise.
+func databasePath() string {
+	if p := os.Getenv(databaseEnv); p != "" {
+		return p
+	}
+
+	return defaultDatabasePath
+}
+
+// openDB opens the SQLite database at path with WAL journaling and a
+// busy timeout, so a report can run while a cron scan is in progress.
+// It does not create or verify the schema.
+func openDB(path string) (*sql.DB, error) {
+	dsn := "file:" + path +
+		"?_pragma=busy_timeout(10000)" +
+		"&_pragma=journal_mode(WAL)" +
+		"&_pragma=synchronous(NORMAL)"
+
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return nil, fmt.Errorf("open database %s: %w", path, err)
+	}
+
+	// A single connection avoids SQLITE_BUSY between this process's
+	// own connections; concurrency lives in the worker pools, not in
+	// parallel database access.
+	db.SetMaxOpenConns(1)
+
+	return db, nil
+}
+
+// openScanDatabase opens the database for the scan subcommand, creating
+// the file, its parent directory, and the schema as needed.
+func openScanDatabase(path string) (*sql.DB, error) {
+	err := os.MkdirAll(filepath.Dir(path), dbDirPerm)
+	if err != nil {
+		return nil, fmt.Errorf("create database directory: %w", err)
+	}
+
+	db, err := openDB(path)
+	if err != nil {
+		return nil, err
+	}
+
+	err = initSchema(db)
+	if err != nil {
+		_ = db.Close()
+
+		return nil, fmt.Errorf("database %s: %w", path, err)
+	}
+
+	return db, nil
+}
+
+// openReportDatabase opens an existing database for the report and
+// trees subcommands. A missing database file is an error directing the
+// user to run scan first; the schema version must match exactly.
+func openReportDatabase(path string) (*sql.DB, error) {
+	_, err := os.Stat(path)
+	if errors.Is(err, fs.ErrNotExist) {
+		return nil, fmt.Errorf("%s: %w", path, errNoDatabase)
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("database: %w", err)
+	}
+
+	db, err := openDB(path)
+	if err != nil {
+		return nil, err
+	}
+
+	v, err := userVersion(db)
+	if err != nil {
+		_ = db.Close()
+
+		return nil, fmt.Errorf("database %s: %w", path, err)
+	}
+
+	if v != schemaVersion {
+		_ = db.Close()
+
+		return nil, fmt.Errorf("database %s: version %d, want %d: %w",
+			path, v, schemaVersion, errSchemaVersion)
+	}
+
+	return db, nil
+}
+
+// initSchema creates the schema on a fresh database and verifies the
+// schema version on an existing one.
+func initSchema(db *sql.DB) error {
+	v, err := userVersion(db)
+	if err != nil {
+		return err
+	}
+
+	switch v {
+	case 0:
+		return createSchema(db)
+	case schemaVersion:
+		return nil
+	default:
+		return fmt.Errorf("version %d, want %d: %w",
+			v, schemaVersion, errSchemaVersion)
+	}
+}
+
+// createSchema applies the schema to a fresh database and stamps the
+// schema version.
+func createSchema(db *sql.DB) error {
+	ctx := context.Background()
+
+	_, err := db.ExecContext(ctx, createTableSQL)
+	if err != nil {
+		return fmt.Errorf("create schema: %w", err)
+	}
+
+	_, err = db.ExecContext(ctx,
+		"PRAGMA user_version = "+strconv.Itoa(schemaVersion))
+	if err != nil {
+		return fmt.Errorf("set schema version: %w", err)
+	}
+
+	return nil
+}
+
+// userVersion reads the database's PRAGMA user_version.
+func userVersion(db *sql.DB) (int, error) {
+	var v int
+
+	err := db.QueryRowContext(context.Background(),
+		"PRAGMA user_version").Scan(&v)
+	if err != nil {
+		return 0, fmt.Errorf("read schema version: %w", err)
+	}
+
+	return v, nil
+}
+
+// loadFileRows reads every record from the files table.
+func loadFileRows(db *sql.DB) ([]scanRec, error) {
+	rows, err := db.QueryContext(context.Background(),
+		"SELECT path, size, mtime, head, tail FROM files")
+	if err != nil {
+		return nil, fmt.Errorf("read records: %w", err)
+	}
+
+	defer func() { _ = rows.Close() }()
+
+	var recs []scanRec
+
+	for rows.Next() {
+		var (
+			path []byte
+			r    scanRec
+		)
+
+		err = rows.Scan(&path, &r.size, &r.mtime, &r.head, &r.tail)
+		if err != nil {
+			return nil, fmt.Errorf("read record: %w", err)
+		}
+
+		r.path = string(path)
+		recs = append(recs, r)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return nil, fmt.Errorf("read records: %w", err)
+	}
+
+	return recs, nil
+}
+
+// applyChanges writes one scan's database changes — upserts for new and
+// changed files, deletes for vanished ones — in a single transaction,
+// so a concurrent report never observes a half-finished scan. Progress
+// is rendered on prog (one increment per change).
+func applyChanges(db *sql.DB, upserts []scanRec, deletes []string,
+	prog *progress,
+) error {
+	ctx := context.Background()
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin transaction: %w", err)
+	}
+
+	defer func() { _ = tx.Rollback() }()
+
+	err = execUpserts(ctx, tx, upserts, prog)
+	if err != nil {
+		return err
+	}
+
+	err = execDeletes(ctx, tx, deletes, prog)
+	if err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("commit: %w", err)
+	}
+
+	return nil
+}
+
+// execUpserts inserts or updates one record per new or changed file.
+func execUpserts(ctx context.Context, tx *sql.Tx, upserts []scanRec,
+	prog *progress,
+) error {
+	st, err := tx.PrepareContext(ctx, upsertSQL)
+	if err != nil {
+		return fmt.Errorf("prepare upsert: %w", err)
+	}
+
+	defer func() { _ = st.Close() }()
+
+	for _, r := range upserts {
+		_, err = st.ExecContext(ctx,
+			[]byte(r.path), r.size, r.mtime, r.head, r.tail)
+		if err != nil {
+			return fmt.Errorf("upsert %s: %w", r.path, err)
+		}
+
+		prog.increment()
+	}
+
+	return nil
+}
+
+// execDeletes removes the records for paths no longer present.
+func execDeletes(ctx context.Context, tx *sql.Tx, deletes []string,
+	prog *progress,
+) error {
+	st, err := tx.PrepareContext(ctx, "DELETE FROM files WHERE path = ?")
+	if err != nil {
+		return fmt.Errorf("prepare delete: %w", err)
+	}
+
+	defer func() { _ = st.Close() }()
+
+	for _, p := range deletes {
+		_, err = st.ExecContext(ctx, []byte(p))
+		if err != nil {
+			return fmt.Errorf("delete %s: %w", p, err)
+		}
+
+		prog.increment()
+	}
+
+	return nil
+}