urus/README.md

# 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](https://github.com/Markk116/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

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

### Minimal Example

```rust
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:
```bash
cargo run --example hello
curl http://localhost:8080/
```

### Routing with Path Parameters

```rust
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:

```rust
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:

```rust
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:

```rust
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:

```rust
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, runtime_config, pipeline)` — the urus
`Config` holds endpoint knobs, the `smarm::Config` holds runtime knobs
(they are separate because an endpoint placed in someone else's tree
cannot dictate the runtime):

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

let cfg = Config {
    listener_pool:      2,                        // Supervised accept-loop actors
    name:               "urus",                   // Endpoint registry name (unique per endpoint)
    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, smarm::Config::exact(2), 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.

### Your Own Supervision Tree

The real API is `urus::endpoint(config, pipeline)`, which binds the socket
and hands back a supervisable child body. Your application owns the
runtime and the root supervisor; urus is one child among your own actors:

```rust
use smarm::{ChildSpec, OneForOne, Restart, supervisor::Shutdown};

// Binds here: an address-in-use error is a startup failure, not an actor
// crash. The fds outlive any restart of the endpoint child.
let endpoint = urus::endpoint(cfg, pipeline)?;

let rt = smarm::init(smarm::Config::default());
rt.run(move || {
    let sup = smarm::spawn(move || {
        OneForOne::new()
            // Your state actor FIRST: reverse-order shutdown therefore
            // stops it LAST, after HTTP has finished draining.
            .child(ChildSpec::new(Restart::Permanent, my_store))
            .child(ChildSpec::new(Restart::Permanent, endpoint)
                // The endpoint bounds its own drain with drain_timeout;
                // a shorter supervisor deadline would cut it in half.
                .shutdown(Shutdown::Infinity))
            .run()
    });
    let _ = sup.join();
});
```

Shutdown is then whatever your app already does — `request_shutdown` on
the root supervisor (from a SIGTERM handler via `rt.handle()`, say). The
endpoint's `handle_shutdown` stops the listener pool, closes idle
keep-alive connections, drains in-flight requests up to `drain_timeout`,
force-stops stragglers past it, and exits only when the last connection is
gone — so "the endpoint child has stopped" *is* "every connection is
gone". See `examples/crud.rs` for a complete app in this shape.

Address a running endpoint by name for introspection:
`urus::endpoint::whereis("urus")`.

### Graceful Shutdown Without a Tree

If serving is all your process does, let `serve*` own the runtime and use
`Handle`, which triggers the same sequence from any thread:

```rust
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, smarm::Config::default(), pipeline, signal).unwrap();
// Returns once the runtime has fully wound down.
```

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

1. Stop accepting — the listener pool is stopped; 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):

```rust
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:

```rust
|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):

```rust
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:

```rust
const BUS: PubSub<String> = PubSub::new("chat");  // a name; spawns nothing
// BUS.child() goes in your supervision tree (or serve_with*'s child vec)
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.

Composition (enforced by `shutdown_with_open_chat_terminates` in the
integration suite):

1. **The handle is an address, not the table.** `PubSub<M>` is a name:
   `const`, `Copy`, spawns nothing, fine in a `static` or outside the
   runtime. The actor is `PubSub::child()`, a `ChildSpec` for your
   supervision tree — or for `serve_with*`'s app-children vec, which puts
   it ahead of the endpoint so it stops only after the endpoint has
   drained. Operations resolve the name per call, so a restarted table is
   reached transparently.
2. Relay/producer actors hold the `Receiver` (plus e.g. a `WsSender`
   clone). They may hold the handle too — it pins nothing — but usually
   have no use for one.

Prior to v0.8 both of these read the other way round: `PubSub::new()`
spawned the table, the handle owned its life, and a non-static
`Arc<OnceLock<PubSub<M>>>` lazily built from the first handler was the
required idiom. smarm 0.7 made a server's lifetime its own, and that
whole apparatus went away with it.

See [`examples/ws_chat.rs`](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.

```rust
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 }))),
        }
    }
}

// A description: spawns nothing. `children` are the ChildSpecs it needs
// (bus table + one registry per session route) — hand them to serve_with*.
let (hub, children) =
    ChannelHub::new("chat-bus", 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
`Arc`s 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.

```rust
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`](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`](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:
```bash
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:

```bash
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`](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