fix: retry a failed upload instead of abandoning the file

pikpak commits an upload as a server-side async task, and rclone polls that
task only as long as its low-level retries last. A task still in
PHASE_TYPE_PENDING when that budget runs out is reported as "can't verify the
task is completed" — and checking the live archive ten minutes later, those
tasks had not committed: the objects were simply absent.

The transfer itself was fine; only the confirmation timed out. Abandoning the
file meant re-downloading it from Telegram on the next pass, which for this
archive can be two gigabytes, so a run against a slow remote paid for the same
bytes repeatedly. The staged copy is still on disk when a move fails — rclone
says so in the same breath — so another attempt costs seconds instead.

Every attempt after the first checks the remote before re-uploading. A pending
task may have committed during the backoff, and pikpak allows two files under
one name, so re-uploading blind is how one file becomes two — which
verification then reports as an ambiguous basename on every future run.

rclone's log now goes through the renderer. It writes to stderr on its own
schedule, so its error lines were landing mid-redraw and shredding the display
exactly when there was most to read.

Also: "capped at uncapped" now reads "uncapped", and the ETA column fits the
three-digit hour counts a slow remote produces.

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,

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)
 		}

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.

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")
 	}
 }

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

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))
+	}
+}

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)
 }