perf: zero-copy server wire path, allocation-free hot paths

AGENTS.md

@@ -52,7 +52,16 @@ Workspace members:
 - Server I/O: one thread owns the ROUTER socket and only moves bytes; handlers
   run on tokio (DB work on the blocking pool) and send replies back over an
   inproc PUSH/PULL pair. Never touch the ROUTER from a handler (zmq sockets are
-  not thread-safe).
+  not thread-safe). Frames travel as `zmq::Message`s end to end (no copies).
+- Zero-copy wire path (`wire.rs` "Server-side views"): the server reads only the
+  header + variant first (`RequestKind`), then decodes the body as `RequestIn`
+  (Complete results / Insert payloads borrow from the received message; bulk
+  requests are decoded on the blocking pool), and encodes replies as `Reply`
+  (`LeasedView` writes leased tasks straight from the chunk leases, `Ids` from
+  `TaskId`s). These mirror `Request`/`Query`/`Response` positionally (postcard):
+  **any change to those types must be mirrored there**; the `wire::tests`
+  byte-equality tests catch drift. Hot `Vec<u8>` fields use `#[serde(with =
+  "bytes")]` (same bytes on the wire, one memcpy instead of per-byte).
 - Transport-level guards (`transport/server.rs`): at most `max_inflight`
   handlers run at once (the I/O thread stops reading the ROUTER when full);
   `Renew`/`Complete`/`CancelLease` reject a lease held by another session uid
@@ -134,6 +143,8 @@ trait method) > engine default (`chunk_size` 1024, no size limits). Resolved in
 ### Task lifecycle (`enginelib/src/nucleus/*`)
 
 - `submit(bytes, type) -> task_id` / `submit_batch(payloads, type) -> [Result<task_id>]`
+  (`submit_batch_ids` / `submit_derived_ids` return `chunk::TaskId`s; the
+  transport uses those so no id `String` is built per task)
   — verifies each payload (+ `max_payload`), packs the valid ones of *this call*
   into fresh chunks of `chunk_size`, reserves seqs, writes all chunks in one
   `WriteBatch`, and only then returns ids. Never merges across calls. Bumps the
