Middleware

Middleware is Capi’s client-side request pipeline: a phase-ordered set of members that a config registers once and every request under that config traverses. A member can inspect and rewrite the request on the way down, inspect and rewrite the response on the way up, or answer without ever reaching the wire — all working on headers and metadata, below the decoder. Request IDs, cookies, and redirect following live here.

One traversal

One traversal is one trip. capi_transport’s Traversal drives it end to end, in this order:

A query is one or more traversals; a runner decides how many.

Three things about that order are load-bearing:

• The gate sits after mutation and before the seal. The rate-limit wait happens before signing, so a signature’s timestamps are computed after the wait rather than staled by it — and the seal resolves the credential at the moment of use, renewing a seated one that the wait left near expiry.
• Nothing mutates after the seal. There is no step between the seal and the wire for a member to occupy, so what the signer signed is what crosses the socket. Transmit’s Host strip and hop-by-hop validation are the traversal’s own, and they run per attempt.
• Every exchange goes through one traversal. The transmit invariants live inside it, so no re-entry path — a runner’s re-dispatch, a decoder’s follow-up, a credential mint — can skip or re-implement them.

Within a phase, insertion order is execution order: “before X in phase P” is expressed by inserting earlier in P, and crossing a boundary by choosing the phase.

The three traits

A member implements whichever halves it needs. All three are plain synchronous methods in every lane — there is no async variant — and each call must be non-blocking and bounded, because the traversal owns the loop and the clock.

pub trait RequestMiddleware: MaybeSend + MaybeSync + 'static {
    fn on_request(&self, ctx: &ApiContext, head: &mut RequestHead, body: &mut Body)
        -> Result<RequestFlow, MwError>;

    /// Whether this member serves `lane`. Defaults to `true` everywhere.
    fn serves_lane(&self, lane: LaneId) -> bool { true }
}

pub trait ResponseMiddleware: MaybeSend + MaybeSync + 'static {
    fn on_response(&self, ctx: &ApiContext, head: &mut ResponseHead, body: &mut Body)
        -> Result<ResponseFlow, MwError>;

    fn serves_lane(&self, lane: LaneId) -> bool { true }
}

pub trait ExchangeMiddleware: MaybeSend + MaybeSync + 'static {
    /// What the request half hands its own response half.
    type State: MaybeSend + 'static;

    fn on_request(&self, ctx: &ApiContext, head: &mut RequestHead, body: &mut Body)
        -> Result<(RequestFlow, Self::State), MwError>;
    fn on_response(&self, state: Self::State, ctx: &ApiContext,
                   head: &mut ResponseHead, body: &mut Body)
        -> Result<ResponseFlow, MwError>;

    fn serves_lane(&self, lane: LaneId) -> bool { true }
}

ExchangeMiddleware is the paired form: its request half hands its own response half a typed State, so the two are matched by the compiler rather than by a lookup. The two shipped exchange members show both uses of that State: CookieMiddleware carries the request’s URL and the instant it was sent (CookieExchange), so its response half files Set-Cookie against the right origin, and FollowRedirects carries a replay handle (RedirectExchange) for the hop it may re-send.

The verbs

A request half returns a RequestFlow:

pub enum RequestFlow {
    /// Proceed to the next member.
    Continue,
    /// This response is the traversal's result. The wire is never reached.
    Respond(Box<Response>),          // built with RequestFlow::respond(response)
}

A response half returns a ResponseFlow:

pub enum ResponseFlow {
    /// Hand the response to the next member on the way out.
    Continue,
    /// Send this request: another exchange, gated, sealed and transmitted like
    /// any other.
    Resend(Box<Request>),            // built with ResponseFlow::resend(request)
}

The exceptional payloads are boxed so the common Continue verdict stays pointer-sized at the call boundary; the respond and resend constructors do the boxing. An Err from either half aborts the traversal and travels outward, unchanged, as the query’s error; the traversal’s event record names the member it stopped at.

A Resend resumes the request walk at the re-sending member’s own seat: the request it handed back already carries what the members before it applied, so they do not run again, while that member and the ones after it do. Their pending response halves are the new ones; the halves of earlier seats stay untouched and run once, on the response that ends the chain. The seat is therefore what a member pairs with — per wire exchange from the re-sending seat onward, once per operation before it — and the request’s plan carries the resend_budget that bounds the chain. A member takes the next exchange’s body from Body::replay_handle() on the way down.

Iteration that re-describes the exchange — retry with backoff, payment, reactive credential renewal — is runner work. A runner holds the endpoint and re-dispatches by endpoint value, so each attempt re-derives its description from the endpoint rather than replaying a request someone else built. Redirect following has one of each: capi_extras::middleware:: FollowRedirects for the config-armed seat (and for an endpoint that cannot be described twice), api.follow_redirects(n) for the caller-installed one.

