29 · The async pipeline: non-blocking emit and backpressure

If handling an HTTP request takes tens of microseconds, you can't afford to spend hundreds of microseconds writing an audit log to disk. This chapter walks through how Quipu-Log turns emit into a non-blocking path that queues the event and returns immediately.

The producer-consumer pattern

This is exactly the queue you already know. A single channel sits between the producers (the app threads handling requests) and the consumer (the writer thread that writes to disk). A producer's job ends the moment it drops the event into the channel; the consumer picks events out one by one and writes them to disk in order.

Rust's standard library sync_channel is that channel. Bounded means there's a cap on how many events can wait inside it (4096 by default). Without a cap, memory usage would be unbounded.

The core code: what emit actually does

AuditHandle is the cloneable handle that app code holds directly. Looking at emit_unchecked shows just how short the hot path really is.

crates/quipu-middleware/src/pipeline.rspub fn emit_unchecked(&self, event: AuditEvent) -> Result<(), MiddlewareError> {
    if self.reject_emit_when_disk_full && self.health.disk_full() { ... }
    self.metrics.queue_inc();
    match self.tx.try_send(Command::Emit(Box::new(event))) {
        Ok(()) => Ok(()),
        Err(TrySendError::Full(Command::Emit(ev))) => {
            self.metrics.rejected_queue_full();
            Err(MiddlewareError::QueueFull(ev))
        }
        Err(_) => Err(MiddlewareError::WorkerGone),
    }
}

try_send returns Err immediately — without blocking — when the channel is full. That produces a QueueFull error, and the event comes back to the caller (so the caller can drop it or route it to a separate fallback). This is backpressure — the system's way of saying "I'm too busy right now, I can't take any more."

AuditPipeline and AuditHandle

AuditPipeline owns the writer thread. Exactly one exists for the lifetime of the server; at shutdown you call shutdown() to drain whatever's left in the queue and clean up the thread. AuditHandle is the lightweight clone you get by calling handle() on the pipeline. It wraps an Arc<SyncSender>, so cloning is essentially free — pass it wherever you need it.

crates/quipu-middleware/src/pipeline.rs// Start the pipeline once, at server initialization
let pipeline = AuditPipeline::start(store, root, permissions, cfg, fallback)?;

let handle = pipeline.handle(); // cloneable AuditHandle
// pass handle anywhere via axum State, Arc, etc.
handle.emit(&role, event)?;   // non-blocking