@@ -147,7 +158,7 @@ trait method) > engine default (`chunk_size` 1024, no size limits). Resolved in
   Requests a refill below the 25% watermark, both before waiting and again after
   taking a chunk (so workers still parked on a small queue aren't left waiting).
 - `renew` / `renew_batch` — extends the chunk lease named by `(task_id, token)`.
-- `complete` / `complete_batch(items)` — verifies each result (+ `max_result`),
+- `complete` / `complete_batch(items)` / `complete_refs(borrowed items)` — verifies each result (+ `max_result`),
   groups items per (type, chunk, token) and merges each group into the `f:`
   record in one `WriteBatch` (partial completes allowed). When the chunk's last
   open task is done the same batch deletes the `t:` record and the lease + dedup
@@ -290,6 +301,11 @@ Report any warnings/errors that indicate risk (panic paths, serialization, auth
 
 ## Notes for future agents
 
+- Hot paths are kept allocation-free per task (server: ~0.1 allocations/task
+  in `bench`). `Error` messages are `Cow<'static, str>`: use string literals for
+  fixed messages; task ids format via `TaskId::to_hex()` (stack buffer).
+  Measure with `bench` (see the lifecycle playbook) before and after changes.
+
 - Existing code has several warnings and `unwrap()` calls in runtime paths.
 - Prefer minimal, targeted edits.
 - Preserve current external behavior unless task explicitly requests behavior change.

docs/protocol.md

@@ -32,9 +32,12 @@ Envelope { req_id: u64, session: u128, body: Request | Response | ProxyRequest |
   IPv6 is enabled on every socket.
 - Messages larger than `max_message_bytes` drop the connection.
 
-Server internals: one I/O thread owns the ROUTER and only moves bytes. Each
-message is decoded and handled on the tokio runtime (DB work on the blocking
-pool); replies return to the I/O thread over an inproc PUSH/PULL pair. At most
+Server internals: one I/O thread owns the ROUTER and only moves bytes (frames
+are handed over as ZeroMQ messages, never copied). Each message is decoded and
+handled on the tokio runtime (DB work, including decoding `Complete` and
+`Query` bodies, on the blocking pool); payloads and results are read in place
+from the received message and leased tasks are encoded straight from the chunk
+leases. Replies return to the I/O thread over an inproc PUSH/PULL pair. At most
 `max_inflight` requests (default 65 536, parked `Lease` waits included) are
 handled at once; at the cap the I/O thread stops reading the ROUTER until a
 handler finishes, so ZeroMQ's high-water marks and TCP push back on clients.

enginelib/macros/src/lib.rs

@@ -47,10 +47,10 @@ pub fn module(_attr: TokenStream, item: TokenStream) -> TokenStream {
             env!("CARGO_PKG_NAME"),
         );),
     );
-    item_fn.block.stmts.insert(
-        0,
-        parse_quote!(::enginelib::api::init_tracing();),
-    );
+    item_fn
+        .block
+        .stmts
+        .insert(0, parse_quote!(::enginelib::api::init_tracing();));
     item_fn
         .attrs
         .push(parse_quote!(#[unsafe(export_name="run")]));

enginelib/src/api.rs

@@ -871,7 +871,9 @@ impl Registry<dyn Task> for EngineTaskRegistry {
         );
         self.tasks.insert(identifier, task);
     }
-    #[instrument(skip_all, fields(namespace = %identifier.0, name = %identifier.1))]
+    // Runs on every insert / complete / typed query: TRACE, so release builds
+    // (INFO) don't build and format a span per request.
+    #[instrument(level = "trace", skip_all, fields(namespace = %identifier.0, name = %identifier.1))]
     fn get(&self, identifier: &Identifier) -> Option<Box<dyn Task>> {
         self.tasks.get(identifier).map(|obj| obj.clone_box())
     }

enginelib/src/chunk.rs

@@ -45,24 +45,75 @@ impl TaskId {
     /// Parses a 24-char hex task id. Anything else is InvalidArgument
     /// `"InvalidTaskId"`.
     pub fn parse(s: &str) -> Result<Self, Error> {
-        let invalid = || Error::invalid_argument(format!("InvalidTaskId: {s}"));
-        if s.len() != TASK_ID_HEX_LEN || !s.bytes().all(|b| b.is_ascii_hexdigit()) {
-            return Err(invalid());
+        let bytes: &[u8; TASK_ID_HEX_LEN] =
+            s.as_bytes().try_into().map_err(|_| Self::invalid(s))?;
+        let mut raw = [0u8; 12];
+        for (out, pair) in raw.iter_mut().zip(bytes.as_chunks::<2>().0) {
+            let (Some(hi), Some(lo)) = (hex_value(pair[0]), hex_value(pair[1])) else {
+                return Err(Self::invalid(s));
+            };
+            *out = hi << 4 | lo;
+        }
+        Ok(Self::new(
+            u16::from_be_bytes([raw[0], raw[1]]),
+            u64::from_be_bytes(raw[2..10].try_into().expect("8 bytes")),
+            u16::from_be_bytes([raw[10], raw[11]]),
+        ))
+    }
+
+    #[cold]
+    fn invalid(s: &str) -> Error {
+        Error::invalid_argument(format!("InvalidTaskId: {s}"))
+    }
+
+    /// The wire form, hex-encoded into a stack buffer (no allocation).
+    pub fn to_hex(&self) -> TaskIdHex {
+        const DIGITS: &[u8; 16] = b"0123456789abcdef";
+        let mut raw = [0u8; 12];
+        raw[0..2].copy_from_slice(&self.server_id.to_be_bytes());
+        raw[2..10].copy_from_slice(&self.chunk_seq.to_be_bytes());
+        raw[10..12].copy_from_slice(&self.idx.to_be_bytes());
+        let mut hex = [0u8; TASK_ID_HEX_LEN];
+        for (pair, byte) in hex.as_chunks_mut::<2>().0.iter_mut().zip(raw) {
+            pair[0] = DIGITS[(byte >> 4) as usize];
+            pair[1] = DIGITS[(byte & 0xf) as usize];
         }
-        let server_id = u16::from_str_radix(&s[0..4], 16).map_err(|_| invalid())?;
-        let chunk_seq = u64::from_str_radix(&s[4..20], 16).map_err(|_| invalid())?;
-        let idx = u16::from_str_radix(&s[20..24], 16).map_err(|_| invalid())?;
-        Ok(Self::new(server_id, chunk_seq, idx))
+        TaskIdHex(hex)
+    }
+}
+
+fn hex_value(digit: u8) -> Option<u8> {
+    match digit {
+        b'0'..=b'9' => Some(digit - b'0'),
+        b'a'..=b'f' => Some(digit - b'a' + 10),
+        b'A'..=b'F' => Some(digit - b'A' + 10),
+        _ => None,
     }
 }
 
 impl fmt::Display for TaskId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(
-            f,
-            "{:04x}{:016x}{:04x}",
-            self.server_id, self.chunk_seq, self.idx
-        )
+        f.write_str(&self.to_hex())
+    }
+}
+
+/// A task id's 24 lowercase hex chars, on the stack. Derefs to `str`.
+#[derive(Clone, Copy)]
+pub struct TaskIdHex([u8; TASK_ID_HEX_LEN]);
+
+impl std::ops::Deref for TaskIdHex {
+    type Target = str;
+
+    fn deref(&self) -> &str {
+        // Only ever filled with ASCII hex digits.
+        std::str::from_utf8(&self.0).expect("hex digits are ASCII")
+    }
+}
+
+/// Serializes as the wire `task_id` string, without building a `String`.
+impl Serialize for TaskId {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_str(&self.to_hex())
     }
 }
 
