aquifer

package module
v0.10.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 7, 2026 License: MIT Imports: 35 Imported by: 0

README

Aquifer — Load balancer for agentic workloads

Increase your rate limit without DDoSing your backend.

Distributed agents call tools and APIs in bursts. Your backend gets overwhelmed on inbound. Your app gets 429s on outbound. One slow dependency takes everything else down with it, and the retries agents fire off while they wait only make it worse — wasted utilization and higher cost on one end, outages reactive autoscaling alone can't prevent on the other.

Aquifer gives those agents a coordination layer: a self-hosted load balancer that absorbs the burst, queues requests durably (SQLite by default, or Pebble — see below), and releases them at a rate you configure — or a slower one, if the destination service asks for it.

What's usually behind that backend can't scale instantly either — a GPU, a database, a CI runner. Aquifer buys time for more of it to come online; it's overkill if that ceiling is fixed for good.

Exposed through pluggable adapters — an MCP server for agent tool-calling, a plain HTTP API, or an A2A (Agent2Agent protocol) agent — with cryptographic agent identity via the L8 protocol for trustless webhook delivery.

Benchmarked: 10x traffic spikes absorbed with zero failures, 30/30 jobs surviving a kill -9 mid-drain, and clean 429 admission shedding under sustained overload — including a real GPU under load, where the ORCA fallback signal cut peak backend queue depth from 449 to 8 waiting requests. See benchmark.md for throughput ceilings, crash recovery, memory behavior, capacity by machine size, and the GPU/vLLM run.

Use cases

MCP tools — coordinate distributed agents

agents / MCP clients  →  aquifer_enqueue_job  →  Aquifer queue  →  target API

Agents call Aquifer as an MCP server instead of racing each other directly against the same backend or external API. Aquifer returns a job id immediately, dispatches the request at a controlled rate, and delivers the result to your webhook.

HTTP API — protect your API

agents / clients  →  POST /jobs to Aquifer  →  your backend (at controlled RPS)

Agents hammering your API over HTTP? Aquifer queues their requests and drains them to your backend at a pace it can handle. Your backend returns X-Aqueduct-Rps headers to signal how fast it wants traffic in real time.

Outbound — a durable checkpoint in front of a rate-limited resource

your app  →  POST /jobs to Aquifer  →  database / CI runner / OpenAI / Stripe / any rate-limited API

Calling something with its own capacity limit — a database read replica, a CI runner, a third-party API? Aquifer queues the calls durably and dispatches them at your configured rate, so a burst from your own side never becomes the thing that takes the downstream down. Works especially well closed-loop: if the downstream already speaks X-Aqueduct-* headers, it can tell Aquifer to back off in real time instead of you guessing a static rate.

Edge load balancer → gateway — pace and route at the edge

your users  →  POST /proxy to Aquifer  →  your resources (paced, routed, at the speed you can handle)

Point Aquifer at your resources like a normal reverse proxy, close to the caller. It tries the request directly first — a healthy resource sees no queue at all — and only falls back to durable queuing when something's actually overloaded, on the same connection, staying in queue mode until that domain's backlog is genuinely drained (not just until a cooldown timer expires) — see POST /proxy for the details, including the header an upstream can use to request queuing proactively.

Sleep through partial outages. On Fly.io, set AQUIFER_FLY_REGIONS and that same overload signal drives real cross-region redirect: other regions Aquifer is deployed to get tried live, over Fly's private network, before this instance ever falls back to its own local queue — nearest-first by measured latency, deterministic enough that two callers racing the same job converge on the same region instead of each chasing their own nearest option. One region degrading routes around itself instead of paging you at 3am. If every known region is down too, that's a real fleet-wide problem — Aquifer says so with a 429 and a long Retry-After rather than quietly queueing it somewhere and hoping. See POST /proxy's "Cross-region redirect" section for the full mechanics, including the one honestly-documented tradeoff it doesn't try to hide.

Why not just Envoy or nginx rate limiting? A static rate limit dispatches as fast as the number allows, with no memory and no way to adapt if the backend is struggling worse than that number assumed — and nothing durable, so a rejected or in-flight request is just gone. Aquifer persists every job before dispatching it, adapts its pace live from what the backend actually says, and can retry against a sibling region if the whole region degrades.

In all four, the upstream can lower the dispatch pace via response headers — see Dynamic Pacing for how the ceiling, backoff, and recovery actually work.

Long-term protocol goal: if more services emit X-Aqueduct-*, agents can respond to capacity signals instead of independently guessing retry and concurrency behavior. Aquifer works today without ecosystem adoption; broader protocol adoption is the longer-term goal.

How it works

  1. Client submits a job through an adapter (MCP tool or HTTP endpoint) and moves on
  2. Aquifer persists it to SQLite — survives crashes, re-dispatches on restart
  3. A per-upstream worker dispatches at your configured RPS with jitter
  4. On completion Aquifer POSTs your webhook with the response body and status
  5. The upstream can adjust the rate live via X-Aqueduct-* response headers

Delivery semantics: Aquifer provides at-least-once dispatch and webhook delivery, not exactly-once execution. If Aquifer crashes after a dispatch succeeds but before it records that completion, the recovered job dispatches to the upstream again on restart — so it's not just the webhook that can repeat, the upstream call itself can. Make both your upstream endpoint and your webhook handler idempotent on job_id (or idempotent_key) anywhere duplicate execution isn't safe, the same contract Stripe and GitHub webhooks already ask of you.

Quick start

Binary

go install github.com/rjpruitt16/aquifer/cmd/aquifer@latest
aquifer

Docker

docker run -p 8080:8080 -v $(pwd)/data:/data \
  -e AQUIFER_ADAPTER=http \
  -e DB_PATH=/data/aquifer.db \
  ghcr.io/rjpruitt16/aquifer

Fly.io

git clone https://github.com/rjpruitt16/aquifer
cd aquifer
flyctl launch --name my-aquifer --no-deploy
flyctl volumes create aquifer_data --size 1 --region iad
flyctl deploy

Cross-platform release binaries (linux/darwin, amd64/arm64) are attached to every GitHub Release — no Go toolchain required if you'd rather grab one directly.

Configuration

Rate limits are set per upstream hostname via a YAML file (CONFIG_PATH) and admission/runtime behavior via env vars.

Full config reference — YAML shape, env var table, admission defaults

Set CONFIG_PATH to a YAML file to configure rate limits per upstream hostname:

# aquifer.yml — copy from aquifer.example.yml
defaults:
  rps: 2
  max_concurrent: 1

upstreams:
  api.openai.com:
    rps: 10
    max_concurrent: 3
  api.stripe.com:
    rps: 20
    max_concurrent: 5
  your-backend.internal:
    rps: 50
    max_concurrent: 10
