diff --git a/cmd/tgexport/list.go b/cmd/tgexport/list.go index 7afb020..81981af 100644 --- a/cmd/tgexport/list.go +++ b/cmd/tgexport/list.go @@ -19,8 +19,11 @@ import ( // // It is the smallest thing that exercises the whole read path — resolve a chat, // walk its history, derive a name — so a naming or paging problem shows up here -// rather than halfway through an archive run. The output is tab-separated on -// purpose: filenames contain spaces, commas and quotes, but not tabs. +// rather than halfway through an archive run. The output is tab-separated, and +// the name is quoted: it comes from DocumentAttributeFilename, so whoever +// uploaded the file chose it, and a raw tab would shift the columns while a raw +// newline would split the record. Quoting also renders escape sequences inert +// rather than letting them redraw the operator's terminal. func listCmd(ctx context.Context, args []string) error { fs := flag.NewFlagSet("list", flag.ContinueOnError) var ( @@ -71,7 +74,7 @@ func listCmd(ctx context.Context, args []string) error { } count++ total += it.Size() - if _, err := fmt.Fprintf(out, "%d\t%d\t%s\n", it.MessageID, it.Size(), it.Name); err != nil { + if _, err := fmt.Fprintf(out, "%d\t%d\t%q\n", it.MessageID, it.Size(), it.Name); err != nil { return err } } diff --git a/cmd/tgexport/main.go b/cmd/tgexport/main.go index 7314d8f..582e3b6 100644 --- a/cmd/tgexport/main.go +++ b/cmd/tgexport/main.go @@ -30,6 +30,7 @@ const ( exitIncomplete = 1 exitUsage = 2 exitRemoteError = 3 + exitStalled = 4 exitSIGINT = 130 exitSIGTERM = 143 ) @@ -41,6 +42,12 @@ var errUsage = errors.New("usage") // errIncomplete marks a run that finished cleanly but left work outstanding. var errIncomplete = errors.New("incomplete") +// errStalled marks a run with work outstanding that no future run can do — every +// remaining file has a name that cannot be written. It is distinct from +// errIncomplete because a driver retrying on "incomplete" would otherwise walk +// the whole history and re-index the whole remote forever, achieving nothing. +var errStalled = errors.New("stalled") + func main() { os.Exit(run()) } @@ -102,6 +109,8 @@ func exitCode(err error, sig os.Signal) int { return exitSIGINT case errors.Is(err, errUsage): return exitUsage + case errors.Is(err, errStalled): + return exitStalled case errors.Is(err, errIncomplete): return exitIncomplete default: diff --git a/cmd/tgexport/main_test.go b/cmd/tgexport/main_test.go index dfd0de3..3aa2ad2 100644 --- a/cmd/tgexport/main_test.go +++ b/cmd/tgexport/main_test.go @@ -8,6 +8,8 @@ import ( "os" "syscall" "testing" + + "github.com/tiennm99dev/telegram-exporter/internal/pipeline" "time" ) @@ -27,6 +29,13 @@ func TestExitCode(t *testing.T) { {"wrapped help", fmt.Errorf("parse: %w", flag.ErrHelp), nil, exitOK}, {"usage mistake", fmt.Errorf("%w: bad flag", errUsage), nil, exitUsage}, {"run left work outstanding", fmt.Errorf("%w: 12 files", errIncomplete), nil, exitIncomplete}, + // Distinct from incomplete on purpose: a driver retrying on 1 would + // walk the whole chat forever for work that can never be done. + {"nothing left that can be fetched", fmt.Errorf("%w: 2 files", errStalled), nil, exitStalled}, + // And a destination that stopped accepting uploads is a failure, not a + // retry: the next pass would be refused identically. + {"destination refusing uploads", + fmt.Errorf("run: %w", pipeline.ErrDestinationFailing), nil, exitRemoteError}, {"remote failure", errors.New("pikpak unreachable"), nil, exitRemoteError}, {"cancelled without a signal", context.Canceled, nil, exitSIGINT}, {"wrapped cancellation", fmt.Errorf("download: %w", context.Canceled), nil, exitSIGINT}, diff --git a/cmd/tgexport/sync.go b/cmd/tgexport/sync.go index d28b726..b87f6c1 100644 --- a/cmd/tgexport/sync.go +++ b/cmd/tgexport/sync.go @@ -6,6 +6,7 @@ import ( "flag" "fmt" "iter" + "math" "os" "github.com/iyear/tdl/core/dcpool" @@ -108,7 +109,10 @@ func syncCmd(ctx context.Context, args []string) error { return err } - var final verify.Report + var ( + final verify.Report + runErr error + ) if err := sess.Run(ctx, func(ctx context.Context, pool dcpool.Pool) error { api := pool.Default(ctx) @@ -130,6 +134,7 @@ func syncCmd(ctx context.Context, args []string) error { if err != nil { return err } + warnCollisions(idx) before := verify.Check(items, idx) fmt.Fprintf(os.Stderr, "%d media messages, %d already archived, %d to fetch\n", before.Expected, before.Present, len(before.Todo())) @@ -151,7 +156,8 @@ func syncCmd(ctx context.Context, args []string) error { fmt.Fprintf(os.Stderr, "fetching %d file(s), %.1f GiB\n", len(todo), float64(todoBytes)/(1<<30)) rep := report.New(os.Stderr, len(todo), todoBytes) - res, runErr := pipeline.Run(ctx, sliceSeq(todo), pipeline.Options{ + var res pipeline.Result + res, runErr = pipeline.Run(ctx, sliceSeq(todo), pipeline.Options{ Pool: pool, Dst: dst, Staging: *staging, @@ -162,6 +168,12 @@ func syncCmd(ctx context.Context, args []string) error { Confirm: *confirm, Takeout: *takeout, Report: rep.Update, + // Re-checked during the run, not only before it: an archive of this + // size runs for hours, and the destination can fill in the middle. + FreeBytes: func(ctx context.Context) (int64, bool) { + return remote.FreeBytes(ctx, dst) + }, + MinFree: *minFree * (1 << 30), }) rep.Finish(res.Stats) @@ -173,6 +185,7 @@ func syncCmd(ctx context.Context, args []string) error { // and hit one transient upload error has still made progress, and // suppressing the report would leave the operator — and any driver // reading the exit code — unable to tell that from a total failure. + // It is returned after the report, below, so the exit code is right. fmt.Fprintf(os.Stderr, "run ended early: %v\n", runErr) } @@ -191,7 +204,17 @@ func syncCmd(ctx context.Context, args []string) error { fmt.Fprintln(os.Stderr) final.Write(os.Stdout) - if !final.Complete() { + switch { + case errors.Is(runErr, pipeline.ErrDestinationFailing): + // Exit 3, not 1. The destination refused upload after upload, and it + // will refuse them next pass too — a driver retrying on "incomplete" + // would walk 18k messages and re-download gigabytes into a remote that + // cannot take a byte, indefinitely. + return runErr + case final.Stalled(): + return fmt.Errorf("%w: %d file(s) remain, none of which can be fetched", + errStalled, len(final.Todo())) + case !final.Complete(): return fmt.Errorf("%w: %d file(s) still to fetch", errIncomplete, len(final.Todo())) } return nil @@ -257,6 +280,23 @@ func validateBudget(budget int64, todo []tgsource.Item) error { return nil } +// warnCollisions reports basenames the index found at more than one path. +// +// verify printed this and sync did not, which was backwards: an ambiguous +// snapshot makes the presence verdict for those names unreliable, and sync is +// the command that acts on the verdict by moving data. +func warnCollisions(idx *remote.Index) { + dup := idx.Collisions() + if len(dup) == 0 { + return + } + fmt.Fprintf(os.Stderr, "warning: %d basename(s) appear at more than one path; "+ + "verdicts for them may flip between runs:\n", len(dup)) + for _, name := range dup[:min(5, len(dup))] { + fmt.Fprintf(os.Stderr, " %q\n", name) + } +} + // checkFree refuses to start when the remote is nearly full. // // A backend that cannot report a quota is treated as unlimited rather than as a @@ -317,10 +357,19 @@ func parseSize(s string) (int64, error) { if r < '0' || r > '9' { return 0, fmt.Errorf("%q is not a size", s) } + // Checked rather than allowed to wrap. A wrapped value is negative or + // zero, and both mean "no cap" downstream — so a typo would silently + // remove the staging limit instead of being refused. + if n > (math.MaxInt64-int64(r-'0'))/10 { + return 0, fmt.Errorf("%q is too large", s) + } n = n*10 + int64(r-'0') } if n <= 0 { return 0, fmt.Errorf("must be greater than zero") } + if n > math.MaxInt64/mult { + return 0, fmt.Errorf("%q is too large", s) + } return n * mult, nil } diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 63a2487..a0d6cc7 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -36,6 +36,12 @@ type Options struct { // Zero uses the shell pipeline's default of 5. MaxFailures int + // FreeBytes and MinFree stop the run when the destination fills mid-way. + // Both must be set for the check to happen; a backend that cannot report a + // quota counts as unlimited. + FreeBytes func(context.Context) (int64, bool) + MinFree int64 + Report func(Stats) } @@ -56,6 +62,14 @@ func (r Result) Failed() []Outcome { return out } +// ErrDestinationFailing marks a run stopped because the destination refused +// upload after upload. It is distinct from an ordinary upload failure because +// the right response differs: a transient error is worth retrying, while a +// remote that is full, unreachable, or refusing credentials will refuse the next +// pass identically, and a driver that retries walks the whole chat and downloads +// gigabytes for nothing every time. +var ErrDestinationFailing = errors.New("destination stopped accepting uploads") + // maxRecordedErrors bounds what a run keeps from a failing remote. Past this, // the pattern is established and joining thousands of identical strings just // makes the final message unreadable. @@ -102,6 +116,7 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R up := &uploader{local: local, dst: o.Dst, confirm: o.Confirm} budget := newBudget(o.Budget) + guard := newSpaceGuard(o.FreeBytes, o.MinFree) uploads := make(chan tgsource.Item, o.Uploads) var ( @@ -126,7 +141,12 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R go func() { defer wg.Done() for it := range uploads { - err := up.upload(ctx, it) + // A full destination is not worth another multi-gigabyte + // attempt, so the file is dropped rather than uploaded. + err := guard.check(ctx) + if err == nil { + err = up.upload(ctx, it) + } if err != nil { // MoveFile leaves the local copy in place when it fails, so @@ -146,7 +166,9 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R } nErrs++ streak++ - if streak >= o.MaxFailures && !tripped { + // A full remote trips immediately: unlike a transient + // error, waiting for a streak just wastes the downloads. + if (streak >= o.MaxFailures || errors.Is(err, ErrDestinationFailing)) && !tripped { tripped = true stopDownloads.Store(true) } @@ -195,8 +217,8 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R var upErr error switch { case trip: - upErr = fmt.Errorf("stopped after %d consecutive upload failures (%d total): %w", - o.MaxFailures, total, joined) + upErr = fmt.Errorf("%w: stopped after %d upload failure(s): %w", + ErrDestinationFailing, total, joined) case total > 0: upErr = fmt.Errorf("%d upload(s) failed: %w", total, joined) } diff --git a/internal/pipeline/run_test.go b/internal/pipeline/run_test.go index e0841e3..44f9f7b 100644 --- a/internal/pipeline/run_test.go +++ b/internal/pipeline/run_test.go @@ -7,7 +7,6 @@ import ( "iter" "os" "path/filepath" - "strings" "sync/atomic" "testing" @@ -239,8 +238,10 @@ func TestRunStopsDownloadingAfterConsecutiveUploadFailures(t *testing.T) { if err == nil { t.Fatal("Run returned nil, want the breaker's error") } - if !strings.Contains(err.Error(), "consecutive upload failures") { - t.Errorf("error does not mention the breaker: %v", err) + // The sentinel, not the wording: the caller maps this to a distinct exit + // code so a driver stops instead of retrying against a dead remote. + if !errors.Is(err, ErrDestinationFailing) { + t.Errorf("error is not ErrDestinationFailing: %v", err) } // The point of stopping the iterator rather than cancelling uploads: the // download side must not have walked the whole chat. diff --git a/internal/pipeline/space.go b/internal/pipeline/space.go new file mode 100644 index 0000000..c2125d1 --- /dev/null +++ b/internal/pipeline/space.go @@ -0,0 +1,66 @@ +package pipeline + +import ( + "context" + "fmt" + "sync" + "time" +) + +// spaceCheckInterval is how often the destination's free space is re-read +// mid-run. About is a network round trip, so it is not worth doing per file. +const spaceCheckInterval = time.Minute + +// spaceGuard watches the destination's free space during a run. +// +// Checking only before starting is not enough on a long archive: an 18k-message +// chat runs for hours, and a remote that was fine at the start can fill in the +// middle — from this run's own uploads, or from anything else using the account. +// Without this the run discovers it by failing several multi-gigabyte uploads in +// a row, which costs the download bandwidth for all of them. +// +// A backend that cannot report a quota is treated as unlimited, matching the +// pre-flight check and the shell pipeline before it. +type spaceGuard struct { + free func(context.Context) (int64, bool) + minFree int64 + + mu sync.Mutex + lastCheck time.Time + failed error +} + +func newSpaceGuard(free func(context.Context) (int64, bool), minFree int64) *spaceGuard { + if free == nil || minFree <= 0 { + return nil + } + return &spaceGuard{free: free, minFree: minFree, lastCheck: time.Now()} +} + +// check reports an error once the destination has dropped below the floor. +// +// The verdict is sticky: once the remote is known to be full, every later call +// says so without another round trip, because the run is ending either way. +func (g *spaceGuard) check(ctx context.Context) error { + if g == nil { + return nil + } + g.mu.Lock() + defer g.mu.Unlock() + + if g.failed != nil { + return g.failed + } + if time.Since(g.lastCheck) < spaceCheckInterval { + return nil + } + g.lastCheck = time.Now() + + free, ok := g.free(ctx) + if !ok || free >= g.minFree { + return nil + } + g.failed = fmt.Errorf("%w: only %.1f GiB free, below the %.1f GiB floor", + ErrDestinationFailing, float64(free)/(1<<30), float64(g.minFree)/(1<<30)) + return g.failed +} diff --git a/internal/pipeline/space_test.go b/internal/pipeline/space_test.go new file mode 100644 index 0000000..87c4057 --- /dev/null +++ b/internal/pipeline/space_test.go @@ -0,0 +1,76 @@ +package pipeline + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestSpaceGuard(t *testing.T) { + const floor = 10 << 30 + + t.Run("unset guard never complains", func(t *testing.T) { + if err := newSpaceGuard(nil, floor).check(t.Context()); err != nil { + t.Errorf("check = %v, want nil when no reporter is configured", err) + } + }) + + t.Run("a backend without a quota counts as unlimited", func(t *testing.T) { + g := newSpaceGuard(func(context.Context) (int64, bool) { return 0, false }, floor) + g.lastCheck = time.Now().Add(-2 * spaceCheckInterval) + if err := g.check(t.Context()); err != nil { + t.Errorf("check = %v, want nil when the backend cannot report", err) + } + }) + + t.Run("plenty of room is fine", func(t *testing.T) { + g := newSpaceGuard(func(context.Context) (int64, bool) { return 100 << 30, true }, floor) + g.lastCheck = time.Now().Add(-2 * spaceCheckInterval) + if err := g.check(t.Context()); err != nil { + t.Errorf("check = %v, want nil with 100 GiB free", err) + } + }) + + t.Run("below the floor stops the run", func(t *testing.T) { + var calls int + g := newSpaceGuard(func(context.Context) (int64, bool) { + calls++ + return 1 << 30, true + }, floor) + g.lastCheck = time.Now().Add(-2 * spaceCheckInterval) + + err := g.check(t.Context()) + if !errors.Is(err, ErrDestinationFailing) { + t.Fatalf("check = %v, want ErrDestinationFailing", err) + } + // Sticky, and without another round trip: the run is ending either way, + // and every upload worker calls this. + for range 5 { + if !errors.Is(g.check(t.Context()), ErrDestinationFailing) { + t.Fatal("the verdict did not stick") + } + } + if calls != 1 { + t.Errorf("free space was read %d times, want 1", calls) + } + }) + + t.Run("checks are throttled", func(t *testing.T) { + var calls int + g := newSpaceGuard(func(context.Context) (int64, bool) { + calls++ + return 100 << 30, true + }, floor) + // Freshly built, so the interval has not elapsed. About is a network + // round trip and every worker calls this per file. + for range 20 { + if err := g.check(t.Context()); err != nil { + t.Fatalf("check = %v", err) + } + } + if calls != 0 { + t.Errorf("free space was read %d times inside the interval, want 0", calls) + } + }) +} diff --git a/internal/remote/fs.go b/internal/remote/fs.go index 1ab57b0..f84861a 100644 --- a/internal/remote/fs.go +++ b/internal/remote/fs.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "os" + "sort" "strings" "sync" @@ -33,7 +34,13 @@ func DefaultTunables() Tunables { // installOnce guards configfile.Install, which swaps unsynchronised package // globals in rclone's config package. Calling it twice is harmless on its own, // but racing it against an Fs resolution is not. -var installOnce sync.Once +var ( + installOnce sync.Once + // installErr is remembered, not just returned once. A second Init would + // otherwise skip the Do body and report success against a config that was + // never loaded. + installErr error +) // Init loads the user's rclone.conf and applies tunables to a derived context. // @@ -48,16 +55,21 @@ var installOnce sync.Once // code this tool defines as "incomplete", which would send a driver into an // endless retry. Loading up front turns that into an ordinary error. func Init(ctx context.Context, t Tunables) (context.Context, error) { - var err error installOnce.Do(func() { configfile.Install() if lerr := config.Data().Load(); lerr != nil && !errors.Is(lerr, config.ErrorConfigFileNotFound) { - err = fmt.Errorf("cannot read rclone config %q: %w "+ + installErr = fmt.Errorf("cannot read rclone config %q: %w "+ "(an encrypted config needs RCLONE_CONFIG_PASS)", config.GetConfigPath(), lerr) + return } + // Marks the data loaded so rclone's own lazy path is never taken. That + // path ends in fs.Fatalf -> os.Exit(1), past every defer and with an + // exit code this tool defines as "incomplete"; loading here and not + // flagging it would leave the window open until the first NewFs. + config.LoadedData() }) - if err != nil { - return ctx, err + if installErr != nil { + return ctx, installErr } ctx, ci := fs.AddConfig(ctx) @@ -93,11 +105,43 @@ func Resolve(ctx context.Context, remote string) (fs.Fs, error) { return nil, fmt.Errorf("rclone remote %q is not configured; configured remotes: %s", remote, strings.Join(sections(), ", ")) } + if missing := missingBackend(err); missing != "" { + // Otherwise this reads as a credentials problem, which sends the + // operator to rclone config for a fault that is in the build. + return nil, fmt.Errorf("remote %q needs the %q backend, which this binary does "+ + "not contain — it registers only %s. A build without -tags slim includes "+ + "every rclone backend: %w", remote, missing, strings.Join(backends(), ", "), err) + } return nil, fmt.Errorf("cannot reach %q — check credentials and connectivity: %w", remote, err) } return f, nil } +// missingBackend names the backend rclone could not find, if that is what went +// wrong. fs.Find returns a bare fmt.Errorf with no sentinel to match, so the +// text is all there is to go on. +func missingBackend(err error) string { + const prefix = "didn't find backend called " + msg := err.Error() + i := strings.Index(msg, prefix) + if i < 0 { + return "" + } + name := strings.TrimPrefix(msg[i+len(prefix):], "\"") + name, _, _ = strings.Cut(name, "\"") + return name +} + +// backends lists the backends compiled into this binary. +func backends() []string { + out := make([]string, 0, len(fs.Registry)) + for _, info := range fs.Registry { + out = append(out, info.Name) + } + sort.Strings(out) + return out +} + // EnsureDir creates the destination if it is not there yet. // // A destination that does not exist yet is the normal case for a first run, and diff --git a/internal/remote/fs_test.go b/internal/remote/fs_test.go index 3dbd30c..310f712 100644 --- a/internal/remote/fs_test.go +++ b/internal/remote/fs_test.go @@ -3,6 +3,8 @@ package remote import ( "context" "os" + "os/exec" + "slices" "strings" "testing" @@ -92,18 +94,61 @@ func TestInitAppliesTunables(t *testing.T) { } } -// An operator's environment override must survive Init, so a tunable is only -// applied when the corresponding variable is absent. +// rclone reads RCLONE_TRANSFERS into its global config at package init, so an +// in-process t.Setenv cannot observe whether Init honours it — the assertion +// would pass with the guard removed. A subprocess is the only real check. func TestInitLeavesEnvOverridesAlone(t *testing.T) { - unset(t, "RCLONE_TRANSFERS") - t.Setenv("RCLONE_TRANSFERS", "7") - - ctx, err := Init(context.Background(), Tunables{Transfers: 2, LowLevelRetries: 20}) - if err != nil { - t.Fatalf("Init: %v", err) + if os.Getenv("GO_INIT_ENV_CHILD") == "1" { + ctx, err := Init(t.Context(), Tunables{Transfers: 2, LowLevelRetries: 20}) + if err != nil { + t.Fatalf("Init: %v", err) + } + ci := fs.GetConfig(ctx) + if ci.Transfers != 7 { + t.Errorf("Transfers = %d, want the operator's 7", ci.Transfers) + } + if ci.LowLevelRetries != 99 { + t.Errorf("LowLevelRetries = %d, want the operator's 99", ci.LowLevelRetries) + } + return } - if got := fs.GetConfig(ctx).Transfers; got == 2 { - t.Errorf("Transfers = 2; Init overwrote the RCLONE_TRANSFERS override") + + cmd := exec.Command(os.Args[0], "-test.run=TestInitLeavesEnvOverridesAlone", "-test.v") + cmd.Env = append(os.Environ(), + "GO_INIT_ENV_CHILD=1", + "RCLONE_TRANSFERS=7", + "RCLONE_LOW_LEVEL_RETRIES=99", + ) + if out, err := cmd.CombinedOutput(); err != nil { + t.Errorf("Init overrode the operator's environment:\n%s", out) + } +} + +// And with nothing set, the pikpak-derived tunables are what apply. +func TestInitAppliesTunablesWhenTheEnvIsQuiet(t *testing.T) { + if os.Getenv("GO_INIT_QUIET_CHILD") == "1" { + ctx, err := Init(t.Context(), Tunables{Transfers: 2, LowLevelRetries: 20}) + if err != nil { + t.Fatalf("Init: %v", err) + } + ci := fs.GetConfig(ctx) + if ci.Transfers != 2 || ci.LowLevelRetries != 20 { + t.Errorf("Transfers=%d LowLevelRetries=%d, want 2 and 20", + ci.Transfers, ci.LowLevelRetries) + } + return + } + + cmd := exec.Command(os.Args[0], "-test.run=TestInitAppliesTunablesWhenTheEnvIsQuiet", "-test.v") + cmd.Env = append(os.Environ(), "GO_INIT_QUIET_CHILD=1") + // Cleared rather than assumed absent: the parent's own environment may + // carry them, which would make this assert the opposite of what it says. + cmd.Env = slices.DeleteFunc(cmd.Env, func(kv string) bool { + return strings.HasPrefix(kv, "RCLONE_TRANSFERS=") || + strings.HasPrefix(kv, "RCLONE_LOW_LEVEL_RETRIES=") + }) + if out, err := cmd.CombinedOutput(); err != nil { + t.Errorf("Init did not apply its tunables:\n%s", out) } } diff --git a/internal/report/progress.go b/internal/report/progress.go index 69cf2c7..4d31ec9 100644 --- a/internal/report/progress.go +++ b/internal/report/progress.go @@ -93,12 +93,16 @@ func humanBytes(n int64) string { if n < unit { return fmt.Sprintf("%d B", n) } + // The unit table runs to exabytes so the index cannot escape it. A PiB is + // not reachable from a Telegram chat, but a panic in the progress line would + // take down a run that was working. + const units = "KMGTPE" div, exp := int64(unit), 0 - for v := n / unit; v >= unit; v /= unit { + for v := n / unit; v >= unit && exp < len(units)-1; v /= unit { div *= unit exp++ } - return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGT"[exp]) + return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), units[exp]) } // isTerminal reports whether w is a character device. diff --git a/internal/tgsource/iterate.go b/internal/tgsource/iterate.go index 53d2d66..87c90db 100644 --- a/internal/tgsource/iterate.go +++ b/internal/tgsource/iterate.go @@ -76,6 +76,16 @@ func Walk(ctx context.Context, api *tg.Client, peer peers.Peer) iter.Seq2[Item, if err := it.Err(); err != nil { yield(Item{}, fmt.Errorf("walk chat history: %w", err)) } + + // There is no cross-check that the walk saw the whole history, and the + // obvious one does not work. gotd's iterator ends with a nil error if a + // fetch yields an empty buffer (messages/iter.go:97,103,155-160), so a + // truncated walk is indistinguishable from a complete one — but + // Iterator.Total is the server's history count, which includes the + // deleted slots that Next skips at iter.go:159-162. Comparing the two + // reports a short read on any chat that has ever had a message deleted, + // which is nearly all of them. A real check would need a count of + // non-empty messages, which the API does not offer. } } diff --git a/internal/verify/verify.go b/internal/verify/verify.go index c72c72e..cac0b29 100644 --- a/internal/verify/verify.go +++ b/internal/verify/verify.go @@ -63,6 +63,13 @@ type Tiny struct { Size int64 } +// Mismatch is a present file whose stored size is not the size Telegram reports. +type Mismatch struct { + MessageID int + Name string + Want, Got int64 +} + // Report is the outcome of comparing a chat against a remote. type Report struct { Expected int // media messages in the chat @@ -72,25 +79,71 @@ type Report struct { Absent []int // not on the remote under the wanted name ZeroByte []int // present but empty - Misnamed []Misnamed - Unsafe []Unsafe - Tiny []Tiny + Misnamed []Misnamed + Unsafe []Unsafe + Tiny []Tiny + Mismatched []Mismatch + + // checked records that Check actually ran. Without it a zero Report claims + // the archive is complete — nothing expected, nothing missing — which is the + // value a command holds before its Telegram callback has populated it. Any + // path that returns early therefore reports success on an untouched chat. + checked bool } +// Ran reports whether this came from a Check rather than being a zero value. +func (r Report) Ran() bool { return r.checked } + // Todo lists the message ids needing another fetch, in ascending order. // -// Zero-byte files are included: rclone overwrites a size-mismatched destination, -// so simply fetching again repairs them. +// Zero-byte and wrong-size files are included: rclone overwrites a +// size-mismatched destination, so simply fetching again repairs them. func (r Report) Todo() []int { - todo := make([]int, 0, len(r.Absent)+len(r.ZeroByte)) + todo := make([]int, 0, len(r.Absent)+len(r.ZeroByte)+len(r.Mismatched)) todo = append(todo, r.Absent...) todo = append(todo, r.ZeroByte...) + for _, m := range r.Mismatched { + todo = append(todo, m.MessageID) + } sort.Ints(todo) return todo } +// Fetchable lists the outstanding ids a run could actually retrieve. +// +// It is Todo minus the unsafe names, which are in Todo because they are not +// archived and out of this because no run will ever archive them. The gap +// between the two is what tells "keep going" apart from "this is as far as it +// goes". +func (r Report) Fetchable() []int { + if len(r.Unsafe) == 0 { + return r.Todo() + } + blocked := make(map[int]struct{}, len(r.Unsafe)) + for _, u := range r.Unsafe { + blocked[u.MessageID] = struct{}{} + } + todo := r.Todo() + out := todo[:0:0] + for _, id := range todo { + if _, skip := blocked[id]; !skip { + out = append(out, id) + } + } + return out +} + // Complete reports whether every expected file is present and non-empty. -func (r Report) Complete() bool { return len(r.Todo()) == 0 } +func (r Report) Complete() bool { return r.checked && len(r.Todo()) == 0 } + +// Stalled reports that work remains and none of it can ever be done. +// +// This is the state a drive-until-complete loop cannot detect for itself: the +// report is identical on every pass, so a driver retrying on "incomplete" walks +// the whole history and indexes the whole remote forever, achieving nothing. +func (r Report) Stalled() bool { + return r.checked && len(r.Todo()) > 0 && len(r.Fetchable()) == 0 +} // Check compares the wanted items against an index of the remote. // @@ -99,7 +152,7 @@ func (r Report) Complete() bool { return len(r.Todo()) == 0 } // and the near-misses are collected into Misnamed rather than being quietly // accepted, because the re-download lands beside them and both copies stay. func Check(items []tgsource.Item, idx Index) Report { - r := Report{Expected: len(items)} + r := Report{Expected: len(items), checked: true} for _, it := range items { r.Bytes += it.Size() @@ -123,6 +176,16 @@ func Check(items []tgsource.Item, idx Index) Report { } case size == 0: r.ZeroByte = append(r.ZeroByte, it.MessageID) + case size != it.Size(): + // The remaining way a report could say "complete" when it is not. + // An upload that died partway leaves a plausible object under the + // right name, and matching on name and non-zero size alone would + // count it archived permanently. Telegram's size is known here, so + // there is no reason not to use it; rclone overwrites a mismatched + // destination, so fetching again repairs it. + r.Mismatched = append(r.Mismatched, Mismatch{ + MessageID: it.MessageID, Name: it.Name, Want: it.Size(), Got: size, + }) default: r.Present++ if size < tinyThreshold { @@ -140,6 +203,13 @@ func (r Report) Write(w io.Writer) { fmt.Fprintf(w, "present and intact : %d\n", r.Present) fmt.Fprintf(w, " absent : %d\n", len(r.Absent)) fmt.Fprintf(w, " zero-byte : %d\n", len(r.ZeroByte)) + if len(r.Mismatched) > 0 { + fmt.Fprintf(w, " wrong size : %d\n", len(r.Mismatched)) + for _, m := range r.Mismatched[:min(5, len(r.Mismatched))] { + fmt.Fprintf(w, " id %d %d B on the remote, expected %d %q\n", + m.MessageID, m.Got, m.Want, m.Name) + } + } if len(r.Tiny) > 0 { fmt.Fprintf(w, " under 1KiB (check, not retried): %d\n", len(r.Tiny)) @@ -149,8 +219,8 @@ func (r Report) Write(w io.Writer) { } if len(r.Unsafe) > 0 { - fmt.Fprintf(w, "\nunsafe filenames : %d\n", len(r.Unsafe)) - fmt.Fprintf(w, " these cannot be written to a path and are never fetched:\n") + fmt.Fprintf(w, "\nunarchivable : %d\n", len(r.Unsafe)) + fmt.Fprintf(w, " these names cannot be written and will never be fetched:\n") for _, u := range r.Unsafe { fmt.Fprintf(w, " id %d %v\n", u.MessageID, u.Reason) } @@ -168,9 +238,15 @@ func (r Report) Write(w io.Writer) { } } - if todo := r.Todo(); len(todo) > 0 { + todo := r.Todo() + switch { + case r.Stalled(): + fmt.Fprintf(w, "\nSTALLED: %d file(s) remain, none of which can be fetched.\n", len(todo)) + case len(todo) > 0: fmt.Fprintf(w, "\nneeds another pass : %d (ids %d–%d)\n", len(todo), todo[0], todo[len(todo)-1]) - return + case !r.checked: + fmt.Fprintf(w, "\nno chat was checked.\n") + default: + fmt.Fprintf(w, "\nCOMPLETE: every media message is present and non-empty.\n") } - fmt.Fprintf(w, "\nCOMPLETE: every media message is present and non-empty.\n") } diff --git a/internal/verify/verify_test.go b/internal/verify/verify_test.go index 6978b77..7196a61 100644 --- a/internal/verify/verify_test.go +++ b/internal/verify/verify_test.go @@ -157,9 +157,17 @@ func TestCheckFlagsUnsafeNames(t *testing.T) { var sb strings.Builder r.Write(&sb) - if !strings.Contains(sb.String(), "unsafe filenames") { + if !strings.Contains(sb.String(), "unarchivable") { t.Errorf("report should flag the unsafe name, got:\n%s", sb.String()) } + // Nothing else is outstanding, so this run is as far as it can get. Saying + // "needs another pass" here is what makes a driver loop forever. + if !r.Stalled() { + t.Error("a report whose only outstanding item is unarchivable must be stalled") + } + if !strings.Contains(sb.String(), "STALLED") { + t.Errorf("report should say it is stalled, got:\n%s", sb.String()) + } } // An unsafe name must be rejected even when something is stored under that @@ -179,3 +187,56 @@ func TestReportTodoIsSorted(t *testing.T) { t.Errorf("Todo() = %v, want %v", r.Todo(), want) } } + +// A zero Report is the value a command holds before its Telegram callback has +// run. It must not claim the archive is complete. +func TestZeroReportIsNotComplete(t *testing.T) { + var r Report + if r.Complete() { + t.Error("a Report that never ran reports the archive complete") + } + if r.Ran() { + t.Error("Ran() is true on a zero Report") + } + if r.Stalled() { + t.Error("a Report that never ran reports itself stalled") + } + + var sb strings.Builder + r.Write(&sb) + if strings.Contains(sb.String(), "COMPLETE") { + t.Errorf("a Report that never ran printed COMPLETE:\n%s", sb.String()) + } +} + +// A remote object under the right name but the wrong size is the last way a +// report could say complete when it is not: an upload that died partway leaves +// exactly that, and the local copy is already gone. +func TestCheckTreatsAWrongSizeAsOutstanding(t *testing.T) { + items := []tgsource.Item{item(4242, "clip.mp4", 4096)} + idx := fakeIndex{"1234567890_4242_clip.mp4": 400} + + r := Check(items, idx) + if r.Present != 0 { + t.Errorf("Present = %d, want 0 for a truncated object", r.Present) + } + if len(r.Mismatched) != 1 { + t.Fatalf("Mismatched = %v, want one entry", r.Mismatched) + } + if got := r.Mismatched[0]; got.Want != 4096 || got.Got != 400 { + t.Errorf("Mismatch = %+v, want want=4096 got=400", got) + } + if !slices.Contains(r.Todo(), 4242) { + t.Errorf("Todo = %v, want it to include 4242", r.Todo()) + } + if r.Complete() { + t.Error("a truncated object was counted as a complete archive") + } + // It is fetchable, unlike an unsafe name — re-uploading overwrites it. + if !slices.Contains(r.Fetchable(), 4242) { + t.Errorf("Fetchable = %v, want it to include 4242", r.Fetchable()) + } + if r.Stalled() { + t.Error("a repairable file must not be reported as stalled") + } +}