From 72d052eac34e7c8c8b8db88b8731959c4027b835 Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 7 Aug 2026 16:59:29 +0000 Subject: [PATCH] Update golangci-lint to v2.12.2 with canonical config Add the canonical .golangci.yml (v2 layout, default: all with the standard disable list) and pin golangci-lint v2.12.2 in the Makefile deps target using the v2 module path, replacing the @latest install on the old v1 path. The Dockerfile lint stage already pinned the v2.12.2 alpine image by digest and is unchanged. Fix all issues surfaced by the strict config: - err113: introduce wrapped static sentinel errors - gocognit/funlen/nestif: split ExtractDay, Run, DumpAndCompress, VerifyOutput, and the CLI flag parsing into smaller helpers - noctx: use ExecContext/QueryContext/BeginTx and CommandContext - noinlineerr: replace inline if-err assignments with plain ones - wsl_v5/nlreturn: add required blank lines - lll: wrap long lines and SQL strings - gosec: filepath.Clean on file open/create; bounded uint64 conversion in CheckFreeSpace; named output dir permission constant - mnd: use unix.FADV_* constants and named size constants - revive: add package and exported symbol comments - testpackage/paralleltest: move smoke test to bsdaily_test with t.Parallel() - nonamedreturns: drop non-error named returns The remaining nolint directives (gosec subprocess launches with internal paths, unqueryvet full-row SELECT * copies whose schema is defined by the source database) are each justified inline. --- .golangci.yml | 34 +++ Makefile | 2 +- TODO.md | 4 + cmd/bsdaily/main.go | 138 +++++++---- internal/bsdaily/bsdaily_test.go | 23 +- internal/bsdaily/config.go | 7 +- internal/bsdaily/copy.go | 39 +++- internal/bsdaily/copy_linux.go | 11 +- internal/bsdaily/disk.go | 24 +- internal/bsdaily/dump.go | 115 +++++++--- internal/bsdaily/extract.go | 382 +++++++++++++++++++++---------- internal/bsdaily/run.go | 264 ++++++++++++++------- internal/bsdaily/snapshot.go | 33 ++- internal/bsdaily/verify.go | 100 ++++++-- 14 files changed, 821 insertions(+), 355 deletions(-) create mode 100644 .golangci.yml diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000..26b1610 --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,34 @@ +version: "2" + +# Config schema uses the golangci-lint v2 layout (settings live under +# linters.settings, not top-level linters-settings) so that the +# thresholds below are actually applied by golangci-lint >= v2. + +run: + timeout: 5m + modules-download-mode: readonly + +linters: + default: all + disable: + # Genuinely incompatible with project patterns + - exhaustruct # Requires all struct fields + - depguard # Dependency allow/block lists + - godot # Requires comments to end with periods + - wsl # Deprecated, replaced by wsl_v5 + - wrapcheck # Too verbose for internal packages + - varnamelen # Short names like db, id are idiomatic Go + settings: + lll: + line-length: 88 + funlen: + lines: 80 + statements: 50 + cyclop: + max-complexity: 15 + dupl: + threshold: 100 + +issues: + max-issues-per-linter: 0 + max-same-issues: 0 diff --git a/Makefile b/Makefile index 04ff721..9025f4f 100644 --- a/Makefile +++ b/Makefile @@ -46,7 +46,7 @@ clean: # Install dependencies. deps: go mod download - go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest + go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.12.2 # Run tests with coverage. test-coverage: diff --git a/TODO.md b/TODO.md index 611a0b9..e48f403 100644 --- a/TODO.md +++ b/TODO.md @@ -20,6 +20,10 @@ green with the new lint config. # Completed Steps +- 2026-08-07: Added canonical `.golangci.yml`; pinned golangci-lint + v2.12.2 in the `Makefile` `deps` target (v2 module path, replacing + `@latest` on the old v1 path); fixed all lint issues surfaced by the + strict config across `cmd/bsdaily` and `internal/bsdaily`. - 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, Makefile shims, README Entrypoints section - 2026-06-28: Fixed errcheck lint failures; added compilation smoke test; diff --git a/cmd/bsdaily/main.go b/cmd/bsdaily/main.go index 7d5409e..aba1888 100644 --- a/cmd/bsdaily/main.go +++ b/cmd/bsdaily/main.go @@ -1,6 +1,9 @@ +// Package main implements bsdaily, a tool that extracts a single day's +// data from the latest daily snapshot. package main import ( + "errors" "fmt" "log/slog" "os" @@ -10,75 +13,110 @@ import ( "github.com/spf13/cobra" ) +var ( + errDateExclusive = errors.New("--date and --from/--to are mutually exclusive") + errFromRequiresTo = errors.New("--from requires --to") + errToRequiresFrom = errors.New("--to requires --from") + errFromAfterTo = errors.New("--from is after --to") +) + +func parseTargetDates(dateFlag, fromFlag, toFlag string) ([]time.Time, error) { + hasDate := dateFlag != "" + hasFrom := fromFlag != "" + hasTo := toFlag != "" + + // Validate mutual exclusivity + if hasDate && (hasFrom || hasTo) { + return nil, errDateExclusive + } + + if hasFrom != hasTo { + if hasFrom { + return nil, errFromRequiresTo + } + + return nil, errToRequiresFrom + } + + if hasDate { + t, err := time.Parse("2006-01-02", dateFlag) + if err != nil { + return nil, fmt.Errorf( + "invalid --date %q (expected YYYY-MM-DD): %w", dateFlag, err) + } + + return []time.Time{t}, nil + } + + if !hasFrom { + // nil → Run() defaults to snapshot date minus one + return nil, nil + } + + from, err := time.Parse("2006-01-02", fromFlag) + if err != nil { + return nil, fmt.Errorf( + "invalid --from %q (expected YYYY-MM-DD): %w", fromFlag, err) + } + + to, err := time.Parse("2006-01-02", toFlag) + if err != nil { + return nil, fmt.Errorf( + "invalid --to %q (expected YYYY-MM-DD): %w", toFlag, err) + } + + if from.After(to) { + return nil, fmt.Errorf( + "%w (--from %s, --to %s)", errFromAfterTo, fromFlag, toFlag) + } + + var targetDates []time.Time + + for d := from; !d.After(to); d = d.AddDate(0, 0, 1) { + targetDates = append(targetDates, d) + } + + return targetDates, nil +} + func main() { logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{ Level: slog.LevelInfo, })) slog.SetDefault(logger) - var dateFlag string - var fromFlag string - var toFlag string + var dateFlag, fromFlag, toFlag string rootCmd := &cobra.Command{ Use: "bsdaily", Short: "Extract a single day's data from the latest daily snapshot", SilenceUsage: true, - RunE: func(cmd *cobra.Command, args []string) error { - hasDate := dateFlag != "" - hasFrom := fromFlag != "" - hasTo := toFlag != "" - - // Validate mutual exclusivity - if hasDate && (hasFrom || hasTo) { - return fmt.Errorf("--date and --from/--to are mutually exclusive") - } - if hasFrom != hasTo { - if hasFrom { - return fmt.Errorf("--from requires --to") - } - return fmt.Errorf("--to requires --from") - } - - var targetDates []time.Time - - if hasDate { - t, err := time.Parse("2006-01-02", dateFlag) - if err != nil { - return fmt.Errorf("invalid --date %q (expected YYYY-MM-DD): %w", dateFlag, err) - } - targetDates = []time.Time{t} - } else if hasFrom { - from, err := time.Parse("2006-01-02", fromFlag) - if err != nil { - return fmt.Errorf("invalid --from %q (expected YYYY-MM-DD): %w", fromFlag, err) - } - to, err := time.Parse("2006-01-02", toFlag) - if err != nil { - return fmt.Errorf("invalid --to %q (expected YYYY-MM-DD): %w", toFlag, err) - } - if from.After(to) { - return fmt.Errorf("--from %s is after --to %s", fromFlag, toFlag) - } - for d := from; !d.After(to); d = d.AddDate(0, 0, 1) { - targetDates = append(targetDates, d) - } - } - // else: targetDates remains nil → Run() defaults to snapshot date minus one - - if err := bsdaily.Run(targetDates); err != nil { + RunE: func(_ *cobra.Command, _ []string) error { + targetDates, err := parseTargetDates(dateFlag, fromFlag, toFlag) + if err != nil { return err } + + err = bsdaily.Run(targetDates) + if err != nil { + return err + } + slog.Info("completed successfully") + return nil }, } - rootCmd.Flags().StringVarP(&dateFlag, "date", "d", "", "target date to extract (YYYY-MM-DD); defaults to snapshot date minus one day") - rootCmd.Flags().StringVar(&fromFlag, "from", "", "start of date range to extract (YYYY-MM-DD, inclusive); use with --to") - rootCmd.Flags().StringVar(&toFlag, "to", "", "end of date range to extract (YYYY-MM-DD, inclusive); use with --from") + rootCmd.Flags().StringVarP(&dateFlag, "date", "d", "", + "target date to extract (YYYY-MM-DD); defaults to snapshot date minus one") + rootCmd.Flags().StringVar(&fromFlag, "from", "", + "start of date range to extract (YYYY-MM-DD, inclusive); use with --to") + rootCmd.Flags().StringVar(&toFlag, "to", "", + "end of date range to extract (YYYY-MM-DD, inclusive); use with --from") - if err := rootCmd.Execute(); err != nil { + err := rootCmd.Execute() + if err != nil { os.Exit(1) } } diff --git a/internal/bsdaily/bsdaily_test.go b/internal/bsdaily/bsdaily_test.go index 279e81f..4deda3b 100644 --- a/internal/bsdaily/bsdaily_test.go +++ b/internal/bsdaily/bsdaily_test.go @@ -1,21 +1,32 @@ -package bsdaily +package bsdaily_test -import "testing" +import ( + "testing" + + "git.eeqj.de/sneak/bsdaily/internal/bsdaily" +) // TestCompiles is a minimal smoke test that references the package's exported // surface so that `go test` fails if the package stops compiling. It does not // touch the filesystem or any of the hard-coded production paths. func TestCompiles(t *testing.T) { - if DBFilename == "" || WALFilename == "" || SHMFilename == "" { + t.Parallel() + + if bsdaily.DBFilename == "" || bsdaily.WALFilename == "" || + bsdaily.SHMFilename == "" { t.Fatal("expected database filename constants to be set") } - if SnapshotBase == "" || TmpBase == "" || DailiesBase == "" { + + if bsdaily.SnapshotBase == "" || bsdaily.TmpBase == "" || + bsdaily.DailiesBase == "" { t.Fatal("expected base path constants to be set") } - if MinTmpFreeBytes == 0 || MinDailiesFreeBytes == 0 { + + if bsdaily.MinTmpFreeBytes == 0 || bsdaily.MinDailiesFreeBytes == 0 { t.Fatal("expected free-space thresholds to be set") } - if ErrNoPosts == nil { + + if bsdaily.ErrNoPosts == nil { t.Fatal("expected ErrNoPosts sentinel to be set") } } diff --git a/internal/bsdaily/config.go b/internal/bsdaily/config.go index b534ea8..3f355c5 100644 --- a/internal/bsdaily/config.go +++ b/internal/bsdaily/config.go @@ -1,7 +1,11 @@ +// Package bsdaily extracts single-day compressed SQL dumps from the +// latest daily ZFS snapshot of the firehose database. package bsdaily import "regexp" +// Filesystem locations and tuning constants for the extraction +// pipeline. const ( SnapshotBase = "/srv/berlin.sneak.fs.blueskyarchive/.zfs/snapshot" TmpBase = "/srv/storage/tmp" @@ -27,4 +31,5 @@ const ( verificationHeadLines = 20 ) -var snapshotPattern = regexp.MustCompile(`^zfs-auto-snap_daily-(\d{4}-\d{2}-\d{2})-\d{4}$`) +var snapshotPattern = regexp.MustCompile( + `^zfs-auto-snap_daily-(\d{4}-\d{2}-\d{2})-\d{4}$`) diff --git a/internal/bsdaily/copy.go b/internal/bsdaily/copy.go index 901ca0b..1b3eacf 100644 --- a/internal/bsdaily/copy.go +++ b/internal/bsdaily/copy.go @@ -1,28 +1,39 @@ package bsdaily import ( + "errors" "fmt" "io" "log/slog" "os" + "path/filepath" "time" ) const ( - copyBufferSize = 256 * 1024 * 1024 // 256MB buffer for large file copies from fast storage + // 256MB buffer for large file copies from fast storage + copyBufferSize = 256 * 1024 * 1024 + oneMB = 1024 * 1024 oneGB = 1024 * 1024 * 1024 ) +var errShortCopy = errors.New("short copy") + +// CopyFile copies src to dst using a large buffer, pre-allocating the +// destination and fsyncing it before returning. func CopyFile(src, dst string) (err error) { startTime := time.Now() + slog.Info("copying file", "src", src, "dst", dst) - srcFile, err := os.Open(src) + srcFile, err := os.Open(filepath.Clean(src)) if err != nil { return fmt.Errorf("opening source %s: %w", src, err) } + defer func() { - if cerr := srcFile.Close(); cerr != nil { + cerr := srcFile.Close() + if cerr != nil { slog.Warn("failed to close source file", "src", src, "error", cerr) } }() @@ -37,40 +48,48 @@ func CopyFile(src, dst string) (err error) { applyFileAdvice(srcFile, srcInfo.Size()) } - dstFile, err := os.Create(dst) + dstFile, err := os.Create(filepath.Clean(dst)) if err != nil { return fmt.Errorf("creating destination %s: %w", dst, err) } + defer func() { - if cerr := dstFile.Close(); cerr != nil && err == nil { + cerr := dstFile.Close() + if cerr != nil && err == nil { err = fmt.Errorf("closing destination %s: %w", dst, cerr) } }() // Pre-allocate space for the destination file to avoid fragmentation - if err := dstFile.Truncate(srcInfo.Size()); err != nil { - slog.Warn("failed to pre-allocate destination file", "error", err) + terr := dstFile.Truncate(srcInfo.Size()) + if terr != nil { + slog.Warn("failed to pre-allocate destination file", "error", terr) } // Use a much larger buffer for NVMe-speed copies buf := make([]byte, copyBufferSize) + written, err := io.CopyBuffer(dstFile, srcFile, buf) if err != nil { return fmt.Errorf("copying data: %w", err) } if written != srcInfo.Size() { - return fmt.Errorf("short copy: wrote %d bytes, expected %d", written, srcInfo.Size()) + return fmt.Errorf("%w: wrote %d bytes, expected %d", + errShortCopy, written, srcInfo.Size()) } - if err := dstFile.Sync(); err != nil { + err = dstFile.Sync() + if err != nil { return fmt.Errorf("syncing destination %s: %w", dst, err) } elapsed := time.Since(startTime) - throughputMBps := float64(written) / elapsed.Seconds() / (1024 * 1024) + throughputMBps := float64(written) / elapsed.Seconds() / oneMB + slog.Info("file copied", "dst", dst, "bytes", written, "elapsed", elapsed.Round(time.Millisecond), "throughput_mbps", fmt.Sprintf("%.1f", throughputMBps)) + return nil } diff --git a/internal/bsdaily/copy_linux.go b/internal/bsdaily/copy_linux.go index d513062..eec53ba 100644 --- a/internal/bsdaily/copy_linux.go +++ b/internal/bsdaily/copy_linux.go @@ -3,14 +3,15 @@ package bsdaily import ( - "golang.org/x/sys/unix" "os" + + "golang.org/x/sys/unix" ) +// applyFileAdvice hints to the kernel that file will be read +// sequentially and should be prefetched into the page cache. func applyFileAdvice(file *os.File, size int64) { fd := int(file.Fd()) - // POSIX_FADV_SEQUENTIAL = 2 - _ = unix.Fadvise(fd, 0, size, 2) - // POSIX_FADV_WILLNEED = 3 - prefetch file into cache - _ = unix.Fadvise(fd, 0, size, 3) + _ = unix.Fadvise(fd, 0, size, unix.FADV_SEQUENTIAL) + _ = unix.Fadvise(fd, 0, size, unix.FADV_WILLNEED) } diff --git a/internal/bsdaily/disk.go b/internal/bsdaily/disk.go index 60cd32e..8786c97 100644 --- a/internal/bsdaily/disk.go +++ b/internal/bsdaily/disk.go @@ -1,26 +1,42 @@ package bsdaily import ( + "errors" "fmt" "log/slog" "golang.org/x/sys/unix" ) +var errInsufficientSpace = errors.New("insufficient disk space") + +// CheckFreeSpace returns an error when the filesystem containing path +// has fewer than minBytes bytes available. func CheckFreeSpace(path string, minBytes uint64, label string) error { var stat unix.Statfs_t - if err := unix.Statfs(path, &stat); err != nil { + + err := unix.Statfs(path, &stat) + if err != nil { return fmt.Errorf("statfs %s (%s): %w", path, label, err) } - free := uint64(stat.Bavail) * uint64(stat.Bsize) + + var blockSize uint64 + if stat.Bsize > 0 { + blockSize = uint64(stat.Bsize) + } + + free := stat.Bavail * blockSize freeGB := float64(free) / float64(bytesPerGB) minGB := float64(minBytes) / float64(bytesPerGB) + slog.Info("disk space check", "label", label, "path", path, "free_gb", fmt.Sprintf("%.1f", freeGB), "required_gb", fmt.Sprintf("%.1f", minGB)) + if free < minBytes { - return fmt.Errorf("insufficient disk space on %s (%s): %.1f GB free, need %.1f GB", - path, label, freeGB, minGB) + return fmt.Errorf("%w on %s (%s): %.1f GB free, need %.1f GB", + errInsufficientSpace, path, label, freeGB, minGB) } + return nil } diff --git a/internal/bsdaily/dump.go b/internal/bsdaily/dump.go index 8cbe19e..e6d1673 100644 --- a/internal/bsdaily/dump.go +++ b/internal/bsdaily/dump.go @@ -1,6 +1,8 @@ package bsdaily import ( + "context" + "errors" "fmt" "log/slog" "os" @@ -9,61 +11,43 @@ import ( "strings" ) +var errEmptyOutput = errors.New("compressed output is empty") + +// DumpAndCompress dumps the SQLite database at dbPath to SQL text via +// sqlite3 and compresses it with zstdmt into outputPath. func DumpAndCompress(dbPath, outputPath string) (err error) { for _, tool := range []string{"sqlite3", "zstdmt"} { - if _, err := exec.LookPath(tool); err != nil { - return fmt.Errorf("required tool %q not found in PATH: %w", tool, err) + _, lerr := exec.LookPath(tool) + if lerr != nil { + return fmt.Errorf("required tool %q not found in PATH: %w", tool, lerr) } } - if err := CheckFreeSpace(filepath.Dir(outputPath), MinDailiesFreeBytes, "dailiesBase (pre-dump)"); err != nil { + err = CheckFreeSpace( + filepath.Dir(outputPath), MinDailiesFreeBytes, "dailiesBase (pre-dump)") + if err != nil { return err } - outFile, err := os.Create(outputPath) + outFile, err := os.Create(filepath.Clean(outputPath)) if err != nil { return fmt.Errorf("creating output file: %w", err) } + defer func() { - if cerr := outFile.Close(); cerr != nil && err == nil { + cerr := outFile.Close() + if cerr != nil && err == nil { err = fmt.Errorf("closing output: %w", cerr) } }() - // Dump all tables but use INSERT OR IGNORE for mergeable imports - // This preserves all data while allowing multiple dumps to be merged - // Users should import with: zstdcat *.sql.zst | sed 's/INSERT INTO/INSERT OR IGNORE INTO/g' | sqlite3 merged.db - dumpCmd := exec.Command("sqlite3", dbPath, ".dump") - zstdCmd := exec.Command("zstdmt", fmt.Sprintf("-%d", zstdCompressionLevel)) - - pipe, err := dumpCmd.StdoutPipe() + err = runDumpPipeline(context.Background(), dbPath, outFile) if err != nil { - return fmt.Errorf("creating dump stdout pipe: %w", err) - } - zstdCmd.Stdin = pipe - zstdCmd.Stdout = outFile - - var dumpStderr, zstdStderr strings.Builder - dumpCmd.Stderr = &dumpStderr - zstdCmd.Stderr = &zstdStderr - - slog.Info("starting sqlite3 dump and zstdmt compression") - - if err := zstdCmd.Start(); err != nil { - return fmt.Errorf("starting zstdmt: %w", err) - } - if err := dumpCmd.Start(); err != nil { - return fmt.Errorf("starting sqlite3 dump: %w", err) + return err } - if err := dumpCmd.Wait(); err != nil { - return fmt.Errorf("sqlite3 dump failed: %w; stderr: %s", err, dumpStderr.String()) - } - if err := zstdCmd.Wait(); err != nil { - return fmt.Errorf("zstdmt failed: %w; stderr: %s", err, zstdStderr.String()) - } - - if err := outFile.Sync(); err != nil { + err = outFile.Sync() + if err != nil { return fmt.Errorf("syncing output: %w", err) } @@ -71,12 +55,69 @@ func DumpAndCompress(dbPath, outputPath string) (err error) { if err != nil { return fmt.Errorf("stat output: %w", err) } + const bytesPerMB = 1024 * 1024 + slog.Info("compressed output written", "path", outputPath, "size_bytes", info.Size(), "size_mb", info.Size()/bytesPerMB) if info.Size() == 0 { - return fmt.Errorf("compressed output is empty") + return errEmptyOutput + } + + return nil +} + +// runDumpPipeline streams a sqlite3 .dump of dbPath through zstdmt +// into outFile. +func runDumpPipeline(ctx context.Context, dbPath string, outFile *os.File) error { + // Dump all tables but use INSERT OR IGNORE for mergeable imports. + // This preserves all data while allowing multiple dumps to be + // merged. Users should import with: + // zstdcat *.sql.zst \ + // | sed 's/INSERT INTO/INSERT OR IGNORE INTO/g' \ + // | sqlite3 merged.db + zstdArg := fmt.Sprintf("-%d", zstdCompressionLevel) + + //nolint:gosec // sqlite3 with internally constructed path + dumpCmd := exec.CommandContext(ctx, "sqlite3", dbPath, ".dump") + //nolint:gosec // zstdmt with fixed compression argument + zstdCmd := exec.CommandContext(ctx, "zstdmt", zstdArg) + + pipe, err := dumpCmd.StdoutPipe() + if err != nil { + return fmt.Errorf("creating dump stdout pipe: %w", err) + } + + zstdCmd.Stdin = pipe + zstdCmd.Stdout = outFile + + var dumpStderr, zstdStderr strings.Builder + + dumpCmd.Stderr = &dumpStderr + zstdCmd.Stderr = &zstdStderr + + slog.Info("starting sqlite3 dump and zstdmt compression") + + err = zstdCmd.Start() + if err != nil { + return fmt.Errorf("starting zstdmt: %w", err) + } + + err = dumpCmd.Start() + if err != nil { + return fmt.Errorf("starting sqlite3 dump: %w", err) + } + + err = dumpCmd.Wait() + if err != nil { + return fmt.Errorf("sqlite3 dump failed: %w; stderr: %s", + err, dumpStderr.String()) + } + + err = zstdCmd.Wait() + if err != nil { + return fmt.Errorf("zstdmt failed: %w; stderr: %s", err, zstdStderr.String()) } return nil diff --git a/internal/bsdaily/extract.go b/internal/bsdaily/extract.go index 18dafc0..a2a25df 100644 --- a/internal/bsdaily/extract.go +++ b/internal/bsdaily/extract.go @@ -1,185 +1,315 @@ package bsdaily import ( + "context" "database/sql" "errors" "fmt" "log/slog" "time" + // Register the pure-Go sqlite driver with database/sql. _ "modernc.org/sqlite" ) +// ErrNoPosts is returned when the source database contains no posts +// for the target day. var ErrNoPosts = errors.New("no posts found for target day") +var errPostCountMismatch = errors.New("post count mismatch") + // ExtractDay opens a new empty database at dstDBPath, attaches srcDBPath, // and copies only the target day's data into it. This is much faster than // pruning a full copy because it only reads/writes the small slice of data // being kept. func ExtractDay(srcDBPath, dstDBPath string, targetDay time.Time) error { + ctx := context.Background() dayStart := targetDay.Format("2006-01-02") + "T00:00:00" dayEnd := targetDay.AddDate(0, 0, 1).Format("2006-01-02") + "T00:00:00" slog.Info("extracting day", "from", dayStart, "until", dayEnd) - // Maximum performance pragmas - we don't care about crash safety for temp files - // Use WAL mode for the source attachment to avoid locking issues - pragmas := fmt.Sprintf("?_pragma=journal_mode(WAL)&_pragma=synchronous(OFF)&_pragma=cache_size(%d)&_pragma=foreign_keys(OFF)&_pragma=temp_store(MEMORY)&_pragma=busy_timeout(5000)", sqliteCacheSizeKB) + // Maximum performance pragmas - we don't care about crash safety + // for temp files. Use WAL mode for the source attachment to avoid + // locking issues. + pragmas := fmt.Sprintf( + "?_pragma=journal_mode(WAL)&_pragma=synchronous(OFF)"+ + "&_pragma=cache_size(%d)&_pragma=foreign_keys(OFF)"+ + "&_pragma=temp_store(MEMORY)&_pragma=busy_timeout(5000)", + sqliteCacheSizeKB) + db, err := sql.Open("sqlite", dstDBPath+pragmas) if err != nil { return fmt.Errorf("opening destination database: %w", err) } + defer func() { - if cerr := db.Close(); cerr != nil { - slog.Warn("failed to close destination database", "path", dstDBPath, "error", cerr) + cerr := db.Close() + if cerr != nil { + slog.Warn("failed to close destination database", + "path", dstDBPath, "error", cerr) } }() // Attach source database - if _, err := db.Exec("ATTACH DATABASE ? AS src", srcDBPath); err != nil { + _, err = db.ExecContext(ctx, "ATTACH DATABASE ? AS src", srcDBPath) + if err != nil { return fmt.Errorf("attaching source database: %w", err) } - // Copy table DDL from source - slog.Info("copying table DDL from source") - rows, err := db.Query("SELECT sql FROM src.sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name") + err = copySchema(ctx, db) if err != nil { - return fmt.Errorf("reading source schema: %w", err) - } - defer func() { - if cerr := rows.Close(); cerr != nil { - slog.Warn("failed to close schema rows", "error", cerr) - } - }() - - var ddlStatements []string - for rows.Next() { - var ddl string - if err := rows.Scan(&ddl); err != nil { - return fmt.Errorf("scanning DDL: %w", err) - } - ddlStatements = append(ddlStatements, ddl) - } - if err := rows.Err(); err != nil { - return fmt.Errorf("iterating DDL rows: %w", err) + return err } - for _, ddl := range ddlStatements { - if _, err := db.Exec(ddl); err != nil { - return fmt.Errorf("creating table: %w\nDDL: %s", err, ddl) - } - } - - // Begin transaction for bulk inserts - tx, err := db.Begin() + postCount, err := insertDayData(ctx, db, targetDay, dayStart, dayEnd) if err != nil { - return fmt.Errorf("beginning transaction: %w", err) + return err } - defer func() { - if err != nil { - if rerr := tx.Rollback(); rerr != nil && !errors.Is(rerr, sql.ErrTxDone) { - slog.Warn("failed to roll back transaction", "error", rerr) - } - } - }() - // Insert target day's data - slog.Info("inserting posts for target day") - result, err := tx.Exec("INSERT INTO posts SELECT * FROM src.posts WHERE timestamp >= ? AND timestamp < ?", dayStart, dayEnd) + err = createIndexes(ctx, db) if err != nil { - return fmt.Errorf("inserting posts: %w", err) - } - postCount, _ := result.RowsAffected() - slog.Info("inserted posts", "count", postCount) - - if postCount == 0 { - return fmt.Errorf("%w %s - aborting to avoid producing empty output", - ErrNoPosts, targetDay.Format("2006-01-02")) - } - - slog.Info("inserting junction and lookup tables") - if _, err := tx.Exec("INSERT INTO posts_hashtags SELECT * FROM src.posts_hashtags WHERE post_id IN (SELECT id FROM posts)"); err != nil { - return fmt.Errorf("inserting posts_hashtags: %w", err) - } - - if _, err := tx.Exec("INSERT INTO posts_urls SELECT * FROM src.posts_urls WHERE post_id IN (SELECT id FROM posts)"); err != nil { - return fmt.Errorf("inserting posts_urls: %w", err) - } - - if _, err := tx.Exec("INSERT INTO hashtags SELECT * FROM src.hashtags WHERE id IN (SELECT hashtag_id FROM posts_hashtags)"); err != nil { - return fmt.Errorf("inserting hashtags: %w", err) - } - - if _, err := tx.Exec("INSERT INTO urls SELECT * FROM src.urls WHERE id IN (SELECT url_id FROM posts_urls)"); err != nil { - return fmt.Errorf("inserting urls: %w", err) - } - - if _, err := tx.Exec("INSERT INTO users SELECT * FROM src.users WHERE did IN (SELECT user_did FROM posts)"); err != nil { - return fmt.Errorf("inserting users: %w", err) - } - - // Check if media table exists in source and copy if present - var mediaTableExists int - if err := tx.QueryRow("SELECT COUNT(*) FROM src.sqlite_master WHERE type='table' AND name='media'").Scan(&mediaTableExists); err != nil { - slog.Warn("checking for media table", "error", err) - } else if mediaTableExists > 0 { - slog.Info("inserting media entries") - // Get post blob_cids for this day's posts - if _, err := tx.Exec("INSERT INTO media SELECT * FROM src.media WHERE content_hash IN (SELECT blob_cids FROM posts WHERE blob_cids IS NOT NULL)"); err != nil { - slog.Warn("inserting media (may not have matching entries)", "error", err) - } - } - - // Commit the transaction before any further database operations - if err := tx.Commit(); err != nil { - return fmt.Errorf("committing transaction: %w", err) - } - tx = nil // Clear tx to ensure defer doesn't try to rollback - - // Create indexes after bulk insert for speed - slog.Info("creating indexes") - idxRows, err := db.Query("SELECT sql FROM src.sqlite_master WHERE type='index' AND name NOT LIKE 'sqlite_%' AND sql IS NOT NULL ORDER BY name") - if err != nil { - return fmt.Errorf("reading source indexes: %w", err) - } - defer func() { - if cerr := idxRows.Close(); cerr != nil { - slog.Warn("failed to close index rows", "error", cerr) - } - }() - - var idxStatements []string - for idxRows.Next() { - var idxSQL string - if err := idxRows.Scan(&idxSQL); err != nil { - return fmt.Errorf("scanning index DDL: %w", err) - } - idxStatements = append(idxStatements, idxSQL) - } - if err := idxRows.Err(); err != nil { - return fmt.Errorf("iterating index rows: %w", err) - } - - for _, idxSQL := range idxStatements { - if _, err := db.Exec(idxSQL); err != nil { - return fmt.Errorf("creating index: %w\nDDL: %s", err, idxSQL) - } + return err } // Detach source - if _, err := db.Exec("DETACH DATABASE src"); err != nil { + _, err = db.ExecContext(ctx, "DETACH DATABASE src") + if err != nil { return fmt.Errorf("detaching source database: %w", err) } // Verify post count var verifyCount int64 - if err := db.QueryRow("SELECT COUNT(*) FROM posts").Scan(&verifyCount); err != nil { + + err = db.QueryRowContext(ctx, "SELECT COUNT(*) FROM posts").Scan(&verifyCount) + if err != nil { return fmt.Errorf("verifying post count: %w", err) } + if verifyCount != postCount { - return fmt.Errorf("post count mismatch: inserted %d but found %d", postCount, verifyCount) + return fmt.Errorf("%w: inserted %d but found %d", + errPostCountMismatch, postCount, verifyCount) } + slog.Info("extraction complete", "posts", verifyCount) return nil } + +// collectSQL runs a query returning a single text column and collects +// the non-NULL results in order. +func collectSQL(ctx context.Context, db *sql.DB, query string) ([]string, error) { + rows, err := db.QueryContext(ctx, query) + if err != nil { + return nil, err + } + + defer func() { + cerr := rows.Close() + if cerr != nil { + slog.Warn("failed to close rows", "error", cerr) + } + }() + + var statements []string + + for rows.Next() { + var stmt string + + err = rows.Scan(&stmt) + if err != nil { + return nil, fmt.Errorf("scanning row: %w", err) + } + + statements = append(statements, stmt) + } + + err = rows.Err() + if err != nil { + return nil, fmt.Errorf("iterating rows: %w", err) + } + + return statements, nil +} + +// copySchema copies the table DDL from the attached src database into +// the destination database. +func copySchema(ctx context.Context, db *sql.DB) error { + slog.Info("copying table DDL from source") + + ddlStatements, err := collectSQL(ctx, db, + "SELECT sql FROM src.sqlite_master WHERE type='table' "+ + "AND name NOT LIKE 'sqlite_%' ORDER BY name") + if err != nil { + return fmt.Errorf("reading source schema: %w", err) + } + + for _, ddl := range ddlStatements { + _, err = db.ExecContext(ctx, ddl) + if err != nil { + return fmt.Errorf("creating table: %w\nDDL: %s", err, ddl) + } + } + + return nil +} + +// insertLookupTables copies the junction and lookup rows related to +// the already-inserted posts from the attached src database. +// +//nolint:unqueryvet // full-row copies; schema is defined by source DB +func insertLookupTables(ctx context.Context, tx *sql.Tx) error { + inserts := []struct { + label string + query string + }{ + {"posts_hashtags", "INSERT INTO posts_hashtags " + + "SELECT * FROM src.posts_hashtags " + + "WHERE post_id IN (SELECT id FROM posts)"}, + {"posts_urls", "INSERT INTO posts_urls " + + "SELECT * FROM src.posts_urls " + + "WHERE post_id IN (SELECT id FROM posts)"}, + {"hashtags", "INSERT INTO hashtags " + + "SELECT * FROM src.hashtags " + + "WHERE id IN (SELECT hashtag_id FROM posts_hashtags)"}, + {"urls", "INSERT INTO urls " + + "SELECT * FROM src.urls " + + "WHERE id IN (SELECT url_id FROM posts_urls)"}, + {"users", "INSERT INTO users " + + "SELECT * FROM src.users " + + "WHERE did IN (SELECT user_did FROM posts)"}, + } + + for _, ins := range inserts { + _, err := tx.ExecContext(ctx, ins.query) + if err != nil { + return fmt.Errorf("inserting %s: %w", ins.label, err) + } + } + + return nil +} + +// insertDayData copies the target day's posts and their related +// junction and lookup rows from the attached src database inside a +// single transaction. It returns the number of posts inserted. +func insertDayData( + ctx context.Context, + db *sql.DB, + targetDay time.Time, + dayStart, dayEnd string, +) (int64, error) { + // Begin transaction for bulk inserts + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("beginning transaction: %w", err) + } + + committed := false + + defer func() { + if committed { + return + } + + rerr := tx.Rollback() + if rerr != nil && !errors.Is(rerr, sql.ErrTxDone) { + slog.Warn("failed to roll back transaction", "error", rerr) + } + }() + + // Insert target day's data + slog.Info("inserting posts for target day") + + //nolint:unqueryvet // full-row copy; schema is defined by source DB + result, err := tx.ExecContext(ctx, + "INSERT INTO posts SELECT * FROM src.posts "+ + "WHERE timestamp >= ? AND timestamp < ?", dayStart, dayEnd) + if err != nil { + return 0, fmt.Errorf("inserting posts: %w", err) + } + + postCount, _ := result.RowsAffected() + + slog.Info("inserted posts", "count", postCount) + + if postCount == 0 { + return 0, fmt.Errorf("%w %s - aborting to avoid producing empty output", + ErrNoPosts, targetDay.Format("2006-01-02")) + } + + slog.Info("inserting junction and lookup tables") + + err = insertLookupTables(ctx, tx) + if err != nil { + return 0, err + } + + copyMediaTable(ctx, tx) + + // Commit the transaction before any further database operations + err = tx.Commit() + if err != nil { + return 0, fmt.Errorf("committing transaction: %w", err) + } + + committed = true + + return postCount, nil +} + +// copyMediaTable copies the media table when it exists in the source. +// Failures are logged rather than fatal because the source may not +// have a media table or matching entries. +func copyMediaTable(ctx context.Context, tx *sql.Tx) { + // Check if media table exists in source and copy if present + var mediaTableExists int + + err := tx.QueryRowContext(ctx, + "SELECT COUNT(*) FROM src.sqlite_master "+ + "WHERE type='table' AND name='media'").Scan(&mediaTableExists) + if err != nil { + slog.Warn("checking for media table", "error", err) + + return + } + + if mediaTableExists == 0 { + return + } + + slog.Info("inserting media entries") + + // Get post blob_cids for this day's posts + //nolint:unqueryvet // full-row copy; schema is defined by source DB + _, err = tx.ExecContext(ctx, + "INSERT INTO media SELECT * FROM src.media "+ + "WHERE content_hash IN "+ + "(SELECT blob_cids FROM posts WHERE blob_cids IS NOT NULL)") + if err != nil { + slog.Warn("inserting media (may not have matching entries)", "error", err) + } +} + +// createIndexes recreates the source database's indexes after the bulk +// insert, which is faster than inserting into indexed tables. +func createIndexes(ctx context.Context, db *sql.DB) error { + // Create indexes after bulk insert for speed + slog.Info("creating indexes") + + idxStatements, err := collectSQL(ctx, db, + "SELECT sql FROM src.sqlite_master WHERE type='index' "+ + "AND name NOT LIKE 'sqlite_%' AND sql IS NOT NULL ORDER BY name") + if err != nil { + return fmt.Errorf("reading source indexes: %w", err) + } + + for _, idxSQL := range idxStatements { + _, err = db.ExecContext(ctx, idxSQL) + if err != nil { + return fmt.Errorf("creating index: %w\nDDL: %s", err, idxSQL) + } + } + + return nil +} diff --git a/internal/bsdaily/run.go b/internal/bsdaily/run.go index 2093d6f..10aba9f 100644 --- a/internal/bsdaily/run.go +++ b/internal/bsdaily/run.go @@ -9,21 +9,33 @@ import ( "time" ) +var errEmptySource = errors.New("source file is empty") + +// outputDirPerm is world-readable because the dailies tree is +// published for download. +const outputDirPerm = 0o755 + // cleanup removes a temporary file, logging a warning if removal fails so // that leaked scratch files are surfaced rather than silently ignored. A // missing file is not an error. func cleanup(path string) { - if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + err := os.Remove(path) + if err != nil && !os.IsNotExist(err) { slog.Warn("failed to remove temporary file", "path", path, "error", err) } } +// Run extracts each requested day from the latest daily snapshot into +// a compressed SQL dump. When targetDates is empty it defaults to the +// snapshot date minus one day. func Run(targetDates []time.Time) error { snapshotDir, snapshotDate, err := FindLatestDailySnapshot() if err != nil { return fmt.Errorf("finding latest snapshot: %w", err) } - slog.Info("found latest daily snapshot", "dir", snapshotDir, "snapshot_date", snapshotDate.Format("2006-01-02")) + + slog.Info("found latest daily snapshot", "dir", snapshotDir, + "snapshot_date", snapshotDate.Format("2006-01-02")) if len(targetDates) == 0 { targetDates = []time.Time{snapshotDate.AddDate(0, 0, -1)} @@ -34,10 +46,13 @@ func Run(targetDates []time.Time) error { "last", targetDates[len(targetDates)-1].Format("2006-01-02")) // Check disk space - if err := CheckFreeSpace(TmpBase, MinTmpFreeBytes, "tmpBase"); err != nil { + err = CheckFreeSpace(TmpBase, MinTmpFreeBytes, "tmpBase") + if err != nil { return err } - if err := CheckFreeSpace(DailiesBase, MinDailiesFreeBytes, "dailiesBase"); err != nil { + + err = CheckFreeSpace(DailiesBase, MinDailiesFreeBytes, "dailiesBase") + if err != nil { return err } @@ -46,15 +61,54 @@ func Run(targetDates []time.Time) error { if err != nil { return fmt.Errorf("creating temp directory in %s: %w", TmpBase, err) } + slog.Info("created temp directory", "path", tmpDir) + defer func() { slog.Info("cleaning up temp directory", "path", tmpDir) - if err := os.RemoveAll(tmpDir); err != nil { - slog.Error("failed to remove temp directory", "path", tmpDir, "error", err) + + rerr := os.RemoveAll(tmpDir) + if rerr != nil { + slog.Error("failed to remove temp directory", + "path", tmpDir, "error", rerr) } }() // Copy database files from snapshot to temp + dstDB, err := copySnapshotFiles(snapshotDir, tmpDir) + if err != nil { + return err + } + + // Process each day completely before moving to the next. This + // ensures we don't have multiple SQLite operations competing for + // the same source database. + processed := 0 + skipped := 0 + + for _, targetDay := range targetDates { + didProcess, perr := processDay(tmpDir, dstDB, targetDay) + if perr != nil { + return perr + } + + if didProcess { + processed++ + } else { + skipped++ + } + } + + slog.Info("run summary", "processed", processed, + "skipped", skipped, "total", len(targetDates)) + + return nil +} + +// copySnapshotFiles copies the database, WAL, and (if present) SHM +// files from the snapshot directory into tmpDir and returns the path +// of the copied database. +func copySnapshotFiles(snapshotDir, tmpDir string) (string, error) { srcDB := filepath.Join(snapshotDir, DBFilename) srcWAL := filepath.Join(snapshotDir, WALFilename) srcSHM := filepath.Join(snapshotDir, SHMFilename) @@ -65,100 +119,136 @@ func Run(targetDates []time.Time) error { for _, f := range []string{srcDB, srcWAL} { info, err := os.Stat(f) if err != nil { - return fmt.Errorf("source file missing: %s: %w", f, err) + return "", fmt.Errorf("source file missing: %s: %w", f, err) } + if info.Size() == 0 { - return fmt.Errorf("source file is empty: %s", f) + return "", fmt.Errorf("%w: %s", errEmptySource, f) } + slog.Info("source file", "path", f, "size_bytes", info.Size()) } - if err := CopyFile(srcDB, dstDB); err != nil { - return fmt.Errorf("copying database: %w", err) - } - if err := CopyFile(srcWAL, dstWAL); err != nil { - return fmt.Errorf("copying WAL: %w", err) - } - if _, err := os.Stat(srcSHM); err == nil { - if err := CopyFile(srcSHM, dstSHM); err != nil { - return fmt.Errorf("copying SHM: %w", err) - } + err := CopyFile(srcDB, dstDB) + if err != nil { + return "", fmt.Errorf("copying database: %w", err) } - // Process each day completely before moving to the next - // This ensures we don't have multiple SQLite operations competing for the same source database - processed := 0 - skipped := 0 + err = CopyFile(srcWAL, dstWAL) + if err != nil { + return "", fmt.Errorf("copying WAL: %w", err) + } - for _, targetDay := range targetDates { - dayStr := targetDay.Format("2006-01-02") - slog.Info("processing day", "date", dayStr) - - // Check if output already exists - outputDir := filepath.Join(DailiesBase, targetDay.Format("2006-01")) - outputFinal := filepath.Join(outputDir, dayStr+".sql.zst") - if _, err := os.Stat(outputFinal); err == nil { - slog.Info("output already exists, skipping", "path", outputFinal) - skipped++ - continue - } - - // Extract target day into a per-day database - extractedDB := filepath.Join(tmpDir, "extracted-"+dayStr+".db") - slog.Info("extracting target day", "src", dstDB, "dst", extractedDB) - if err := ExtractDay(dstDB, extractedDB, targetDay); err != nil { - if errors.Is(err, ErrNoPosts) { - slog.Warn("no posts found, skipping day", "date", dayStr) - cleanup(extractedDB) - skipped++ - continue - } - return fmt.Errorf("extracting day %s: %w", dayStr, err) - } - - // Dump to SQL and compress - if err := os.MkdirAll(outputDir, 0755); err != nil { - cleanup(extractedDB) - return fmt.Errorf("creating output directory %s: %w", outputDir, err) - } - - outputTmp := filepath.Join(outputDir, "."+dayStr+".sql.zst.tmp") - - slog.Info("dumping and compressing", "tmp_output", outputTmp) - if err := DumpAndCompress(extractedDB, outputTmp); err != nil { - cleanup(outputTmp) - cleanup(extractedDB) - return fmt.Errorf("dump and compress for %s: %w", dayStr, err) - } - - slog.Info("verifying compressed output") - if err := VerifyOutput(outputTmp); err != nil { - cleanup(outputTmp) - cleanup(extractedDB) - return fmt.Errorf("verification failed for %s: %w", dayStr, err) - } - - // Atomic rename to final path - slog.Info("renaming to final output", "from", outputTmp, "to", outputFinal) - if err := os.Rename(outputTmp, outputFinal); err != nil { - cleanup(outputTmp) - cleanup(extractedDB) - return fmt.Errorf("atomic rename for %s: %w", dayStr, err) - } - - info, err := os.Stat(outputFinal) + _, err = os.Stat(srcSHM) + if err == nil { + err = CopyFile(srcSHM, dstSHM) if err != nil { - cleanup(extractedDB) - return fmt.Errorf("stat final output: %w", err) + return "", fmt.Errorf("copying SHM: %w", err) } - slog.Info("day completed", "date", dayStr, "path", outputFinal, "size_bytes", info.Size()) - - // Remove extracted DB to reclaim space immediately - cleanup(extractedDB) - processed++ } - slog.Info("run summary", "processed", processed, "skipped", skipped, "total", len(targetDates)) + return dstDB, nil +} + +// processDay extracts, dumps, compresses, verifies, and publishes a +// single day. It reports whether the day was processed; false means it +// was skipped (already present or no posts). +func processDay(tmpDir, dstDB string, targetDay time.Time) (bool, error) { + dayStr := targetDay.Format("2006-01-02") + + slog.Info("processing day", "date", dayStr) + + // Check if output already exists + outputDir := filepath.Join(DailiesBase, targetDay.Format("2006-01")) + outputFinal := filepath.Join(outputDir, dayStr+".sql.zst") + + _, err := os.Stat(outputFinal) + if err == nil { + slog.Info("output already exists, skipping", "path", outputFinal) + + return false, nil + } + + // Extract target day into a per-day database + extractedDB := filepath.Join(tmpDir, "extracted-"+dayStr+".db") + + slog.Info("extracting target day", "src", dstDB, "dst", extractedDB) + + err = ExtractDay(dstDB, extractedDB, targetDay) + if err != nil { + if errors.Is(err, ErrNoPosts) { + slog.Warn("no posts found, skipping day", "date", dayStr) + cleanup(extractedDB) + + return false, nil + } + + return false, fmt.Errorf("extracting day %s: %w", dayStr, err) + } + + // Dump to SQL and compress + err = os.MkdirAll(outputDir, outputDirPerm) + if err != nil { + cleanup(extractedDB) + + return false, fmt.Errorf("creating output directory %s: %w", outputDir, err) + } + + outputTmp := filepath.Join(outputDir, "."+dayStr+".sql.zst.tmp") + + slog.Info("dumping and compressing", "tmp_output", outputTmp) + + err = DumpAndCompress(extractedDB, outputTmp) + if err != nil { + cleanup(outputTmp) + cleanup(extractedDB) + + return false, fmt.Errorf("dump and compress for %s: %w", dayStr, err) + } + + slog.Info("verifying compressed output") + + err = VerifyOutput(outputTmp) + if err != nil { + cleanup(outputTmp) + cleanup(extractedDB) + + return false, fmt.Errorf("verification failed for %s: %w", dayStr, err) + } + + err = publishOutput(outputTmp, outputFinal, dayStr) + if err != nil { + cleanup(extractedDB) + + return false, err + } + + // Remove extracted DB to reclaim space immediately + cleanup(extractedDB) + + return true, nil +} + +// publishOutput atomically renames the temporary output file to its +// final path and logs the completed day. +func publishOutput(outputTmp, outputFinal, dayStr string) error { + // Atomic rename to final path + slog.Info("renaming to final output", "from", outputTmp, "to", outputFinal) + + err := os.Rename(outputTmp, outputFinal) + if err != nil { + cleanup(outputTmp) + + return fmt.Errorf("atomic rename for %s: %w", dayStr, err) + } + + info, err := os.Stat(outputFinal) + if err != nil { + return fmt.Errorf("stat final output: %w", err) + } + + slog.Info("day completed", "date", dayStr, + "path", outputFinal, "size_bytes", info.Size()) return nil } diff --git a/internal/bsdaily/snapshot.go b/internal/bsdaily/snapshot.go index a8a3824..0caa1b3 100644 --- a/internal/bsdaily/snapshot.go +++ b/internal/bsdaily/snapshot.go @@ -1,6 +1,7 @@ package bsdaily import ( + "errors" "fmt" "log/slog" "os" @@ -9,10 +10,15 @@ import ( "time" ) -func FindLatestDailySnapshot() (dir string, snapshotDate time.Time, err error) { +var errNoSnapshots = errors.New("no daily snapshots found") + +// FindLatestDailySnapshot locates the newest daily ZFS snapshot that +// contains the firehose database and returns its directory and date. +func FindLatestDailySnapshot() (string, time.Time, error) { entries, err := os.ReadDir(SnapshotBase) if err != nil { - return "", time.Time{}, fmt.Errorf("reading snapshot directory %s: %w", SnapshotBase, err) + return "", time.Time{}, fmt.Errorf( + "reading snapshot directory %s: %w", SnapshotBase, err) } type snapshot struct { @@ -21,24 +27,30 @@ func FindLatestDailySnapshot() (dir string, snapshotDate time.Time, err error) { } var snapshots []snapshot + for _, e := range entries { if !e.IsDir() { continue } + m := snapshotPattern.FindStringSubmatch(e.Name()) if m == nil { continue } - d, err := time.Parse("2006-01-02", m[1]) - if err != nil { - slog.Warn("skipping snapshot with unparseable date", "name", e.Name(), "error", err) + + d, perr := time.Parse("2006-01-02", m[1]) + if perr != nil { + slog.Warn("skipping snapshot with unparseable date", + "name", e.Name(), "error", perr) + continue } + snapshots = append(snapshots, snapshot{name: e.Name(), date: d}) } if len(snapshots) == 0 { - return "", time.Time{}, fmt.Errorf("no daily snapshots found in %s", SnapshotBase) + return "", time.Time{}, fmt.Errorf("%w in %s", errNoSnapshots, SnapshotBase) } sort.Slice(snapshots, func(i, j int) bool { @@ -46,11 +58,14 @@ func FindLatestDailySnapshot() (dir string, snapshotDate time.Time, err error) { }) latest := snapshots[0] - dir = filepath.Join(SnapshotBase, latest.name) + dir := filepath.Join(SnapshotBase, latest.name) dbPath := filepath.Join(dir, DBFilename) - if _, err := os.Stat(dbPath); err != nil { - return "", time.Time{}, fmt.Errorf("database not found in snapshot %s: %w", dir, err) + + _, err = os.Stat(dbPath) + if err != nil { + return "", time.Time{}, fmt.Errorf( + "database not found in snapshot %s: %w", dir, err) } return dir, latest.date, nil diff --git a/internal/bsdaily/verify.go b/internal/bsdaily/verify.go index c419c3d..bfd9793 100644 --- a/internal/bsdaily/verify.go +++ b/internal/bsdaily/verify.go @@ -1,6 +1,7 @@ package bsdaily import ( + "context" "errors" "fmt" "log/slog" @@ -9,6 +10,11 @@ import ( "strings" ) +var ( + errEmptyDecompressed = errors.New("decompressed content is empty") + errNotSQL = errors.New("decompressed content does not look like SQL") +) + // killCat terminates the zstdcat process, ignoring the benign case where it // has already exited (e.g. after receiving SIGPIPE when head closed the pipe) // and logging any other failure. @@ -16,70 +22,126 @@ func killCat(cmd *exec.Cmd) { if cmd.Process == nil { return } - if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + + err := cmd.Process.Kill() + if err != nil && !errors.Is(err, os.ErrProcessDone) { slog.Warn("failed to kill zstdcat process", "error", err) } } +// VerifyOutput checks that the compressed file at path passes a zstdmt +// integrity test and that its decompressed head looks like SQL text. func VerifyOutput(path string) error { + ctx := context.Background() + slog.Info("running zstdmt integrity check") - testCmd := exec.Command("zstdmt", "--test", path) + + //nolint:gosec // zstdmt with internally constructed path + testCmd := exec.CommandContext(ctx, "zstdmt", "--test", path) + var testStderr strings.Builder + testCmd.Stderr = &testStderr - if err := testCmd.Run(); err != nil { - return fmt.Errorf("zstdmt --test failed: %w; stderr: %s", err, testStderr.String()) + + err := testCmd.Run() + if err != nil { + return fmt.Errorf("zstdmt --test failed: %w; stderr: %s", + err, testStderr.String()) } + slog.Info("zstdmt integrity check passed") slog.Info("verifying SQL content") - catCmd := exec.Command("zstdcat", path) - headCmd := exec.Command("head", fmt.Sprintf("-%d", verificationHeadLines)) + + content, err := readDecompressedHead(ctx, path) + if err != nil { + return err + } + + err = verifySQLContent(content) + if err != nil { + return err + } + + slog.Info("SQL content verification passed") + + return nil +} + +// readDecompressedHead returns the first verificationHeadLines lines of +// the decompressed file at path via a zstdcat | head pipeline. +func readDecompressedHead(ctx context.Context, path string) (string, error) { + headArg := fmt.Sprintf("-%d", verificationHeadLines) + + //nolint:gosec // zstdcat with internally constructed path + catCmd := exec.CommandContext(ctx, "zstdcat", path) + //nolint:gosec // head with fixed numeric argument + headCmd := exec.CommandContext(ctx, "head", headArg) pipe, err := catCmd.StdoutPipe() if err != nil { - return fmt.Errorf("creating zstdcat pipe: %w", err) + return "", fmt.Errorf("creating zstdcat pipe: %w", err) } + headCmd.Stdin = pipe var headOut strings.Builder + headCmd.Stdout = &headOut - if err := catCmd.Start(); err != nil { - return fmt.Errorf("starting zstdcat: %w", err) + err = catCmd.Start() + if err != nil { + return "", fmt.Errorf("starting zstdcat: %w", err) } - if err := headCmd.Start(); err != nil { + + err = headCmd.Start() + if err != nil { killCat(catCmd) // Clean up if head fails to start - return fmt.Errorf("starting head: %w", err) + + return "", fmt.Errorf("starting head: %w", err) } // Wait for head first (it will exit when it has enough lines) - if err := headCmd.Wait(); err != nil { + err = headCmd.Wait() + if err != nil { killCat(catCmd) - return fmt.Errorf("head command failed: %w", err) + + return "", fmt.Errorf("head command failed: %w", err) } // Kill zstdcat since head closed the pipe (expected SIGPIPE) killCat(catCmd) + _ = catCmd.Wait() // Reap the process - content := headOut.String() + return headOut.String(), nil +} + +// verifySQLContent checks that content is non-empty and contains a +// recognizable SQL marker. +func verifySQLContent(content string) error { if len(content) == 0 { - return fmt.Errorf("decompressed content is empty") + return errEmptyDecompressed } hasSQLMarker := false - for _, marker := range []string{"BEGIN TRANSACTION", "CREATE TABLE", "INSERT INTO", "PRAGMA"} { + markers := []string{"BEGIN TRANSACTION", "CREATE TABLE", "INSERT INTO", "PRAGMA"} + + for _, marker := range markers { if strings.Contains(content, marker) { hasSQLMarker = true + break } } + const verificationSampleBytes = 200 + if !hasSQLMarker { - return fmt.Errorf("decompressed content does not look like SQL; first %d bytes: %s", - verificationSampleBytes, content[:min(verificationSampleBytes, len(content))]) + return fmt.Errorf("%w; first %d bytes: %s", errNotSQL, + verificationSampleBytes, + content[:min(verificationSampleBytes, len(content))]) } - slog.Info("SQL content verification passed") return nil }