diff --git a/docs/advanced/data-export.md b/docs/advanced/data-export.md index 6860ebd3..09593198 100644 --- a/docs/advanced/data-export.md +++ b/docs/advanced/data-export.md @@ -60,8 +60,36 @@ make through it is wrapped in the takeout session. This is the [`takeout`](https://github.com/gotd/td/tree/main/examples/takeout) example. -:::note -Takeout sessions are subject to a server-imposed delay before they start -(`TAKEOUT_INIT_DELAY`). The [floodwait middleware](../helpers/middleware.md) recognises -this error and waits it out. -::: +## Long exports + +A takeout session lives until it is finished, so a long export can save its ID and +resume it after a restart instead of starting a new session (which may require the +user to confirm again): + +```go +t, err := takeout.Init(ctx, client, cfg) +if err != nil { + return err +} +saveTakeoutID(t.ID()) + +// After restart: +t = takeout.NewClient(loadTakeoutID(), client) +// ... export ... +return t.Finish(ctx, true) +``` + +New accounts must confirm the export on another device, or wait: init fails with +`TAKEOUT_INIT_DELAY_X`. Set `OnInitDelay` to keep polling until the user confirms: + +```go +cfg := takeout.Config{ + Contacts: true, + OnInitDelay: func(ctx context.Context, d time.Duration) error { + fmt.Printf("Confirm the export in Telegram (or wait %s)\n", d) + return nil // keep polling every InitPollInterval (30s) + }, +} +``` + +`takeout.Run` finishes the session even if its context is canceled. diff --git a/docs/advanced/long-running.md b/docs/advanced/long-running.md new file mode 100644 index 00000000..5cca7848 --- /dev/null +++ b/docs/advanced/long-running.md @@ -0,0 +1,84 @@ +--- +sidebar_position: 7 +--- + +# Long-running clients + +Services that run for weeks, often with many accounts in one process, need to detect +broken connections and lost authorizations instead of hanging or retrying forever. + +## Lost authorization + +When a session is terminated from another device, requests start failing with +`AUTH_KEY_UNREGISTERED`, `SESSION_REVOKED` and similar errors. `OnAuthLost` is called once +when this happens to an authorized client. Return an error to stop `Run` with +`*telegram.AuthLostError`, so one account can be stopped and reported while others keep +running: + +```go +client := telegram.NewClient(appID, appHash, telegram.Options{ + OnAuthLost: func(ctx context.Context, err error) error { + alert(account, err) + return err // stop Run + }, +}) + +err := client.Run(ctx, run) +var lost *telegram.AuthLostError +if errors.As(err, &lost) { + // Ask the user to sign in again. +} +``` + +Return `nil` to keep running, e.g. to sign in again in the same `Run`. The callback runs +on its own goroutine with `Run`'s context, so the failed request returns right away, and +`Run` waits for it before returning. A client counts as authorized after any successful +request that needs authorization, including one restored from session storage. When the +connect-time `Self` call finds a lost authorization, `OnAuthLost` is called first, then +`OnSelfError` if it returned `nil`. Use `telegram.IsAuthLost(err)` to check errors +yourself. + +## Connection events + +`OnConnectionState` reports the primary connection state (connecting, ready, +disconnected), and `OnDead` reports dead connections. To know which DC and which kind +of connection an event is about, use the event variants: + +```go +telegram.Options{ + OnConnectionStateEvent: func(e telegram.ConnectionStateEvent) { + // e.DC, e.State + }, + OnConnDead: func(e telegram.ConnDeadEvent) { + // e.DC, e.Kind (ConnKindPrimary, ConnKindData, ConnKindCDN), e.Err + }, +} +``` + +## Stuck requests + +A request that the server acknowledged but never answered (a half-dead connection, a +slow proxy) waits as long as its context allows. With `RequestTimeout`, the client +reconnects when the deadline passes and resends the request once on the new connection, +like Telegram Desktop does: + +```go +telegram.Options{ + RequestTimeout: func(input bin.Encoder) time.Duration { + switch input.(type) { + case *tg.MessagesSendMessageRequest: + return 0 // never send twice + default: + return 30 * time.Second + } + }, + // Liveness of connections. + PingInterval: 30 * time.Second, + PingTimeout: 10 * time.Second, +} +``` + +## Files on other datacenters + +Use `client.FileDC` for file requests to other datacenters: pools are cached, use your +middlewares and follow `FILE_MIGRATE`. See [Downloading files](../helpers/downloading-files.md#other-datacenters-and-cdn). diff --git a/docs/advanced/observability.md b/docs/advanced/observability.md new file mode 100644 index 00000000..7254f172 --- /dev/null +++ b/docs/advanced/observability.md @@ -0,0 +1,222 @@ +--- +sidebar_position: 8 +--- + +# Observability + +gotd can report what it does through three channels: + +- [**Metrics**](#metrics) via OpenTelemetry. Off unless you pass a `MeterProvider`. +- [**Logs**](#logs) via [`gotd/log`](https://github.com/gotd/log). Off unless you pass a + `Logger`. +- [**Hooks**](#hooks), which are callbacks for events your application should react to. + +Nothing is recorded by default. gotd never falls back to the global OpenTelemetry +providers, and with no provider set, instrumentation costs no allocations on the request +path. + +## Setup + +Pass the same providers to the client, the updates manager and the downloader: + +```go +import ( + "github.com/gotd/log" + "github.com/gotd/log/logzap" + "go.opentelemetry.io/otel" + + "github.com/gotd/td/telegram" + "github.com/gotd/td/telegram/downloader" + "github.com/gotd/td/telegram/updates" +) + +mp := otel.GetMeterProvider() // or your SDK MeterProvider +tp := otel.GetTracerProvider() +lg := logzap.New(zapLogger) + +gaps := updates.New(updates.Config{ + Handler: dispatcher, + Logger: log.Named(lg, "updates"), + MeterProvider: mp, + TracerProvider: tp, +}) + +client := telegram.NewClient(appID, appHash, telegram.Options{ + UpdateHandler: gaps, + Logger: lg, + MeterProvider: mp, + TracerProvider: tp, +}) + +// client.Downloader() and client.Download* use the client's MeterProvider and Logger. +// The updates manager needs its own MeterProvider (above). For your own Downloader: +d := downloader.NewDownloader(). + WithMeterProvider(mp). + WithLogger(log.Named(lg, "downloader")) +``` + +Most events are logged at debug level, so with an info-level logger you only see the +abnormal ones (see [Logs](#logs)). Raise the level when you need to see more. + +Metric attributes never contain user, chat or channel IDs, file locations or message +content, so the number of series stays bounded. IDs only appear in logs. + +## Metrics + +### Client (`telegram.Options.MeterProvider`) + +| Metric | Type | Attributes | Meaning | +| --- | --- | --- | --- | +| `gotd.client.connection.state` | counter | `dc`, `state` (`connecting`, `ready`, `disconnected`) | Primary connection state changes. | +| `gotd.client.connection.dead` | counter | `dc`, `kind` (`primary`, `data`, `cdn`) | Connections found dead (pings, read errors). | +| `gotd.client.connection.reconnects` | counter | `dc` | Reconnects of the primary connection. | +| `gotd.client.connection.restarts` | counter | `reason` (`timeout`, `migration`) | Restarts requested by the client itself; each is also counted in `connection.reconnects`. | +| `gotd.client.migrations` | counter | `dc` (target) | Home DC changes (`*_MIGRATE`). | +| `gotd.client.auth.lost` | counter | `reason` | Authorized → lost transitions (see `OnAuthLost`). | +| `gotd.client.request.timeouts` | counter | `result` (`resent`, `failed`), `method` | Requests that missed their `RequestTimeout` deadline. | +| `gotd.client.pool.created` | counter | `dc`, `kind` (`file`, `migrate`), `result` (`ok`, `error`, `canceled`) | Connection pools created for other DCs. | +| `gotd.client.pool.create.duration` | histogram, s | `kind`, `result` | Time to create a pool (connect and authorization transfer). | +| `gotd.client.pool.evicted` | counter | `dc`, `kind`, `reason` | Cached pools dropped, e.g. after `AUTH_KEY_UNREGISTERED`. | +| `gotd.client.file.migrations` | counter | `dc` (target), `source` | `FILE_MIGRATE` hops. | +| `gotd.client.pools` | gauge | `kind` (`file`, `migrate`, `cdn`) | Open pools while `Run` runs. | +| `gotd.client.invoke.duration` | histogram, s | `result` (`ok`, `rpc_error` with `code`, `canceled`, `error`) | Duration of every request, including `client.API()` calls. | + +`invoke.duration` has no method attribute: one series per TL method would be too many. +For per-method metrics, add a [middleware](../helpers/middleware). + +### Updates (`updates.Config.MeterProvider`) + +| Metric | Type | Attributes | Meaning | +| --- | --- | --- | --- | +| `gotd.updates.difference.requests` | counter | `kind` (`common`, `channel`), `result` (`empty`, `slice`, `full`, `too_long`, `error`, `inaccessible`) | `getDifference` / `getChannelDifference` calls. | +| `gotd.updates.difference.duration` | histogram, s | `kind`, `result` | Their duration. | +| `gotd.updates.difference.triggers` | counter | `kind`, `reason` (`startup`, `pts-gap-timeout`, `idle-timeout`, `channel-too-long`, …) | Why a difference was requested. | +| `gotd.updates.gaps` | counter | `kind`, `seq` (`pts`, `qts`, `seq`), `action` (`detected`, `resolved_by_waiting`, `resolved_by_pending`) | Sequence gaps and how they were closed. | +| `gotd.updates.channel.too_long` | counter | `dialog` (`ok`, `stale`, `invalid`) | `channelDifferenceTooLong` gaps, one per `OnChannelGap` call. | +| `gotd.updates.channel.gap.size` | histogram, pts | | `RemotePts - LocalPts` of those gaps. | +| `gotd.updates.channel.gap.messages` | histogram | | Number of messages returned with a gap. | +| `gotd.updates.before_run` | counter, batches | `action` (`passed`, `buffered`, `dropped`, `replayed`, `passed_raw`) | Updates received before `Run`, see `BeforeRun`. | +| `gotd.updates.channel.track` | counter | `source` (`api`, `stored`, `update`, `too_long`), `result` | Channels starting to be tracked. | +| `gotd.updates.channels.tracked` | up-down counter | | Channels tracked now. | +| `gotd.updates.channel.inaccessible` | counter | | Channels dropped after `CHANNEL_PRIVATE`. | +| `gotd.updates.stopped` | counter | `op` | Calls rejected with `updates.ErrStopped`. | +| `gotd.updates.dispatched` | counter | `source` (`live`, `difference`) | Updates passed to the handler. | + +### Downloader (`Downloader.WithMeterProvider`) + +| Metric | Type | Attributes | Meaning | +| --- | --- | --- | --- | +| `gotd.downloader.downloads` | counter | `result` (`ok`, `error`, `canceled`) | Finished downloads. | +| `gotd.downloader.requests` | counter | `kind` (`master`, `cdn`, `web`), `result` (`ok`, `flood_wait`, `timeout`, `canceled`, `error`) | Part requests. | +| `gotd.downloader.request.duration` | histogram, s | `kind` | Part request duration. | +| `gotd.downloader.bytes` | counter, By | `kind` | Bytes downloaded. | +| `gotd.downloader.resumes` | counter | | Downloads started with `WithOffset`. | +| `gotd.downloader.cdn.redirects` | counter | | Files served from a CDN DC. | +| `gotd.downloader.cdn.reuploads` | counter | | `upload.reuploadCdnFile` requests. | +| `gotd.downloader.hash.mismatches` | counter | `kind` (`cdn`, `verify`) | Parts that failed hash verification. | +| `gotd.downloader.file_reference.refreshes` | counter | `result` (`ok`, `not_found`, `error`, `limit`) | `WithRefresh` attempts. | +| `gotd.downloader.verify` | counter | `result` (`complete`, `partial`, `none`, `error`) | `Builder.Verify` results. | +| `gotd.downloader.retries` | counter | `operation` | Retried part requests. | +| `gotd.downloader.retries.exhausted` | counter | `operation` | Requests that ran out of retries. | +| `gotd.downloader.scheduler.backoffs` | counter | `reason` (`flood_wait`, `timeout`) | Backoffs reported to a `Scheduler`. | +| `gotd.downloader.scheduler.window` | gauge | your attributes | Current window of a `Scheduler`. | +| `gotd.downloader.scheduler.in_flight` | gauge | your attributes | Part requests in flight. | +| `gotd.downloader.scheduler.waiting` | gauge | your attributes | Part requests waiting for a slot. | +| `gotd.downloader.scheduler.decreases` | counter | your attributes | Window decreases. | + +Scheduler gauges are registered by `Scheduler.Instrument`, with your own attributes, for +example the DC. Schedulers the client creates from `Options.DownloadScheduler` are +registered with a `dc` attribute while `Run` runs. For your own schedulers: + +```go +reg, err := sched.Instrument(meter, attribute.Int("dc", 2)) +defer reg.Unregister() +``` + +### What to watch + +- **The client is healthy:** + - `connection.state{state="ready"}` grows only on reconnects; + - `connection.dead` and `request.timeouts` stay near zero; + - `invoke.duration` p95 is stable. +- **Updates are complete:** + - `difference.requests{result="too_long"}` and `channel.too_long` are rare; + - `gaps{action="detected"}` is followed by a resolution; + - `before_run{action="dropped"}` is zero, unless you chose `BeforeRunDrop`. +- **Authorization:** any increase of `auth.lost` needs attention. +- **Downloads keep up:** + - `downloads{result="error"}`, `hash.mismatches` and `retries.exhausted` stay near zero; + - `scheduler.window` recovers after `scheduler.backoffs`. + +## Logs + +Events are logged through the `Logger` you pass. Attributes are structured fields, and +IDs (channel, DC, pts) appear here, never in metrics. + +| Level | Component | Events | +| --- | --- | --- | +| Debug | client | Connection state, dead data and CDN connections, pool created, `FILE_MIGRATE` hop, restart requested | +| Info | client | Home DC migration (from, to) | +| Warn | client | Dead primary connection, pool creation failed, pool evicted after `AUTH_KEY_UNREGISTERED`, lost authorization, request timeout | +| Debug | updates | Each difference request with result and duration, channel tracked, update received before `Run` | +| Info | updates | Channel gap (channel, local and remote pts, top message, number of messages), replay of updates buffered before `Run` | +| Warn | updates | `updates.differenceTooLong`, channel pts that did not advance, failed `TrackChannel` | +| Debug | downloader | Part requests, CDN redirect, reupload, file reference refreshed | +| Warn | downloader | Hash mismatch, failed refresh, retries exhausted | + +## Hooks + +Use hooks to act on events. Use metrics and logs to watch them. + +### Client (`telegram.Options`) + +| Hook | Called when | +| --- | --- | +| `OnConnectionStateEvent func(ConnectionStateEvent)` | The primary connection state changes; `DC`, `State`. Prefer it to `OnConnectionState`. | +| `OnConnDead func(ConnDeadEvent)` | Any connection is dead; `DC`, `Kind`, `Err`. Prefer it to `OnDead`. | +| `OnAuthLost func(ctx, err) error` | An authorized session is lost. Runs on its own goroutine with `Run`'s context. Return an error to stop `Run` with `*AuthLostError`. See [Long-running clients](long-running). | +| `OnSelfError` / `OnSelfSuccess` | The `Self` call after connecting fails or succeeds. | +| `OnTransfer` | Authorization is transferred to another DC. | +| `RequestTimeout func(bin.Encoder) time.Duration` | Per-request response deadline; one resend after a reconnect. Return 0 for requests that must not be sent twice. | +| `PingInterval`, `PingTimeout` | Liveness checks of connections. | +| `Middlewares` | Wrap every request, e.g. for per-method metrics or rate limits. | + +### Updates (`updates.Config`) + +| Hook | Called when | +| --- | --- | +| `OnChannelGap func(ChannelGap)` | A channel gap can't be recovered. Has the pts range, the top message, and the latest `Messages` with `Users` and `Chats`; backfill with `messages.getHistory`. | +| `OnChannelTooLong func(channelID)` | Same event, ID only. | +| `OnChannelInaccessible func(channelID)` | Access to a channel is lost (`CHANNEL_PRIVATE`); it stops being tracked. | +| `OnTooLong func()` | The common state gap can't be recovered (`updates.differenceTooLong`). | +| `OnLoadUserStateFailed func()` | No stored state on start, so a full resync follows. | +| `OnLoadChannelStateFailed func(channelID)` | A stored channel can't be loaded; track it again with `Manager.TrackChannel`. | +| `BeforeRun`, `BeforeRunBufferSize` | What happens to updates received before `Run`: pass, buffer or drop. | + +### Dispatcher (`tg.UpdateDispatcher`) + +| Hook | Called when | +| --- | --- | +| `OnBefore func(ctx, e, update) error` | Every update, before its handler, including updates without one. Return an error to skip the handler. Useful for per-type counters or tracing. | +| `OnFallback` | An update has no handler. | +| `OnAny*` | One handler for private chats, groups and channels; see [Handling updates](../basics/handling-updates). | + +### Downloads + +| Hook | Called when | +| --- | --- | +| `Builder.WithProgress(func(done, total int64))` | Bytes are written. | +| `Builder.WithRefresh(RefreshFunc)` | A file reference expired; return a fresh location (see `downloader/fileref`). | +| `Scheduler.Snapshot()` | Read the window, in-flight, waiting and decreases at any time. | + +### Other packages + +| Hook | Package | +| --- | --- | +| `Config.OnInitDelay` | `telegram/takeout`: `TAKEOUT_INIT_DELAY` while waiting for the export to be allowed. | +| `Waiter.WithCallback(func(ctx, FloodWait))`, `Waiter.Snapshot()` | `gotd/contrib/middleware/floodwait`: every flood wait (type, wait, error) and per-type state. | + +## Tracing + +`TracerProvider` in `telegram.Options` and `updates.Config` adds spans for requests and +update processing; see [Debugging](debugging). diff --git a/docs/authentication/qr-login.md b/docs/authentication/qr-login.md index cfda58b7..c807f4d8 100644 --- a/docs/authentication/qr-login.md +++ b/docs/authentication/qr-login.md @@ -54,8 +54,19 @@ if _, err := client.QR().Auth(ctx, loggedIn, show); err != nil { ## Handling 2FA -If the account has a 2FA password, `Auth` fails with `SESSION_PASSWORD_NEEDED`. Detect -it and complete the sign-in with the password: +If the account has a 2FA password, pass a password callback, and `Import`/`Auth` complete +the sign-in with it: + +```go +qr := qrlogin.NewQR(client.API(), appID, appHash, qrlogin.Options{ + Password: func(ctx context.Context) (string, error) { + return promptPassword(ctx) + }, +}) +``` + +Without the callback, `Auth` fails with `SESSION_PASSWORD_NEEDED`. Detect it and +complete the sign-in with the password: ```go import "github.com/gotd/td/tgerr" diff --git a/docs/basics/handling-updates.md b/docs/basics/handling-updates.md index c85d83ef..69f6bb66 100644 --- a/docs/basics/handling-updates.md +++ b/docs/basics/handling-updates.md @@ -44,6 +44,42 @@ Common handlers: There are handlers for the full update set — see the [`tg` reference](https://ref.gotd.dev/pkg/github.com/gotd/td/tg.html). +The dispatcher can be passed by value: copies share handlers, so you can register +handlers (including `OnFallback`) after passing it to the client, even while updates +are being handled. + +### One handler for all chats + +Private chats, basic groups and channels have separate update types. The `OnAny*` +helpers register one handler for all of them: + +```go +dispatcher.OnAnyNewMessage(func(ctx context.Context, e tg.Entities, msg tg.MessageClass) error { + return archive(msg) +}) +dispatcher.OnAnyEditMessage(onEdit) +dispatcher.OnAnyDeleteMessages(func(ctx context.Context, e tg.Entities, channelID int64, ids []int) error { + return nil // channelID is 0 for private chats and basic groups +}) +dispatcher.OnAnyReadInbox(func(ctx context.Context, e tg.Entities, r tg.AnyRead) error { + // r.Peer, r.MaxID; r.TopMsgID for forum topics and comment threads. + return nil +}) // also OnAnyReadOutbox +dispatcher.OnAnyTyping(func(ctx context.Context, e tg.Entities, t tg.AnyTyping) error { + // t.Peer, t.From, t.Action; t.TopMsgID for forum topics. + return nil +}) + +// Called for every dispatched update before its handler, e.g. for metrics or filtering. +// A non-nil error skips the update's handler and is returned by Handle. +dispatcher.OnBefore(func(ctx context.Context, e tg.Entities, u tg.UpdateClass) error { + updatesTotal.Inc() + return nil +}) +``` + +Each event struct also keeps the source update in its `Update` field. + ## The `Entities` argument Updates reference users and chats by ID. The `tg.Entities` value passed to each handler @@ -64,6 +100,10 @@ dispatcher.OnNewMessage(func(ctx context.Context, e tg.Entities, u *tg.UpdateNew }) ``` +To call handlers yourself with results of other methods (e.g. replaying stored updates or +a `getDifference` result), build the entities with +`tg.EntitiesFrom(users, chats)` or `tg.EntitiesFromResult(result)`. + ## Filtering your own messages For userbots, skip outgoing messages or you may echo yourself into a loop: diff --git a/docs/helpers/contrib-packages.md b/docs/helpers/contrib-packages.md index 1c95a146..56d113ec 100644 --- a/docs/helpers/contrib-packages.md +++ b/docs/helpers/contrib-packages.md @@ -75,7 +75,11 @@ See [User authentication](../authentication/user.md). ## Storage — sessions, peers & state `storage` defines the common peer abstractions; the rest are backend -implementations of the session, peer and update-state storage interfaces. See +implementations of the session, peer and update-state storage interfaces. `bbolt` and +`pebble` also provide persistent updates state and access hashers for the +[updates engine](./updates-recovery.md#persisting-state). The pebble backends commit +with `WithWriteOptions` (synced by default; `pebble.NoSync` for speed), and +`storage.UpdateHook` writes all peers of an update in one batch. See [Sessions](../authentication/sessions.md) for how session storage fits into the client. diff --git a/docs/helpers/downloading-files.md b/docs/helpers/downloading-files.md index 7e0a2651..e6bd490d 100644 --- a/docs/helpers/downloading-files.md +++ b/docs/helpers/downloading-files.md @@ -61,6 +61,108 @@ d := downloader.NewDownloader(). CDN downloads must also be permitted on the client via `Options.AllowCDN`. +## Resuming, size and progress + +To resume a partial download, start from an offset. `ToPath` keeps the existing file +content before the offset, so a `.part` file can be continued after a restart. Pass the +known size (e.g. `Document.Size`) so workers stop at the end of the file instead of +probing past it, and a progress callback: + +```go +_, err := client.Download(location). + WithOffset(offset). // divisible by the part size (4 KB with WithVerify(true)) + WithSize(doc.Size). + WithProgress(func(done, total int64) { + fmt.Printf("%d/%d\n", done, total) + }). + ToPath(ctx, "video.mp4.part") +``` + +`done` counts from the start of the file, including the offset. + +To check what is already on disk without downloading it, verify the local data +against the file hashes from the server (`upload.getFileHashes`). `Verify` returns the +length of the longest verified prefix, which is a safe offset to resume from: + +```go +f, _ := os.Open("video.mp4.part") +st, _ := f.Stat() +valid, err := client.Download(location).Verify(ctx, f, st.Size()) +// Round down to the part size unless verification is enabled. +offset := valid - valid%(512*1024) +``` + +## Expired file references + +File references expire, and a long download (or a download from an old message) +fails with `FILE_REFERENCE_EXPIRED`. Give the builder a refresh function: on a file +reference error the location is refreshed once for all workers and the download +continues from the same offset. The +[`fileref`](https://ref.gotd.dev/pkg/github.com/gotd/td/telegram/downloader/fileref.html) +package fetches the origin of the file again: + +```go +import "github.com/gotd/td/telegram/downloader/fileref" + +_, err := client.Download(location). + WithRefresh(fileref.Message(client.API(), peer, msgID, location)). + ToPath(ctx, path) +``` + +There are also `fileref.Story` and `fileref.UserPhoto`, and `downloader.IsFileReferenceError` +to detect the error yourself. + +## Other datacenters and CDN + +Files live on the datacenter in their `DCID`. `client.FileDC(ctx, dc)` returns an invoker +for that DC: the pool is created once and cached, wrapped with your +`Options.Middlewares`, uses the primary session for the home DC, and follows +`FILE_MIGRATE`. Pool size is `Options.FilePoolSize` (default 4). + +Downloads through such an invoker can still use CDN redirects with `DownloadVia`: + +```go +inv, err := client.FileDC(ctx, doc.DCID) +if err != nil { + return err +} +_, err = client.DownloadVia(inv, location).ToPath(ctx, path) +``` + +For a custom `downloader.Client`, wrap it with +`downloader.WithCDNProvider(c, client.CDNProvider())`. + +## Sharing a download window + +`WithThreads` limits workers of one download. To share a budget between concurrent +downloads to the same DC, use a `Scheduler`: an adaptive limit of in-flight part +requests that grows on success and halves on flood waits and timeouts, like the +download window of Telegram Desktop: + +```go +sched := downloader.NewScheduler(downloader.SchedulerOptions{Start: 4, Max: 16}) +d := client.Downloader().WithScheduler(sched) // one scheduler per DC + +_, err := d.Download(api, location).WithThreads(16).ToPath(ctx, path) + +// For metrics. +snap := sched.Snapshot() // Window, InFlight, Waiting, Decreases +``` + +To use schedulers for `client.Download` and `client.DownloadWeb`, set the options on the +client. It creates one scheduler per DC on first use: + +```go +client := telegram.NewClient(appID, appHash, telegram.Options{ + DownloadScheduler: &downloader.SchedulerOptions{Start: 4, Max: 16}, +}) + +snap := client.DownloadScheduler(dc).Snapshot() +``` + +`DownloadVia` can't know the DC of its invoker, so it uses a scheduler only when you +pass one with `WithScheduler`. + ## Bulk downloads To back up many files, combine the downloader with a diff --git a/docs/helpers/middleware.md b/docs/helpers/middleware.md index bb1d3044..8329b8b4 100644 --- a/docs/helpers/middleware.md +++ b/docs/helpers/middleware.md @@ -51,6 +51,36 @@ return waiter.Run(ctx, func(ctx context.Context) error { For one-shot scripts, `floodwait.NewSimpleWaiter()` is a simpler timer-based variant that needs no `Run` wrapper. Both support `WithMaxRetries` and `WithMaxWait`. +### Long-running programs + +The `Waiter` keys flood waits by the method, unwrapping `invokeWith*` (so a flood +wait inside a [takeout session](../advanced/data-export.md) throttles only that +method), and dispatches requests concurrently. Further options: + +```go +waiter := floodwait.NewWaiter(). + // Space out requests of a method after it hits FLOOD_WAIT: the interval + // doubles on each flood wait and decays on success. + WithAdaptivePacing(100*time.Millisecond, 10*time.Second). + // Requests with their own pacing bypass the waiter. + WithExempt(tg.UploadGetFileRequestTypeID). + // Retry transient 5xx errors with exponential backoff. + WithServerErrorRetries(3). + WithCallback(func(ctx context.Context, wait floodwait.FloodWait) { + log.Printf("%s: wait %s", wait.Name, wait.Duration) + }) + +// Per-request policy: background jobs wait as long as it takes, +// interactive ones fail fast. +bg := floodwait.WithPolicy(ctx, floodwait.Policy{MaxWait: -1, MaxRetries: -1}) +ui := floodwait.WithPolicy(ctx, floodwait.Policy{NoRetry: true}) + +// Pacing state for metrics. +for _, s := range waiter.Snapshot() { + fmt.Println(s.Name, s.Wait, s.Interval) +} +``` + ## Rate limiting To stay *under* the limits in the first place, add the diff --git a/docs/helpers/query-iterators.md b/docs/helpers/query-iterators.md index daa88517..d87005c7 100644 --- a/docs/helpers/query-iterators.md +++ b/docs/helpers/query-iterators.md @@ -35,6 +35,11 @@ err := query.GetDialogs(client.API()).ForEach(ctx, ) ``` +`elem.Peer` is an input peer you can pass to other methods, or `InputPeerEmpty` if the +peer is known only as a [min constructor](https://core.telegram.org/api/min). For such +peers use `elem.Entities.ExtractPeerFromMessage`, and `dialogs.DialogKey.InputPeer()` to +turn a stored key back into an input peer. + ## Message history Walk a chat's history newest-first: @@ -55,6 +60,35 @@ err := query.Messages(client.API()). `Elem` also has `elem.File()` to extract downloadable media — see [Downloading files](./downloading-files.md). +Use `MinID`/`MaxID` to fetch only a range of messages, e.g. new messages since the last +archived one: + +```go +query.Messages(client.API()).GetHistory(peer).MinID(lastArchivedID).ForEach(ctx, cb) +``` + +To fetch messages by ID together with their users and chats, use +`GetMessagesWithEntities(ctx, peer, ids...)`. + +### Resuming + +Iterators expose their position as a `Cursor`, which you can persist and use to continue +after a restart from the element after the last one you processed: + +```go +iter := query.Messages(api).GetHistory(peer).Iter() +for iter.Next(ctx) { + process(iter.Value()) + save(iter.Cursor()) +} + +// After restart. +iter = query.Messages(api).GetHistory(peer).Iter().Resume(load()) +``` + +Dialogs iterators have the same `Cursor`/`Resume`. To iterate the main list and the +archive in one pass, use `query.NewQuery(api).Dialogs().ForEachFolder(ctx, 100, cb)`. + ### Stopping early Return a sentinel error from the callback and ignore it after the loop: @@ -104,6 +138,28 @@ err := q.ForEach(ctx, func(ctx context.Context, p participants.Elem) error { }) ``` +## Stories + +`query.NewQuery(api).Stories()` pages through stories: + +```go +q := query.NewQuery(client.API()).Stories() + +// Active stories of all followed peers; pass the returned state next time to +// fetch only changes. +state, err := q.GetAllStories().ForEach(ctx, func(ctx context.Context, s tg.PeerStories, e peer.Entities) error { + return nil +}) + +// Pinned stories and the stories archive of a peer. +err = q.GetPinnedStories(peer).ForEach(ctx, cb) +err = q.GetStoriesArchive(&tg.InputPeerSelf{}).ForEach(ctx, cb) + +// Skipped stories by ID, and views of own story. +stories, entities, err := q.GetStoriesByID(ctx, peer, ids...) +err = q.GetStoryViewsList(&tg.InputPeerSelf{}, storyID).ForEach(ctx, onView) +``` + ## Other iterators `query.NewQuery(api)` is the umbrella builder; alongside dialogs, messages and diff --git a/docs/helpers/updates-recovery.md b/docs/helpers/updates-recovery.md index 4a04123f..a7b650e8 100644 --- a/docs/helpers/updates-recovery.md +++ b/docs/helpers/updates-recovery.md @@ -94,24 +94,79 @@ stream. ## Persisting state By default the engine keeps state in memory, so a restart re-syncs from scratch. For a -long-running service, give it a `Storage` (and access-hash storage) so it resumes from -where it left off. The -[`gotd/contrib`](https://github.com/gotd/contrib) module provides bbolt-backed -implementations: +long-running service, give it a `Storage` and access-hash storage so it resumes from +where it left off: without stored channel access hashes it can't fetch channel +differences after a restart. The +[`gotd/contrib`](https://github.com/gotd/contrib) module provides bbolt- and +pebble-backed implementations: ```go +import "github.com/gotd/contrib/pebble" + gaps := updates.New(updates.Config{ - Handler: d, - Storage: boltstor.NewStateStorage(boltdb), + Handler: d, + Storage: pebble.NewStateStorage(db), + AccessHasher: pebble.NewAccessHasher(db), + UserAccessHasher: pebble.NewAccessHasher(db), +}) +``` + +The bbolt equivalents are `bbolt.NewStateStorage(db)` and `bbolt.NewAccessHasher(db)`. + +## Tracking channels + +The engine tracks channels with a stored state, and learns about other channels from +their next update. To track channels joined while the client was offline (or reported +via `OnLoadChannelStateFailed`), pass them with the pts of their dialog. Stored pts take +precedence, so missed updates are fetched: + +```go +gaps.Run(ctx, api, self.ID, updates.AuthOptions{ + OnStart: func(ctx context.Context) { + for _, d := range dialogs { // e.g. from messages.getDialogs + _ = gaps.TrackChannel(ctx, d.ChannelID, d.AccessHash, d.Pts) + } + }, }) ``` +## Updates before `Run` + +Updates can arrive before the manager's `Run` has loaded the state, e.g. while the +client connects. By default they are handled right away (`BeforeRunPass`). To handle +them in order once the state is loaded, buffer them, or drop them and let gap +recovery fetch them again: + +```go +updates.Config{ + BeforeRun: updates.BeforeRunBuffer, // or updates.BeforeRunDrop + BeforeRunBufferSize: 100, // batches; extra ones are dropped +} +``` + +## Channel gaps + +When a channel gap is too long to recover (`updates.channelDifferenceTooLong`), the +engine moves to the new state and reports the gap with the latest messages returned by +the server. They are not passed to your handler, because they may include messages it +has already received. Backfill what you missed with `messages.getHistory`: + +```go +updates.Config{ + OnChannelGap: func(gap updates.ChannelGap) { + // gap.ChannelID, gap.AccessHash, gap.LocalPts, gap.RemotePts, gap.TopMessage + // gap.Messages are the latest messages, with gap.Users and gap.Chats. + }, +} +``` + The [`userbot`](https://github.com/gotd/td/tree/main/examples/userbot) example uses persistent state; the [`updates`](https://github.com/gotd/td/tree/main/examples/updates) example is a minimal in-memory version. :::note[Known limits] -The engine relies on the server for `getDifference` correctness and cannot recover from -`ChannelDifferenceTooLong` (it resyncs that channel instead). Stateless updates can't be -ordered. These are inherent to the MTProto update model. +The engine relies on the server for `getDifference` correctness, and messages missed in +a `ChannelDifferenceTooLong` gap must be fetched by the application +(see [Channel gaps](#channel-gaps)). Stateless updates can't be ordered. These are +inherent to the MTProto update model. :::