Env var Default Description
AQUIFER_ADAPTER http for binary, mcp-stdio in Docker image Runtime adapter: http, mcp-stdio, or a2a
AQUIFER_A2A_PUBLIC_URL http://localhost:$PORT A2A adapter only — externally-reachable base URL advertised in the Agent Card
PORT 8080 HTTP listen port
DB_PATH aquifer.db Storage path — a SQLite file, or a directory if AQUIFER_STORE_BACKEND=pebble
CONFIG_PATH (none) Path to rate limit config YAML
AQUIFER_STORE_BACKEND sqlite Storage engine: sqlite or pebble (opt-in, pure-Go LSM store — see benchmark.md for why you might want it)
AQUIFER_PEBBLE_WAL_SYNC_INTERVAL_MS 5 Pebble only — batches concurrent durable writes into fewer real fsyncs under load (Pebble's own group-commit); each caller still blocks until its own write is actually durable
AQUIFER_MEMORY_LIMIT_MB (none, disabled) Reject new jobs with 429 once process memory exceeds this many MB
AQUIFER_MAX_BODY_BYTES 1048576 (1MB) Reject oversized request bodies with 413
AQUIFER_DB_MAX_BYTES 838860800 (800MB) Reject new jobs with 429 once the SQLite file exceeds this size
AQUIFER_RETRY_AFTER_SECONDS 5 Base Retry-After value sent on 429 admission rejections
AQUIFER_IDLE_TIMEOUT_SECONDS 300 (5min) How long a per-tenant/per-domain queue can sit idle before self-tearing-down — see drain mode for why this gates a real drain flush
AQUIFER_ALLOWED_URL_DOMAINS (none, unrestricted) Comma-separated hostnames url-routed jobs are permitted to target — see POST /jobs
AQUIFER_FLY_REGIONS (none, feature off) Comma-separated Fly region codes this app is deployed to — enables /proxy's cross-region redirect on Fly. See POST /proxy
AQUIFER_FLY_POLL_INTERVAL_SECONDS 30 How often to poll sibling regions over Fly's private network for liveness
AQUIFER_REDIRECT_GATE_COOLDOWN_SECONDS 500 How long to stop attempting cross-region redirect after a tour finds no reachable region at all, before trying again — internal probe throttling, not what's told to the caller
AQUIFER_REDIRECT_EXHAUSTED_RETRY_AFTER_SECONDS 900 (15min) Retry-After sent to the caller when cross-region redirect is configured but exhausted — no known-live region could serve or queue the request. Request is rejected (429), not queued locally. See POST /proxy

Body-size and DB-size admission are on by default; memory admission stays off until you set a limit, since a safe default depends on your own deployment, not Aquifer's disk usage. Retry-After backs off exponentially under sustained rejection (5s → 10s → 20s → 40s → capped at 60s, resets on the next allowed request). See CONFIGURATION.md for the full rationale and benchmark.md for the numbers behind these defaults.

Framework adapters

Aquifer has a framework-neutral core — idempotency, persistence, rate control, dispatch, SSE events, L8 signing, webhook delivery — with pluggable front doors:

Adapter Env Purpose
HTTP AQUIFER_ADAPTER=http REST/SSE API on PORT (the default)
MCP stdio AQUIFER_ADAPTER=mcp-stdio MCP server exposing Aquifer tools over stdio (the published Docker image's default)
A2A AQUIFER_ADAPTER=a2a Agent2Agent protocol (v1.0) agent over JSON-RPC/HTTPS

ADAPTERS.md has the full reference for each built-in adapter (MCP tool list, A2A Agent Card details), plus how to write your own FrameworkAdapter, storage backend, or metrics adapter.

Dynamic Pacing

Terminology: Aquifer is this implementation; Aqueduct is the implementation-agnostic header protocol (X-Aqueduct-*) it speaks, so other services could speak it too.

The upstream controls pace at runtime via response headers — X-Aqueduct-Rps, X-Aqueduct-Max-Concurrent, and per-tenant queue isolation — and Aquifer honors a lower pace immediately, recovering gradually once pressure clears. For backends that can't speak Aqueduct directly, Aquifer also reads the real open ORCA standard as a fallback signal — vLLM and Triton/TensorRT-LLM both work today, verified against their actual source.

Full pacing reference — header table, account-queue isolation, ORCA details
Header (preferred) Alias Effect
X-Aqueduct-Rps X-Aquifer-Rps Reduce dispatch rate to this value
X-Aqueduct-Max-Concurrent X-Aquifer-Max-Concurrent Reduce max in-flight requests
X-Aqueduct-Account-Queue X-Aquifer-Account-Queue enabled — isolate each tenant's queue
X-Aqueduct-Slow-Start X-Aquifer-Slow-Start true — new queues ramp up instead of firing at full rate immediately

Aquifer reads both namespaces, preferring X-Aqueduct-* when both are present.

With X-Aqueduct-Account-Queue: enabled, each (user_id, api_key) pair gets its own independently paced queue, so one tenant's burst can't slow down another. Each queue's pace still stays inside the upstream's actual budget — a background check throttles the sum of every active tenant queue proportionally if too many are active at once, so isolation never means an unbounded copy of the full rate per tenant.

A backend can lower RPS at any time via these headers when it's under pressure; Aquifer honors the lower pace immediately and recovers gradually toward the configured ceiling once pressure clears.

With X-Aqueduct-Slow-Start: true, a new queue starts at a low floor rate instead of its full configured rate and climbs toward that ceiling using the same gradual-recovery mechanism above, rather than firing at full speed from its very first dispatch. This applies per domain: since a queue's first-ever dispatch has no prior response to read the signal from, the setting takes effect on the next new queue created for that domain once any response has carried it — not the request that carried the header itself, and not retroactively for queues already running.

Use the pacing headers for intentional backpressure. A 5xx response is treated as a failed dispatch attempt and, for pool members, lowers that member's reputation. If a service is alive but overloaded, prefer 429 and/or X-Aqueduct-Rps / X-Aqueduct-Max-Concurrent so Aquifer slows down without interpreting the member as broken.

ORCA fallback for backends that can't speak Aqueduct directly. Some backends already report load in a different, real open standard — ORCA (Open Request Cost Aggregation), the gRPC/Envoy ecosystem's convention for backends to report utilization. Aquifer sends endpoint-load-metrics-format: text on every dispatch (the request-side opt-in both verified backends require — there's no server startup flag for this), and a backend that understands it replies with an endpoint-load-metrics header carrying a KV-cache utilization fraction. If a response carries no X-Aqueduct-Rps/X-Aquifer-Rps, Aquifer reads this header as a fallback and paces down as utilization rises: full configured rate below 70%, 2 RPS at 70-90%, 0.5 RPS at 90-97%, 0.25 RPS above that — never dropping to zero, same pacing-down-gracefully philosophy as everywhere else. An explicit X-Aqueduct-Rps always wins if present; this only fires when the backend hasn't opted into speaking Aqueduct's own headers.

Two backends verified directly against their own source, not assumed: vLLM (vllm/entrypoints/serve/utils/orca_metrics.py, metric name kv_cache_usage_perc, case-insensitive opt-in) and Triton/TensorRT-LLM (src/orca_http.cc, metric name kv_cache_utilization, case-sensitive lowercase-only opt-in — Aquifer sends lowercase specifically so both work). Aquifer tries both known metric names, so whichever backend you're running, this works without any configuration.

API

POST /jobs
{
  "user_id":        "user-123",
  "idempotent_key": "invoice-42-notify",
  "url":            "https://api.openai.com/v1/chat/completions",
  "method":         "POST",
  "headers":        { "Authorization": "Bearer sk-..." },
  "body":           "{\"model\":\"gpt-4o\",\"messages\":[...]}",
  "webhook_url":    "https://yourapp.com/webhooks/aquifer"
}

Do not expose Aquifer directly to untrusted callers. url is dispatched as a real HTTP request — if an arbitrary or untrusted party can set it, Aquifer becomes an open relay/SSRF vector, using Aquifer's own network position and identity to reach anything the machine can reach. The intended caller is your own trusted backend or gateway code dispatching to a destination it already knows about — not an agent, end user, or any other party choosing the destination itself.

API.md has the full reference: GET /jobs/:id, the SSE stream, POST /proxy (edge-gateway mode — see Use cases), GET /health, webhook payload shapes, and the autoscaling headers.

Agent-native load balancing

Instead of dispatching to a fixed url, a job can target a named pool — a group of registered service instances Aquifer picks from at dispatch time, weighted by declared capacity and live reputation. Useful when you have several interchangeable backends instead of one fixed endpoint, and it grows or shrinks automatically as members register, degrade, or drop out — no need to reconfigure Aquifer as your fleet autoscales.

Full pool reference — registration, dispatch, reputation model, scenario harness

Registering a member:

curl -X POST https://your-aquifer/pools/writers/members \
  -d '{"member_id": "writer-1", "address": "http://10.0.1.5:8080", "capacity_rps": 20, "heartbeat_interval_seconds": 30}'

The same call is both initial registration and heartbeat — call it again periodically (at roughly your declared heartbeat_interval_seconds) to stay in the pool. Missing several consecutive expected heartbeats evicts a member. A member can register under more than one pool id.

Dispatching to a pool:

{
  "user_id": "user-123",
  "idempotent_key": "job-1",
  "pool_id": "writers",
  "method": "POST",
  "webhook_url": "https://yourapp.com/webhooks/aquifer"
}

pool_id and url are mutually exclusive — a job sets exactly one.

How a member gets picked: proportional to capacity_rps × reputation, not equal-split round robin — a member declaring 100 RPS gets roughly 4x the dispatches of one declaring 25. The pool's aggregate ceiling is the live sum of every member's current effective rate, so it grows and shrinks automatically as members register, degrade, or drop out — no need to reconfigure Aquifer as your fleet autoscales.

Reputation: a dispatch failure halves a member's effective share; a successful dispatch nudges it back up, and heartbeats recover it more slowly after a restart. A member isn't evicted on one bad response — only once its reputation has stayed at the floor continuously, with no interrupting success, for a sustained window. This avoids flapping a member in and out of the pool over a single transient error.

Treat 5xx carefully. Aquifer interprets connection errors and 5xx responses as reliability signals for the selected member. One 5xx does not fail the job by itself — Aquifer records failure for that member and retries another member when possible. If every retry across the pool still ends in connection errors or 5xx, the job is marked failed. That behavior is intentional for reliability, but it means application bugs that accidentally return 5xx on a new code path can reduce that member's traffic share or eventually remove it from the pool.

For overload, prefer explicit backpressure over generic server errors:

Situation Recommended signal
Instance is healthy but needs less traffic X-Aqueduct-Rps or X-Aqueduct-Max-Concurrent
Request should be retried later due to pressure 429 with Retry-After
Instance/code path is actually failing 5xx

Roll new members into a pool gradually. Start new versions with a conservative capacity_rps, send a small share of traffic first, watch /health reputation and your own error metrics, then raise capacity as confidence grows. Blue/green or canary rollout matters more here than with a blind round-robin balancer because Aquifer uses runtime failures as routing input.

Set capacity_rps conservatively, not at your true theoretical max. Aquifer only learns a member died via a failed dispatch or a missed heartbeat, both of which lag the actual failure — leaving headroom in what you declare gives real slack for that detection delay. Reputation decay is a second line of defense on top of this: a member that's silently struggling gets throttled down by observed failures even if its last-declared capacity was optimistic.

Watch the model locally: Aquifer includes a local scenario harness that starts one fake backend server with multiple logical workers, registers them as pool members, and prints per-second traffic, failures, dynamic header values, and reputation.

go run ./cmd/aquifer-scenario --scenario mixed --workers 10 --jobs 500 --duration 30s --rps 50

Scenarios: steady, weighted, flapping, backpressure, recovering, mixed, and harsh. The harsh scenario penalizes sustained overload by adding latency, then 5xx, then simulated crash windows. Add --mode regular to compare against a simple round-robin load balancer model that retries on 5xx but ignores Aquifer reputation and dynamic pacing headers.

Pool state isn't shared across Aquifer instances — see Deployment model for how that constrains a given pool_id to one instance.

GET /health reports every pool's current members, their declared capacity, and current reputation.

L8 Protocol — trustless webhook delivery

Traditional webhook security shares an HMAC secret between sender and receiver, stored in a database on both sides — something that can be stolen, logged accidentally, or forgotten during rotation, letting anyone forge deliveries forever once it leaks. Aquifer implements L8 v0.1, a lightweight challenge-response protocol that replaces the shared secret with public key cryptography: the receiver publishes a public key, a one-time handshake proves both sides own their private keys, and every delivery afterward carries a signature verified locally in microseconds — no database lookup, no round-trip to any authority.

The full protocol rationale, wire format, and a reference receiver implementation live at the L8 spec — also served locally at GET /l8-spec for an agent/script with only network access to this instance. Set L8_PRIVATE_KEY for a stable identity across restarts, or let Aquifer auto-generate one on first start.

Reliability

Durable queue, automatic crash recovery, panic isolation per job — see benchmark.md for the numbers behind these claims.

Full reliability reference — mechanisms, job TTLs
  • Durable queue — jobs persist to the configured storage backend on every write
  • Crash recovery — queued jobs re-dispatched automatically on restart
  • In-flight tracking — jobs marked in_flight before dispatch; recovered immediately on panic without waiting for full restart
  • Stale job safety net — in-flight jobs older than 5 min automatically reset to queued
  • Per-job panic isolation — a panic in one job marks it failed and delivers the webhook; the worker keeps running

Job TTLs:

Status TTL
queued 24 h
completed 30 min
failed 2 h

Partitioning strategies

Running one instance for everything works fine until you have multiple tenants or multiple upstreams sharing it — then one tenant's burst, or one upstream's own rate limit, ends up affecting everyone else on that same instance. Two ways to split traffic apart so that doesn't happen, not mutually exclusive:

Static partitioning — decided once, at deploy time: dedicate one instance to a single protected resource — a CI runner, a database, a GPU, or a rate-limited external API you want to be nice to — so that resource only ever sees traffic paced the way you configured, up to whatever it can actually bear. Multiple tenants can safely share that same instance: turn on account-queue isolation and each tenant gets their own independently-paced queue, so one tenant's burst doesn't starve another's, and the resource itself never sees more aggregate load than it's rated for. The mistake to avoid: pointing multiple instances at the same resource instead of routing everyone through this one pacing checkpoint — that just multiplies your total request rate against it. Same rule for pools: a given pool_id should belong to exactly one instance, since pool state isn't shared across instances.

Optional HTTP cluster routing can hash user_id across a static member list so callers can hit any node and still land on that user's owner. See API.md for config and caveats.

For a regional deployment that does not need a separate control plane, combine static cluster routing with Valkey remote idempotency: any node can receive the request, consistent hashing keeps a user's normal traffic on one owner, and AQUIFER_DRAIN_SINK=valkey publishes completed/failed idempotency records under the generic aqueduct:idempotency: prefix so another node can reject duplicates before dispatch. See API.md for the exact key contract.

Dynamic partitioning (drain mode) — off by default, for a more specific shape: instead of deciding every assignment up front, an instance gets handed to one tenant at a time, absorbs and drains whatever burst that tenant sends, then frees itself up to be handed to a different tenant next — useful when you want dedicated capacity per user without hand-assigning it at deploy time. Aquifer can stream completed/failed job ledger events in acknowledged batches to a webhook or Valkey, and when idle for AQUIFER_DRAIN_TIMER_SECONDS, it flushes anything remaining before clearing local state and moving through an active → draining → unassigned state machine visible via GET /health. See DRAIN_MODE.md for the full state machine, env vars, and payload shape.

The two combine: a fleet can partition statically by upstream domain, while individual instances within a partition cycle through tenants dynamically via drain mode.

External registration — off by default, and orthogonal to the above: AQUIFER_REGISTRY_URL makes an instance periodically report its own listening port to an external control plane (deciding tenant assignment, scaling, etc. is entirely that service's job, not Aquifer's). See REGISTRATION.md for the env vars and ping payload shape.

Deployment model

Aquifer runs four ways: as a sidecar alongside your app, as a standalone service multiple services point to, embedded directly as a Go library in your own process (see Framework adapters), or as an extension behind a Gateway API proxy like Envoy Gateway in Kubernetes — the proxy owns routing and TLS, Aquifer owns the queue behind it (see examples/kubernetes). Each instance persists to its own SQLite volume — no external database or coordination service to run.

People run Aquifer in front of things like: internal coding platforms (GitLab, Forgejo), CI runners, database read replicas, and MCP servers — anywhere a burst of agent or service traffic needs to hit something that has its own capacity limit.

Full deployment reference — partitioning, scaling, security note

See Partitioning strategies above for how to assign tenants to instances, statically or dynamically.

See the security warning under POST /jobs — the same untrusted-caller risk applies regardless of deployment shape.

Choosing a machine size: see benchmark.md for current throughput, capacity by machine size, and the benchmark methodology.

Writing

License

MIT

Built by Rahmi Pruitt — open to AI infra consulting, founding engineer, and contract work.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrJobNotFound = errors.New("job not found")

Functions

func RunAdapter 

func RunAdapter(ctx context.Context, adapter FrameworkAdapter, opts RuntimeOptions) error

Types

type AccountQueue 

type AccountQueue struct {
	// contains filtered or unexported fields
}

func NewAccountQueue 

func NewAccountQueue(key, upstream string, rps float64, maxConc int, pool *Pool, store JobStore, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, enqueueWebhook webhookEnqueuer, onIdle func(string), slowStart bool, onSlowStartSignal func(bool)) *AccountQueue

func (*AccountQueue) Active added in v0.9.0 

func (q *AccountQueue) Active() bool

Active reports whether this queue currently has real backlog (queued or in-flight work) — used by proxy mode to decide whether a domain should keep routing through the durable queue even after its circuit breaker's cooldown has elapsed, since a cooldown timer alone doesn't know whether the backlog it caused has actually finished draining yet.

func (*AccountQueue) Enqueue 

func (q *AccountQueue) Enqueue(job *Job)

func (*AccountQueue) RPS 

func (q *AccountQueue) RPS() float64

func (*AccountQueue) Stop added in v0.10.0 

func (q *AccountQueue) Stop()

func (*AccountQueue) Throttle added in v0.4.0 

func (q *AccountQueue) Throttle(rps float64)

Throttle pushes an external rate adjustment into the queue's dispatch loop, reusing the same jobDoneMsg channel that header-driven pacing already uses — the loop doesn't need to know whether a lower rate came from the upstream's own response header or from the URLWorker capping this queue's share of a shared aggregate budget, it's the same signal either way.

type AdmissionController added in v0.3.0 

type AdmissionController struct {
	// contains filtered or unexported fields
}

AdmissionController evaluates memory and DB size limits at request time. Body size is enforced separately at the transport layer via http.MaxBytesReader, since that has to happen before the body is even read.

func NewAdmissionController added in v0.3.0 

func NewAdmissionController(limits AdmissionLimits, dbPath string) *AdmissionController

func (*AdmissionController) AnyLimitConfigured added in v0.3.0 

func (c *AdmissionController) AnyLimitConfigured() bool

AnyLimitConfigured reports whether at least one admission limit is actually active, as opposed to merely whether a controller instance exists (the runtime always constructs one, even with everything at zero).

func (*AdmissionController) Check added in v0.3.0 

func (c *AdmissionController) Check() AdmissionDecision

func (*AdmissionController) MaxBodyBytes added in v0.3.0 

func (c *AdmissionController) MaxBodyBytes() int64

func (*AdmissionController) RetryAfterSeconds added in v0.3.0 

func (c *AdmissionController) RetryAfterSeconds() int

RetryAfterSeconds returns the configured base value on the first rejection, then doubles for each additional consecutive rejection (capped at maxRetryAfterSeconds), resetting the moment a request is allowed again.

func (*AdmissionController) Snapshot added in v0.3.0 

func (c *AdmissionController) Snapshot() map[string]any

Snapshot reports current admission pressure for /health, independent of whether any request is currently being rejected.

type AdmissionDecision added in v0.3.0 

type AdmissionDecision struct {
	Allowed bool
	Reason  string // "memory" or "db_size"
	Limit   int64
	Current int64
}

AdmissionDecision is the result of an admission check on a new (non-duplicate) job.

type AdmissionLimits added in v0.3.0 

type AdmissionLimits struct {
	MemoryLimitMB     int64 // AQUIFER_MEMORY_LIMIT_MB
	MaxBodyBytes      int64 // AQUIFER_MAX_BODY_BYTES
	DBMaxBytes        int64 // AQUIFER_DB_MAX_BYTES
	RetryAfterSeconds int   // AQUIFER_RETRY_AFTER_SECONDS
}

AdmissionLimits are operator-configured ceilings that protect Aquifer itself from the traffic it's meant to be absorbing. A zero value disables that particular check. Memory is opt-in (LoadAdmissionLimits defaults it to disabled); body size and DB size default on — see LoadAdmissionLimits.

func LoadAdmissionLimits added in v0.3.0 

func LoadAdmissionLimits() AdmissionLimits

LoadAdmissionLimits reads the AQUIFER_* admission env vars. Body size and DB size fall back to sane, non-zero defaults when unset (see the default* constants above) rather than being disabled — Aquifer's whole purpose is protecting the process from the traffic it absorbs, so it protects itself by default rather than requiring that to be opted into. An explicit "0" still disables a given check. Memory has no safe one-size-fits-all default (it depends on the deployment's own memory budget, not Aquifer's benchmarked disk usage), so it stays disabled unless set explicitly — LoadAdmissionLimits logs a warning when it is.

type AdmissionRejectedError added in v0.3.0 

type AdmissionRejectedError struct {
	Decision AdmissionDecision
}

AdmissionRejectedError is returned by Aquifer.Enqueue when a genuinely new job is rejected due to memory or DB size pressure. Callers (the HTTP server) type-assert on this to build a 429 with Retry-After.

func (*AdmissionRejectedError) Error added in v0.3.0 

func (e *AdmissionRejectedError) Error() string

type Aquifer 

type Aquifer struct {
	// contains filtered or unexported fields
}

func NewAquifer 

func NewAquifer(store JobStore, registry *Registry, broker *Broker, l8 *L8Registry, admission *AdmissionController, pools *PoolRegistry) *Aquifer

func (*Aquifer) AdmissionSnapshot added in v0.3.0 

func (a *Aquifer) AdmissionSnapshot() map[string]any

AdmissionSnapshot reports current admission pressure for /health. Returns enabled:false if admission control isn't configured.

func (*Aquifer) AttemptDirect added in v0.9.0 

func (a *Aquifer) AttemptDirect(ctx context.Context, req JobRequest, timeout time.Duration) ProxyOutcome

AttemptDirect is proxy mode's entry point: persist the job (same idempotency/admission path Enqueue uses), then — for URL-based jobs only, see the PoolID check below — try dispatching it directly and synchronously before ever touching the durable queue. A direct attempt is skipped entirely if this job's target already has its circuit breaker open (see URLWorker.BreakerOpen), so a known-bad upstream doesn't cost every subsequent request the latency of a doomed attempt.

func (*Aquifer) Close added in v0.10.0 

func (a *Aquifer) Close()

func (*Aquifer) Dispatch added in v0.9.0 

func (a *Aquifer) Dispatch(job *Job, accountQueueHeader string)

Dispatch hands an already-persisted, admission-approved job to the durable paced queue — Enqueue's last step, exposed for a caller (proxy mode) that already ran PrepareJob itself.

func (*Aquifer) Enqueue 

func (a *Aquifer) Enqueue(req JobRequest) (EnqueueResult, error)

func (*Aquifer) GetJob 

func (a *Aquifer) GetJob(id string) (*Job, error)

func (*Aquifer) HandleL8Challenge 

func (a *Aquifer) HandleL8Challenge(req L8ChallengeReq) (*L8ChallengeResp, error)

func (*Aquifer) Health 

func (a *Aquifer) Health() map[string]any

func (*Aquifer) L8Metadata 

func (a *Aquifer) L8Metadata(host string) L8Meta

func (*Aquifer) MaxBodyBytes added in v0.3.0 

func (a *Aquifer) MaxBodyBytes() int64

MaxBodyBytes returns the configured request body ceiling, or 0 if unconfigured (unlimited).

func (*Aquifer) PrepareJob added in v0.9.0 

func (a *Aquifer) PrepareJob(req JobRequest) (job *Job, duplicate *EnqueueResult, err error)

PrepareJob validates, persists (idempotency-checked), and admission-checks a request without dispatching it — Enqueue's first two steps, exposed separately so a caller (proxy mode) can attempt a direct dispatch in between persistence and the durable-queue handoff. Returns a non-nil duplicate result if this idempotent_key already exists; job is nil in that case. Behavior is otherwise identical to Enqueue's first half.

func (*Aquifer) RegisterPoolMember added in v0.4.0 

func (a *Aquifer) RegisterPoolMember(poolID, memberID, address string, declaredRPS float64, heartbeatIntervalSeconds int) error

RegisterPoolMember adds or refreshes (heartbeats) a member of a pool. The same call serves both roles — re-registering resets the member's liveness TTL and updates its declared capacity.

func (*Aquifer) RetryAfterSeconds added in v0.3.0 

func (a *Aquifer) RetryAfterSeconds() int

RetryAfterSeconds returns the configured Retry-After value for 429 responses, defaulting to 5 seconds if admission control isn't configured.

func (*Aquifer) SetClusterRouter added in v0.10.0 

func (a *Aquifer) SetClusterRouter(router *ClusterRouter)

func (*Aquifer) SetRegionAdapter added in v0.9.0 

func (a *Aquifer) SetRegionAdapter(adapter RegionAdapter)

SetRegionAdapter wires in a RegionAdapter after construction -- kept separate from NewAquifer's constructor so adding this opt-in feature doesn't change NewAquifer's signature for every existing caller. A nil Aquifer.regionAdapter (the default) behaves as NoopRegionAdapter via regionAdapterOrDefault.

func (*Aquifer) SetRemoteIdempotency added in v0.10.0 

func (a *Aquifer) SetRemoteIdempotency(remote RemoteIdempotency)

func (*Aquifer) SubscribeJob 

func (a *Aquifer) SubscribeJob(id string) (*Job, <-chan SSEEvent, func(), error)

type Broker 

type Broker struct {
	// contains filtered or unexported fields
}

Broker is the pub/sub layer for SSE streams. Each job gets its own set of subscriber channels.

func NewBroker 

func NewBroker() *Broker

func (*Broker) Publish 

func (b *Broker) Publish(jobID string, event SSEEvent)

func (*Broker) Subscribe 

func (b *Broker) Subscribe(jobID string) (<-chan SSEEvent, func())

type ClusterConfig added in v0.10.0 

type ClusterConfig struct {
	Enabled           bool
	Self              ClusterMember
	Members           []ClusterMember
	PartitionCount    int
	ReplicationFactor int
	Load              float64
}

func LoadClusterConfig added in v0.10.0 

func LoadClusterConfig() ClusterConfig

type ClusterMember added in v0.10.0 

type ClusterMember struct {
	ID      string `json:"id"`
	Address string `json:"address"`
}

func (ClusterMember) String added in v0.10.0 

func (m ClusterMember) String() string

type ClusterRouter added in v0.10.0 

type ClusterRouter struct {
	// contains filtered or unexported fields
}

func NewClusterRouter added in v0.10.0 

func NewClusterRouter(cfg ClusterConfig) *ClusterRouter

func (*ClusterRouter) IsOwner added in v0.10.0 

func (r *ClusterRouter) IsOwner(key string) bool

func (*ClusterRouter) OwnerFor added in v0.10.0 

func (r *ClusterRouter) OwnerFor(key string) (ClusterMember, bool)

func (*ClusterRouter) Snapshot added in v0.10.0 

func (r *ClusterRouter) Snapshot() map[string]any

type Config 

type Config struct {
	Defaults  RateConfig            `yaml:"defaults"`
	Upstreams map[string]RateConfig `yaml:"upstreams"`
}

func LoadConfig 

func LoadConfig(path string) *Config

func (*Config) ForURL 

func (c *Config) ForURL(rawURL string) RateConfig

type DrainConfig added in v0.6.0 

type DrainConfig struct {
	Enabled              bool  // AQUIFER_DRAIN_ENABLED
	TimerSeconds         int64 // AQUIFER_DRAIN_TIMER_SECONDS
	Sink                 string
	WebhookURL           string
	BatchEnabled         bool  // AQUIFER_DRAIN_BATCH_ENABLED
	BatchIntervalSeconds int64 // AQUIFER_DRAIN_BATCH_INTERVAL_SECONDS
	BatchMaxEvents       int   // AQUIFER_DRAIN_BATCH_MAX_EVENTS
}

func LoadDrainConfig added in v0.6.0 

func LoadDrainConfig() DrainConfig

LoadDrainConfig reads the AQUIFER_DRAIN_* env vars. Enabled defaults to false. If enabled but no webhook URL is configured, that's treated as disabled (with a warning) rather than a flush attempt with nowhere to send it -- fail-safe toward "do nothing," never toward "clear the ledger anyway."

type DrainEvent added in v0.10.0 

type DrainEvent struct {
	Sequence   int64  `json:"sequence"`
	HashKey    string `json:"idempotent_key_hash"`
	JobID      string `json:"job_id"`
	Status     Status `json:"status"`
	RecordedAt int64  `json:"recorded_at"`
}

DrainEvent is the durable, acknowledged unit used by batched drain streaming. It intentionally carries the same hash-only ledger data as LedgerEntry, plus a monotonic local sequence so a receiver can treat batches idempotently and Aquifer can delete only acknowledged events.

type DrainState added in v0.6.0 

type DrainState string

DrainState is the instance's explicit position in drain mode's lifecycle, visible via GET /health (see Registry.DrainSnapshot). Only meaningful when drain mode is enabled. 

const (
	// DrainStateActive: at least one worker has live work. The normal
	// state for any instance, drain mode enabled or not.
	DrainStateActive DrainState = "active"
	// DrainStateDraining: every worker has gone idle, but either the drain
	// timer hasn't elapsed yet, or a flush attempt is in flight/being
	// retried. Not yet safe to hand off.
	DrainStateDraining DrainState = "draining"
	// DrainStateUnassigned: the ledger was successfully flushed (or there
	// was nothing to flush) and local state is clear. Safe to hand off to
	// a different tenant. Reverts to Active the instant new work arrives.
	DrainStateUnassigned DrainState = "unassigned"
)

type EnqueueResult 

type EnqueueResult struct {
	JobID     string `json:"job_id"`
	Status    Status `json:"status"`
	Duplicate bool   `json:"duplicate,omitempty"`
}

type FlyRegionAdapter added in v0.9.0 

type FlyRegionAdapter struct {
	// contains filtered or unexported fields
}

FlyRegionAdapter polls sibling regions over Fly's private 6PN network to know which are currently live, backing /proxy's cross-region redirect (proxy.go's AttemptDirect). Region enumeration is explicit config only (AQUIFER_FLY_REGIONS) -- Aquifer never silently polls regions a deployer didn't say to use.

func NewFlyRegionAdapter added in v0.9.0 

func NewFlyRegionAdapter() *FlyRegionAdapter

NewFlyRegionAdapter constructs the adapter and starts its background polling loop. Returns nil if AQUIFER_FLY_REGIONS isn't set -- the feature stays off unless explicitly configured, matching every other opt-in feature in this codebase. Callers should treat a nil return as "use NoopRegionAdapter" (ensureRegionAdapter already does this for a nil RegionAdapter passed to Aquifer.SetRegionAdapter).

func (*FlyRegionAdapter) Close added in v0.10.0 

func (a *FlyRegionAdapter) Close()

func (*FlyRegionAdapter) LiveRegions added in v0.9.0 

func (a *FlyRegionAdapter) LiveRegions() []string

func (*FlyRegionAdapter) SelfRegion added in v0.9.0 

func (a *FlyRegionAdapter) SelfRegion() string

type FrameworkAdapter 

type FrameworkAdapter interface {
	Name() string
	Start(ctx context.Context, aquifer *Aquifer) error
}

type HTTPAdapter 

type HTTPAdapter struct {
	// contains filtered or unexported fields
}

func NewHTTPAdapter 

func NewHTTPAdapter(addr string) *HTTPAdapter

func (*HTTPAdapter) Name 

func (a *HTTPAdapter) Name() string

func (*HTTPAdapter) Start 

func (a *HTTPAdapter) Start(ctx context.Context, aquifer *Aquifer) error

type Job 

type Job struct {
	ID            string            `json:"id"`
	UserID        string            `json:"user_id"`
	IdempotentKey string            `json:"idempotent_key"`
	URL           string            `json:"url,omitempty"`
	PoolID        string            `json:"pool_id,omitempty"`
	Method        string            `json:"method"`
	Headers       map[string]string `json:"headers,omitempty"`
	Body          string            `json:"body,omitempty"`
	WebhookURL    string            `json:"webhook_url"`
	Status        Status            `json:"status"`
	CreatedAt     int64             `json:"created_at"`

	// Cross-region /proxy redirect fields — see proxy.go's AttemptDirect.
	// Absent/zero on a fresh top-level request; that absence IS the signal
	// "I'm the origin, nobody redirected this to me." OriginMachineID set
	// to someone else's ID means this instance must NOT itself originate a
	// further redirect tour — it just runs the normal local direct-attempt-
	// then-queue path, exactly as if this feature didn't exist.
	OriginMachineID string   `json:"origin_machine_id,omitempty"`
	OriginRegion    string   `json:"origin_region,omitempty"`
	VisitedRegions  []string `json:"visited_regions,omitempty"`
	RerouteCount    int      `json:"reroute_count,omitempty"`
}

func NewJob 

func NewJob(r *JobRequest) *Job

type JobRequest 

type JobRequest struct {
	UserID        string            `json:"user_id"`
	IdempotentKey string            `json:"idempotent_key"`
	URL           string            `json:"url,omitempty"`
	PoolID        string            `json:"pool_id,omitempty"`
	Method        string            `json:"method"`
	Headers       map[string]string `json:"headers,omitempty"`
	Body          string            `json:"body,omitempty"`
	WebhookURL    string            `json:"webhook_url"`

	// Cross-region /proxy redirect fields — see Job's own doc comment and
	// proxy.go's AttemptDirect. Only ever set on an internal redirect hop
	// (one Aquifer instance calling another's /proxy directly); a real
	// caller never sets these.
	OriginMachineID string   `json:"origin_machine_id,omitempty"`
	OriginRegion    string   `json:"origin_region,omitempty"`
	VisitedRegions  []string `json:"visited_regions,omitempty"`
	RerouteCount    int      `json:"reroute_count,omitempty"`

	// DirectOnly is set on every redirect hop except the final one (the
	// deterministic-hash-selected target that's allowed to actually queue).
	// It tells the receiving instance "try a direct dispatch, but if you
	// can't, tell me cleanly — do NOT fall back to your own local queue."
	// Without this, a tour that moved on to a second candidate after a
	// first candidate quietly queued the job locally would leave the job
	// committed in two places at once — a real duplicate-delivery bug,
	// entirely within one origin's own tour, independent of the separate
	// cross-origin race already documented as an accepted gap. See
	// region_redirect.go.
	DirectOnly bool `json:"direct_only,omitempty"`

	// AccountQueueMode is never read from the request body — it's set by the
	// HTTP adapter from the X-Aqueduct-Account-Queue / X-Aquifer-Account-Queue
	// request header, the only source of truth for this setting. Empty means
	// "no opinion, leave the upstream's current mode unchanged."
	AccountQueueMode string `json:"-"`
}

func (*JobRequest) Validate 

func (r *JobRequest) Validate() string

type JobStore added in v0.4.0 

type JobStore interface {
	Path() string
	Close() error
	CheckOrInsert(job *Job) (string, bool)
	SetQueueKey(jobID, queueKey string)
	DeleteJob(jobID string)
	MarkInFlight(jobID string)
	RecoverInFlight(queueKey string) []*Job
	UpdateStatus(jobID string, status Status)
	Counts() StoreCounts
	GetJob(jobID string) *Job
	GetQueuedJobs() []*Job

	// ListIdempotentKeys and ClearIdempotentKeys back drain mode (see
	// drain.go) -- an opt-in feature, off by default, so calling these on
	// a deployment that never enables it is never reached.
	ListIdempotentKeys() []LedgerEntry
	ClearIdempotentKeys()
	ListDrainEvents(limit int) []DrainEvent
	AcknowledgeDrainEventsThrough(sequence int64)
}

JobStore is the storage backend behind everything that persists a job: idempotency, status transitions, crash recovery, and admission control's db-size check. *Store (SQLite/WAL) is the default and only implementation until now; *PebbleStore is an opt-in alternative for benchmarking whether a memory-first LSM store changes the throughput ceiling the way it did for the Elixir/Mnesia sibling of this project. Selected via AQUIFER_STORE_BACKEND ("sqlite", the default, or "pebble") — existing deployments that don't set it see no change at all.

func NewJobStore added in v0.4.0 

func NewJobStore(path string) JobStore

NewJobStore constructs whichever backend AQUIFER_STORE_BACKEND names. path is the same value that used to go straight to NewStore — for SQLite it's a file path, for Pebble it's a directory.

type L8ChallengeReq 

type L8ChallengeReq struct {
	ChallengeID     string `json:"challenge_id"`
	Nonce           string `json:"nonce"`
	Timestamp       int64  `json:"timestamp"`
	SenderPublicKey string `json:"sender_public_key"`
	Signature       string `json:"signature"`
}

type L8ChallengeResp 

type L8ChallengeResp struct {
	ChallengeID       string `json:"challenge_id"`
	Nonce             string `json:"nonce"`
	ReceiverSignature string `json:"receiver_signature"`
	ReceiverPublicKey string `json:"receiver_public_key"`
}

type L8Meta 

type L8Meta struct {
	ProtocolVersion   string   `json:"protocol_version"`
	ServiceName       string   `json:"service_name"`
	PublicKey         string   `json:"public_key"`
	ChallengeEndpoint string   `json:"challenge_endpoint"`
	SupportedAlgos    []string `json:"supported_algorithms"`
	Capabilities      []string `json:"capabilities"`
	SpecURL           string   `json:"spec_url"`
}

type L8Registry 

type L8Registry struct {
	PubB64 string // exported so server can embed in responses
	// contains filtered or unexported fields
}

func NewL8Registry 

func NewL8Registry(keyPath, trustDir string) *L8Registry

func (*L8Registry) Close added in v0.10.0 

func (r *L8Registry) Close()

func (*L8Registry) EnsureTrust 

func (r *L8Registry) EnsureTrust(webhookURL string)

EnsureTrust runs the L8 handshake with the webhook domain if not already trusted. Silently does nothing if the receiver doesn't support L8 — delivery still proceeds unsigned.

func (*L8Registry) HandleChallenge 

func (r *L8Registry) HandleChallenge(req L8ChallengeReq) (*L8ChallengeResp, error)

HandleChallenge verifies the sender's ownership proof and returns Aquifer's signed response.

func (*L8Registry) IsTrusted 

func (r *L8Registry) IsTrusted(webhookURL string) bool

IsTrusted returns true if the L8 handshake has been completed for this webhook domain.

func (*L8Registry) Meta 

func (r *L8Registry) Meta(host string) L8Meta

func (*L8Registry) SignHeaders 

func (r *L8Registry) SignHeaders(body []byte) map[string]string

SignHeaders returns X-L8-* headers to attach to an outgoing webhook delivery.

type LedgerEntry added in v0.6.0 

type LedgerEntry struct {
	HashKey string `json:"idempotent_key_hash"`
	JobID   string `json:"job_id"`
	Status  Status `json:"status"`
}

LedgerEntry is one row of the drain-mode idempotency ledger -- hash-only, never the plaintext idempotent key. See drain.go.

type MCPStdioAdapter 

type MCPStdioAdapter struct {
	// contains filtered or unexported fields
}

func NewMCPStdioAdapter 

func NewMCPStdioAdapter(in io.Reader, out io.Writer) *MCPStdioAdapter

func (*MCPStdioAdapter) Name 

func (a *MCPStdioAdapter) Name() string

func (*MCPStdioAdapter) Start 

func (a *MCPStdioAdapter) Start(ctx context.Context, aquifer *Aquifer) error

type MetricsAdapter 

type MetricsAdapter interface {
	JobQueued(userID, upstream string)
	JobDispatched(userID, upstream string)
	JobCompleted(userID, upstream string, durationMs int64)
	JobFailed(userID, upstream string, reason string)
	WebhookDelivered(url string, attempt int)
	WebhookFailed(url string, attempts int)
	QueueDepth(upstream string, depth int)
	FlowRate(upstream string, rps float64)
	// DrainFlushSucceeded/DrainFlushFailed only ever fire when drain mode
	// is enabled (see drain.go) -- unreached on a deployment that never
	// turns it on.
	DrainFlushSucceeded(instanceKey string, ledgerSize int)
	DrainFlushFailed(instanceKey string, ledgerSize int)
}

type NoopMetricsAdapter 

type NoopMetricsAdapter struct{}

func (NoopMetricsAdapter) DrainFlushFailed added in v0.6.0 

func (NoopMetricsAdapter) DrainFlushFailed(instanceKey string, ledgerSize int)

func (NoopMetricsAdapter) DrainFlushSucceeded added in v0.6.0 

func (NoopMetricsAdapter) DrainFlushSucceeded(instanceKey string, ledgerSize int)

func (NoopMetricsAdapter) FlowRate 

func (NoopMetricsAdapter) FlowRate(upstream string, rps float64)

func (NoopMetricsAdapter) JobCompleted 

func (NoopMetricsAdapter) JobCompleted(userID, upstream string, durationMs int64)

func (NoopMetricsAdapter) JobDispatched 

func (NoopMetricsAdapter) JobDispatched(userID, upstream string)

func (NoopMetricsAdapter) JobFailed 

func (NoopMetricsAdapter) JobFailed(userID, upstream string, reason string)

func (NoopMetricsAdapter) JobQueued 

func (NoopMetricsAdapter) JobQueued(userID, upstream string)

func (NoopMetricsAdapter) QueueDepth 

func (NoopMetricsAdapter) QueueDepth(upstream string, depth int)

func (NoopMetricsAdapter) WebhookDelivered 

func (NoopMetricsAdapter) WebhookDelivered(url string, attempt int)

func (NoopMetricsAdapter) WebhookFailed 

func (NoopMetricsAdapter) WebhookFailed(url string, attempts int)

type NoopRegionAdapter added in v0.9.0 

type NoopRegionAdapter struct{}

NoopRegionAdapter is the default: no known regions, cross-region redirect never triggers. This is what "the feature is off" looks like -- AttemptDirect checks LiveRegions() being empty as its gate for even considering redirect, so a deployment that never configures a real RegionAdapter sees zero behavior change from this feature existing in the codebase.

func (NoopRegionAdapter) LiveRegions added in v0.9.0 

func (NoopRegionAdapter) LiveRegions() []string

func (NoopRegionAdapter) SelfRegion added in v0.9.0 

func (NoopRegionAdapter) SelfRegion() string

type PebbleStore added in v0.4.0 

type PebbleStore struct {
	// contains filtered or unexported fields
}

func NewPebbleStore added in v0.4.0 

func NewPebbleStore(path string) *PebbleStore

func (*PebbleStore) AcknowledgeDrainEventsThrough added in v0.10.0 

func (s *PebbleStore) AcknowledgeDrainEventsThrough(sequence int64)

func (*PebbleStore) CheckOrInsert added in v0.4.0 

func (s *PebbleStore) CheckOrInsert(job *Job) (string, bool)

CheckOrInsert mirrors Store.CheckOrInsert's contract exactly: :ok for a fresh job, or the existing job ID if the (user_id, idempotent_key) pair was already accepted. The shard lock is what makes this atomic instead of a racy Get-then-Set — see the package doc comment above.

func (*PebbleStore) ClearIdempotentKeys added in v0.6.0 

func (s *PebbleStore) ClearIdempotentKeys()

ClearIdempotentKeys wipes both the job: and idem: prefixes -- only ever called by drain mode's watchdog after a successful ledger-flush webhook delivery, never on a normal (non-drain-mode) deployment. Takes every shard lock before wiping so a concurrent CheckOrInsert (which only holds one shard's lock) can't race a mid-wipe read/write and reintroduce a row.

func (*PebbleStore) Close added in v0.4.0 

func (s *PebbleStore) Close() error

func (*PebbleStore) Counts added in v0.4.0 

func (s *PebbleStore) Counts() StoreCounts

func (*PebbleStore) DeleteJob added in v0.4.0 

func (s *PebbleStore) DeleteJob(jobID string)

DeleteJob removes both the job row and its idempotency index entry — dropping only the job: key would leave a dangling idem: entry pointing at a job that no longer exists, the same ghost-row risk DeleteJob exists to prevent on the SQLite side.

func (*PebbleStore) GetJob added in v0.4.0 

func (s *PebbleStore) GetJob(jobID string) *Job

func (*PebbleStore) GetQueuedJobs added in v0.4.0 

func (s *PebbleStore) GetQueuedJobs() []*Job

func (*PebbleStore) ListDrainEvents added in v0.10.0 

func (s *PebbleStore) ListDrainEvents(limit int) []DrainEvent

func (*PebbleStore) ListIdempotentKeys added in v0.6.0 

func (s *PebbleStore) ListIdempotentKeys() []LedgerEntry

ListIdempotentKeys backs drain mode's ledger export. Unlike SQLite, pebbleRecord.Job retains the plaintext IdempotentKey (for unrelated reasons -- see the package doc comment), but this must never surface it: the hash is recomputed the same way CheckOrInsert derives it, and only the hash/job_id/status ever go into the returned LedgerEntry. Excludes webhook-delivery jobs (WebhookURL == "", see Job.isWebhookDeliveryJob) -- those are internal delivery bookkeeping, not real user-submitted work, and have no business appearing in a ledger meant for tenant-handoff dedup.

func (*PebbleStore) MarkInFlight added in v0.4.0 

func (s *PebbleStore) MarkInFlight(jobID string)

func (*PebbleStore) Path added in v0.4.0 

func (s *PebbleStore) Path() string

func (*PebbleStore) RecoverInFlight added in v0.4.0 

func (s *PebbleStore) RecoverInFlight(queueKey string) []*Job

func (*PebbleStore) SetQueueKey added in v0.4.0 

func (s *PebbleStore) SetQueueKey(jobID, queueKey string)

func (*PebbleStore) UpdateStatus added in v0.4.0 

func (s *PebbleStore) UpdateStatus(jobID string, status Status)

type Pool added in v0.4.0 

type Pool struct {
	// contains filtered or unexported fields
}

Pool is one named group of registered members. Selection uses virtual-time weighted round robin: each pick advances the chosen member's virtual position by 1/weight (a higher-weight member advances less per pick, so it comes back to the front of the heap sooner and gets picked proportionally more often), giving O(log n) per dispatch — the naive nginx-style smooth-weighted-round-robin algorithm (recompute every member's running weight on every pick) is O(n) per pick instead; this avoids that by only ever touching the single member actually chosen.

func (*Pool) Pick added in v0.4.0 

func (p *Pool) Pick() *PoolMember

Pick selects a member proportional to declared_rate x reputation and advances its virtual position. Returns nil if the pool has no members.

func (*Pool) RecordFailure added in v0.4.0 

func (p *Pool) RecordFailure(id string) bool

RecordFailure halves a member's reputation — the same shape as the existing Retry-After backoff (doubles per consecutive event), just applied as a share reduction instead of a delay increase. Once reputation has been at or below the floor continuously for defaultFloorEvictionWindow, the member is evicted. Returns whether this call caused an eviction.

func (*Pool) RecordSuccess added in v0.4.0 

func (p *Pool) RecordSuccess(id string)

RecordSuccess nudges a member's reputation back toward full trust and unconditionally clears any in-progress floor-eviction timer — the eviction criterion is sustained badness with zero successes in between, not a numeric reputation threshold held for a duration, so any success resets the clock even if the reputation number itself hasn't recovered above the floor yet.

func (*Pool) Register added in v0.4.0 

func (p *Pool) Register(id, address string, declaredRPS float64, heartbeatInterval time.Duration)

Register adds a new member or refreshes an existing one. The same call serves as both initial registration and heartbeat — re-calling it resets LastHeartbeat and updates declared capacity, so a member that wants to lower or raise its reported rate just re-registers.

func (*Pool) Size added in v0.4.0 

func (p *Pool) Size() int

func (*Pool) Snapshot added in v0.4.0 

func (p *Pool) Snapshot() []map[string]any

func (*Pool) TotalCapacity added in v0.4.0 

func (p *Pool) TotalCapacity() float64

TotalCapacity is the live sum of every member's current effective weight — the pool's aggregate ceiling grows and shrinks automatically as members register, degrade, or drop out, rather than being a static number an operator configures once.

type PoolMember added in v0.4.0 

type PoolMember struct {
	ID                string
	Address           string
	DeclaredRPS       float64
	HeartbeatInterval time.Duration
	LastHeartbeat     time.Time
	// contains filtered or unexported fields
}

PoolMember is one registered target within a pool — a service instance that pinged in with its own address and declared capacity.

func (*PoolMember) Reputation added in v0.4.0 

func (m *PoolMember) Reputation() float64

type PoolRegistry added in v0.4.0 

type PoolRegistry struct {
	// contains filtered or unexported fields
}

PoolRegistry owns every named pool. One registry per Aquifer instance.

func NewPoolRegistry added in v0.4.0 

func NewPoolRegistry() *PoolRegistry

func (*PoolRegistry) Get added in v0.4.0 

func (pr *PoolRegistry) Get(poolID string) *Pool

Get returns the named pool, creating an empty one if nobody has registered to it yet. A job dispatched to a pool that exists but has zero members is a different, correctly-handled state (fails cleanly with "no pool members registered") from a job that isn't pool-backed at all — returning nil here would collapse that distinction and let a pool-backed job silently fall through to a non-pool dispatch path with an empty URL instead.

func (*PoolRegistry) Register added in v0.4.0 

func (pr *PoolRegistry) Register(poolID, memberID, address string, declaredRPS float64, heartbeatInterval time.Duration)

func (*PoolRegistry) Snapshot added in v0.4.0 

func (pr *PoolRegistry) Snapshot() map[string]any

func (*PoolRegistry) Stop added in v0.4.0 

func (pr *PoolRegistry) Stop()

type ProxyFallbackInfo added in v0.9.0 

type ProxyFallbackInfo struct {
	Reason string
	Status int
}

ProxyFallbackInfo carries the one extra SSE event proxy mode's fallback path emits before the normal queued/dispatching/terminal sequence — see streamEvents. Status is 0 when a real upstream status was never received (a skipped attempt or a timeout), non-zero when the upstream actually responded (e.g. 429/503).

type ProxyOutcome added in v0.9.0 

type ProxyOutcome struct {
	Err         error
	Duplicate   bool
	ExistingJob *Job
	Job         *Job
	Direct      bool
	Status      int
	Header      http.Header
	Body        []byte

	// FallbackReason and FallbackStatus are set whenever Job is set but
	// Direct is false — a short, stable label for why a direct attempt
	// wasn't completed, surfaced to the caller as a proxy_fallback SSE
	// event before the normal queued/dispatching/terminal sequence, so a
	// client watching the stream (a browser, an agent with no server of
	// its own to explain this some other way) knows it's in the queue
	// because something specific happened, not just "queued" with no
	// context. FallbackStatus is the upstream's real status code when one
	// was actually received (0 for a skipped or timed-out attempt).
	FallbackReason string
	FallbackStatus int

	// RelayFrom is set when this request was redirected to another region
	// (region_redirect.go) and that region accepted it into its own
	// durable queue rather than completing directly — the caller
	// (server.go's proxyJob) should relay every SSE event read from this
	// response body onto the original caller's connection in real time,
	// rather than subscribing to a local job's events the normal way.
	// Job/Direct/FallbackReason are meaningless in this case; the real job
	// now lives on the target region under its own ID — the caller learns
	// it from the relayed stream itself, the same place it always would
	// for any fallback. The caller owns closing RelayFrom.Body once the
	// stream ends. RerouteRegion is always set alongside it — which region
	// actually ended up owning the job — so the caller can tell the client
	// via a synthetic "rerouted" event before relaying, same rationale as
	// FallbackReason/proxy_fallback: a client with no server of its own to
	// explain this (a browser, an agent) shouldn't have to wonder why its
	// connection is still open or where the response actually came from.
	RelayFrom     *http.Response
	RerouteRegion string
}

ProxyOutcome is AttemptDirect's result. Exactly one of the following holds: Err is set (validation/admission failure, same contract as Enqueue); Duplicate is true (ExistingJob holds the prior job); or Job is set, with Direct indicating whether it was completed synchronously (Status/Header/Body then valid) or needs the caller to Dispatch it and stream the result the normal way.

type RateConfig 

type RateConfig struct {
	RPS           float64 `yaml:"rps"`
	MaxConcurrent int     `yaml:"max_concurrent"`
}

type RedirectExhaustedError added in v0.9.0 

type RedirectExhaustedError struct {
	JobID string
}

RedirectExhaustedError is returned by Aquifer.AttemptDirect (via fallbackOutcome) when cross-region redirect is configured and was actually attempted (or is still within its post-exhaustion gate cooldown — see redirectGate) but no known-live region could serve the request directly or accept it into its own queue. Callers (the HTTP server) type-assert on this to build a 429 with a long Retry-After, the same way AdmissionRejectedError already does for admission control.

This is deliberately NOT the same as "queue it locally instead" — queueing on total redirect exhaustion was never actually decided; the default is to fail loudly so a caller (or its own retry/alerting logic) knows the whole fleet is degraded, not just this one instance. A future per-deployment option to queue locally instead (e.g. for an Aquifer instance dedicated to a single customer, where "queue and eventually deliver" might be preferable to erroring) is a real possibility, just not the default and not built here.

func (*RedirectExhaustedError) Error added in v0.9.0 

func (e *RedirectExhaustedError) Error() string

type RegionAdapter added in v0.9.0 

type RegionAdapter interface {
	// LiveRegions returns the currently known-live regions, excluding
	// SelfRegion. Safe to call frequently; implementations should return a
	// cached snapshot, not block on a live check.
	LiveRegions() []string
	// SelfRegion returns this instance's own region identifier, or "" if
	// unknown/not applicable.
	SelfRegion() string
}

RegionAdapter reports which regions Aquifer is currently deployed to and reachable in, backing /proxy's cross-region redirect (see proxy.go's AttemptDirect). Same shape as MetricsAdapter/JobStore: a small interface, a no-op default, the deployer plugs in a real implementation for their platform. Off unless explicitly configured -- matches every other opt-in feature in this codebase.

type RegistrationConfig added in v0.10.0 

type RegistrationConfig struct {
	URL             string // AQUIFER_REGISTRY_URL -- presence enables this feature
	Port            string // PORT (same env var and "8080" default region_adapter_fly.go already uses)
	IntervalSeconds int64  // AQUIFER_REGISTRY_INTERVAL_SECONDS
}

func LoadRegistrationConfig added in v0.10.0 

func LoadRegistrationConfig() RegistrationConfig

LoadRegistrationConfig reads the AQUIFER_REGISTRY_* env vars (plus the already-established PORT var, not a new one of its own).

func (RegistrationConfig) Enabled added in v0.10.0 

func (cfg RegistrationConfig) Enabled() bool

type Registry 

type Registry struct {
	// contains filtered or unexported fields
}

func NewRegistry 

func NewRegistry(store JobStore, cfg *Config, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, pools *PoolRegistry) *Registry

NewRegistry reads drain mode's config from AQUIFER_DRAIN_* env vars (LoadDrainConfig) — disabled unless AQUIFER_DRAIN_ENABLED is explicitly set, matching NewPebbleStore's existing precedent of reading its own opt-in env vars internally. Callers wanting a programmatic override (RuntimeOptions.DrainConfig) call ConfigureDrain after construction.

func (*Registry) Close added in v0.10.0 

func (r *Registry) Close()

func (*Registry) ConfigureDrain added in v0.6.0 

func (r *Registry) ConfigureDrain(cfg DrainConfig)

ConfigureDrain overrides drain mode's config after construction (used by RuntimeOptions.DrainConfig) and starts the watchdog if the override enables it and the constructor-time env-based config hadn't already started one. There is no supported path to stop an already-running watchdog at runtime — disabling drain mode requires a restart, same as every other env-var-driven config in this codebase.

func (*Registry) DrainSnapshot added in v0.6.0 

func (r *Registry) DrainSnapshot() map[string]any

DrainSnapshot reports drain mode's current state for GET /health, or nil when drain mode isn't enabled — an instance that never turned this on shouldn't see a new key appear in its health output.

func (*Registry) DrainState added in v0.6.0 

func (r *Registry) DrainState() DrainState

DrainState is the instance's current position in drain mode's lifecycle (active/draining/unassigned) — meaningful only when drain mode is enabled, but always safe to call (returns DrainStateActive otherwise, since the watchdog that would ever move it elsewhere never runs).

func (*Registry) Enqueue 

func (r *Registry) Enqueue(job *Job, accountQueueHeader string)

Enqueue queues a job on the URLWorker for its upstream domain, or for its target pool if the job carries a PoolID instead of a URL. accountQueueHeader is the raw X-Aqueduct-Account-Queue/X-Aquifer-Account-Queue value from the originating HTTP request, or "" if this job has no live request behind it (e.g. recovered from disk at startup). An empty value leaves the worker's current account-queue mode unchanged rather than forcing it off — the mode is shared per upstream domain, so one request that doesn't care about it shouldn't be able to flip it off for every other concurrent tenant relying on it being on.

func (*Registry) EnqueueWebhook added in v0.7.0 

func (r *Registry) EnqueueWebhook(originalJobID, userID, webhookURL string, payload map[string]any)

EnqueueWebhook queues a webhook delivery through the same domain-keyed account-queue pacing and backpressure machinery as forward dispatch (RPS/concurrency limits, X-Aqueduct-* response-header throttling) instead of firing immediately with a fixed retry schedule — a slow or rate-limited webhook receiver can now shed load exactly the way an upstream API already can, and delivery is durable across a restart the same way a real job is (the underlying webhook-delivery Job is persisted via CheckOrInsert, not just an in-memory retry loop).

originalJobID scopes the idempotent key (see Job.isWebhookDeliveryJob) so a given job's webhook is enqueued at most once even if this were somehow called twice for it.

func (*Registry) JobDispatched 

func (r *Registry) JobDispatched()

func (*Registry) JobDone 

func (r *Registry) JobDone()

func (*Registry) SetDrainRemote added in v0.10.0 

func (r *Registry) SetDrainRemote(remote RemoteIdempotency)

type RemoteIdempotency added in v0.10.0 

type RemoteIdempotency interface {
	Lookup(hash string) (RemoteIdempotencyEntry, bool)
	Record(events []DrainEvent) bool
}

func NewValkeyRemoteIdempotency added in v0.10.0 

func NewValkeyRemoteIdempotency(cfg RemoteIdempotencyConfig) RemoteIdempotency

type RemoteIdempotencyConfig added in v0.10.0 

type RemoteIdempotencyConfig struct {
	Enabled    bool
	URL        string
	Timeout    time.Duration
	Prefix     string
	TTLSeconds int64
}

func LoadRemoteIdempotencyConfig added in v0.10.0 

func LoadRemoteIdempotencyConfig() RemoteIdempotencyConfig

type RemoteIdempotencyEntry added in v0.10.0 

type RemoteIdempotencyEntry struct {
	JobID      string `json:"job_id"`
	Status     Status `json:"status"`
	RecordedAt int64  `json:"recorded_at"`
	Source     string `json:"source"`
}

type Runtime 

type Runtime struct {
	Aquifer   *Aquifer
	Store     JobStore
	Broker    *Broker
	Registry  *Registry
	L8        *L8Registry
	Config    *Config
	Admission *AdmissionController
	Pools     *PoolRegistry
}

func NewRuntime 

func NewRuntime(opts RuntimeOptions) *Runtime

func (*Runtime) Close added in v0.10.0 

func (r *Runtime) Close()

func (*Runtime) DBPath 

func (r *Runtime) DBPath() string

func (*Runtime) RecoverQueuedJobs 

func (r *Runtime) RecoverQueuedJobs(dbPath string)

type RuntimeOptions 

type RuntimeOptions struct {
	DBPath          string
	ConfigPath      string
	Config          *Config
	L8KeyPath       string
	L8TrustDir      string
	Metrics         MetricsAdapter
	AdmissionLimits *AdmissionLimits
	// Store overrides the storage backend entirely. If nil, NewRuntime falls
	// back to NewJobStore(DBPath), selecting sqlite/pebble via
	// AQUIFER_STORE_BACKEND as before. Set this to plug in a custom JobStore
	// implementation (e.g. Postgres, rqlite) without needing to bypass
	// NewRuntime and wire the lower-level constructors by hand.
	Store JobStore
	// DrainConfig overrides drain mode's config (see drain.go). If nil,
	// the Registry reads AQUIFER_DRAIN_* env vars itself — disabled unless
	// AQUIFER_DRAIN_ENABLED is explicitly set to true.
	DrainConfig *DrainConfig
	// RegionAdapter backs /proxy's cross-region redirect (proxy.go,
	// region_adapter.go). If nil, NewRuntime tries NewFlyRegionAdapter,
	// which itself only activates if AQUIFER_FLY_REGIONS is set — same
	// zero-code, env-var-only activation pattern as AQUIFER_DRAIN_ENABLED.
	// Set this to plug in a RegionAdapter for a different platform.
	RegionAdapter RegionAdapter
	// ClusterRouter optionally routes HTTP /jobs and /proxy requests to the
	// Aquifer node that owns the request's partition. If nil, NewRuntime
	// loads static env config from AQUIFER_CLUSTER_*.
	ClusterRouter *ClusterRouter
	// RemoteIdempotency optionally checks a shared remote ledger after the
	// local store accepts a new key but before the job is dispatched. If
	// nil, NewRuntime loads AQUIFER_REMOTE_IDEMPOTENCY_* env config.
	RemoteIdempotency RemoteIdempotency
}

type SSEEvent 

type SSEEvent struct {
	Event string
	Data  map[string]any
}

type Server 

type Server struct {
	// contains filtered or unexported fields
}

func NewServer 

func NewServer(aquifer *Aquifer) *Server

func (*Server) Routes 

func (s *Server) Routes() http.Handler

type Status 

type Status string
const (
	StatusQueued    Status = "queued"
	StatusInFlight  Status = "in_flight"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
)

type Store 

type Store struct {
	// contains filtered or unexported fields
}

func NewStore 

func NewStore(path string) *Store

func (*Store) AcknowledgeDrainEventsThrough added in v0.10.0 

func (s *Store) AcknowledgeDrainEventsThrough(sequence int64)

func (*Store) CheckOrInsert 

func (s *Store) CheckOrInsert(job *Job) (string, bool)

CheckOrInsert inserts job unless its (user_id, idempotent_key) pair already exists, in which case it reports the existing job's ID as a duplicate. The duplicate/fresh decision is read directly off the INSERT's own RowsAffected, not a follow-up SELECT — a SELECT-after-INSERT here would race under real concurrency (multiple open connections): if the INSERT is still in flight relative to another goroutine's read, or a transient busy-retry delays it, the SELECT can find nothing and this would misreport a brand-new job as an empty-ID "duplicate," silently losing it. RowsAffected==1 is authoritative and atomic: it's exactly the row this call just wrote, no read-your-own-write race possible.

func (*Store) ClearIdempotentKeys added in v0.6.0 

func (s *Store) ClearIdempotentKeys()

ClearIdempotentKeys wipes the whole table -- only ever called by drain mode's watchdog after a successful ledger-flush webhook delivery, never on a normal (non-drain-mode) deployment.

func (*Store) Close added in v0.4.0 

func (s *Store) Close() error

Close releases the underlying connection pool. In WAL mode, SQLite keeps -wal/-shm files alongside the main database file for as long as any connection is open; callers that manage a Store's lifetime explicitly (tests especially, cleaning up a t.TempDir()) should call this before their directory is removed.

func (*Store) Counts 

func (s *Store) Counts() StoreCounts

func (*Store) DeleteJob added in v0.3.0 

func (s *Store) DeleteJob(jobID string)

DeleteJob removes a job row outright. Used when a freshly-inserted, non-duplicate job is rejected by admission control — CheckOrInsert already wrote the row before duplicate status was known, so a rejected job must be deleted here or it would sit as a ghost "queued" row that never dispatches.

func (*Store) GetJob 

func (s *Store) GetJob(jobID string) *Job

func (*Store) GetQueuedJobs 

func (s *Store) GetQueuedJobs() []*Job

func (*Store) ListDrainEvents added in v0.10.0 

func (s *Store) ListDrainEvents(limit int) []DrainEvent

func (*Store) ListIdempotentKeys added in v0.6.0 

func (s *Store) ListIdempotentKeys() []LedgerEntry

ListIdempotentKeys backs drain mode's ledger export -- hash-only, matches what this table has always stored (the plaintext idempotent key was never persisted here, only its hash, see hashKey/CheckOrInsert). Excludes webhook-delivery jobs (webhook_url = ”, see Job.isWebhookDeliveryJob) -- those are internal delivery bookkeeping, not real user-submitted work, and have no business appearing in a ledger meant for tenant-handoff dedup.

func (*Store) MarkInFlight 

func (s *Store) MarkInFlight(jobID string)

func (*Store) Path 

func (s *Store) Path() string

func (*Store) RecoverInFlight 

func (s *Store) RecoverInFlight(queueKey string) []*Job

func (*Store) SetQueueKey 

func (s *Store) SetQueueKey(jobID, queueKey string)

func (*Store) UpdateStatus 

func (s *Store) UpdateStatus(jobID string, status Status)

type StoreCounts 

type StoreCounts struct {
	TotalJobs  int64
	QueueDepth int64
}

type URLWorker 

type URLWorker struct {
	// contains filtered or unexported fields
}

func NewURLWorker 

func NewURLWorker(domain string, rps float64, maxConc int, pool *Pool, store JobStore, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, enqueueWebhook webhookEnqueuer, onIdle func(string)) *URLWorker

func (*URLWorker) BreakerKind added in v0.9.0 

func (w *URLWorker) BreakerKind() string

BreakerKind reports which kind of signal tripped the breaker last — "queue" or "reroute" (see classifyOverload) — only meaningful while BreakerOpen is true. A subsequent request arriving while the breaker is still open has no fresh response of its own to classify, so it reuses whichever kind actually tripped it: a domain breaker-tripped by a 429 stays queue-only on every retry during that cooldown, not reroute-eligible just because SOME overload happened.

func (*URLWorker) BreakerOpen added in v0.9.0 

func (w *URLWorker) BreakerOpen() bool

BreakerOpen reports whether proxy mode should skip a direct dispatch attempt to this worker's domain entirely and fall straight back to the durable queue — set by TripBreaker after an overload signal, cleared automatically once the cooldown elapses.

func (*URLWorker) Enqueue 

func (w *URLWorker) Enqueue(job *Job)

func (*URLWorker) QueueActive added in v0.9.0 

func (w *URLWorker) QueueActive() bool

QueueActive reports whether any of this domain's account queues currently has real backlog (queued or in-flight work). Distinct from BreakerOpen: a breaker cooldown is a fixed clock that can expire while a real backlog is still draining, letting proxy mode resume direct dispatch against an upstream that's still catching up from the very overload that tripped the breaker. QueueActive self-corrects instead — it stays true for exactly as long as there's real work in flight, independent of any timer, and goes false the instant the backlog is actually empty.

func (*URLWorker) Stop added in v0.10.0 

func (w *URLWorker) Stop()

func (*URLWorker) TripBreaker added in v0.9.0 

func (w *URLWorker) TripBreaker(cooldown time.Duration, kind string)

TripBreaker opens the breaker for cooldown, recording which kind of signal caused it (see BreakerKind). No separate half-open state is needed: once breakerUntil passes, BreakerOpen naturally returns false again, so the next request after the cooldown is itself a real probe against the live upstream — success leaves the breaker closed, a repeat overload signal re-trips it via another TripBreaker call.

type ValkeyRemoteIdempotency added in v0.10.0 

type ValkeyRemoteIdempotency struct {
	// contains filtered or unexported fields
}

func (*ValkeyRemoteIdempotency) Lookup added in v0.10.0 

func (v *ValkeyRemoteIdempotency) Lookup(hash string) (RemoteIdempotencyEntry, bool)

func (*ValkeyRemoteIdempotency) Record added in v0.10.0 

func (v *ValkeyRemoteIdempotency) Record(events []DrainEvent) bool

Source Files

View all Source files

Directories

Path Synopsis
Package a2aadapter exposes Aquifer as an A2A (Agent2Agent protocol, v1.0) agent over JSON-RPC/HTTPS.
 Package a2aadapter exposes Aquifer as an A2A (Agent2Agent protocol, v1.0) agent over JSON-RPC/HTTPS.
cmd
aquifer command
aquifer-scenario command
examples
custom_adapter command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.