The body view

Body is the body as the member can see it — one type serves both directions. The surface is the body’s identity and its dressing: swap the stream (replace, take), wrap it in a codec (with_codec), register an observation (observe), take a replay handle on it for an exchange that may be sent again (replay_handle), and read what it declares about itself (body_source, content_length, content_coding, is_empty). There is no read method: pulling bytes is flow-time, and flow-time belongs to the byte plane, which runs in the stream’s own lane once the traversal has handed the request to the wire.

A decision that genuinely needs body content belongs at the description altitude (the endpoint, the codec), where the content is still typed; a decision that needs to watch bytes move belongs to the transfer plane.

Registering members

A config names its pipeline once:

fn middleware(&self) -> Pipeline {
    self.inner
        .middleware()
        .with_request(Phase::Early, InjectRequestId::new("x-request-id"))
}

A decorator forwards self.inner.middleware() and seats its own members on top, so the wrapped config’s members ride too. Pipeline clones by reference count, so a config with a large layout can hold one and return a clone.

The three seating verbs come in a push_* form (on &mut self) and a consuming with_* form: push_request / with_request, push_response / with_response, and push_exchange / with_exchange.

Per request, a caller attaches a delta that merges at Phase::User:

api.scoped()
    .with_middleware(my_request_member)          // or with_response_middleware,
    .query(endpoint).await?;                     // or with_exchange_middleware

The delta is a MiddlewareOverride (in the facade prelude): with_request / with_response / with_exchange seat a member at Phase::User, and the with_request_in(phase, ..) / with_response_in / with_exchange_in forms name the phase. It rides the whole operation’s lineage: a paged iterator’s later pages, a resumable stream’s reconnects, and a decoder’s follow-ups all apply it freshly at their own assembly. A transfer stage that must be shared across requests rather than built per request — one Throttle as a process-wide cap — is seated with with_shared_transfer_stage(Arc<dyn TransferStage>).

The other two entries

Every ordinary exchange takes the traversal above. Two entries take a shortened one, and both are about not doing something the ordinary walk does:

• A credential mint (Traversal::credential_exchange) runs the phases and the transmit invariants exactly as any exchange, but the seal places whatever the store holds without renewing it — a mint must not depend on the credential it exists to produce — and the gate is skipped: a config’s limiter models the API service, not the identity provider, and queueing a mint behind the requests waiting for it can deadlock freshness.
• A capture (Traversal::capture) is the shape of an exchange without the exchange: everything a send does short of the wire — the phases, the seal, the transmit invariants — and no response, so no unwind. It is how a request is finalized for a batch envelope or a tunneled frame (the capability plane).

Lanes

One pipeline serves every transport lane the config’s endpoints use — plain HTTP, WebSocket upgrades, gRPC. A member that is genuinely wrong on some lane says so:

fn serves_lane(&self, lane: LaneId) -> bool {
    lane == LaneId::of::<Http>()
}

The traversal asks every member this before any of them runs, and refuses an off-lane one by name, phase, position, and lane. The boundary is behavioral, not cosmetic: a trigger that simply never fires on some lane is honest inertness and needs no restriction, while a member that would faithfully measure the wrong thing there — progress-metering a WebSocket upgrade’s empty handshake instead of the session it was attached for — declares its lanes and is refused elsewhere.

Watching a traversal

The traversal is the only thing that can attribute time to one member, so per-member observation is its job rather than an instrumentation member’s. A traversal emits capi_transport::traversal::TraversalEvents as it goes — Started, MemberEntered and MemberExited, Gated, Sealed, Transmitted, ResponseReceived, ShortCircuited, Resent, Aborted — and closes with Completed(TraversalInfo), the record: which members ran, on which half, how long the gate waited, how long the seal took, how many transmit attempts, and the outcome. Subscribe on the config’s base context and every traversal — including a decoder’s follow-ups and a runner’s re-dispatches — bubbles up. A context with no notification channel builds no record and allocates nothing.

The plane below: transfer stages

There is one thing no member can do, however it is written: watch a byte move. A member computes on a request and a response as values, at plan time, and the bytes flow afterwards. Bandwidth caps, progress reporting, byte accounting, stall detection and wire capture all want that later moment, and it is a different plane.

A transfer stage runs there:

pub trait TransferStage: Debug + MaybeSend + MaybeSync + 'static {
    fn begin(&self, transfer: &TransferDescriptor) {}
    fn pace(&self, transfer: &TransferDescriptor) -> Result<(), Duration> { Ok(()) }
    fn on_chunk(&self, transfer: &TransferDescriptor, len: usize) -> Result<(), TransferAbort> { Ok(()) }
    fn on_bytes(&self, transfer: &TransferDescriptor, bytes: &[u8]) -> Result<(), TransferAbort> { Ok(()) }
    fn complete(&self, transfer: &TransferDescriptor) {}
}