@@ -340,50 +391,17 @@ impl FinishedChunk {
     }
 
     pub fn encode(&self) -> Vec<u8> {
-        let n = self.results.len();
-        let size: usize = self.results.iter().flatten().map(|r| r.len() + 4).sum();
-        let mut out = Vec::with_capacity(5 + n.div_ceil(8) + size);
-        out.push(FORMAT_V1);
-        put_u32(&mut out, n);
-        let start = out.len();
-        out.resize(start + n.div_ceil(8), 0);
-        for (idx, result) in self.results.iter().enumerate() {
-            if result.is_some() {
-                out[start + idx / 8] |= 1 << (idx % 8);
-            }
-        }
-        for result in self.results.iter().flatten() {
-            put_u32(&mut out, result.len());
-            out.extend_from_slice(result);
-        }
-        out
+        encode_finished(self.results.iter().map(Option::as_deref))
     }
 
     pub fn decode(data: &[u8]) -> Result<Self, Error> {
-        let mut at = 0;
-        if data.first() != Some(&FORMAT_V1) {
-            return Err(corrupt("finished"));
-        }
-        at += 1;
-        let n = read_u32(data, &mut at).ok_or_else(|| corrupt("finished"))?;
-        if n > MAX_CHUNK_SIZE as usize {
-            return Err(corrupt("finished"));
-        }
-        let bitmap = data
-            .get(at..at + n.div_ceil(8))
-            .ok_or_else(|| corrupt("finished"))?;
-        at += bitmap.len();
-        let mut results = vec![None; n];
-        for (idx, result) in results.iter_mut().enumerate() {
-            if bitmap[idx / 8] & (1 << (idx % 8)) == 0 {
-                continue;
-            }
-            let len = read_u32(data, &mut at).ok_or_else(|| corrupt("finished"))?;
-            let bytes = data.get(at..at + len).ok_or_else(|| corrupt("finished"))?;
-            *result = Some(bytes.to_vec());
-            at += len;
-        }
-        Ok(Self { results })
+        let slots = decode_finished(data)?;
+        Ok(Self {
+            results: slots
+                .into_iter()
+                .map(|slot| slot.map(<[u8]>::to_vec))
+                .collect(),
+        })
     }
 
     /// Walks the results of an encoded record in index order without copying
@@ -441,6 +459,59 @@ impl FinishedChunk {
     }
 }
 
