import "github.com/kinorai/omnifeed/internal/httpx/redislimit"
Package redislimit paces outbound requests with the pacing state held in Redis, so every replica of a deployment counts against one limit instead of one limit each. It implements httpx.Limiter and mirrors *httpx.DomainLimiter: the minimum delay is measured from the previous request’s COMPLETION, the rolling-window quota keeps the admission instants still inside the window (not a token bucket, so a burst cannot be spent all at once), and a nonzero wait carries jitter.
A caller that abandons its wait books nothing at all, so cancellation needs no cleanup — the price is that cross-pod queueing is jittered retry rather than the local limiter’s future-booked FIFO order.
The divergence MinDelay buys depends on ClusterConcurrency.
With ClusterConcurrency 0 (the default) the acquire script sets the next-allowed-at at admission and release bumps it again from completion, so with MinDelay > 0 admissions to one host are strictly serialized across the whole cluster: same-host concurrency collapses to 1 however many MaxConcurrent slots each pod holds. That is politer than the in-process limiter’s N concurrent requests, and slower — deliberate for a politeness control, and crash-safe, since a pod that dies mid-request leaves correct spacing behind.
With ClusterConcurrency > 0 that serialization is the thing being escaped, and the two controls split apart. MinDelay then spaces SENDS: the acquire script still books next-allowed-at at admission, but release no longer bumps it, so admissions leave MinDelay apart while up to N requests run at once. The N itself moves into Redis as a ZSET of leases scored by deadline, which is the TTL-lease machinery the per-pod semaphore existed to avoid. It needs no heartbeat: a lease is purged when its deadline passes, so a pod that dies mid-request returns its slot at LeaseTTL instead of never. Set LeaseTTL from the caller’s own request timeout — too short frees a slot a live request still holds, too long strands one after a crash.
Choose by what the upstream punishes. Serialized spacing suits an upstream that counts gaps between hits. A cap with spaced sends suits one that counts requests in a window, which is what search engines do, and it is the only mode that turns N replicas into N times the throughput.
Both scripts take the clock from Redis’ TIME rather than from the pod. That was gated on a question about the test double: redis.call(‘TIME’) inside miniredis’ EVAL does honour miniredis.SetTime (verified), so tests drive the same server-clock path production uses. Note that miniredis.FastForward expires keys WITHOUT moving that clock, so a test that advances time must call both.
Redis is never allowed to break retrieval: every backend failure comes back wrapped in httpx.ErrLimiterUnavailable for *httpx.FallbackLimiter to absorb.
Config describes one limiter scope. Scope separates the key spaces of limiters that share a Redis instance (“domain” for crawling, “searxng” for queries), so a crawl and a search never consume each other’s quota.
type Config struct {
// Client is a UniversalClient rather than *redis.Client so a deployment can
// point the same wiring at a standalone, sentinel or cluster Redis. This
// package only needs EVALSHA, which every variant serves.
Client redis.UniversalClient
Scope string // key-space separator, e.g. "domain" or "searxng"
Prefix string // key namespace, e.g. "omnifeed:ratelimit"
MaxConcurrent int // per-pod, see the package doc
MinDelay time.Duration // minimum gap between completion and next send
Quota int // 0 disables the rolling-window cap
Window time.Duration // width of that window
// ClusterConcurrency caps requests in flight to one host across every
// replica. 0 keeps the pre-cap behaviour: no lease bookkeeping, and MinDelay
// serializes admissions cluster-wide. Above 0 it becomes the authoritative
// bound and MinDelay spaces sends instead of gaps. See the package doc.
//
// It also raises the per-pod semaphore when MaxConcurrent is smaller, since
// a pod that admits fewer than the cluster allows would bottleneck the cap
// and make it look broken.
ClusterConcurrency int
// LeaseTTL is how long one in-flight slot stays booked without a release.
// Only read when ClusterConcurrency > 0. It is a crash-recovery bound, not a
// request timeout: the request's own ctx already bounds the happy path.
// Defaults to defaultLeaseTTL.
LeaseTTL time.Duration
// ConcurrencyRetry is how long a caller waits before re-attempting when the
// in-flight cap is full. Only read when ClusterConcurrency > 0. Short,
// because a slot frees on release and the script cannot predict when.
// Defaults to defaultConcurrencyRetry.
ConcurrencyRetry time.Duration
}
Limiter admits outbound requests against state shared through Redis.
type Limiter struct {
// OnWait, when non-nil, is called on every Acquire exit that made a pacing
// decision, with the engine that waited, the outcome ("acquired",
// "canceled" when the caller's ctx died, or "budget_exceeded" when the wait
// Redis asked for was longer than the caller's remaining deadline) and the
// time spent blocked. Same contract as DomainLimiter.OnWait. A backend failure deliberately emits
// NOTHING: FallbackLimiter immediately re-runs the acquire on the in-process
// limiter, which observes the whole wait itself, and a second observation
// here would double-count every acquire made while Redis is down. Set once
// at wiring time.
OnWait func(engine, outcome string, waited time.Duration)
// OnError, when non-nil, is called on every backend failure with the
// operation that failed ("acquire", "release" or "penalize"). Release and
// penalize failures reach the caller no other way. Set once at wiring time.
OnError func(op string)
// OnErrorDetail, when non-nil, is called alongside OnError with the error
// itself. OnError carries only the op, which makes a rising release-failure
// counter undiagnosable: the operation is fire-and-forget, so the error
// reaches nothing else. Wire this to a logger. Set once at wiring time.
OnErrorDetail func(op string, err error)
// contains filtered or unexported fields
}
func New(cfg Config) *Limiter
New returns a limiter for one scope. Per-operation timeouts are NOT this package’s business: they belong to the redis client’s options, set once at wiring time. An Acquire legitimately sleeps for minutes between attempts, so there is no timeout this package could impose that would not be wrong.
func (l *Limiter) Acquire(ctx context.Context, engine, rawURL string) (func(), error)
Acquire blocks until Redis admits this request, or until ctx is done. The returned release func must be called when the request finishes: it is what starts the minimum delay for the next one. When ctx carries a deadline and the wait Redis asks for is longer than what is left of it, Acquire returns *httpx.WaitBudgetError immediately instead of sleeping it out.
func (l *Limiter) Penalize(rawURL string, d time.Duration)
Penalize holds the host back for d across every replica — what an upstream’s Retry-After on a 429 or 503 is worth. It is the same next-allowed-at bump a release makes, with the upstream’s delay instead of the configured one, so it only ever extends the hold and never shortens one another replica reserved.
Fire and forget, like release: a failed penalty costs pacing accuracy, and the caller is a response handler with nothing to do about it. A non-positive d does nothing.
Generated by gomarkdoc