urus

A cowboy/bandit-style HTTP library for the smarm actor runtime.

Overview

urus is a lightweight, actor-first HTTP/1.1 library designed to integrate seamlessly with the smarm actor runtime. Instead of traditional shared mutable state and locking patterns, urus embraces the actor model: connection handling is fully concurrent, and request processing pipelines are message-passing all the way down.

Key design principles:

• No locks: State is owned by actors; concurrency is via channels.
• Actor-native: Built from the ground up for smarm; each connection is an actor.
• Pluggable pipelines: Compose HTTP request handling logic with Plug and Pipeline.
• Fast HTTP/1.1 parsing: Uses httparse for robust, battle-tested RFC 7230 compliance.

Quick Start

[dependencies]
urus = { path = "../urus" }
smarm = { path = "../smarm" }

Minimal Example

use urus::{Pipeline, Router, Conn, Next, serve};

let pipeline = Pipeline::new().plug(
    Router::new()
        .get("/", |c: Conn, _n: Next| {
            c.put_status(200).put_body("hello")
        })
);

serve("0.0.0.0:8080", pipeline).unwrap();

Then:

cargo run --example hello
curl http://localhost:8080/

Routing with Path Parameters

Router::new()
    .get("/users/:id", |c: Conn, _n: Next| {
        let id = c.params.get("id").unwrap_or("").to_string();
        c.put_status(200).put_body(format!("user: {}", id))
    })

Path parameters are extracted into c.params: HashMap<String, String>.

Core Concepts

Conn — The Connection Object

The Conn struct represents a single HTTP request/response pair:

pub struct Conn {
    pub method:        Method,
    pub path:          String,
    pub params:        HashMap<String, String>,  // Path parameters (e.g., :id)
    pub assigns:       Assigns,                  // Request-scoped state
    pub body:          Body,                     // Request body
    pub headers:       HeaderMap,                // Request headers
    pub status:        u16,                      // Response status (default 200)
    pub resp_headers:  HeaderMap,                // Response headers
    pub resp_body:     RespBody,                 // Response body
    // ... (internal fields)
}

Builders make it ergonomic to modify response state:

