diff --git a/.claude/agent-memory/code-reviewer/MEMORY.md b/.claude/agent-memory/code-reviewer/MEMORY.md new file mode 100644 index 0000000..7098a1a --- /dev/null +++ b/.claude/agent-memory/code-reviewer/MEMORY.md @@ -0,0 +1,4 @@ +# Agent Memory Index + +- [rclone embedding traps](rclone-embedding-traps.md) — `NewFilter(nil)` is not neutral; `RCLONE_*` reaches globals at package init, so `t.Setenv` tests prove nothing. +- [Exit-code contract](repo-exit-code-contract.md) — exit 1 is the driver's retry signal; a non-retryable state that exits 1 loops forever. diff --git a/.claude/agent-memory/code-reviewer/rclone-embedding-traps.md b/.claude/agent-memory/code-reviewer/rclone-embedding-traps.md new file mode 100644 index 0000000..5ebc642 --- /dev/null +++ b/.claude/agent-memory/code-reviewer/rclone-embedding-traps.md @@ -0,0 +1,36 @@ +--- +name: rclone-embedding-traps +description: Non-obvious rclone-as-a-library behaviours verified against rclone v1.75.1 source that keep biting this repo's remote/index code +metadata: + type: project +--- + +rclone v1.75.1 embedded as a library reads `RCLONE_*` environment variables into +package-level global option structs at **package init**, via +`fs.RegisterGlobalOptions` -> `OptionsInfo.load()` -> `fs.ConfigMap` -> +`optionEnvVars.Get` (`fs/registry.go:498-527`, `fs/configmap.go:115-152`). No +cobra/pflag wiring is needed for this to happen. + +Two consequences that are easy to get backwards, both verified empirically: + +- `filter.NewFilter(nil)` copies `filter.Opt` (`fs/filter/filter.go:196-202`), + which already holds the env-derived values. It is **not** a neutral filter. + `RCLONE_EXCLUDE` / `RCLONE_FILTER` / `RCLONE_MIN_SIZE` / `RCLONE_MAX_AGE` all + survive it. A genuinely neutral filter needs explicit + `&filter.Options{MinAge: fs.DurationOff, MaxAge: fs.DurationOff, MinSize: -1, MaxSize: -1}` + — the zero-value `filter.Options{}` errors with "min-age can't be larger than + max-age". +- `fs.AddConfig(ctx)` + assigning `ci.MaxDepth` *does* override + `RCLONE_MAX_DEPTH`, and env values for `Transfers` / `LowLevelRetries` really + are present in `fs.GetConfig(ctx)` without any flag parsing. + +**Why:** an index built through a narrowed listing does not fail — it reports +archived files as absent and re-downloads them, which is the exact multi-GiB +failure this project exists to remove. + +**How to apply:** when reviewing anything that lists a remote, check the filter +is neutralised explicitly rather than with `NewFilter(nil)`, and prove it with a +subprocess run under the env var rather than a `t.Setenv` test — `t.Setenv` runs +long after rclone's package init, so such a test passes regardless. + +Related: [[repo-exit-code-contract]] diff --git a/.claude/agent-memory/code-reviewer/repo-exit-code-contract.md b/.claude/agent-memory/code-reviewer/repo-exit-code-contract.md new file mode 100644 index 0000000..724fe7e --- /dev/null +++ b/.claude/agent-memory/code-reviewer/repo-exit-code-contract.md @@ -0,0 +1,25 @@ +--- +name: repo-exit-code-contract +description: tgexport's exit codes are a machine contract an external until-complete driver loops on; miscategorising one causes an infinite retry loop +metadata: + type: project +--- + +`tgexport` exit codes: 0 complete, 1 ran but files remain, 2 usage, 3 +remote/Telegram failure, 130/143 interrupted. Code 1 is the *retry* signal — an +external driver re-invokes on 1 and stops on 0 or 3. + +**Why:** the retired `export-until-complete.sh` had two stop conditions the Go +rewrite deliberately dropped (`plans/.../phase-06-cli-resume-and-observability.md` +lines 46-57): a pass counter, and a no-progress detector. The Go tool converges +in one invocation instead, so the exit code is now the *only* thing that can +stop a driver. + +**How to apply:** when reviewing any new error path in `cmd/tgexport`, ask which +exit code it produces and whether that state is actually retryable. Two states +have no representation in the contract and therefore fall through to 1 forever: +work that can never succeed (an unwritable filename, permanently unavailable +media) and a remote that is full or broken mid-run. Treat "exit 1 with no +possible progress" as a blocking defect, not a cosmetic one. + +Related: [[rclone-embedding-traps]] diff --git a/internal/naming/safe.go b/internal/naming/safe.go index feebf6f..a4e7c72 100644 --- a/internal/naming/safe.go +++ b/internal/naming/safe.go @@ -5,11 +5,23 @@ import ( "path/filepath" "strconv" "strings" + + "github.com/rclone/rclone/lib/encoder" ) -// maxNameBytes is NAME_MAX on Linux: the longest single path component ext4 and -// friends accept. It is a byte count, not a rune count. -const maxNameBytes = 255 +// maxNameBytes is the longest stored name that can actually be written. +// +// NAME_MAX on Linux is 255 bytes for a single path component, but the name is +// not what lands on disk first: a download is written to name+".part" and +// renamed afterwards, so the suffix has to fit inside the limit too. Checking +// the bare 255 would pass a name whose part file then fails with ENAMETOOLONG — +// after the item had already been queued, which stalls the walk on the same +// message on every pass. +const maxNameBytes = 255 - len(PartSuffix) + +// PartSuffix marks a download still in flight. It lives here because Safe's +// length limit has to account for it. +const PartSuffix = ".part" // Safe reports whether a stored name can be joined onto a directory path. // @@ -48,6 +60,26 @@ func Safe(name string) error { // Catches embedded separators, trailing slashes, and any ".." segment, // since Base of all of those differs from the original. return fmt.Errorf("filename is not a single path element: %q", name) + case encoder.OS.FromStandardName(name) != name: + // rclone does not address files by the bytes on disk. Every name given + // to an Fs is run through the backend's encoder, and every name listed + // back is re-encoded to the standard set — neither of which os.OpenFile + // performs. So a name containing one of the characters those encoders + // rewrite is written verbatim, then looked up under a different string: + // the upload fails with "object not found" forever, or it succeeds and + // the index records a name the presence check will never match. + // + // That is the original bug exactly — one name derived two ways — with + // rclone's encoder in the place filenamify used to occupy. Rejecting + // rather than encoding is the same choice made everywhere else here: + // encoding would give the two derivations a chance to disagree again. + return fmt.Errorf("filename is rewritten by rclone's path encoder: %q", name) + case encoder.Standard.Encode(encoder.Standard.Decode(name)) != name: + // The listing side of the same problem, and it is not backend-specific: + // the re-encode to the standard set happens above the backend encoder, + // so it applies to every remote. Control characters and DEL are the + // common case, and both are trivially settable in a Telegram filename. + return fmt.Errorf("filename is rewritten when rclone lists it back: %q", name) } return nil } diff --git a/internal/naming/safe_test.go b/internal/naming/safe_test.go index 531202d..766b873 100644 --- a/internal/naming/safe_test.go +++ b/internal/naming/safe_test.go @@ -1,6 +1,8 @@ package naming import ( + "fmt" + "os" "path/filepath" "strings" "testing" @@ -120,21 +122,30 @@ func TestSplitStoredDoesNotMatchPrefixOverlap(t *testing.T) { // An over-long name is the one input that can hang a drive-until-complete loop: // os.Create rejects it, so the download fails forever while verify keeps // reporting it absent. It must be refused up front, not discovered per pass. +// +// The limit leaves room for the ".part" suffix, because that is what is opened +// first. A name that fits in NAME_MAX but whose part file does not would pass +// this check, be queued, and then fail to open every single pass. func TestSafeRejectsNamesOverTheFilesystemLimit(t *testing.T) { prefix := "1234567890_42_" - fill := 255 - len(prefix) + fill := maxNameBytes - len(prefix) atLimit := prefix + strings.Repeat("a", fill) if err := Safe(atLimit); err != nil { t.Errorf("Safe(%d bytes) = %v, want nil at exactly the limit", len(atLimit), err) } + // The part file for a name at the limit must actually be creatable, which is + // the property the limit exists to guarantee. + if err := os.WriteFile(filepath.Join(t.TempDir(), atLimit+PartSuffix), nil, 0o600); err != nil { + t.Errorf("a name Safe accepted cannot be written as a part file: %v", err) + } overLimit := prefix + strings.Repeat("a", fill+1) err := Safe(overLimit) if err == nil { t.Fatalf("Safe(%d bytes) = nil, want an error past the limit", len(overLimit)) } - if !strings.Contains(err.Error(), "over the 255-byte limit") { + if !strings.Contains(err.Error(), fmt.Sprintf("over the %d-byte limit", maxNameBytes)) { t.Errorf("error should name the limit, got: %v", err) } @@ -145,3 +156,45 @@ func TestSafeRejectsNamesOverTheFilesystemLimit(t *testing.T) { len(multibyte), len([]rune(multibyte))) } } + +// rclone addresses files by an encoded name, not by the bytes os.OpenFile +// wrote. A name either encoder rewrites is the original two-derivations bug in +// a new place: written verbatim, then looked up or listed under a different +// string, so the file is re-fetched on every pass forever. +// +// These were confirmed against the real local backend before the check existed: +// NewObject on a written "a‛b.jpg" reported "object not found", and a written +// "a\nb.jpg" listed back as "a␊b.jpg". +func TestSafeRejectsNamesRcloneRewrites(t *testing.T) { + rejected := map[string]string{ + "the encoder's own escape character": "1234567890_42_a\u201bb.jpg", + "a symbol-for-control glyph": "1234567890_42_a\u2421b.jpg", + "a raw newline": "1234567890_42_a\nb.jpg", + "a raw DEL": "1234567890_42_a\x7fb.jpg", + "a raw control byte": "1234567890_42_a\x01b.jpg", + } + for label, name := range rejected { + t.Run(label, func(t *testing.T) { + if err := Safe(name); err == nil { + t.Errorf("Safe(%q) = nil; rclone rewrites this name", name) + } + }) + } + + // Rejecting too much would be its own bug: these are ordinary Telegram + // filenames and every one must survive. + accepted := []string{ + "1234567890_42_ünïcödé 12-08 🍓.mp4", + "1234567890_42_a#b%c!d[e]{f}.mp4", + "1234567890_42_ㅋㅋㅋ 😀.png", + "1234567890_42_a/b.jpg", // fullwidth solidus, not a separator + "1234567890_42_trailing. ", + "1234567890_42_'quoted' \"double\".mp4", + "1234567890_42_ünïcödé, spaces & commas.webm", + } + for _, name := range accepted { + if err := Safe(name); err != nil { + t.Errorf("Safe(%q) = %v, want nil for an ordinary filename", name, err) + } + } +} diff --git a/internal/pipeline/elem.go b/internal/pipeline/elem.go index 4738202..7e24c63 100644 --- a/internal/pipeline/elem.go +++ b/internal/pipeline/elem.go @@ -24,7 +24,11 @@ import ( // because upload is triggered by a download returning rather than by a filter // over a directory. A distinct suffix just keeps a staging directory shared with // a legacy tdl run unambiguous during the cutover. -const partSuffix = ".part" +// +// It is defined in naming because naming.Safe's length limit has to leave room +// for it — a name that fits but whose part file does not would pass the check +// and then fail to open, stalling the walk on that message forever. +const partSuffix = naming.PartSuffix // elem adapts one media item to the downloader's element interface. type elem struct { diff --git a/internal/remote/index.go b/internal/remote/index.go index 40acc9c..4e97e5c 100644 --- a/internal/remote/index.go +++ b/internal/remote/index.go @@ -54,14 +54,32 @@ type Index struct { // a narrowed listing here does not fail — it silently reports archived files as // absent and re-downloads every one of them. The transfer tunables in Init are // deliberately env-overridable; this is not. +// +// Note that filter.NewFilter(nil) does NOT give a neutral filter: it copies the +// package-level filter.Opt (filter.go:198-201), which rclone has already +// populated from RCLONE_* at init via RegisterGlobalOptions. Passing nil here +// reproduces exactly the inherited filter this is trying to escape, which is +// why every field that can narrow a listing is set explicitly. A zero-value +// Options is not a substitute either — it fails validation, because MinAge and +// MaxAge both being 0 reads as "min > max". func BuildIndex(ctx context.Context, f fs.Fs, dialogID int64) (*Index, error) { ctx, ci := fs.AddConfig(ctx) ci.MaxDepth = -1 - unfiltered, err := filter.NewFilter(nil) + unfiltered, err := filter.NewFilter(&filter.Options{ + MinAge: fs.DurationOff, + MaxAge: fs.DurationOff, + MinSize: -1, + MaxSize: -1, + }) if err != nil { return nil, fmt.Errorf("build an empty filter: %w", err) } + if !unfiltered.InActive() { + // Cheap and worth keeping: this is the assertion whose absence let a + // no-op neutralisation stand. + return nil, fmt.Errorf("internal: listing filter is not neutral") + } ctx = filter.ReplaceConfig(ctx, unfiltered) idx := &Index{ @@ -92,7 +110,13 @@ func BuildIndex(ctx context.Context, f fs.Fs, dialogID int64) (*Index, error) { // A destination that does not exist yet holds nothing. That is an empty // index, not a failure — it is what a first run against a new path looks // like, and treating it as an error would make verify unusable there. - if errors.Is(err, fs.ErrorDirNotFound) { + // + // Only when nothing was listed, though. rclone's walk records a failed + // directory and keeps going (walk.go:168-183), returning the error at the + // end, so this same error also means "one subdirectory could not be + // listed" — and swallowing that would return a partial index as + // authoritative, reporting everything under it absent. + if errors.Is(err, fs.ErrorDirNotFound) && len(idx.byName) == 0 { return idx, nil } return nil, fmt.Errorf("list %s: %w", f.String(), err) diff --git a/internal/remote/index_env_test.go b/internal/remote/index_env_test.go new file mode 100644 index 0000000..1fb4124 --- /dev/null +++ b/internal/remote/index_env_test.go @@ -0,0 +1,64 @@ +package remote + +import ( + "os" + "os/exec" + "path/filepath" + "testing" + + _ "github.com/rclone/rclone/backend/local" + "github.com/rclone/rclone/fs" +) + +// rclone reads RCLONE_* into its global config at package init, so an in-process +// t.Setenv is too late to prove anything. A subprocess is the only way to +// observe what BuildIndex actually does under an operator's environment — which +// is exactly why the original no-op neutralisation went unnoticed. +func TestBuildIndexIgnoresInheritedFilters(t *testing.T) { + if os.Getenv("GO_INDEX_ENV_CHILD") == "1" { + indexChild(t) + return + } + for _, env := range []string{ + "RCLONE_EXCLUDE=*.mp4", + "RCLONE_FILTER=- *.mp4", + "RCLONE_MIN_SIZE=1M", + "RCLONE_MAX_AGE=1h", + "RCLONE_MAX_DEPTH=1", + } { + t.Run(env, func(t *testing.T) { + cmd := exec.Command(os.Args[0], "-test.run=TestBuildIndexIgnoresInheritedFilters", "-test.v") + cmd.Env = append(os.Environ(), "GO_INDEX_ENV_CHILD=1", env) + out, err := cmd.CombinedOutput() + if err != nil { + t.Errorf("%s narrowed the index:\n%s", env, out) + } + }) + } +} + +func indexChild(t *testing.T) { + dir := t.TempDir() + for _, rel := range []string{"a.mp4", "b.txt", "sub/c.mp4"} { + p := filepath.Join(dir, rel) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte("x"), 0o600); err != nil { + t.Fatal(err) + } + } + f, err := fs.NewFs(t.Context(), dir) + if err != nil { + t.Fatal(err) + } + idx, err := BuildIndex(t.Context(), f, 1234567890) + if err != nil { + t.Fatalf("BuildIndex: %v", err) + } + for _, want := range []string{"a.mp4", "b.txt", "c.mp4"} { + if _, ok := idx.Lookup(want); !ok { + t.Errorf("%q missing from the index", want) + } + } +} diff --git a/internal/tgsource/chat.go b/internal/tgsource/chat.go index 7729c84..9e913c4 100644 --- a/internal/tgsource/chat.go +++ b/internal/tgsource/chat.go @@ -26,10 +26,10 @@ var telegramHosts = []string{"t.me/", "telegram.me/", "telegram.dog/"} // // This is parsed rather than pattern-matched because the two HTTPS shapes overlap // in a way a regex gets wrong: a private link is t.me/c// and a public -// one is t.me//, so "t.me/c/1234567890" — a perfectly good private -// *channel* link — looks exactly like a public message link with the username -// "c". The distinction is whether a trailing numeric component follows the chat, -// and where that component sits depends on the "c" marker. +// one is t.me//, so "t.me/c/1234567890" — a private channel link — +// looks exactly like a public message link with the username "c". The +// distinction is whether a trailing numeric component follows the chat, and +// where that component sits depends on the leading marker. func isMessageLink(s string) bool { lower := strings.ToLower(s) @@ -51,29 +51,82 @@ func isMessageLink(s string) bool { rest, _, _ = strings.Cut(rest, "#") parts := strings.Split(strings.Trim(rest, "/"), "/") - if len(parts) >= 1 && parts[0] == "c" { - // c/ is the channel; c// is one message in it. + if len(parts) >= 1 && (parts[0] == "c" || parts[0] == "s") { + // Both put the chat in the second component, so a message is the third: + // c// and s//. return len(parts) >= 3 && isDigits(parts[2]) } - // t.me/joinchat/ and t.me/s/ are chats, not messages, and their - // second component is not a bare number — except for a hypothetical all-digit - // invite hash, which is not worth mis-parsing every real link to guard. - if len(parts) >= 1 && (parts[0] == "joinchat" || parts[0] == "s") { + // t.me/joinchat/ is a chat, and its second component is not a bare + // number — except for a hypothetical all-digit invite hash, which is not + // worth mis-parsing every real link to guard. + if len(parts) >= 1 && parts[0] == "joinchat" { return false } return len(parts) >= 2 && isDigits(parts[1]) } -// afterHost returns the path following a Telegram host, if s names one. -func afterHost(lower string) (string, bool) { - for _, host := range telegramHosts { - if i := strings.Index(lower, host); i >= 0 { - return lower[i+len(host):], true +// linkChat extracts the chat from a link whose first path component is a marker +// rather than the chat itself. +// +// Without this the marker *is* the chat as far as the resolver is concerned. +// gotd's deeplink parser takes the first path component as the domain and drops +// the rest (deeplink.go:106-148), and ValidateDomain accepts a single letter, so +// "t.me/s/mychannel" resolves the username "s" — either a hard-to-read +// USERNAME_NOT_OCCUPIED, or, since one-character usernames exist, somebody +// else's chat archived into the operator's remote. +// +// t.me/s/ is the preview page for a public channel, and the form most +// likely to be copied out of a browser. t.me/c/ carries the bare MTProto +// channel id — the same value a -100 Bot API id strips to. +func linkChat(s string) (string, bool) { + lower := strings.ToLower(s) + if strings.HasPrefix(lower, "tg://") { + return "", false + } + i, ok := hostEnd(lower) + if !ok { + return "", false + } + rest := s[i:] + rest, _, _ = strings.Cut(rest, "?") + rest, _, _ = strings.Cut(rest, "#") + + parts := strings.Split(strings.Trim(rest, "/"), "/") + if len(parts) < 2 || parts[1] == "" { + return "", false + } + switch strings.ToLower(parts[0]) { + case "c": + if isDigits(parts[1]) { + return parts[1], true } + case "s": + return parts[1], true } return "", false } +// afterHost returns the path following a Telegram host, if s names one. +func afterHost(lower string) (string, bool) { + i, ok := hostEnd(lower) + if !ok { + return "", false + } + return lower[i:], true +} + +// hostEnd returns the offset just past a Telegram host in an already-lowercased +// string. Offsets rather than a substring, so a caller can slice the original +// and keep the chat's real case. +func hostEnd(lower string) (int, bool) { + for _, host := range telegramHosts { + if i := strings.Index(lower, host); i >= 0 { + return i + len(host), true + } + } + return 0, false +} + func isDigits(s string) bool { if s == "" { return false @@ -105,6 +158,10 @@ func NormalizeChat(chat string) (string, error) { "pass the chat's username or id instead", chat) } + if c, ok := linkChat(chat); ok { + return c, nil + } + if m := botAPIID.FindStringSubmatch(chat); m != nil { return m[1], nil } diff --git a/internal/tgsource/chat_test.go b/internal/tgsource/chat_test.go index d13deff..8354846 100644 --- a/internal/tgsource/chat_test.go +++ b/internal/tgsource/chat_test.go @@ -21,12 +21,17 @@ func TestNormalizeChat(t *testing.T) { {"tg protocol link", "tg://resolve?domain=mychannel", "tg://resolve?domain=mychannel"}, {"bot api id loses the -100 prefix", "-1001234567890", "1234567890"}, {"surrounding whitespace is trimmed", " mychannel\n", "mychannel"}, - {"private channel link without a message", "https://t.me/c/1234567890", "https://t.me/c/1234567890"}, - // Invite and preview links are chats; their second component is not a - // bare message number and must not be read as one. + // The "c" and "s" markers are not the chat. Passed through whole, gotd + // takes the first path component as the username and resolves "c" or + // "s" — a confusing failure at best, and somebody else's chat at worst, + // since one-character usernames exist. + {"private channel link yields the channel id", "https://t.me/c/1234567890", "1234567890"}, + {"preview link yields the username", "https://t.me/s/mychannel", "mychannel"}, + {"preview link keeps the username's case", "https://t.me/s/MyChannel", "MyChannel"}, + // Invite links are chats; their second component is not a bare message + // number and must not be read as one. {"invite link", "https://t.me/+AbCd_1234", "https://t.me/+AbCd_1234"}, {"joinchat link", "https://t.me/joinchat/AbCd1234", "https://t.me/joinchat/AbCd1234"}, - {"preview link", "https://t.me/s/mychannel", "https://t.me/s/mychannel"}, {"other telegram host, no message", "https://telegram.dog/mychannel", "https://telegram.dog/mychannel"}, {"tg link without a post parameter", "tg://resolve?domain=mychannel", "tg://resolve?domain=mychannel"}, } @@ -50,6 +55,7 @@ func TestNormalizeChatRejectsMessageLinks(t *testing.T) { for _, in := range []string{ "https://t.me/c/1234567890/4242", "t.me/c/1234567890/4242", + "https://t.me/s/mychannel/4242", "https://t.me/mychannel/4242", // gotd accepts all three Telegram hosts and its parser keeps only the // domain, silently dropping the message number — so missing one of these diff --git a/internal/tgsource/client.go b/internal/tgsource/client.go index 72de17d..72074d5 100644 --- a/internal/tgsource/client.go +++ b/internal/tgsource/client.go @@ -133,7 +133,7 @@ func (s *Session) Run(ctx context.Context, fn func(context.Context, dcpool.Pool) tclient.NewDefaultMiddlewares(ctx, s.timeout)...) defer func() { _ = pool.Close() }() - return fn(ctx, pool) + return fn(ctx, withSafeTakeout(pool)) }) // gotd swallows cancellation: telegram.Client.Run ends with diff --git a/internal/tgsource/takeout.go b/internal/tgsource/takeout.go new file mode 100644 index 0000000..5b763b2 --- /dev/null +++ b/internal/tgsource/takeout.go @@ -0,0 +1,62 @@ +package tgsource + +import ( + "context" + "sync" + + "github.com/gotd/td/tg" + + "github.com/iyear/tdl/core/dcpool" + "github.com/iyear/tdl/core/middlewares/takeout" +) + +// safeTakeout wraps a pool to make its takeout path survive a failed init. +// +// core's own dcpool.Takeout deadlocks on that path. It holds the pool's mutex +// for the whole call, and its recovery from a failed init is to return +// p.Client(ctx, dc) — which locks the same mutex again (dcpool.go:113-121 and +// :57-58). sync.Mutex is not reentrant, so the worker blocks forever, then every +// other worker blocks behind it, and the process hangs with no output and no +// response to cancellation, since the goroutine is parked on a mutex rather than +// a select. +// +// That is not an exotic path. Telegram answers account.initTakeoutSession with +// TAKEOUT_INIT_DELAY when a takeout was started recently — tdl's own "ignore +// init delay error" comment shows it expects exactly this — and takeout is on by +// default, so running two exports in succession is enough to trigger it. +// +// Probing before the run is not an alternative: a probe would consume an init +// and make the pool's own init the one that gets the delay error. So the takeout +// session is established here instead, once, and the pool's Takeout is never +// called at all. +type safeTakeout struct { + dcpool.Pool + + once sync.Once + id int64 + ok bool +} + +func withSafeTakeout(p dcpool.Pool) dcpool.Pool { return &safeTakeout{Pool: p} } + +// Takeout returns a takeout-scoped client, or an ordinary one if no takeout +// session could be established. +// +// Falling back rather than failing matches what core intended: takeout raises +// rate limits and reaches older history, but a download works without it. The +// difference is that this fallback returns. +func (s *safeTakeout) Takeout(ctx context.Context, dc int) *tg.Client { + base := s.Pool.Client(ctx, dc) + + s.once.Do(func() { + id, err := takeout.Takeout(ctx, base.Invoker()) + if err != nil { + return // ok stays false; every caller gets a plain client + } + s.id, s.ok = id, true + }) + if !s.ok { + return base + } + return tg.NewClient(takeout.Middleware(s.id).Handle(base.Invoker())) +} diff --git a/internal/tgsource/takeout_test.go b/internal/tgsource/takeout_test.go new file mode 100644 index 0000000..dd9274f --- /dev/null +++ b/internal/tgsource/takeout_test.go @@ -0,0 +1,64 @@ +package tgsource + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/gotd/td/bin" + "github.com/gotd/td/tg" + + "github.com/iyear/tdl/core/dcpool" +) + +// failingInvoker refuses every call, which is what Telegram does to +// account.initTakeoutSession with TAKEOUT_INIT_DELAY. +type failingInvoker struct{ calls atomic.Int64 } + +func (f *failingInvoker) Invoke(context.Context, bin.Encoder, bin.Decoder) error { + f.calls.Add(1) + return errors.New("TAKEOUT_INIT_DELAY_86400") +} + +type fakePool struct{ inv tg.Invoker } + +func (p *fakePool) Client(context.Context, int) *tg.Client { return tg.NewClient(p.inv) } +func (p *fakePool) Takeout(context.Context, int) *tg.Client { + panic("upstream Takeout must not be called") +} +func (p *fakePool) Default(context.Context) *tg.Client { return tg.NewClient(p.inv) } +func (p *fakePool) Close() error { return nil } + +// core's dcpool.Takeout holds the pool mutex and recovers from a failed init by +// calling Client, which locks the same mutex — so the upstream version of this +// test hangs instead of failing. The wrapper must return a usable client. +func TestSafeTakeoutFallsBackWhenInitFails(t *testing.T) { + inv := &failingInvoker{} + var pool dcpool.Pool = withSafeTakeout(&fakePool{inv: inv}) + + done := make(chan *tg.Client, 1) + go func() { done <- pool.Takeout(t.Context(), 2) }() + + select { + case got := <-done: + if got == nil { + t.Fatal("Takeout returned nil after a failed init") + } + case <-time.After(5 * time.Second): + t.Fatal("Takeout deadlocked after a failed init") + } + + // The init is attempted once, not once per file: a failing init that ran on + // every element would add a round trip to each of 18k downloads. + before := inv.calls.Load() + for range 5 { + if pool.Takeout(t.Context(), 2) == nil { + t.Fatal("Takeout returned nil") + } + } + if got := inv.calls.Load(); got != before { + t.Errorf("takeout init retried %d times after failing; want no retries", got-before) + } +}