All checks were successful
check / check (push) Successful in 1m2s
hashPhase returned the moment recordRun failed and left the pool running: the feeder parked forever on a full jobs channel and every worker on a full results channel. Until #4 landed the process exited before that mattered; now that runScan returns an error and unwinds, the goroutines are a real leak. The pool is now an owned hashPool. Its context is derived from the scan's, every blocking send in the feeder and the workers selects on ctx.Done(), the feeder closes jobs on every path out so the workers' range always terminates, and hashPhase defers pool.stop(), which cancels and then drains results until the last goroutine has exited. Draining is the half that matters: a worker already parked on a send cannot observe the cancellation until a receiver frees it. ctx comes from cmd.Context() and is threaded through runScan, syncScan, both worker pools and the database layer as the first parameter throughout, so graceful interrupt handling has a path to hook into rather than a pool to rewrite. The walk pool never leaked, because walkPhase always drains its events to close, but it has the same unbounded-send shape and gets the same treatment, together with a ctx.Err() guard after the walk: a cancelled walk leaves a partial size census, and the update phase would read every file it never reached as vanished and delete its record. Tests drive the scan entry point against a database whose insert trigger aborts, with a fixture large enough that the failure lands partway through the hash phase with more runs queued than either pool channel can hold, and assert that the scan fails instead of hanging and that runtime.NumGoroutine polls back to its pre-scan baseline.
171 lines
4.0 KiB
Go
171 lines
4.0 KiB
Go
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"slices"
|
|
"strings"
|
|
)
|
|
|
|
// ioBufSize is the buffer size for the buffered stdout writers.
|
|
const ioBufSize = 1 << 20
|
|
|
|
// minGroupSize is the smallest number of members that makes a
|
|
// duplicate group.
|
|
const minGroupSize = 2
|
|
|
|
// scanRec is one file record from the database. The signature (size,
|
|
// head, tail) is the duplicate key; mtime is informational only and
|
|
// used by scan for change detection.
|
|
type scanRec struct {
|
|
size int64
|
|
mtime int64
|
|
head string
|
|
tail string
|
|
path string
|
|
}
|
|
|
|
// loadRecords opens the database and reads every file record for the
|
|
// report and trees subcommands. Any database problem — including a
|
|
// missing database — is fatal. The error is returned rather than
|
|
// exiting, so that the deferred close — which checkpoints the SQLite
|
|
// WAL — always runs; the database is closed before the caller formats
|
|
// its output, so it stays closed even if that output fails.
|
|
func loadRecords(ctx context.Context) ([]scanRec, error) {
|
|
dbPath := databasePath()
|
|
|
|
db, err := openReportDatabase(ctx, dbPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
defer func() { _ = db.Close() }()
|
|
|
|
recs, err := loadFileRows(ctx, db)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("database %s: %w", dbPath, err)
|
|
}
|
|
|
|
return recs, nil
|
|
}
|
|
|
|
// dupeGroup is one set of candidate-duplicate files: identical size,
|
|
// head hash, and tail hash. paths is sorted lexicographically; the
|
|
// first entry is the group's "first", the rest are dupes.
|
|
type dupeGroup struct {
|
|
size int64
|
|
paths []string
|
|
}
|
|
|
|
// runReport implements the report subcommand: it reads every record
|
|
// from the database and prints the file-level duplicates report as TSV
|
|
// on stdout. It never touches the scanned filesystem; its only I/O is
|
|
// the database, stdout, and stderr.
|
|
func runReport(ctx context.Context) error {
|
|
recs, err := loadRecords(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
dupes := collectDupeGroups(recs)
|
|
|
|
out := bufio.NewWriterSize(os.Stdout, ioBufSize)
|
|
|
|
_, err = fmt.Fprintln(out, "first\tdupe\tsize")
|
|
if err != nil {
|
|
return fmt.Errorf("write stdout: %w", err)
|
|
}
|
|
|
|
dupeFiles := 0
|
|
|
|
var reclaimable int64
|
|
|
|
for _, g := range dupes {
|
|
for _, p := range g.paths[1:] {
|
|
_, err = fmt.Fprintf(out, "%s\t%s\t%d\n",
|
|
g.paths[0], p, g.size)
|
|
if err != nil {
|
|
return fmt.Errorf("write stdout: %w", err)
|
|
}
|
|
|
|
dupeFiles++
|
|
reclaimable += g.size
|
|
}
|
|
}
|
|
|
|
err = out.Flush()
|
|
if err != nil {
|
|
return fmt.Errorf("write stdout: %w", err)
|
|
}
|
|
|
|
fmt.Fprintf(os.Stderr,
|
|
"report: %d records read, %d duplicate groups, %d dupe files, "+
|
|
"%s reclaimable\n",
|
|
len(recs), len(dupes), dupeFiles, humanBytes(reclaimable))
|
|
|
|
return nil
|
|
}
|
|
|
|
// collectDupeGroups groups records by signature and returns every group
|
|
// with two or more paths, each group's paths sorted lexicographically,
|
|
// groups ordered by size descending then by first path ascending.
|
|
func collectDupeGroups(recs []scanRec) []dupeGroup {
|
|
groups := make(map[fileSig][]string)
|
|
|
|
for _, r := range recs {
|
|
// A record without hashes (its size was unique when last
|
|
// scanned) has unknown content and is never reported as a
|
|
// duplicate.
|
|
if r.head == "" {
|
|
continue
|
|
}
|
|
|
|
k := fileSig{size: r.size, head: r.head, tail: r.tail}
|
|
groups[k] = append(groups[k], r.path)
|
|
}
|
|
|
|
var dupes []dupeGroup
|
|
|
|
for k, paths := range groups {
|
|
if len(paths) < minGroupSize {
|
|
continue
|
|
}
|
|
|
|
slices.Sort(paths)
|
|
dupes = append(dupes, dupeGroup{size: k.size, paths: paths})
|
|
}
|
|
|
|
// Biggest reclaimable space first; ties broken by first path.
|
|
slices.SortFunc(dupes, func(a, b dupeGroup) int {
|
|
if a.size != b.size {
|
|
if a.size > b.size {
|
|
return -1
|
|
}
|
|
|
|
return 1
|
|
}
|
|
|
|
return strings.Compare(a.paths[0], b.paths[0])
|
|
})
|
|
|
|
return dupes
|
|
}
|
|
|
|
// humanBytes formats a byte count in human units (binary prefixes).
|
|
func humanBytes(n int64) string {
|
|
const unit = 1024
|
|
if n < unit {
|
|
return fmt.Sprintf("%d B", n)
|
|
}
|
|
|
|
div, exp := int64(unit), 0
|
|
for m := n / unit; m >= unit; m /= unit {
|
|
div *= unit
|
|
exp++
|
|
}
|
|
|
|
return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp])
|
|
}
|