c.put_status(201)
 .put_header("content-type", "application/json")
 .put_body(r#"{"ok": true}"#)

c.assigns is a request-scoped key-value store (similar to Plug's Assigns in Elixir/Phoenix) for passing data between plugs in a pipeline.

Pipeline — Composable Handlers

A pipeline chains plugs (middleware/handlers) together. Each plug receives a Conn, can modify it, and passes it to the next plug via the Next callback:

use urus::{Pipeline, Plug, Conn, Next};

struct LoggingPlug;

impl Plug for LoggingPlug {
    fn call(&self, mut c: Conn, n: Next) -> Conn {
        println!("{} {}", c.method, c.path);
        n.call(c)
    }
}

let pipeline = Pipeline::new()
    .plug(LoggingPlug)
    .plug(Router::new().get("/", |c, _| c.put_body("ok")));

Router — Path Matching

The built-in router matches HTTP methods and paths:

Router::new()
    .get("/", handler_fn)
    .post("/users", create_user)
    .get("/users/:id", get_user)
    .put("/users/:id", update_user)
    .delete("/users/:id", delete_user)

Handlers are closures taking (Conn, Next) -> Conn. Call Next::call(c) to continue to the next plug; omitting it short-circuits the pipeline (e.g., for authentication failures).

Server Configuration

serve(addr, pipeline) binds and listens on the given address. For more control, use serve_with(config, pipeline):

use urus::{serve_with, Config};
use std::time::Duration;

let cfg = Config {
    listener_pool:      2,                        // Supervised accept-loop actors
    scheduler_threads:  Some(2),                  // smarm worker threads (None = one per CPU)
    keep_alive_timeout: Duration::from_secs(60),  // Idle budget between requests
    request_timeout:    Duration::from_secs(30),  // Whole-request READ deadline
    write_timeout:      Duration::from_secs(30),  // Per-write response budget
    max_header_count:   64,
    read_buf_size:      8 * 1024,
    max_body_bytes:     1 << 20,
    drain_timeout:      Duration::from_secs(30),  // Graceful-shutdown drain budget
    ..Config::new("127.0.0.1:8080".parse().unwrap())
};

serve_with(cfg, pipeline).unwrap();

Timeout semantics:

• keep_alive_timeout — how long a connection may sit idle waiting for the first byte of a request. Expiry closes the socket silently (nothing was in flight). Pipelined leftover bytes count as a started request, not idle.
• request_timeout — a wall-clock deadline from a request's first byte until its head and body are fully read. Expiry mid-head gets a best-effort 408 Request Timeout; expiry mid-body just closes. Deadlines are absolute instants, so a trickling (slowloris-style) client can't reset its budget by sending one byte at a time. Covers the read phase only — a streaming response may legitimately outlive any whole-request clock.
• write_timeout — a per-write budget for response bytes: the fixed head+body write, and each streamed chunk, must complete within it. A client that stops reading is dropped once its socket buffer fills and a write stalls past the budget (the write-side twin of slowloris). Streamed chunks each get a fresh budget; a stream as a whole has no deadline.

Graceful Shutdown

serve_with_shutdown takes a ShutdownSignal; the paired Handle can be triggered from anywhere (another thread, a signal handler):

use urus::{serve_with_shutdown, shutdown_handle, Config};

let (handle, signal) = shutdown_handle();

std::thread::spawn(move || {
    // e.g. wait for SIGTERM / stdin / an admin endpoint...
    handle.shutdown();
});

serve_with_shutdown(cfg, pipeline, signal).unwrap();
// Returns once the runtime has fully wound down.

Handle::shutdown() is idempotent and performs, in order:

1. Stop accepting — every listener exits; no new connections.
2. Close idle keep-alive connections immediately.
3. Drain in-flight requests for up to Config.drain_timeout.
4. Force-stop any stragglers past the deadline (sockets close cleanly on unwind via OwnedFd::drop).

If every Handle is dropped, shutdown can never be signalled and the server runs forever — exactly serve_with's semantics (it does this internally).

Streaming Responses

A handler can return a body it doesn't have yet: hand urus the read end of a channel and feed it from a producer actor (the conn actor owns the socket and all write deadlines; the producer never touches the fd):

use urus::{Conn, Next};

|c: Conn, _n: Next| {
    let (tx, rx) = smarm::channel::<Vec<u8>>();
    smarm::spawn(move || {
        for part in ["chunk one", "chunk two"] {
            if tx.send(part.as_bytes().to_vec()).is_err() {
                return; // connection died — stop producing
            }
        }
        // tx drops: end of stream.
    });
    c.put_status(200).put_body(rx)
}

On HTTP/1.1 the response uses Transfer-Encoding: chunked and the connection is reusable afterwards; on HTTP/1.0 (no chunked framing) bytes go out raw and the connection closes to delimit the body. End of stream is signalled by dropping every Sender. The producer contract: a send error means the connection is gone — exit. Chunked request bodies are decoded transparently; handlers see conn.body either way.

Server-Sent Events

Conn::sse() is the streaming body with the SSE headers and a heartbeat wired up:

|c: Conn, _n: Next| {
    let (c, events) = c.sse(); // or c.sse_with_heartbeat(interval)
    smarm::spawn(move || {
        loop {
            if events.send("tick", "data").is_err() {
                return; // SseClosed: client gone or server draining
            }
            smarm::sleep(std::time::Duration::from_secs(1));
        }
    });
    c
}

While the producer is silent the connection actor emits : keep-alive comment chunks every HEARTBEAT_INTERVAL (15s default). Liveness comes from that ping: a vanished client is detected when a write stalls past write_timeout. No request clock applies to an open SSE stream, and a graceful shutdown force-stops it at the drain deadline.

Named Actors

For introspection and debugging, the server registers itself in smarm's process registry: urus.server (the listener-pool supervisor) and urus.listener.{i} (each accept loop). smarm::whereis(name) resolves them from any actor inside the runtime.

WebSocket (v0.4)

A route handler accepts the upgrade by handing Conn::upgrade a WsHandler; after the 101 the connection actor leaves HTTP and drives the handler from a duplex select loop (smarm RFC 008 fd arms: first-of outbound-channel / fd-readable, one actor, no fd dup):

struct Echo;

impl urus::WsHandler for Echo {
    fn on_message(&mut self, msg: urus::Message, sender: &urus::WsSender) {
        let _ = sender.send(msg); // or .text(..) / .binary(..) / .ping() / .close(code, reason)
    }
    fn on_close(&mut self, code: Option<u16>, reason: &str) {
        // what the PEER said: Some(code) iff its close frame arrived;
        // None for EOF / write failure / handshake timeout.
        let _ = (code, reason);
    }
}

Router::new().get("/echo", |c: Conn, _n: Next| c.upgrade(Echo))

Callbacks run inside the connection actor — a slow on_message stops reads, which is backpressure, not a bug. A handler that wants concurrency spawns its own actor and hands it a WsSender clone (the SSE-producer pattern); every WsSender method returns Err(WsClosed) once the connection ends. While a large outbound frame is mid-write the actor isn't reading — accepted v1 trade-off of the one-actor topology.

Control frames are invisible to the handler in v1: pings are auto-ponged, a peer close is auto-echoed and surfaces as on_close. Protocol violations close with the mapped RFC 6455 code (1002/1009/1007; handler panic → 1011). Caps: Config.max_frame_payload (1 MiB default, enforced from the frame header before payload buffers) and Config.max_message_bytes (4 MiB default, spans fragments).

Like SSE, an open WebSocket has no request clock: idle is fine, a dead client is caught when a write stalls past write_timeout, and graceful shutdown force-stops the connection at the drain deadline. See examples/ws_echo.rs.

The WsHandler trait also has a defaulted on_open(&mut self, sender), called once before the frame loop — the place to subscribe or spawn producers for clients that may never send. Note the client sees the 101 before on_open runs; a client that must know its subscriptions are live uses an application-level ack (any reply proves on_open completed, callbacks being sequential).

Pub/Sub (v0.5)

urus::pubsub is a local-node, phoenix_pubsub-shaped topic broadcaster — independent of HTTP (it imports only smarm) and built for the WebSocket relay pattern:

let bus: PubSub<String> = PubSub::new();          // in-runtime only!
let rx = bus.subscribe("room:lobby")?;            // Receiver<Arc<String>>
bus.broadcast("room:lobby", "hi".to_string())?;
bus.broadcast_from(smarm::self_pid(), "room:lobby", "no echo".into())?;

One generic instance per message domain; payloads broadcast as Arc<M> (one allocation per broadcast). subscribe is idempotent per (pid, topic) and pins the subscription to the calling actor — its death prunes the entry via a monitor; a dropped Receiver is pruned lazily on the next broadcast. subscribe_as(pid, topic) pins to an explicit pid for relay patterns. Mailboxes are unbounded: broadcast never blocks the table, and a slow subscriber's memory bill is bounded by the two cleanup paths above.

Two composition rules that matter (both enforced by shutdown_with_open_chat_terminates in the integration suite):

1. PubSub::new() spawns an actor, so it must run in-runtime — build it lazily from a handler via a non-static Arc<OnceLock<PubSub<M>>> captured by the route closure. A static cell pins the table actor forever and graceful shutdown never returns.
2. Relay/producer actors hold the Receiver (plus e.g. a WsSender clone) — never a PubSub clone, or relay and table keep each other alive past shutdown.

See examples/ws_chat.rs: rooms as topics, on_open subscribes + spawns the relay, on_message uses broadcast_from so the speaker isn't echoed.

Channels (v0.6)

Phoenix-style join/leave/event multiplexing over the WebSocket duplex, pubsub underneath. Two additive features:

• channels — the wire-neutral core (Channel, ChannelHub, PrefixRouter, ChannelSocket, sessions). Zero new dependencies; bring your own codec by implementing Encode/Decode for your payload type.
• phoenix — implies channels, adds serde/serde_json and the phoenix.js V2 wire format via the Json<T> newtype.

use urus::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status};
use urus::channels::phoenix::Json;
use serde_json::{json, Value};

type P = Json<Value>;

#[derive(Default)]
struct Room;

impl Channel<P> for Room {
    fn join(&mut self, topic: &str, payload: P, _s: &ChannelSocket<P>) -> Result<P, P> {
        Ok(Json(json!({ "joined": topic })))   // Err(..) rejects
    }
    fn handle_in(&mut self, event: &str, payload: P, s: &ChannelSocket<P>) {
        match event {
            "shout" => s.broadcast("shouted", payload),
            _ => s.reply(Status::Ok, Json(json!({ "echo": event }))),
        }
    }
}

// in-runtime, non-static — the ws_chat OnceLock pattern applies
let hub = ChannelHub::new(PrefixRouter::new().channel_default::<Room>("room:*"));
// route handler: hub.upgrade(conn)
// from anywhere with a hub handle: hub.broadcast("room:lobby", "news", payload)

One channel actor per joined topic per socket, linked to the connection: callbacks are sequential, a channel panic is a socket event, and the conn-side handler answers transport heartbeats and routes joins — Channel impls never see frames or refs (reply correlates to the in-flight event invisibly).

Session persistence (opt-in). By default channels cold-start per join, Phoenix-server style. Implementing ChannelSession and registering with channel_session makes the channel actor outlive its transport: broadcasts that arrive while detached are buffered (the same Arcs the relay path carries — zero re-serialisation) and drained, in order, under the rejoin's new generation. Buffer cap or TTL exceeded, explicit leave, or a rejected rejoin tears the session down; the next join is a cold start. A second transport claiming the same key evicts the first.

struct ByToken;
impl urus::ChannelSession<P> for ByToken {
    type Key = String;
    fn session_key(topic: &str, join_payload: &P) -> String {
        format!("{topic}/{}", join_payload.0["token"].as_str().unwrap_or(""))
    }
    // fn buffer_cap() -> usize { 128 }            // defaults
    // fn ttl() -> Duration { Duration::from_secs(30) }
}

PrefixRouter::new().channel_session::<ByToken>("session:*", |_: &str| {
    Box::new(Room::default()) as Box<dyn Channel<P>>
})

Session channels must treat join as re-entrant: every reattach calls it again on the same instance (the rejoin ack needs a payload, and the channel gets to re-auth).

See examples/channels_chat.rs (cargo run --example channels_chat --features phoenix): ephemeral room:* and persistent session:* side by side over phoenix.js V2 JSON.

Examples

CRUD with Actor Ownership

See examples/crud.rs for a complete example demonstrating:

• A background "store" actor that owns all data.
• Handlers sending requests to the store via a channel.
• No locks, no shared mutable state.
• Automatic JSON serialization and file persistence.

Run it:

cargo run --example crud
curl -s http://localhost:8080/users
curl -s -X POST -d '{"name":"alice","email":"a@x"}' http://localhost:8080/users
curl -s http://localhost:8080/users/1

Testing

Integration tests spawn the server on an ephemeral port and issue real TCP requests:

cargo test

Tests in tests/integration.rs cover:

• Basic routing, request bodies, headers, path parameters, status codes
• Keep-alive and pipelining
• Listener supervision (a panicking listener restarts without dropping a pending accept)
• Graceful shutdown (idle close, in-flight drain, force-stop at the drain deadline)
• Timeouts (idle keep-alive reaping, slowloris-style slow headers/body)
• Registry names (urus.server, urus.listener.{i} resolve via whereis)

Architecture

Connection Actors

Each incoming TCP connection is handled by a dedicated actor (spawned in conn_actor.rs). The actor:

1. Parses the HTTP request line and headers.
2. Reads the body (if present).
3. Runs the request through the pipeline.
4. Writes the HTTP response to the socket.
5. Closes the connection (or handles pipelining if HTTP/1.1 Keep-Alive is enabled).

All of this happens concurrently with other connections—no thread pool juggling required. The smarm scheduler handles actor fairness.

Parsing

HTTP request parsing uses the robust httparse crate, which handles:

• Chunked transfer encoding (for request bodies).
• Header validation.
• Method and URI parsing.
• HTTP version detection.

No Async/Await

urus uses synchronous code with blocking channels. This is intentional:

• Simpler to reason about; no complex state machines.
• Each actor runs in a smarm worker thread, blocked on I/O or channels.
• The scheduler multiplexes many actors across a thread pool.

This avoids the complexity of async ecosystems while maintaining full concurrency.

Roadmap

See ROADMAP.md. Summary: v0.2 (supervised listener pool, graceful shutdown, enforced timeouts, registry names), v0.3 (streaming bodies, chunked requests, SSE), v0.4 (WebSocket: handshake, frame codec, duplex handler API), v0.5 (PubSub) and v0.6 (Phoenix-style channels with opt-in session persistence) are done.

Refer to urus-spec.md and urus-v1-build-notes.md in the artifact persistence for the original design and implementation notes.

License

MIT