Every callback takes the same one argument, and that is deliberate: the descriptor is where the plane grows. A capability stages turn out to need becomes a field and an accessor there, rather than a breaking change to five signatures. The framework mints a fresh one for each callback it fires, so it is a snapshot of one moment — which is what lets it carry a timestamp without carrying a clock.

Stages attach per request, and the traversal attaches them at Transmit — below every member and below the seal, where nothing shapes the request any more. Three things follow from that placement:

• The body they meter is the one that went on the wire, on the attempt that actually went out. A re-dispatch runs a fresh traversal, so each attempt reports itself.
• The bytes they see are the coded ones. A codec sits between the source and outer planes with opposite orientation per direction, so the framework picks the outer plane for a request and the source plane for a response; a stage is never told a codec is there.
• Nothing replaces the body, so declared content lengths behave exactly as they would with no stage attached.

api.throttle(Throttle::download(1_500_000).with_burst(64 * 1024))
    .query(endpoint).await?;

// the sugars chain, so one request can carry both stages
api.throttle(Throttle::download(1_500_000))
    .progress(Progress::download(sender, |update| println!("{update}")))
    .query(endpoint).await?;

// and `with_transfer_stage` is the general slot every stage rides,
// shipped or third-party
api.scoped().with_transfer_stage(my_stage).query(endpoint).await?;

capi_extras::stages ships two: Throttle, a bytes-per-second cap (curl’s --limit-rate), and Progress, which emits ProgressNotifications a listener runs off the read path. Each has an extension trait carrying its verb (ThrottleExt, ProgressExt); a third party ships its own the same way.

What a stage can reach, and why

A stage runs at the most sensitive position in the traversal — after every credential has been placed. What keeps that safe is not where it sits but what the trait hands it: byte counts, read-only slices, and a TransferDescriptor carrying the direction, the wire-plane length, the context id, and now() — when the framework observed this event, taken once per fire and shared, so every stage watching a chunk agrees about when it arrived. No URL, no headers, no method, and no way to reach the body value. Not the ApiContext either: its extension map is removable, its notification channel is writable, and its RNG is a seeded stream a replay depends on. Not even the clock — only the reading, so sleep is unreachable by construction. A stage cannot produce bytes either, so a signature computed before it runs covers exactly what was transmitted. Asking for content at all means overriding on_bytes, which a reviewer can see in the stage’s source.

Pacing is a verdict, not a sleep: pace returns the wait and the framework observes it. One stage definition therefore compiles sync and async, and no stage ever blocks a thread or holds an executor.

The lane bound is a sealed TransferCapable marker on Http and Grpc. A session lane’s traversal only ever sees the empty upgrade handshake, never the session traffic, so attaching a stage there is a compile error rather than a stage faithfully metering the wrong transfer.

Member, stage, or codec

Codecs transform, stages observe and pace, members decide. Changing bytes is permanently outside the stage plane: a stage runs after signing, and the framework already owns a transformation plane — the codec system, at construction time, under the author’s control, before signing. A stage that wants to change bytes is a codec in the wrong seam. A member that wants a different description of the exchange — a fresh body, another endpoint — is a runner; one that wants the exchange it already holds sent again returns ResponseFlow::Resend.

Cookies

Cookie handling is a pipeline member. capi_cookies provides Cookie, CookieJar, CookieMiddleware, CookieConfigExt (the config onion layer), and HasCookieJar (whose cookie_jar() a form flow reads). CookieMiddleware is an ExchangeMiddleware seated at Phase::Late: its request half applies stored cookies to the final URL, and — because response halves unwind in reverse — its response half is the first to see Set-Cookie. It is a no-op on wasm, where the browser owns the jar. Add the CookieConfigExt layer to your onion and the jar rides every request automatically.

When the cookie is the credential, CookieConfigExt::session_auth(name) also installs a CookieSessionSource — an auth store that reports the named cookie’s presence to the seal and places nothing, leaving the member to write the header — so endpoints behind the session declare Auth::DEFAULT like any other.

A redirect hop re-runs the members seated after the one that took it, so the jar supplies each hop’s own origin its cookies — the runner’s hops, which are whole traversals, get the same treatment for the same reason.

The lightweight alternative

For a one-off header on a single call, you don’t need a member at all — use a scoped request:

pub fn with_conversation_id(&self, id: impl Into<HeaderValue>) -> ScopedRequest<'_, GrokConfig, Client> {
    self.scoped().with_header("x-grok-conv-id", id)
}
// api.with_conversation_id("conv_42").query(endpoint).await?

scoped().with_header(...) attaches per-call headers without touching the config’s pipeline — the right tool when the change is per-request, not client-wide.

Next: rate limiting, the plane the traversal’s gate consults.