import "github.com/kinorai/omnifeed/internal/httpx"
Package httpx provides HTTP utilities shared across engines: a retrying HTTP client wrapper, per-domain rate limiter, and SSRF guards.
ErrLimiterUnavailable wraps backend failures of a distributed limiter (Redis unreachable, script error). It never reaches an engine: FallbackLimiter absorbs it and paces in process instead.
var ErrLimiterUnavailable = errors.New("rate limiter backend unavailable")
func ClassifyClientError(err error, fallback domain.FailureKind) *domain.FetchError
ClassifyClientError translates an error returned by DoRetry into a typed domain.FetchError. A StatusError carries the upstream status after retries: a 429 is unambiguous and becomes KindHTTP429, but a 5xx is ambiguous — it can be a genuine upstream fault OR, on the browser/anti-bot paths, the block itself surfacing as a crawl4ai 5xx (see browser/crawl4ai) — so the caller’s fallback decides (KindUpstreamError for a generic crawl, KindBotBlock for a Reddit navigation). A context deadline becomes KindTimeout; a caller cancellation (client abort) becomes KindCanceled; a limiter’s fast-fail (*WaitBudgetError) becomes KindQuotaExhausted with the retry-after in the message; anything else becomes the fallback. Returns nil when err is nil.
The WaitBudgetError case lives here rather than at each call site so the search path and every crawl path classify a refused pacing wait the same way: every limiter error already reaches a caller through this function.
func HostMatcher(domain string) *regexp.Regexp
HostMatcher returns a case-insensitive regexp matching host and all of its subdomains: HostMatcher(“reddit.com”) matches reddit.com, www.reddit.com, and old.reddit.com, but not evilreddit.com or reddit.com.evil.com. Engines use it to claim URLs; centralizing the pattern keeps this security-sensitive rule from drifting between engines.
func NewGuardedClient(blockPrivate bool, timeout time.Duration) *http.Client
NewGuardedClient returns an http.Client for fetching caller-supplied URLs directly (no crawl4ai in between). When blockPrivate is true, every dial is checked AFTER DNS resolution and refused if the resolved address is private/reserved — unlike ValidateURL’s lookup-then-fetch, this can’t be raced by DNS rebinding, and it covers every redirect hop because the guard sits below the client’s redirect-following.
func NewTransport() *http.Transport
NewTransport returns the outbound transport shared by omnifeed’s HTTP clients: DefaultTransport’s dial/TLS/proxy behavior with a per-host idle pool sized for this proxy’s traffic. Go’s DefaultTransport keeps only 2 idle connections per host, so concurrent requests to the same upstream (crawl4ai, SearXNG, api.github.com) evict each other’s connections and re-handshake TCP + TLS on the next call.
func RemainingBudget(ctx context.Context) (time.Duration, bool)
RemainingBudget reports how long ctx has left, and whether it has a deadline at all. A context without one has no budget to exceed, so pacing queues as before. Exported for the out-of-package limiter implementations (redislimit), which make the same fast-fail decision as *DomainLimiter.
func ValidateURL(rawURL string, blockPrivate bool) error
ValidateURL parses rawURL and enforces the proxy’s outbound-request guard:
When blockPrivate is set, obfuscated IPv4 literals (leading-zero octal 0177.0.0.1, short 127.1, decimal 2130706433) are rejected outright: Go’s net.LookupIP and a browser disagree on what they resolve to, so validating one interpretation and letting the downstream fetcher act on another is a bypass. DNS failures pass through — the downstream fetcher surfaces them.
NOTE: this is a best-effort, app-layer guard only. The actual fetch is done by the crawl4ai upstream’s headless browser, which re-resolves DNS and follows redirects on its own — so a determined attacker can still defeat this via DNS rebinding or a redirect to a private target. The crawl4ai egress NetworkPolicy (which blocks RFC1918 + link-local) is the load-bearing control; this check is defense in depth that rejects the common cases early.
Client wraps http.Client with retry-on-429/5xx and Retry-After honoring.
type Client struct {
HTTP *http.Client
// OnAttempt, when non-nil, is called once per HTTP attempt DoRetry makes,
// with the upstream this client is labeled for (see WithUpstream). retry is
// false for the first try and true for every retry, so a caller can count
// retry volume — the wasted work #2's RetryableStatus veto cuts. Set on the
// shared crawl client; nil elsewhere.
OnAttempt func(upstream string, retry bool)
// OnUpstream, when non-nil, is called once per HTTP attempt with the
// round-trip duration — request start until the response body is fully read
// (or closed), or until the transport error. status is "ok" for 2xx and
// "error" otherwise (non-2xx status, transport error, timeout).
OnUpstream func(upstream, op, status string, duration time.Duration)
// OnRetryAfter, when non-nil, is called once per response that pairs a 429
// or 503 with a parseable Retry-After, with the request URL and the delay
// the upstream asked for. It fires whatever the retry loop then does —
// including when a RetryableStatus veto or exhausted attempts end the
// request — because the point is to keep the upstream's answer AFTER this
// request dies: wired to a limiter, the next caller waits instead of walking
// into the same wall. This client reports the fact; the cap and the policy
// live at the wiring site. The retry loop's own capped backoff is unchanged.
OnRetryAfter func(upstream, rawURL string, wait time.Duration)
// contains filtered or unexported fields
}
func New(c *http.Client) *Client
New returns a Client wrapping the given http.Client. If nil is passed, a 90s-timeout client is used.
func (c *Client) DoRetry(ctx context.Context, method, url string, body []byte, headers map[string]string, cfg RetryConfig) (*http.Response, error)
DoRetry sends an HTTP request with exponential-backoff-with-jitter retries on transient failures. It retries on network errors, 429, and 5xx. 4xx other than 429 and context cancellation are not retried. A non-nil RetryConfig.RetryableStatus can additionally veto retrying a specific 429/5xx (e.g. a non-transient block), returning the StatusError immediately.
Body is passed as a byte slice (or nil) so the helper can rebuild the request on each attempt — http.Request bodies are single-use streams.
Honors Retry-After when the server provides it (capped at MaxDelay). Caller is responsible for closing the returned response body.
func (c *Client) WithUpstream(upstream, op string) *Client
WithUpstream returns a shallow copy of c whose attempts are labeled with the given upstream/op pair (e.g. “crawl4ai”/”crawl”). The copy shares the underlying http.Client and hooks, so adapters declare their identity once at construction. Returns nil when c is nil (engines built without a client).
DomainLimiter caps concurrency and enforces a minimum delay between successive requests to the same domain. The delay carries a small random jitter so bursts don’t synchronize across goroutines. It can additionally cap how many requests are admitted within a rolling window (see NewDomainQuotaLimiter).
type DomainLimiter struct {
// OnWait, when non-nil, is called on every Acquire exit with the engine
// that waited, the outcome, and the time spent blocked (semaphore wait +
// politeness delay); ~0 when uncontended. Set once at wiring time.
//
// Three outcomes: "acquired"; "canceled" when the caller's ctx died in the
// queue — the worst waits, which never acquire; and "budget_exceeded" when
// the computed wait was longer than the caller's remaining deadline, so
// Acquire refused to queue at all (see WaitBudgetError). The last one is
// the cheap failure: it costs ~0 seconds, not the caller's whole budget.
OnWait func(engine, outcome string, waited time.Duration)
// contains filtered or unexported fields
}
func NewDomainLimiter(maxConcurrent int, minDelay time.Duration) *DomainLimiter
NewDomainLimiter returns a limiter with the given concurrency cap and per-domain minimum delay, and no rolling-window cap.
func NewDomainQuotaLimiter(maxConcurrent int, minDelay time.Duration, quota int, window time.Duration) *DomainLimiter
NewDomainQuotaLimiter returns a limiter that also admits at most `quota` requests per domain within any rolling `window`. quota <= 0 disables that cap, making this identical to NewDomainLimiter.
A minimum delay alone cannot express the limit some upstreams actually enforce. Measured against this deployment’s search pool on 2026-08-17: one engine kept answering at a 3s spacing from a quiet start, yet blocked after ~20 requests in ~85s — it counts requests in a window, not the gap between them. A 3s delay run continuously sends 30 requests per 90s and trips it, so the two controls are complementary: the delay shapes the gap, the quota bounds the burst.
func NewSpacedDomainQuotaLimiter(maxConcurrent int, minDelay time.Duration, quota int, window time.Duration) *DomainLimiter
NewSpacedDomainQuotaLimiter returns a limiter that spaces SENDS instead of gaps: minDelay is measured from the previous caller’s admission rather than from its completion, so N callers are admitted minDelay apart and up to maxConcurrent of them run at once.
The distinction only matters above maxConcurrent 1, and there it is the difference between working and not. NewDomainQuotaLimiter measures from lastSend, which is written on Release, so on an idle slot every waiting goroutine reads the same stale instant, computes wait 0 and sends together — maxConcurrent 8 emits an 8-wide burst rather than eight sends 8×minDelay apart. That is why every search limiter built on the completion-spaced constructor passes maxConcurrent 1.
The crawl limiter is the exception and does NOT: it is built completion-spaced with OMNIFEED_PER_DOMAIN_CONCURRENCY (default 2) and a 1500ms delay, so two requests can reach a cold domain at once despite that gap. Left alone on purpose — the gap it protects was measured against the completion-spaced shape, and moving it is a change to crawl politeness that needs its own measurement, not a side effect of this one.
This mirrors redislimit’s ClusterConcurrency > 0 mode, one process instead of a cluster. Prefer it for an upstream that counts requests in a window (search engines do); prefer the completion-spaced limiter for one that punishes short gaps, since serialized spacing is the politer shape.
func (d *DomainLimiter) Acquire(ctx context.Context, engine, rawURL string) (func(), error)
Acquire blocks until a slot is available and the minimum delay since the last request to the same domain has elapsed, or until ctx is done — a canceled caller must not keep queuing behind a slow domain. When ctx carries a deadline and the pacing wait is longer than what is left of it, Acquire returns *WaitBudgetError immediately instead of queueing (fail fast, the wait is reported so the caller can retry later). engine names the caller for the wait metric (the limiter itself only knows hosts). Caller must Release (call the returned func) when done; on error there is nothing to release.
func (d *DomainLimiter) Penalize(rawURL string, dur time.Duration)
Penalize holds the domain back for d, on top of whatever the pacing settings already impose — what an upstream’s Retry-After on a 429 or 503 is worth. It only ever extends an existing hold, so the strictest answer any upstream gave stands, and it never shortens one. A non-positive d does nothing.
Nothing is released here: this is a note left for the next Acquire, not an admission.
FallbackLimiter runs Primary while it is healthy and falls back to Fallback when Primary’s backend is down.
The distributed limiter’s whole point is to share pacing state between replicas, and its backend (Redis) is a single point of failure. Retrieval must never depend on it: a Redis outage degrades pacing to per-pod limits (the behavior omnifeed had before it was introduced), it does not fail a crawl. Hence fail OPEN — Primary errors are absorbed, not returned.
On the first ErrLimiterUnavailable the circuit opens: this acquire and every later one are served by Fallback until Cooldown elapses, then the next acquire probes Primary again. Context errors are the caller’s own timeout, not backend health, so they pass through and leave the circuit alone.
Two costs of failing open, both accepted. Fallback keeps its own state, so a failover lands on a COLD local window (no lastSend, no admissions recorded): the first moments of degradation can burst up to the local limits on top of what Redis already admitted in the same window. It is transient and bounded by the per-pod settings. And the probe is not one request: every concurrent caller passes primaryReady once the cooldown expires, so a dead Redis costs N concurrent timeouts per cooldown, not exactly one.
type FallbackLimiter struct {
Primary Limiter
Fallback Limiter
// Cooldown is how long to stay on Fallback after a Primary failure.
// 0 means defaultLimiterCooldown.
Cooldown time.Duration
// OnDegraded, when non-nil, is called on TRANSITIONS only: true when the
// circuit opens, false when a probe finds Primary healthy again. It is the
// single hook for logging and metrics — this type stays logger-free so it
// can be unit-tested with stubs. Set once at wiring time.
OnDegraded func(down bool)
// contains filtered or unexported fields
}
func (f *FallbackLimiter) Acquire(ctx context.Context, engine, rawURL string) (func(), error)
Acquire admits one request through whichever backend is currently serving. The release func returned always belongs to that same backend, so a caller can never release a slot it did not take.
func (f *FallbackLimiter) Penalize(rawURL string, d time.Duration)
Penalize holds a host back on BOTH backends — what an upstream’s Retry-After on a 429 or 503 is worth. Both, because the fallback’s state must stay warm: a penalty recorded only in Redis is forgotten the moment the circuit opens, which is exactly when an upstream that is already refusing traffic gets hit with per-pod pacing.
The primary is skipped while the circuit is open, for the same reason acquires skip it: with Redis down every penalty would otherwise spend a whole RedisTimeout in the response path. The fallback is penalized always.
A backend that cannot be penalized is skipped rather than refused, so a test stub or a future Limiter without the method still composes.
Limiter admits one outbound request, blocking until the pacing policy allows it or ctx dies. engine names the caller for the wait metric, rawURL selects the domain being paced. The returned release func must be called when the request is done; on error there is nothing to release.
Implementations: *DomainLimiter (in-process, the default), redislimit.Limiter (state shared between replicas via Redis) and *FallbackLimiter (fail-open composite of the two). A nil Limiter means pacing is disabled — every call site checks for it.
type Limiter interface {
Acquire(ctx context.Context, engine, rawURL string) (release func(), err error)
}
RetryConfig controls per-request retry behavior. Zero values use defaults.
type RetryConfig struct {
MaxAttempts int // total attempts including the first try
BaseDelay time.Duration // first backoff interval
MaxDelay time.Duration // cap on any single backoff
// RetryableStatus, when non-nil, is consulted for a retryable status (429 or
// 5xx) before another attempt is scheduled. Returning false stops retries and
// surfaces the StatusError immediately — used to avoid re-driving an expensive
// crawl for a non-transient block that an upstream reports as a 5xx.
RetryableStatus func(status int, body string) bool
}
StatusError reports a non-2xx HTTP status returned by an upstream after retries were exhausted (429 / 5xx). It lets callers classify the failure by code via errors.As instead of parsing error text.
type StatusError struct {
StatusCode int
// Body is a bounded snippet of the upstream's error-response body, captured
// when retries are exhausted. It lets a caller tell an anti-bot block served
// as a 5xx (crawl4ai's detector) from a genuine upstream fault without
// re-reading the response. Empty when the body was absent or unreadable.
Body string
}
func (e *StatusError) Error() string
WaitBudgetError reports that the pacing wait a limiter computed is longer than the time the caller’s context has left, so nothing was admitted and the caller was not made to queue for a slot it could never use. The alternative is what omnifeed did before: hold the caller for its whole budget and then fail it with a deadline error anyway, which costs an agent the wait AND the answer.
It is a pacing VERDICT, not a backend failure: it is deliberately NOT wrapped in ErrLimiterUnavailable, so *FallbackLimiter passes it through untouched instead of absorbing it and opening its circuit against a healthy backend.
RetryAfter is the wait that was refused — how long the caller must leave before the same request can be admitted.
type WaitBudgetError struct {
RetryAfter time.Duration
}
func (e *WaitBudgetError) Error() string
Generated by gomarkdoc