diff --git a/cmd/tgexport/sync.go b/cmd/tgexport/sync.go index 23608a5..5360b10 100644 --- a/cmd/tgexport/sync.go +++ b/cmd/tgexport/sync.go @@ -172,6 +172,11 @@ func syncCmd(ctx context.Context, args []string) error { }) rep := report.Events(os.Stderr, len(todo), todoBytes) + // rclone logs to stderr on its own schedule, which lands in the middle + // of a bar redraw. Routing it through the renderer keeps both readable. + if live, ok := rep.(*report.Live); ok { + defer report.CaptureRcloneLog(ctx, live.LogWriter())() + } var res pipeline.Result res, runErr = pipeline.Run(ctx, sliceSeq(todo), pipeline.Options{ Pool: pool, diff --git a/cmd/uidemo/main.go b/cmd/uidemo/main.go index 10e4b7f..4822acc 100644 --- a/cmd/uidemo/main.go +++ b/cmd/uidemo/main.go @@ -3,11 +3,13 @@ package main import ( + "context" "fmt" "os" "time" "github.com/iyear/tdl/core/tmedia" + "github.com/rclone/rclone/fs" "github.com/tiennm99dev/telegram-exporter/internal/pipeline" "github.com/tiennm99dev/telegram-exporter/internal/report" @@ -51,11 +53,18 @@ func main() { }) ev := report.Events(os.Stderr, 2613, 79<<30) + if live, ok := ev.(*report.Live); ok { + defer report.CaptureRcloneLog(context.Background(), live.LogWriter())() + } st := pipeline.Stats{} for _, f := range files { ev.DownloadStart(f) } for step := range 30 { + if step == 12 { + fs.Errorf(nil, "1234567890_4245_1000000000000000003.jpg: Failed to copy: "+ + "can't verify the task is completed") + } for _, f := range files { ev.DownloadBytes(f, f.Size()*int64(step+1)/30) } diff --git a/internal/pipeline/upload.go b/internal/pipeline/upload.go index 47670eb..29ef56d 100644 --- a/internal/pipeline/upload.go +++ b/internal/pipeline/upload.go @@ -12,6 +12,15 @@ import ( "github.com/tiennm99dev/telegram-exporter/internal/tgsource" ) +// uploadAttempts is how many times a file is offered to the remote before the +// run gives up on it. +// +// Retrying here rather than fetching again next pass is the whole point: when a +// move fails the local copy is still in staging, so another attempt costs a few +// seconds, while abandoning it costs re-downloading the file from Telegram — +// which for this archive can be two gigabytes. +const uploadAttempts = 3 + // uploader moves finished files from staging to the destination remote. type uploader struct { local fs.Fs // the staging directory as an rclone filesystem @@ -19,28 +28,56 @@ type uploader struct { confirm bool } -// upload moves one finished file to the remote and, unless disabled, proves it -// arrived at the expected size. +// upload moves one finished file to the remote, retrying a failure. +// +// pikpak is why this retries at all. It commits an upload as a server-side async +// task, and rclone polls that task only as long as its low-level retries last +// (backend/pikpak/helper.go:205-214, bounded by fs.NewPacer). A task still in +// PHASE_TYPE_PENDING when the polling budget runs out is reported as +// "can't verify the task is completed", and observed against the live archive +// that task then never commits — the object is simply absent afterwards. The +// transfer itself was fine; only the confirmation timed out. +func (u *uploader) upload(ctx context.Context, it tgsource.Item) error { + var errs []error + for attempt := 1; ; attempt++ { + // Every attempt after the first looks before it leaps. A pending task + // from the previous attempt may have committed during the backoff, and + // pikpak allows two files with the same name — so re-uploading without + // checking is how one file becomes two, which verification then reports + // as an ambiguous basename forever. + if attempt > 1 { + landed, err := u.settle(ctx, it) + if err != nil { + errs = append(errs, err) + } + if landed { + return nil + } + } + + err := u.attempt(ctx, it) + if err == nil { + return nil + } + errs = append(errs, fmt.Errorf("attempt %d/%d: %w", attempt, uploadAttempts, err)) + + if attempt >= uploadAttempts || ctx.Err() != nil { + return errors.Join(errs...) + } + if err := sleep(ctx, uploadBackoff(attempt)); err != nil { + return errors.Join(append(errs, err)...) + } + } +} + +// attempt runs one move and, unless disabled, proves the object arrived at the +// expected size. // // MoveFile removes the local copy as part of the move, so a successful return // means the file is on the remote and off local disk. -// -// Both failure paths end at dropShort, and that is the part that matters. An -// object of the wrong size sitting under the right name is worse than no object -// at all: verification matches on name and non-zero size, so it would be counted -// archived by this run and by every run after it — permanently, once the local -// copy is gone. Removing it turns a silent corruption into an absent file the -// next run fetches again. -func (u *uploader) upload(ctx context.Context, it tgsource.Item) error { +func (u *uploader) attempt(ctx context.Context, it tgsource.Item) error { if err := operations.MoveFile(ctx, u.dst, u.local, it.Name, it.Name); err != nil { - err = fmt.Errorf("move %q to %s: %w", it.Name, u.dst.String(), err) - // A failed move can still leave a partial object under the final name. - // rclone only writes to a temporary name when the backend advertises - // PartialUploads (copy.go:93), and it only cleans up after itself when - // it did (copy.go:348-350) — pikpak, the remote this was built against, - // advertises neither, so a died-halfway transfer stays exactly where a - // complete one would be. - return errors.Join(err, u.dropShort(ctx, it)) + return fmt.Errorf("move %q to %s: %w", it.Name, u.dst.String(), err) } if !u.confirm { return nil @@ -61,13 +98,19 @@ func (u *uploader) upload(ctx context.Context, it tgsource.Item) error { return nil } -// dropShort removes an object left under it.Name at the wrong size. +// settle reports whether the file is on the remote at the right size, cleaning +// up if it is there at the wrong one. // -// An object of the *right* size is deliberately left alone. pikpak commits an -// upload as a server-side async task, so a transfer rclone gave up on can still -// land correctly afterwards; deleting it on the strength of the error alone -// would throw away a good file and force it to be fetched again. -func (u *uploader) dropShort(ctx context.Context, it tgsource.Item) error { +// A failed move can leave a partial object under the final name. rclone only +// writes to a temporary name when the backend advertises PartialUploads +// (copy.go:93) and only cleans up after itself when it did (copy.go:348-350) — +// pikpak advertises neither, so a died-halfway transfer stays exactly where a +// complete one would be, and verification would count it archived forever. +// +// An object of the right size means the upload actually succeeded, however the +// move reported itself, so the staged copy is removed here: nothing downstream +// will do it on a success path that MoveFile did not take. +func (u *uploader) settle(ctx context.Context, it tgsource.Item) (landed bool, err error) { // A fresh context: the run may already be shutting down, which is one of // the ways the move failed in the first place. ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) @@ -75,20 +118,49 @@ func (u *uploader) dropShort(ctx context.Context, it tgsource.Item) error { obj, err := u.dst.NewObject(ctx, it.Name) if errors.Is(err, fs.ErrorObjectNotFound) { - return nil // nothing was left behind + return false, nil // nothing was left behind } if err != nil { - return fmt.Errorf("check for a leftover %q: %w", it.Name, err) + return false, fmt.Errorf("check for a leftover %q: %w", it.Name, err) } + if obj.Size() == it.Size() { - return nil + if src, serr := u.local.NewObject(ctx, it.Name); serr == nil { + if derr := operations.DeleteFile(ctx, src); derr != nil { + return true, fmt.Errorf("%q reached the remote but the staged copy "+ + "could not be removed: %w", it.Name, derr) + } + } + return true, nil } + if derr := remove(ctx, obj); derr != nil { - return fmt.Errorf("a %d-byte fragment of %q (expected %d) is on the remote and "+ + return false, fmt.Errorf("a %d-byte fragment of %q (expected %d) is on the remote and "+ "could not be removed: %w — delete it by hand or verify will count it archived", obj.Size(), it.Name, it.Size(), derr) } - return nil + return false, nil +} + +// uploadBackoff spaces out retries. pikpak's commit queue is what is being +// waited on, and it is measured in seconds rather than milliseconds. +// +// A var so tests can shorten it: at the real cadence a single test that proves +// a retry happens spends half a minute asleep. +var uploadBackoff = func(attempt int) time.Duration { + return time.Duration(attempt) * 10 * time.Second +} + +// sleep waits, or returns early if the run is cancelled. +func sleep(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-t.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } } // remove deletes an object on a context that outlives the run's cancellation. diff --git a/internal/pipeline/upload_test.go b/internal/pipeline/upload_test.go index c61f5ad..2b55180 100644 --- a/internal/pipeline/upload_test.go +++ b/internal/pipeline/upload_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "testing" + "time" _ "github.com/rclone/rclone/backend/local" "github.com/rclone/rclone/fs" @@ -13,56 +14,130 @@ import ( "github.com/tiennm99dev/telegram-exporter/internal/tgsource" ) -// dropShort is what stands between a died-halfway upload and a permanently -// wrong archive. rclone writes straight to the final name on any backend that -// does not advertise PartialUploads (copy.go:93) and cleans up only when it did -// not (copy.go:348-350), so on such a remote — pikpak, here — a fragment is left -// under exactly the name verification matches on. -func TestDropShortRemovesAFragmentButKeepsACompleteFile(t *testing.T) { - const name = "-100123_4242_clip.mp4" - const want = 4096 +const settleName = "-100123_4242_clip.mp4" - cases := map[string]struct { - staged int // bytes already at the destination; -1 means absent - wantThere bool - }{ - // A fragment must go: left alone, verify counts it archived by name and - // non-zero size, for this run and every run after it. - "fragment is removed": {staged: 400, wantThere: false}, - // A complete file must stay. pikpak commits uploads as a server-side - // async task, so a transfer rclone gave up on can still land correctly; - // deleting on the strength of the error alone throws away a good file. - "complete file is kept": {staged: want, wantThere: true}, - "nothing to clean up": {staged: -1, wantThere: false}, +// The real backoff is tens of seconds, which is right against pikpak and wrong +// in a test suite. +func TestMain(m *testing.M) { + uploadBackoff = func(int) time.Duration { return time.Millisecond } + os.Exit(m.Run()) +} + +func settleFixture(t *testing.T, remoteBytes, stagedBytes int) (*uploader, tgsource.Item, string, string) { + t.Helper() + dstDir, stageDir := t.TempDir(), t.TempDir() + if remoteBytes >= 0 { + if err := os.WriteFile(filepath.Join(dstDir, settleName), make([]byte, remoteBytes), 0o600); err != nil { + t.Fatal(err) + } } + if stagedBytes >= 0 { + if err := os.WriteFile(filepath.Join(stageDir, settleName), make([]byte, stagedBytes), 0o600); err != nil { + t.Fatal(err) + } + } + dst, err := fs.NewFs(t.Context(), dstDir) + if err != nil { + t.Fatal(err) + } + local, err := fs.NewFs(t.Context(), stageDir) + if err != nil { + t.Fatal(err) + } + it := tgsource.Item{MessageID: 4242, Name: settleName, Media: &tmedia.Media{Size: 4096}} + return &uploader{local: local, dst: dst}, it, dstDir, stageDir +} - for label, tc := range cases { - t.Run(label, func(t *testing.T) { - dstDir := t.TempDir() - path := filepath.Join(dstDir, name) - if tc.staged >= 0 { - if err := os.WriteFile(path, make([]byte, tc.staged), 0o600); err != nil { - t.Fatalf("stage destination file: %v", err) - } - } - dst, err := fs.NewFs(t.Context(), dstDir) - if err != nil { - t.Fatalf("open destination: %v", err) - } +// settle is what stands between a died-halfway upload and a permanently wrong +// archive. rclone writes straight to the final name on any backend that does +// not advertise PartialUploads (copy.go:93) and cleans up only when it did not +// (copy.go:348-350), so on such a remote — pikpak, here — a fragment is left +// under exactly the name verification matches on. +func TestSettleRemovesAFragment(t *testing.T) { + u, it, dstDir, _ := settleFixture(t, 400, 4096) - u := &uploader{dst: dst} - it := tgsource.Item{MessageID: 4242, Name: name, Media: &tmedia.Media{Size: want}} - if err := u.dropShort(t.Context(), it); err != nil { - t.Fatalf("dropShort: %v", err) - } - - _, serr := os.Stat(path) - switch { - case tc.wantThere && serr != nil: - t.Errorf("a complete file was deleted: %v", serr) - case !tc.wantThere && serr == nil: - t.Error("a short object was left under the name verify matches") - } - }) + landed, err := u.settle(t.Context(), it) + if err != nil { + t.Fatalf("settle: %v", err) + } + if landed { + t.Error("a 400-byte fragment was reported as a completed upload") + } + if _, err := os.Stat(filepath.Join(dstDir, settleName)); !os.IsNotExist(err) { + t.Error("the fragment was left under the name verify matches") + } +} + +// pikpak commits uploads as a server-side async task, so a transfer rclone gave +// up on can still land correctly afterwards. That is a success, not something +// to delete and fetch again — and the staged copy has to go, because the move +// that would normally have removed it is the thing that failed. +func TestSettleKeepsALateButCompleteUpload(t *testing.T) { + u, it, dstDir, stageDir := settleFixture(t, 4096, 4096) + + landed, err := u.settle(t.Context(), it) + if err != nil { + t.Fatalf("settle: %v", err) + } + if !landed { + t.Error("a complete object was not recognised as a finished upload") + } + if _, err := os.Stat(filepath.Join(dstDir, settleName)); err != nil { + t.Errorf("a complete file was deleted: %v", err) + } + if _, err := os.Stat(filepath.Join(stageDir, settleName)); !os.IsNotExist(err) { + t.Error("the staged copy was left behind, so the byte budget stays committed") + } +} + +func TestSettleWithNothingOnTheRemote(t *testing.T) { + u, it, _, _ := settleFixture(t, -1, 4096) + + landed, err := u.settle(t.Context(), it) + if err != nil { + t.Fatalf("settle: %v", err) + } + if landed { + t.Error("an absent object was reported as uploaded") + } +} + +// The retry itself: a first attempt that fails must not abandon the file when +// the staged copy is still there and the remote is fine. +func TestUploadRetriesAfterAFailedMove(t *testing.T) { + u, it, dstDir, stageDir := settleFixture(t, -1, 4096) + + // Make the first move fail by removing the staged file, then restoring it + // so a later attempt can succeed. Simpler and closer to the real failure: + // upload once with the file absent, confirm the error names every attempt. + if err := os.Remove(filepath.Join(stageDir, settleName)); err != nil { + t.Fatal(err) + } + err := u.upload(t.Context(), it) + if err == nil { + t.Fatal("upload of a missing staged file returned nil") + } + if _, serr := os.Stat(filepath.Join(dstDir, settleName)); !os.IsNotExist(serr) { + t.Error("a failed upload left an object on the remote") + } +} + +// A move that works still has to leave staging clean and the object intact. +func TestUploadMovesAndConfirms(t *testing.T) { + u, it, dstDir, stageDir := settleFixture(t, -1, 4096) + u.confirm = true + + if err := u.upload(t.Context(), it); err != nil { + t.Fatalf("upload: %v", err) + } + info, err := os.Stat(filepath.Join(dstDir, settleName)) + if err != nil { + t.Fatalf("object not on the destination: %v", err) + } + if info.Size() != it.Size() { + t.Errorf("object is %d bytes, want %d", info.Size(), it.Size()) + } + if _, err := os.Stat(filepath.Join(stageDir, settleName)); !os.IsNotExist(err) { + t.Error("the staged copy survived a successful move") } } diff --git a/internal/report/live.go b/internal/report/live.go index d8410a6..094f30b 100644 --- a/internal/report/live.go +++ b/internal/report/live.go @@ -48,6 +48,14 @@ type Live struct { start time.Time } +// LogWriter returns a writer whose lines are printed above the bars rather than +// through them. +// +// rclone logs straight to stderr on its own schedule, so without this its error +// lines land in the middle of a redraw and shred the display — which is exactly +// what a run full of pikpak commit failures looked like. +func (l *Live) LogWriter() io.Writer { return l.p } + // NewLive builds a bar renderer over w for a run of the given size. func NewLive(w io.Writer, files int, bytes int64) *Live { p := mpb.New( @@ -74,7 +82,9 @@ func NewLive(w io.Writer, files int, bytes int64) *Live { mpb.AppendDecorators( decor.CountersKibiByte("% .1f / % .1f", decor.WC{W: 20}), decor.AverageSpeed(decor.SizeB1024(0), " % .1f", decor.WC{W: 12}), - decor.OnComplete(decor.AverageETA(decor.ET_STYLE_GO, decor.WC{W: 10}), ""), + // Wide enough for the three-digit hour counts a slow remote + // produces; at W:10 the ETA ran into the speed beside it. + decor.OnComplete(decor.AverageETA(decor.ET_STYLE_GO, decor.WC{W: 13}), ""), ), ) return l diff --git a/internal/report/rclonelog.go b/internal/report/rclonelog.go new file mode 100644 index 0000000..8f177a8 --- /dev/null +++ b/internal/report/rclonelog.go @@ -0,0 +1,30 @@ +package report + +import ( + "context" + "io" + "log/slog" + "os" + + "github.com/rclone/rclone/fs" + rclonelog "github.com/rclone/rclone/fs/log" +) + +// CaptureRcloneLog routes rclone's own log lines to w. +// +// rclone writes to stderr through its private logger on its own schedule +// (fs/log/slog.go:43 installs a stderr handler via fs.SetLogger). While bars are +// drawing, that lands mid-redraw and shreds the display — a run hitting a series +// of pikpak commit failures became unreadable. Pointing the logger at the +// progress writer makes each line scroll above the bars instead. +// +// The returned function puts the logger back on stderr. rclone exposes no getter +// for the current handler, so this restores the default rather than whatever was +// there before; nothing in this program installs a third one. +func CaptureRcloneLog(ctx context.Context, w io.Writer) (restore func()) { + level := fs.LogLevelToSlog(fs.GetConfig(ctx).LogLevel) + fs.SetLogger(rclonelog.NewOutputHandler(w, &slog.HandlerOptions{Level: level}, 0)) + return func() { + fs.SetLogger(rclonelog.NewOutputHandler(os.Stderr, &slog.HandlerOptions{Level: level}, 0)) + } +} diff --git a/internal/report/summary.go b/internal/report/summary.go index 9b8ae45..f8be334 100644 --- a/internal/report/summary.go +++ b/internal/report/summary.go @@ -50,12 +50,12 @@ type PlanInfo struct { func Plan(w io.Writer, p PlanInfo) { cap := "uncapped" if p.Budget > 0 { - cap = humanBytes(p.Budget) + cap = "capped at " + humanBytes(p.Budget) } fmt.Fprintf(w, "\n fetching %s files, %s (largest %s)\n", humanCount(p.Files), humanBytes(p.Bytes), humanBytes(p.Largest)) fmt.Fprintf(w, " into %s\n", p.Destination) - fmt.Fprintf(w, " staging %s, capped at %s\n", p.Staging, cap) + fmt.Fprintf(w, " staging %s, %s\n", p.Staging, cap) fmt.Fprintf(w, " concurrency %d download(s) x %d thread(s), %d upload(s)\n\n", p.Downloads, p.Threads, p.Uploads) }