From efb29c9c399e75ecaf37f9d598004ccaf1f98fb1 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Tue, 8 Sep 2026 18:20:24 +0700 Subject: [PATCH] fix: report why a download failed, retry it, and stop a failing source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit core's downloader logs a failed transfer and returns nil, and nothing here installed a logger, so logctx handed out a nop and the reason was destroyed. The size check in finish was all that survived, which reported a failed two-gigabyte fetch as "short download: got 0 bytes" — a symptom with no cause an operator can act on. The logger it reaches for now feeds a core that keeps error entries and stores them on the element they belong to, so finish reports the reason ahead of the byte count. A failed download is also retried inside the run, up to three passes over whatever is still missing and spaced minutes apart, because the failure this repairs is transient: a dead connection takes every transfer in flight with it and all of them are fetchable again afterwards, while leaving them to the next run costs a full re-walk of the chat and a re-index of the remote first. An item is counted once however many attempts it took, and a pass that recovers reports nothing. The download leg gains the breaker the upload leg already had. A source that refuses everything refuses the rest of the list in milliseconds, so the pass gives up after five failures in a row and the run exits 3 rather than spending the entire outstanding list finding that out. --- README.md | 16 ++- cmd/tgexport/sync.go | 12 +- go.mod | 7 +- go.sum | 12 -- internal/pipeline/cause.go | 80 +++++++++++++ internal/pipeline/download.go | 40 ++++++- internal/pipeline/download_test.go | 104 +++++++++++++++++ internal/pipeline/elem.go | 7 ++ internal/pipeline/events.go | 13 ++- internal/pipeline/pipeline.go | 173 ++++++++++++++++++++++++++--- internal/pipeline/progress.go | 31 ++++++ internal/pipeline/run_test.go | 134 ++++++++++++++++++++-- internal/report/live.go | 7 ++ internal/report/progress.go | 7 ++ 14 files changed, 591 insertions(+), 52 deletions(-) create mode 100644 internal/pipeline/cause.go diff --git a/README.md b/README.md index 0e75887..a9d062b 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ a chat. | 0 | complete | | 1 | ran, but files remain — run again | | 2 | usage error | -| 3 | remote or Telegram failure, including a destination that stopped accepting uploads | +| 3 | remote or Telegram failure, including either leg refusing transfer after transfer | | 4 | stalled: files remain, none of which can ever be fetched | | 130 / 143 | interrupted (SIGINT / SIGTERM) | @@ -187,6 +187,20 @@ verifier, which matched on name and non-zero size — on the archive this was built for it found six objects that had been counted complete for months, one of them 221 MiB standing in for a 2 GiB video. Re-running repairs them. +**Failures.** A failed download is retried inside the run: up to three passes +over whatever is still missing, spaced a minute and then two apart. The failure +this exists for is the transient one — a connection that dies takes every +transfer in flight with it, and all of them are fetchable again minutes later — +and leaving them to the next run costs a full re-walk of the chat and a re-index +of the remote before a byte can move. Each failure is reported with the reason +Telegram gave for it, not just the byte count that arrived. + +Either leg gives up once five transfers in a row fail. A source or destination +that is refusing everything will refuse the rest of the list too, in +milliseconds, so the run stops and exits 3 instead of spending the whole +outstanding list finding that out. A run that trips and then recovers on a retry +reports nothing. + ## Replacing the shell pipeline Earlier versions of this repo were three bash scripts — `run.sh`, diff --git a/cmd/tgexport/sync.go b/cmd/tgexport/sync.go index 90bdd94..1b9f306 100644 --- a/cmd/tgexport/sync.go +++ b/cmd/tgexport/sync.go @@ -213,11 +213,13 @@ func syncCmd(ctx context.Context, args []string) error { final.Write(os.Stdout) 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. + case errors.Is(runErr, pipeline.ErrDestinationFailing), + errors.Is(runErr, pipeline.ErrSourceFailing): + // Exit 3, not 1. One half of the run refused transfer after transfer, + // and it will refuse them next pass too — a driver retrying on + // "incomplete" would walk the whole chat and re-index the whole remote + // to achieve nothing, indefinitely. The run already retried the failures + // itself before giving up, so this is not a transient verdict. return runErr case final.Stalled(): return fmt.Errorf("%w: %d file(s) remain, none of which can be fetched", diff --git a/go.mod b/go.mod index 8fbdd96..90baf01 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,9 @@ require ( github.com/gotd/td v0.140.0 github.com/iyear/tdl/core v0.20.4 github.com/rclone/rclone v1.75.1 + github.com/vbauerster/mpb/v8 v8.16.1 go.etcd.io/bbolt v1.5.0 + go.uber.org/zap v1.28.0 golang.org/x/sync v0.22.0 ) @@ -38,7 +40,6 @@ require ( github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect github.com/adrg/xdg v0.5.3 // indirect github.com/anchore/go-lzo v0.1.1 // indirect - github.com/andybalholm/brotli v1.2.2 // indirect github.com/andybalholm/cascadia v1.3.4 // indirect github.com/apache/arrow-go/v18 v18.7.0 // indirect github.com/appscode/go-querystring v0.0.0-20170504095604-0126cfb3f1dc // indirect @@ -127,7 +128,6 @@ require ( github.com/gorilla/schema v1.4.1 // indirect github.com/gotd/contrib v0.20.0 // indirect github.com/gotd/ige v0.3.0 // indirect - github.com/gotd/log v0.1.0 // indirect github.com/gotd/neo v0.1.5 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect @@ -184,7 +184,6 @@ require ( github.com/putdotio/go-putio/putio v0.0.0-20200123120452-16d982cac2b8 // indirect github.com/rclone/Proton-API-Bridge v1.0.5 // indirect github.com/rclone/go-proton-api v1.0.4 // indirect - github.com/refraction-networking/utls v1.8.2 // indirect github.com/relvacode/iso8601 v1.7.0 // indirect github.com/rfjakob/eme v1.2.0 // indirect github.com/sabhiram/go-gitignore v0.0.0-20210923224102-525f6e181f06 // indirect @@ -206,7 +205,6 @@ require ( github.com/ulikunitz/xz v0.5.15 // indirect github.com/unknwon/goconfig v1.0.0 // indirect github.com/vbauerster/cupwriter v0.0.4 // indirect - github.com/vbauerster/mpb/v8 v8.16.1 // indirect github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect @@ -223,7 +221,6 @@ require ( go.opentelemetry.io/otel/trace v1.44.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect - go.uber.org/zap v1.28.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/crypto v0.56.0 // indirect diff --git a/go.sum b/go.sum index af02a09..c0aeb5f 100644 --- a/go.sum +++ b/go.sum @@ -294,16 +294,10 @@ github.com/gotd/contrib v0.20.0 h1:1Wc4+HMQiIKYQuGHVwVksIx152HFTP6B5n88dDe0ZYw= github.com/gotd/contrib v0.20.0/go.mod h1:P6o8W4niqhDPHLA0U+SA/L7l3BQHYLULpeHfRSePn9o= github.com/gotd/ige v0.3.0 h1:4f6LEHWsVDLBG0bT9wWG2/9TZb5aWm265G8ZlTXmRRU= github.com/gotd/ige v0.3.0/go.mod h1:FE9bTaQtvfArizAcZuI4sS6gXaEUBmixdUufVHoCKac= -github.com/gotd/log v0.1.0 h1:4LJUEvafD1xtBwx2QkrlzFnRgbYXTlWqJPDi8BvrLbU= -github.com/gotd/log v0.1.0/go.mod h1:5ilhdu1Ux0QvDY/FF3Ojfw24Ws3SlCtyLwOpXy8KYXs= -github.com/gotd/log/logzap v0.1.1 h1:O6l7d8HUbODe+UMcrM47eXYDwdJ6RNmpQejLjrlcEIQ= -github.com/gotd/log/logzap v0.1.1/go.mod h1:5ObZkITbfhbsBOLzBkzmMk9QxXc0eNQpimau7zRL+Y8= github.com/gotd/neo v0.1.5 h1:oj0iQfMbGClP8xI59x7fE/uHoTJD7NZH9oV1WNuPukQ= github.com/gotd/neo v0.1.5/go.mod h1:9A2a4bn9zL6FADufBdt7tZt+WMhvZoc5gWXihOPoiBQ= github.com/gotd/td v0.140.0 h1:trNBzTnhNtNwHsFp5qwKnNxQRAZJ6/BRE+uH3Lojauk= github.com/gotd/td v0.140.0/go.mod h1:0ZkRxG7N+5ooG7/zdRXcnGautGPM6IKmyPQvdsAeF20= -github.com/gotd/td v0.161.0 h1:krbzsb70cakdrqF+MUIo+W7BkQTVhyB1kNS7X/+BLcY= -github.com/gotd/td v0.161.0/go.mod h1:7HdCs+zeJugdgZAF5iG8f70eOJvuiH2QzjoyUcysXbY= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -386,8 +380,6 @@ github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stg github.com/mattn/go-isatty v0.0.23 h1:cYwCQTQf3HB6xUC+BtyCLZNr7IzbOmoZbmssVNzSyiQ= github.com/mattn/go-isatty v0.0.23/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= github.com/mattn/go-runewidth v0.0.3/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU= -github.com/mattn/go-runewidth v0.0.24 h1:cpokDiIn0MGnhdHwuWnJBITySJ20QyNGnY2kR/ay2DU= -github.com/mattn/go-runewidth v0.0.24/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= github.com/mattn/go-runewidth v0.0.28 h1:rPyg2ybwEKPebvpzVWe1gKBkH8EQFkxO4Y0hjBeLaBU= github.com/mattn/go-runewidth v0.0.28/go.mod h1:3qAiGCV4Koz/yuveO58qUefmUTRm8r0IGEXZ9jeHp/8= github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y= @@ -461,8 +453,6 @@ github.com/rclone/go-proton-api v1.0.4 h1:AJW0e9pB4j0hVK4WqyGErFwaI+5MUQWPCtj5FY github.com/rclone/go-proton-api v1.0.4/go.mod h1:QAlkFfswzrBuxvCORWV8rZdddg52hahMN98CFWoFW1E= github.com/rclone/rclone v1.75.1 h1:kIxQcoDLj2Gke/gMSHK7OnxhX1Gu1cJBLP1kJZoaFp0= github.com/rclone/rclone v1.75.1/go.mod h1:4zmMjGatCkSJPRZDpo+7y3xOl8S29EMUyKvZop5mHr4= -github.com/refraction-networking/utls v1.8.2 h1:j4Q1gJj0xngdeH+Ox/qND11aEfhpgoEvV+S9iJ2IdQo= -github.com/refraction-networking/utls v1.8.2/go.mod h1:jkSOEkLqn+S/jtpEHPOsVv/4V4EVnelwbMQl4vCWXAM= github.com/relvacode/iso8601 v1.7.0 h1:BXy+V60stMP6cpswc+a93Mq3e65PfXCgDFfhvNNGrdo= github.com/relvacode/iso8601 v1.7.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I= github.com/rfjakob/eme v1.2.0 h1:8dAHL+WVAw06+7DkRKnRiFp1JL3QjcJEZFqDnndUaSI= @@ -540,8 +530,6 @@ github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/ github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw= github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM= github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw= -github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= -github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM= github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= diff --git a/internal/pipeline/cause.go b/internal/pipeline/cause.go new file mode 100644 index 0000000..2707470 --- /dev/null +++ b/internal/pipeline/cause.go @@ -0,0 +1,80 @@ +package pipeline + +import ( + "context" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + + "github.com/iyear/tdl/core/logctx" +) + +// captureCauses points core's downloader logging back into this package. +// +// core's Download never returns why a transfer failed. It logs the error and +// returns nil (downloader.go:47-60), so the size check in finish is the only +// evidence anything went wrong — and a size check can say how many bytes +// arrived, never why the rest did not. Nothing here installed a logger either, +// which means logctx handed out zap.NewNop() and the reason was destroyed: a +// failed two-gigabyte fetch was reported as "short download: got 0 bytes" with +// the flood wait, expired file reference, or dead connection behind it gone. +// +// So the error is taken from the one place it exists. The entry carries the +// element it belongs to, which is this package's own *elem, so the error is +// stored on it and finish reports it beside the size. +func captureCauses(ctx context.Context) context.Context { + return logctx.With(ctx, zap.New(&causeCore{})) +} + +// causeCore is a zapcore.Core that keeps error entries and drops everything +// else. Fields are matched on their type rather than their key so a rename +// upstream cannot silently switch the capture off. +type causeCore struct{ fields []zapcore.Field } + +func (c *causeCore) Enabled(l zapcore.Level) bool { return l >= zapcore.ErrorLevel } + +func (c *causeCore) With(fields []zapcore.Field) zapcore.Core { + joined := make([]zapcore.Field, 0, len(c.fields)+len(fields)) + joined = append(joined, c.fields...) + joined = append(joined, fields...) + return &causeCore{fields: joined} +} + +func (c *causeCore) Check(e zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry { + if c.Enabled(e.Level) { + return ce.AddCore(e, c) + } + return ce +} + +// Write pairs a logged error with the element it was logged for. +// +// The write happens on the download worker's own goroutine, immediately before +// the deferred OnDone that reads it, so storing the error on the element needs +// no synchronisation. +func (c *causeCore) Write(_ zapcore.Entry, fields []zapcore.Field) error { + var ( + el *elem + err error + ) + for _, set := range [][]zapcore.Field{c.fields, fields} { + for _, f := range set { + switch f.Type { + case zapcore.ErrorType: + if e, ok := f.Interface.(error); ok { + err = e + } + case zapcore.ReflectType: + if e, ok := f.Interface.(*elem); ok { + el = e + } + } + } + } + if el != nil && err != nil { + el.cause = err + } + return nil +} + +func (c *causeCore) Sync() error { return nil } diff --git a/internal/pipeline/download.go b/internal/pipeline/download.go index daa3718..8ae7130 100644 --- a/internal/pipeline/download.go +++ b/internal/pipeline/download.go @@ -38,6 +38,16 @@ type DownloadOptions struct { // stop, when set, ends iteration cleanly from another goroutine — used to // halt downloads once the destination has stopped accepting uploads. stop *atomic.Bool + // maxFailures ends the pass after this many consecutive failures. Zero means + // no breaker. onTrip, when set, is called once if that happens — the caller + // decides what an abandoned pass means, because a retry that then succeeds + // must not report the source as down. + maxFailures int + onTrip func() + // seed starts the run's counters from an earlier pass's totals, so a retry + // continues the numbers a reporter is already displaying rather than + // restarting them at zero. + seed Stats // onReady hands a completed file to the upload leg; onFailed says nothing // was staged, so whatever acquire reserved must be given back. onReady func(tgsource.Item) @@ -55,7 +65,11 @@ type DownloadOptions struct { // guard, because it could not see inside tdl. // // A failed item does not abort the run: it is recorded in the returned outcomes -// and the rest continue, matching what a partial `tdl dl` pass did. +// and the rest continue, matching what a partial `tdl dl` pass did. Past +// maxFailures failures in a row the pass gives up on the remaining items and +// calls onTrip, because a source that is refusing every file will refuse the +// rest of the list too, in milliseconds, and a pass with no brake spends the +// whole todo list proving it. func Download(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o DownloadOptions) ([]Outcome, Stats, error) { if o.Threads <= 0 { o.Threads = 4 @@ -67,6 +81,11 @@ func Download(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Downlo return nil, Stats{}, fmt.Errorf("create staging directory: %w", err) } + // core's downloader logs why a transfer failed rather than returning it, so + // the logger it reaches for is pointed back into this package before any of + // its work starts. Without this the reason goes to a nop logger. + ctx = captureCauses(ctx) + it := newElemIter(seq, o.Staging, o.Takeout) it.acquire = o.acquire it.release = o.release @@ -88,6 +107,12 @@ func Download(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Downlo } return ferr }, o.Events) + // Set rather than passed: both are this package's own bookkeeping, and no + // worker exists yet, so the mutex the fields normally live under has + // nothing to protect them from here. + prog.stats = o.seed + prog.maxStreak = o.maxFailures + prog.onTrip = func() { it.stopped.Store(true) } err := downloader.New(downloader.Options{ Pool: o.Pool, @@ -103,6 +128,12 @@ func Download(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Downlo if err == nil { err = it.failure } + // A tripped breaker is not an error here. It is reported to the caller, who + // knows whether another pass is coming; the items the pass never reached are + // simply absent from the outcomes. + if prog.brokeCircuit() && o.onTrip != nil { + o.onTrip() + } // Skipped items are reported alongside whatever else happened rather than // instead of it: the run did real work, and the caller still needs to know // these messages were never attempted. @@ -131,6 +162,13 @@ func finish(staging string, e *elem, downloadErr error) error { if downloadErr == nil { if err := checkSize(part, e.item.Size()); err != nil { + // The size is the symptom. e.cause is the reason, when the + // downloader logged one, and it is put first: "got 0 bytes" alone + // says a two-gigabyte fetch failed without saying anything an + // operator can act on. + if e.cause != nil { + err = fmt.Errorf("%w: %w", e.cause, err) + } downloadErr = err } } diff --git a/internal/pipeline/download_test.go b/internal/pipeline/download_test.go index 691fbdb..bd7372f 100644 --- a/internal/pipeline/download_test.go +++ b/internal/pipeline/download_test.go @@ -10,7 +10,9 @@ import ( "github.com/gotd/td/tg" "github.com/iyear/tdl/core/downloader" + "github.com/iyear/tdl/core/logctx" "github.com/iyear/tdl/core/tmedia" + "go.uber.org/zap" "github.com/tiennm99dev/telegram-exporter/internal/naming" "github.com/tiennm99dev/telegram-exporter/internal/tgsource" @@ -429,3 +431,105 @@ func TestElemIterReturnsReservationWhenOpenFails(t *testing.T) { t.Errorf("acquired %d bytes but released %d — the reservation leaked", acquired, released) } } + +// The reason a transfer failed only exists in core's log call, so the capture is +// tested through that exact call rather than by poking the field directly: it is +// the shape of the log entry — a reflected element and a zap error — that this +// depends on, and a change to it must fail here rather than silently go back to +// reporting a byte count with no reason. +func TestFinishReportsWhyTheTransferFailed(t *testing.T) { + staging := t.TempDir() + it := testItem(t, 5, "video.mp4", 1000) + e := openElem(t, staging, it) + + want := errors.New("rpc error code 420: FLOOD_WAIT (60)") + ctx := captureCauses(t.Context()) + logctx.From(ctx).Error("Download error", + zap.Any("element", downloader.Elem(e)), zap.Error(want)) + + // nil, exactly as core reports a transfer it has already logged and given up + // on, with nothing written to the file. + err := finish(staging, e, nil) + if !errors.Is(err, want) { + t.Fatalf("finish = %v, want the logged transfer error", err) + } + if !strings.Contains(err.Error(), "expected 1000") { + t.Errorf("the byte count must survive alongside the reason, got: %v", err) + } +} + +// Below error level nothing is captured, so an ordinary debug entry cannot +// attach a bogus reason to an item that merely arrived short. +func TestCaptureIgnoresNonErrorLogging(t *testing.T) { + staging := t.TempDir() + it := testItem(t, 6, "video.mp4", 1000) + e := openElem(t, staging, it) + + ctx := captureCauses(t.Context()) + logctx.From(ctx).Debug("Start download elem", zap.Any("elem", downloader.Elem(e))) + + if e.cause != nil { + t.Fatalf("cause = %v, want nil for a debug entry", e.cause) + } + if err := finish(staging, e, nil); !strings.Contains(err.Error(), "short download") { + t.Errorf("finish = %v, want the plain size failure", err) + } +} + +// The breaker exists because a source that has stopped serving files fails every +// item in milliseconds: without it a pass spends the entire remaining todo list +// proving the same point. A success in between is what tells a run of bad luck +// apart from a source that is down, so it resets the streak. +func TestProgressBreakerTripsOnConsecutiveFailuresOnly(t *testing.T) { + staging := t.TempDir() + trips := 0 + + p := newProgress(func(*elem, error) error { return nil }, nil) + p.maxStreak = 3 + p.onTrip = func() { trips++ } + + fail := func(id int) { p.OnDone(openElem(t, staging, testItem(t, id, "a.mp4", 10)), errors.New("boom")) } + ok := func(id int) { p.OnDone(openElem(t, staging, testItem(t, id, "a.mp4", 10)), nil) } + + fail(1) + fail(2) + ok(3) // the streak is broken here, so the next two must not trip it + fail(4) + fail(5) + if trips != 0 { + t.Fatalf("tripped after a success reset the streak (trips = %d)", trips) + } + if p.brokeCircuit() { + t.Fatal("brokeCircuit() = true before the limit was reached") + } + + fail(6) + if trips != 1 { + t.Fatalf("trips = %d, want 1 at three failures in a row", trips) + } + if !p.brokeCircuit() { + t.Error("brokeCircuit() = false after the breaker fired") + } + + // Fires once: the callback stops the iterator, and repeating it for every + // item still in flight would be noise. + fail(7) + if trips != 1 { + t.Errorf("trips = %d, want the breaker to fire exactly once", trips) + } +} + +// With no limit set there is no breaker at all, which is what a standalone +// Download must keep doing: every item gets an attempt. +func TestProgressWithoutABreakerNeverTrips(t *testing.T) { + staging := t.TempDir() + p := newProgress(func(*elem, error) error { return nil }, nil) + p.onTrip = func() { t.Error("onTrip called with no limit configured") } + + for id := 1; id <= 20; id++ { + p.OnDone(openElem(t, staging, testItem(t, id, "a.mp4", 10)), errors.New("boom")) + } + if p.brokeCircuit() { + t.Error("brokeCircuit() = true with no limit configured") + } +} diff --git a/internal/pipeline/elem.go b/internal/pipeline/elem.go index 7e24c63..c56b3d6 100644 --- a/internal/pipeline/elem.go +++ b/internal/pipeline/elem.go @@ -35,6 +35,13 @@ type elem struct { item tgsource.Item file *os.File takeout bool + + // cause is why the transfer failed, when the downloader logged a reason. + // It is the only route that reason has to reach finish, because core's + // Download logs it and returns nil; see captureCauses. Written by the log + // call on the download worker's goroutine and read by that worker's OnDone, + // so it needs no lock. + cause error } func (e *elem) File() downloader.File { return mediaFile{e.item} } diff --git a/internal/pipeline/events.go b/internal/pipeline/events.go index 8356ce6..ca6ee5d 100644 --- a/internal/pipeline/events.go +++ b/internal/pipeline/events.go @@ -1,6 +1,10 @@ package pipeline -import "github.com/tiennm99dev/telegram-exporter/internal/tgsource" +import ( + "time" + + "github.com/tiennm99dev/telegram-exporter/internal/tgsource" +) // Events receives a run's per-item lifecycle. // @@ -24,6 +28,12 @@ type Events interface { UploadStart(it tgsource.Item) UploadDone(it tgsource.Item, err error) + + // Retry announces another pass over the items the last one failed, and how + // long the run waits first. It is the one event that is not per-item, and it + // exists because the wait is measured in minutes: a display that went quiet + // for that long with no explanation is indistinguishable from a hang. + Retry(attempt, files int, wait time.Duration) } // nopEvents is used when a caller wants no reporting, so nothing on the hot @@ -36,3 +46,4 @@ func (nopEvents) DownloadBytes(tgsource.Item, int64) {} func (nopEvents) DownloadDone(tgsource.Item, error) {} func (nopEvents) UploadStart(tgsource.Item) {} func (nopEvents) UploadDone(tgsource.Item, error) {} +func (nopEvents) Retry(int, int, time.Duration) {} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index def6a24..351b37c 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -9,6 +9,7 @@ import ( "path/filepath" "sync" "sync/atomic" + "time" "github.com/rclone/rclone/fs" "golang.org/x/sync/semaphore" @@ -72,6 +73,37 @@ func (r Result) Failed() []Outcome { // gigabytes for nothing every time. var ErrDestinationFailing = errors.New("destination stopped accepting uploads") +// ErrSourceFailing marks a run stopped because Telegram failed download after +// download. It is the download leg's counterpart to ErrDestinationFailing and +// means the same thing to a driver — another pass will fail the same way, so +// retrying on "incomplete" only re-walks the chat for nothing — but it points at +// the other half of the run, which is what an operator needs to know first. +// +// It survives the run's own retries: a pass that trips and then recovers reports +// nothing, so this only appears when the last attempt was still failing. +var ErrSourceFailing = errors.New("telegram stopped serving downloads") + +// downloadAttempts is how many passes a run makes over the items it failed to +// fetch. +// +// Retrying inside the run is worth far more than leaving it to the next one: a +// fresh sync re-walks every message and re-indexes the whole remote before it +// can fetch a byte, while a retry here already has both. The failure this exists +// for is the transient one — a dead connection or a source that stops serving +// files for a few minutes takes down every transfer in flight, and every one of +// them is fetchable again afterwards. +const downloadAttempts = 3 + +// retryDelay spaces the passes out, and is a var so tests need not sleep. +// +// Minutes rather than seconds: the client's own recovery gives up on a broken +// connection only after the reconnect timeout (five minutes by default), so a +// pass that starts seconds after the last one failed is a pass into the same +// dead connection. +var retryDelay = func(attempt int) time.Duration { + return time.Duration(attempt) * time.Minute +} + // 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. @@ -187,23 +219,97 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R }() } - dlOutcomes, stats, dlErr := download(ctx, seq, DownloadOptions{ - Pool: o.Pool, - Staging: o.Staging, - Threads: o.Threads, - Limit: o.Limit, - Takeout: o.Takeout, - Events: o.Events, - acquire: budget.acquire, - release: budget.release, - onReady: func(it tgsource.Item) { uploads <- it }, - onFailed: func(it tgsource.Item) { - // Nothing was staged, so the reservation has to come back here - // instead of from an upload that will never happen. - budget.release(it.Size()) - }, - stop: stopDownloads, - }) + var ( + dlOutcomes []Outcome + stats Stats + dlErrs []error + at = make(map[int]int) // message id -> its place in dlOutcomes + ) + pending := seq + for attempt := 1; ; attempt++ { + sourceDown := false + passOutcomes, passStats, err := download(ctx, pending, DownloadOptions{ + Pool: o.Pool, + Staging: o.Staging, + Threads: o.Threads, + Limit: o.Limit, + Takeout: o.Takeout, + Events: o.Events, + acquire: budget.acquire, + release: budget.release, + maxFailures: o.MaxFailures, + seed: stats, + onReady: func(it tgsource.Item) { uploads <- it }, + onFailed: func(it tgsource.Item) { + // Nothing was staged, so the reservation has to come back here + // instead of from an upload that will never happen. + budget.release(it.Size()) + }, + stop: stopDownloads, + onTrip: func() { sourceDown = true }, + }) + stats = passStats + if err != nil { + dlErrs = append(dlErrs, err) + } + // One outcome per item, whatever it took: a later attempt replaces the + // earlier verdict in place, so an item that failed once and then + // arrived is reported as archived rather than as both. + for _, oc := range passOutcomes { + if i, ok := at[oc.Item.MessageID]; ok { + dlOutcomes[i] = oc + continue + } + at[oc.Item.MessageID] = len(dlOutcomes) + dlOutcomes = append(dlOutcomes, oc) + } + + // Read, and the shared stop flag cleared, under the upload leg's own + // lock. The flag is what the download breaker sets to end a pass, so a + // retry needs it clear — but clearing it after the upload breaker has + // tripped in this same window would restart downloads into a + // destination that has stopped accepting them, and that leg never sets + // the flag twice. + mu.Lock() + destDown := tripped + if !destDown { + stopDownloads.Store(false) + } + mu.Unlock() + + // Another pass is worth making only for items that failed on their own. + // A pass that returned an error failed as a whole — the walk broke, a + // name could not be opened, the run was cancelled — and the items it + // never reached are lost to this run either way, so retrying the few + // that failed first would dress that up as a nearly complete run. A + // destination that has stopped accepting uploads rules it out too: + // fetching more would only fill staging with files it will refuse. + again := failedItems(passOutcomes) + if len(again) > 0 && attempt < downloadAttempts && + err == nil && !destDown && ctx.Err() == nil { + // These items are being fetched again, so their first attempt comes + // back out of the totals: one item is one file to fetch, not one per + // attempt. Bytes already transferred stay counted — they were really + // spent, and throughput is the honest figure. + stats = rollback(stats, again) + + wait := retryDelay(attempt) + o.Events.Retry(attempt+1, len(again), wait) + if serr := sleep(ctx, wait); serr == nil { + pending = itemsSeq(again) + continue + } + dlErrs = append(dlErrs, fmt.Errorf("cancelled before retrying %d download(s): %w", + len(again), ctx.Err())) + } + + if sourceDown { + dlErrs = append(dlErrs, fmt.Errorf("%w: gave up after %d consecutive download failure(s)", + ErrSourceFailing, o.MaxFailures)) + } + break + } + dlErr := errors.Join(dlErrs...) // Safe only because Download joined its workers, which is guaranteed by // elemIter.Err always being nil. @@ -232,6 +338,39 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R return Result{Stats: stats, Outcomes: dlOutcomes}, errors.Join(dlErr, upErr) } +// failedItems lists the items a pass did not manage to fetch. +func failedItems(outcomes []Outcome) []tgsource.Item { + var out []tgsource.Item + for _, oc := range outcomes { + if oc.Err != nil { + out = append(out, oc.Item) + } + } + return out +} + +// rollback takes a pass's failures back out of the running totals so the next +// pass counts them once, not twice. +func rollback(s Stats, again []tgsource.Item) Stats { + for _, it := range again { + s.Started-- + s.Failed-- + s.BytesTotal -= it.Size() + } + return s +} + +// itemsSeq feeds a retry pass the items the last one failed. +func itemsSeq(items []tgsource.Item) iter.Seq2[tgsource.Item, error] { + return func(yield func(tgsource.Item, error) bool) { + for _, it := range items { + if !yield(it, nil) { + return + } + } + } +} + // budget bounds how many bytes of downloaded-but-not-yet-uploaded data sit on // local disk. A zero limit means no bound. type budget struct{ sem *semaphore.Weighted } diff --git a/internal/pipeline/progress.go b/internal/pipeline/progress.go index 5ad97da..dade2e2 100644 --- a/internal/pipeline/progress.go +++ b/internal/pipeline/progress.go @@ -35,6 +35,17 @@ type progress struct { outcomes []Outcome inFlight map[int]int64 // message id -> bytes written so far + // maxStreak is how many failures in a row end the pass; zero disables the + // breaker. streak counts them, and onTrip is called once when the limit is + // reached. The download leg needs this for the same reason the upload leg + // does: when the source stops serving files every item fails in + // milliseconds, and without a breaker a pass burns the entire remaining + // todo list on transfers that cannot succeed. + maxStreak int + streak int + tripped bool + onTrip func() + finish func(*elem, error) error events Events } @@ -88,16 +99,36 @@ func (p *progress) OnDone(e downloader.Elem, err error) { delete(p.inFlight, el.item.MessageID) if err != nil { p.stats.Failed++ + p.streak++ } else { p.stats.Done++ + p.streak = 0 } p.outcomes = append(p.outcomes, Outcome{Item: el.item, Err: err}) stats := p.stats + // Decided under the lock and acted on outside it: onTrip stops the + // iterator, and holding this mutex across it would put the callback's + // locking order inside this one's. + trip := p.maxStreak > 0 && p.streak >= p.maxStreak && !p.tripped + if trip { + p.tripped = true + } p.mu.Unlock() + + if trip && p.onTrip != nil { + p.onTrip() + } p.events.DownloadDone(el.item, err) p.events.Stats(stats) } +// brokeCircuit reports whether the failure streak ended this pass. +func (p *progress) brokeCircuit() bool { + p.mu.Lock() + defer p.mu.Unlock() + return p.tripped +} + func (p *progress) results() ([]Outcome, Stats) { p.mu.Lock() defer p.mu.Unlock() diff --git a/internal/pipeline/run_test.go b/internal/pipeline/run_test.go index 44f9f7b..d5b6c30 100644 --- a/internal/pipeline/run_test.go +++ b/internal/pipeline/run_test.go @@ -9,6 +9,7 @@ import ( "path/filepath" "sync/atomic" "testing" + "time" _ "github.com/rclone/rclone/backend/local" "github.com/rclone/rclone/fs" @@ -25,11 +26,18 @@ import ( // pieces are wired together, and a regression in any of them is silent. // runFake substitutes the download step for the duration of a test. +// +// It also collapses the wait between retry passes. A run retries the items it +// failed to fetch, and at the real cadence every test with a failure in it would +// spend minutes asleep. func runFake(t *testing.T, f func(context.Context, iter.Seq2[tgsource.Item, error], DownloadOptions) ([]Outcome, Stats, error)) { t.Helper() - prev := download + prev, prevDelay := download, retryDelay download = f - t.Cleanup(func() { download = prev }) + retryDelay = func(int) time.Duration { return time.Millisecond } + t.Cleanup(func() { + download, retryDelay = prev, prevDelay + }) } // stageItems is a download step that writes each item's bytes into staging and @@ -37,10 +45,10 @@ func runFake(t *testing.T, f func(context.Context, iter.Seq2[tgsource.Item, erro // Items named in fail never reach staging. func stageItems(fail map[int]bool) func(context.Context, iter.Seq2[tgsource.Item, error], DownloadOptions) ([]Outcome, Stats, error) { return func(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o DownloadOptions) ([]Outcome, Stats, error) { - var ( - outcomes []Outcome - stats Stats - ) + var outcomes []Outcome + // Seeded exactly as the real step does, so a retry pass continues the + // run's totals instead of restarting them. + stats := o.seed for it, err := range seq { if err != nil { return outcomes, stats, err @@ -213,10 +221,10 @@ func TestRunStopsDownloadingAfterConsecutiveUploadFailures(t *testing.T) { var dispatched atomic.Int64 runFake(t, func(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o DownloadOptions) ([]Outcome, Stats, error) { - var ( - outcomes []Outcome - stats Stats - ) + var outcomes []Outcome + // Seeded exactly as the real step does, so a retry pass continues the + // run's totals instead of restarting them. + stats := o.seed for it := range seqValues(seq) { if o.stop.Load() { break @@ -305,3 +313,109 @@ func assertEmpty(t *testing.T, dir string) { t.Errorf("staging is not empty: %v", names) } } + +// A failure that clears on a later pass has to be reported as an archived file, +// not as both a failure and a success. The whole point of retrying in-run is +// that the outage which cost these items is usually over minutes later, and the +// next sync would have to re-walk the chat and re-index the remote to find out. +func TestRunRetriesFailedDownloadsWithinTheRun(t *testing.T) { + const size = 512 + items := testItems(3, size) + + var passes atomic.Int64 + runFake(t, func(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o DownloadOptions) ([]Outcome, Stats, error) { + // The middle item fails on the first pass only, as a transient outage + // looks from here. + fail := map[int]bool{2: passes.Add(1) == 1} + return stageItems(fail)(ctx, seq, o) + }) + o, staging, dstDir := runOpts(t, 0) + + res, err := Run(t.Context(), seqOf(items), o) + if err != nil { + t.Fatalf("Run: %v", err) + } + if passes.Load() != 2 { + t.Errorf("passes = %d, want 2 — the failure was not retried", passes.Load()) + } + if got := len(res.Failed()); got != 0 { + t.Errorf("Failed() = %d, want 0: %v", got, res.Failed()) + } + if len(res.Outcomes) != len(items) { + t.Errorf("Outcomes = %d, want %d — a retried item must not be reported twice", + len(res.Outcomes), len(items)) + } + // Counters follow the same rule: an item is one file to fetch however many + // attempts it took, or the closing summary claims more work than the chat + // contains. + if res.Stats.Done != 3 || res.Stats.Failed != 0 || res.Stats.Started != 3 { + t.Errorf("Stats = %+v, want 3 started, 3 done, 0 failed", res.Stats) + } + for _, it := range items { + if _, serr := os.Stat(filepath.Join(dstDir, it.Name)); serr != nil { + t.Errorf("%s should be on the destination: %v", it.Name, serr) + } + } + assertEmpty(t, staging) +} + +// Retrying is bounded. A source that is down stays down for the run, and the +// verdict has to be the distinct sentinel: a driver looping on "incomplete" +// would otherwise re-walk the whole chat forever against a dead source. +func TestRunGivesUpAndNamesTheSourceAfterEveryPassFails(t *testing.T) { + const size = 512 + items := testItems(2, size) + + var passes atomic.Int64 + runFake(t, func(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o DownloadOptions) ([]Outcome, Stats, error) { + passes.Add(1) + outcomes, stats, err := stageItems(map[int]bool{1: true, 2: true})(ctx, seq, o) + // What Download reports once its own breaker has ended the pass. + if o.onTrip != nil { + o.onTrip() + } + return outcomes, stats, err + }) + o, staging, _ := runOpts(t, 0) + o.MaxFailures = 2 + + res, err := Run(t.Context(), seqOf(items), o) + if !errors.Is(err, ErrSourceFailing) { + t.Fatalf("Run = %v, want ErrSourceFailing", err) + } + if got := passes.Load(); got != downloadAttempts { + t.Errorf("passes = %d, want %d", got, downloadAttempts) + } + if got := len(res.Failed()); got != len(items) { + t.Errorf("Failed() = %d, want %d", got, len(items)) + } + assertEmpty(t, staging) +} + +// A trip that the retry then clears must not reach the caller. Reported anyway, +// it would send a driver to the "source is down, stop" exit code on a run that +// finished everything. +func TestRunDoesNotReportASourceThatRecovered(t *testing.T) { + const size = 512 + items := testItems(2, size) + + var passes atomic.Int64 + runFake(t, func(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o DownloadOptions) ([]Outcome, Stats, error) { + first := passes.Add(1) == 1 + outcomes, stats, err := stageItems(map[int]bool{1: first, 2: first})(ctx, seq, o) + if first && o.onTrip != nil { + o.onTrip() + } + return outcomes, stats, err + }) + o, _, _ := runOpts(t, 0) + o.MaxFailures = 2 + + res, err := Run(t.Context(), seqOf(items), o) + if err != nil { + t.Fatalf("Run = %v, want nil after the retry succeeded", err) + } + if got := len(res.Failed()); got != 0 { + t.Errorf("Failed() = %d, want 0", got) + } +} diff --git a/internal/report/live.go b/internal/report/live.go index ec68f64..2f1792c 100644 --- a/internal/report/live.go +++ b/internal/report/live.go @@ -210,6 +210,13 @@ func (l *Live) UploadDone(it tgsource.Item, err error) { l.upTotal.SetCurrent(l.legs.addUpload(it.Size())) } +// Retry prints above the bars rather than through them, so the pass that is +// about to start is announced without shredding the display. +func (l *Live) Retry(attempt, files int, wait time.Duration) { + fmt.Fprintf(l.p, " retrying %s file(s) in %s — attempt %d\n", + humanCount(files), wait.Round(time.Second), attempt) +} + // Finish drains the bars and prints the closing summary. func (l *Live) Finish(s pipeline.Stats) { l.mu.Lock() diff --git a/internal/report/progress.go b/internal/report/progress.go index ee726ee..1d40912 100644 --- a/internal/report/progress.go +++ b/internal/report/progress.go @@ -111,6 +111,13 @@ func (r *Reporter) UploadDone(it tgsource.Item, err error) { fmt.Fprintf(r.w, " archived %-10s %q\n", humanBytes(it.Size()), it.Name) } +func (r *Reporter) Retry(attempt, files int, wait time.Duration) { + r.mu.Lock() + defer r.mu.Unlock() + fmt.Fprintf(r.w, " retrying %s file(s) in %s — attempt %d\n", + humanCount(files), wait.Round(time.Second), attempt) +} + // Finish writes the closing summary. func (r *Reporter) Finish(s pipeline.Stats) { r.mu.Lock()