Documentation
¶
Overview ¶
Package floodwait implements a tg.Invoker that handles flood wait errors.
It provides two implementations: the scheduler-based Waiter, suitable for long-running, highly concurrent programs, and the simpler timer-based SimpleWaiter, suitable for one-off tasks.
Recognized wait errors ¶
Following TDLib's NetQueryDelayer and Telethon's flood handling, both waiters retry on a broader set of "wait" errors than just FLOOD_WAIT:
- FLOOD_WAIT (420) and FLOOD_PREMIUM_WAIT (420): per-method rate limits.
- SLOWMODE_WAIT, 2FA_CONFIRM_WAIT, TAKEOUT_INIT_DELAY, FLOOD_TEST_PHONE_WAIT (420): chat- or operation-specific waits.
- FLOOD_SKIP_FAILED_WAIT (420): argument-less, retried after 1s.
- WORKER_BUSY_TOO_LONG_RETRY (500): transient, retried after 1s.
The reported wait duration is clamped into the [1s, 14d] range, matching TDLib's clamp(timeout, 1, 14*24*60*60).
Per-method versus request-specific waits ¶
FLOOD_WAIT and FLOOD_PREMIUM_WAIT throttle every request of the same type, so the scheduler-based Waiter proactively delays future requests of that type after one is seen. The remaining wait errors are tied to a specific chat or operation rather than the method (Telethon notes "SLOW_MODE_WAIT is chat-specific, not request-specific"), so the Waiter retries only the offending request and leaves unrelated requests of the same type untouched.
Requests wrapped in invokeWith* (e.g. invokeWithTakeout in takeout sessions) are keyed by the inner method, so a flood wait of one method inside a takeout session does not throttle other methods.
Limits ¶
Both waiters bound retries via WithMaxRetries and the per-attempt wait via WithMaxWait: a wait longer than the limit is returned as an error instead of being slept on. Use a context with a deadline to bound the total wait time. WithPolicy overrides the limits per request context, e.g. to wait as long as it takes in background jobs or to fail fast in interactive ones.
Long-running programs ¶
The Waiter dispatches scheduled requests concurrently, so requests that invoke nested requests through the same middleware chain (like the authorization transfer on FILE_MIGRATE in gotd/td) do not deadlock. It also supports:
- WithAdaptivePacing: spacing of requests of a type that grows on each flood wait, decays on success down to the minimum, and is removed after 10 minutes without flood waits;
- WithExempt: request types that bypass the Waiter (e.g. upload.getFile with its own pacing);
- WithServerErrorRetries: exponential backoff for transient 5xx errors;
- Snapshot and the FloodWait callback event with the request type, for metrics.
Behavior changes ¶
Since concurrent dispatch was introduced:
- Scheduled requests are sent concurrently instead of one at a time per tick, so the downstream invoker must be safe for concurrent use (as telegram.Client is).
- The per-type wait decays by one step on every response that is not a wait error, including other errors, as before. Requests re-queued for a server error retry (WithServerErrorRetries) do not decay it.
- Run waits for in-flight requests before returning and fails requests still queued with an error.
- WithClock and WithAdaptivePacing return a Waiter with its own scheduler, so they must be called before Run.
Index ¶
- func WithPolicy(ctx context.Context, p Policy) context.Context
- type FloodWait
- type KeyState
- type Policy
- type SimpleWaiter
- type Waiter
- func (w *Waiter) Handle(next tg.Invoker) telegram.InvokeFunc
- func (w *Waiter) Run(ctx context.Context, f func(ctx context.Context) error) (err error)
- func (w *Waiter) Snapshot() []KeyState
- func (w *Waiter) WithAdaptivePacing(minInterval, maxInterval time.Duration) *Waiter
- func (w *Waiter) WithCallback(f func(ctx context.Context, wait FloodWait)) *Waiter
- func (w *Waiter) WithClock(c clock.Clock) *Waiter
- func (w *Waiter) WithExempt(typeIDs ...uint32) *Waiter
- func (w *Waiter) WithMaxRetries(m int) *Waiter
- func (w *Waiter) WithMaxWait(m time.Duration) *Waiter
- func (w *Waiter) WithServerErrorRetries(n int) *Waiter
- func (w *Waiter) WithTick(t time.Duration) *Waiter
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type FloodWait ¶ added in v0.16.0
type FloodWait struct {
Duration time.Duration
// TypeID is the TL type ID of the request, unwrapped from invokeWith*
// (e.g. the method called in a takeout session).
TypeID uint32
// Name is the TL name of the request, if known.
Name string
// PerType is true if the wait applies to all requests of the type
// (FLOOD_WAIT), false if only to this request (e.g. SLOWMODE_WAIT,
// server errors).
PerType bool
// Err is the error that caused the wait.
Err error
}
FloodWait event.
type KeyState ¶ added in v0.26.0
type KeyState struct {
// TypeID is the TL type ID of the request (unwrapped from invokeWith*).
TypeID uint32
// Name is the TL name of the request, if known.
Name string
// Wait is the delay added to new requests of the type after flood wait.
Wait time.Duration
// Interval is the minimum spacing of requests of the type (adaptive
// pacing).
Interval time.Duration
}
KeyState is a pacing state of a request type, see Waiter.Snapshot.
type Policy ¶ added in v0.26.0
type Policy struct {
// MaxWait limits wait time per attempt, overriding Waiter.WithMaxWait.
// Zero keeps the Waiter limit, negative means no limit ("background: wait
// as long as it takes").
MaxWait time.Duration
// MaxRetries limits number of retries, overriding Waiter.WithMaxRetries.
// Zero keeps the Waiter limit, negative means no limit.
MaxRetries int
// NoRetry makes wait errors returned immediately ("interactive: fail
// fast"). Scheduling delays of the request type still apply.
NoRetry bool
}
Policy overrides Waiter limits for requests made with a context, see WithPolicy.
type SimpleWaiter ¶
type SimpleWaiter struct {
// contains filtered or unexported fields
}
SimpleWaiter is a tg.Invoker that handles flood wait errors on underlying invoker.
This implementation is more suitable for one-off tasks and programs with low level of concurrency and parallelism.
See Waiter for a fully-blown scheduler-based flood wait handler.
func NewSimpleWaiter ¶
func NewSimpleWaiter() *SimpleWaiter
NewSimpleWaiter returns a new invoker that waits on the flood wait errors.
func (*SimpleWaiter) Handle ¶
func (w *SimpleWaiter) Handle(next tg.Invoker) telegram.InvokeFunc
Handle implements telegram.Middleware.
func (*SimpleWaiter) WithClock ¶
func (w *SimpleWaiter) WithClock(c clock.Clock) *SimpleWaiter
WithClock sets clock to use. Default is to use system clock.
func (*SimpleWaiter) WithMaxRetries ¶
func (w *SimpleWaiter) WithMaxRetries(m uint) *SimpleWaiter
WithMaxRetries sets max number of retries before giving up. Default is to keep retrying on flood wait errors indefinitely.
func (*SimpleWaiter) WithMaxWait ¶
func (w *SimpleWaiter) WithMaxWait(m time.Duration) *SimpleWaiter
WithMaxWait limits wait time per attempt. SimpleWaiter will return an error if flood wait time exceeds that limit. Default is to wait without time limit.
To limit total wait time use a context.Context with timeout or deadline set.
type Waiter ¶
type Waiter struct {
// contains filtered or unexported fields
}
Waiter is a tg.Invoker that handles flood wait errors on underlying invoker.
This implementation uses a request scheduler and is more suitable for long-running programs with high level of concurrency and parallelism.
You should use Waiter if unsure which waiter implementation to use.
See SimpleWaiter for a simple timer-based implementation.
func NewWaiter ¶
func NewWaiter() *Waiter
NewWaiter returns a new invoker that waits on the flood wait errors.
NB: You MUST use Run method. Example:
if err := waiter.Run(ctx, func(ctx context.Context) error {
// Client should be started after waiter.
return client.Run(ctx, handler)
}); err != nil {
return errors.Wrap(err, "run client")
}
func (*Waiter) Handle ¶
func (w *Waiter) Handle(next tg.Invoker) telegram.InvokeFunc
Handle implements telegram.Middleware.
func (*Waiter) Run ¶
Run runs send loop.
Example:
if err := waiter.Run(ctx, func(ctx context.Context) error {
// Client should be started after waiter.
return client.Run(ctx, handler)
}); err != nil {
return errors.Wrap(err, "run client")
}
func (*Waiter) Snapshot ¶ added in v0.26.0
Snapshot returns pacing state of request types with active waits or intervals, e.g. for metrics.
func (*Waiter) WithAdaptivePacing ¶ added in v0.26.0
WithAdaptivePacing enables adaptive pacing: each flood wait of a request type doubles the minimum spacing of requests of that type (starting from minInterval, up to maxInterval), and each success decays it down to minInterval, so long-running programs do not hit the same limit again right after the wait. The spacing is removed after 10 minutes without flood waits of the type.
Must be called before Run. Disabled by default. The returned Waiter has its own scheduler: the original Waiter is not changed.
func (*Waiter) WithCallback ¶ added in v0.16.0
WithCallback sets callback for flood wait event.
func (*Waiter) WithClock ¶
WithClock sets clock to use for ticks and scheduling. Default is to use system clock.
Must be called before Run. The returned Waiter has its own scheduler: the original Waiter is not changed.
func (*Waiter) WithExempt ¶ added in v0.26.0
WithExempt makes requests of given types bypass the Waiter, e.g. requests with their own pacing like upload.getFile on file pools (tg.UploadGetFileRequestTypeID). Requests are matched after unwrapping invokeWith* wrappers.
func (*Waiter) WithMaxRetries ¶
WithMaxRetries sets max number of retries before giving up. Default is to retry at most 5 times.
func (*Waiter) WithMaxWait ¶
WithMaxWait limits wait time per attempt. Waiter will return an error if flood wait time exceeds that limit. Default is to wait at most a minute.
To limit total wait time use a context.Context with timeout or deadline set.
func (*Waiter) WithServerErrorRetries ¶ added in v0.26.0
WithServerErrorRetries enables retries of transient server errors (code 500 and above, e.g. INTERNAL or Timeout) with exponential backoff from 1s up to 32s, at most n times. Disabled by default; WORKER_BUSY_TOO_LONG_RETRY is always retried as a wait error.
Source Files