+/// Encodes a finished chunk record from its result slots (`Some` = done), in
+/// index order. Takes borrowed results so callers can merge stored and new
+/// results without copying each one into its own buffer first.
+pub fn encode_finished<'a>(
+    slots: impl ExactSizeIterator<Item = Option<&'a [u8]>> + Clone,
+) -> Vec<u8> {
+    let n = slots.len();
+    let size: usize = slots.clone().flatten().map(|r| r.len() + 4).sum();
+    let mut out = Vec::with_capacity(5 + n.div_ceil(8) + size);
+    out.push(FORMAT_V1);
+    put_u32(&mut out, n);
+    let start = out.len();
+    out.resize(start + n.div_ceil(8), 0);
+    for (idx, slot) in slots.clone().enumerate() {
+        if slot.is_some() {
+            out[start + idx / 8] |= 1 << (idx % 8);
+        }
+    }
+    for result in slots.flatten() {
+        put_u32(&mut out, result.len());
+        out.extend_from_slice(result);
+    }
+    out
+}
+
+/// Decodes a finished chunk record into result slots borrowing from `data`.
+pub fn decode_finished(data: &[u8]) -> Result<Vec<Option<&[u8]>>, Error> {
+    let mut at = 0;
+    if data.first() != Some(&FORMAT_V1) {
+        return Err(corrupt("finished"));
+    }
+    at += 1;
+    let n = read_u32(data, &mut at).ok_or_else(|| corrupt("finished"))?;
+    if n > MAX_CHUNK_SIZE as usize {
+        return Err(corrupt("finished"));
+    }
+    let bitmap = data
+        .get(at..at + n.div_ceil(8))
+        .ok_or_else(|| corrupt("finished"))?;
+    at += bitmap.len();
+    let mut slots = vec![None; n];
+    for (idx, slot) in slots.iter_mut().enumerate() {
+        if bitmap[idx / 8] & (1 << (idx % 8)) == 0 {
+            continue;
+        }
+        let len = read_u32(data, &mut at).ok_or_else(|| corrupt("finished"))?;
+        let end = at.checked_add(len).ok_or_else(|| corrupt("finished"))?;
+        *slot = Some(data.get(at..end).ok_or_else(|| corrupt("finished"))?);
+        at = end;
+    }
+    Ok(slots)
+}
+
 // ---------------------------------------------------------------------------
 // Chunk sequence allocator
 // ---------------------------------------------------------------------------

enginelib/src/error.rs

@@ -1,3 +1,4 @@
+use std::borrow::Cow;
 use std::fmt;
 
 use serde::{Deserialize, Serialize};
@@ -11,36 +12,39 @@ pub enum ErrorKind {
     Unknown,
 }
 
+/// An engine error. The message is a `Cow` so the common static messages
+/// (`"TaskNotFound"`, `"LeaseTokenMismatch"`, ...) never allocate, and cloning
+/// one error into every item of a batch is free.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Error {
     kind: ErrorKind,
-    message: String,
+    message: Cow<'static, str>,
 }
 
 impl Error {
     pub fn new(message: String) -> Error {
         Error {
             kind: ErrorKind::Unknown,
-            message,
+            message: Cow::Owned(message),
         }
     }
 
-    pub fn with_kind(kind: ErrorKind, message: impl Into<String>) -> Error {
+    pub fn with_kind(kind: ErrorKind, message: impl Into<Cow<'static, str>>) -> Error {
         Error {
             kind,
             message: message.into(),
         }
     }
 
-    pub fn not_found(message: impl Into<String>) -> Error {
+    pub fn not_found(message: impl Into<Cow<'static, str>>) -> Error {
         Self::with_kind(ErrorKind::NotFound, message)
     }
 
-    pub fn invalid_argument(message: impl Into<String>) -> Error {
+    pub fn invalid_argument(message: impl Into<Cow<'static, str>>) -> Error {
         Self::with_kind(ErrorKind::InvalidArgument, message)
     }
 
-    pub fn io_error(message: impl Into<String>) -> Error {
+    pub fn io_error(message: impl Into<Cow<'static, str>>) -> Error {
         Self::with_kind(ErrorKind::IOError, message)
     }
 
@@ -57,7 +61,7 @@ impl AsRef<str> for Error {
 
 impl From<Error> for String {
     fn from(e: Error) -> String {
-        e.message
+        e.message.into_owned()
     }
 }