Tauri Rust Backpressure: Bounded vs Unbounded Channels for Chat Floods

The problem: desktop chat floods meet a tiny event loop

Tauri pairs a Rust core with a WebView front end. Firing one UI event per token from an LLM, or per message from a gateway, is trivial. It’s also how you end up buffering millions of events when the WebView can’t drain them fast enough.

The root cause is simple: unbounded channels in the core plus window.emit on every message. Memory climbs without limit the moment the UI slows down (GC, devtools, expensive React work) or inbound traffic bursts.

You need backpressure. In Rust that means bounded channels, explicit overload behavior (block, drop, or coalesce), and batching.

Bounded vs unbounded channels in Rust (Tauri context)

• tokio::sync::mpsc::unbounded_channel

  • Pros: no backpressure; send never fails.
  • Cons: memory grows without limit; collapses under floods.

• tokio::sync::mpsc::channel(cap)

  • Pros: memory bounded; send().await slows producers when full.
  • Cons: if called directly on a hot Tauri command path, you can stall handlers unless you spawn work off-thread.

• tokio::sync::broadcast::channel(cap)

  • Pros: bounded ring buffer; late receivers lose oldest items (drop-oldest) and get RecvError::Lagged(n); send is sync and non-blocking. Good when shedding is better than blocking.
  • Cons: receivers observe gaps when they lag; you must handle Lagged and tolerate missing items.

For a chat viewer, “latest view wins” beats “buffer everything and freeze.” That’s a fit for broadcast with coalescing.

Design goals

• Memory capped even if producers outrun the WebView.
• Keep UI work per frame low; avoid event storms.
• Surface metrics for drops and lag to guide tuning.

Architecture

• Producers (network/LLM) push ChatEvent into a bounded broadcast channel.
• One emitter task subscribes, batches for ~one frame (8–16 ms), and emits to the WebView.
• If the receiver lags, record the drop count and keep going.

Implementation: broadcast with drop-oldest + frame-time batching

Cargo.toml (relevant bits):

[dependencies]
tauri = { version = "1", features = ["api-all"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tracing = "0.1"

Rust (core):

use serde::Serialize;
use std::sync::{Arc, atomic::{AtomicU64, Ordering}};
use tauri::{Manager, State, Window};
use tokio::{sync::broadcast, time::{self, Duration}};

#[derive(Debug, Clone, Serialize)]
pub struct ChatEvent {
  pub room_id: String,
  pub text: String,
  pub ts_ms: u128,
}

#[derive(Clone)]
pub struct ChatHub {
  tx: broadcast::Sender<ChatEvent>,
  dropped: Arc<AtomicU64>,
}

impl ChatHub {
  pub fn new(capacity: usize) -> Self {
    let (tx, _rx) = broadcast::channel(capacity);
    Self { tx, dropped: Arc::new(AtomicU64::new(0)) }
  }

  pub fn publisher(&self) -> broadcast::Sender<ChatEvent> { self.tx.clone() }
  pub fn dropped_total(&self) -> u64 { self.dropped.load(Ordering::Relaxed) }
}

async fn start_emitter(window: Window, hub: ChatHub) {
  let mut rx = hub.tx.subscribe();
  let mut ticker = time::interval(Duration::from_millis(16)); // ~60 FPS
  let mut buf: Vec<ChatEvent> = Vec::with_capacity(256);
  const MAX_BATCH: usize = 256;

  loop {
    tokio::select! {
      _ = ticker.tick() => {
        if !buf.is_empty() {
          let payload = serde_json::json!({ "events": &buf });
          // Emit is synchronous per call; errors usually mean the window is gone.
          let _ = window.emit("chat:delta", payload);
          buf.clear();
        }
      }
      recv = rx.recv() => {
        match recv {
          Ok(ev) => {
            buf.push(ev);
            if buf.len() >= MAX_BATCH {
              let payload = serde_json::json!({ "events": &buf });
              let _ = window.emit("chat:delta", payload);
              buf.clear();
            }
          }
          Err(broadcast::error::RecvError::Lagged(n)) => {
            hub.dropped.fetch_add(n as u64, Ordering::Relaxed);
            // We lost n oldest items; continue with latest.
          }
          Err(broadcast::error::RecvError::Closed) => break,
        }
      }
    }
  }
}

#[tauri::command]
async fn publish_token(hub: State<'_, ChatHub>, room_id: String, text: String) -> Result<(), String> {
  let ev = ChatEvent { room_id, text, ts_ms: 
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() };
  // Non-blocking, bounded, drop-oldest semantics under pressure
  let _ = hub.publisher().send(ev);
  Ok(())
}

pub fn run() {
  tauri::Builder::default()
    .setup(|app| {
      let hub = ChatHub::new(1024); // tuneable capacity (see below)
      let window = app.get_window("main").expect("main window");
      app.manage(hub.clone());
      tauri::async_runtime::spawn(start_emitter(window, hub));
      Ok(())
    })
    .invoke_handler(tauri::generate_handler![publish_token])
    .run(tauri::generate_context!())
    .expect("error while running tauri application");
}

Frontend (TypeScript):

import { listen } from '@tauri-apps/api/event';

type ChatEvent = { room_id: string; text: string; ts_ms: number };

async function bootstrap() {
  await listen<{ events: ChatEvent[] }>('chat:delta', ({ payload }) => {
    // Apply batched deltas in a single render pass.
    for (const ev of payload.events) {
      appendToChat(ev.room_id, ev.text, ev.ts_ms);
    }
  });
}

Why this works under floods:

• Memory is bounded: the broadcast ring caps usage at O(capacity), plus a small batching vector.
• UI work is bounded: coalescing turns many messages into one emit per frame or per 256 events.
• Under overload, old deltas are discarded; users see current state rather than a massive backlog.

Anti-pattern: unbounded channels + per-message emits

This is how you OOM a desktop app under load:

use tokio::sync::mpsc;
let (tx, mut rx) = mpsc::unbounded_channel::<ChatEvent>();
// Producers: tx.send(ev).unwrap(); // infallible => unbounded growth
// Emitter loop:
while let Some(ev) = rx.recv().await {
  window.emit("chat:delta", ev).unwrap(); // emit 100k/s? enjoy
}

Avoid this for UI-bound streams. When the WebView stalls, the buffer grows forever.

Alternative: bounded mpsc to block producers (strict delivery)

If dropping is unacceptable (e.g., feeding persistence that must receive every item), use a bounded mpsc and move the wait off the Tauri command path:

use tokio::sync::mpsc;

#[derive(Clone)]
struct StrictBus { tx: mpsc::Sender<ChatEvent> }

impl StrictBus {
  fn new(cap: usize) -> (Self, mpsc::Receiver<ChatEvent>) {
    let (tx, rx) = mpsc::channel(cap);
    (Self { tx }, rx)
  }

  async fn publish(&self, ev: ChatEvent) {
    // This awaits when full: true backpressure
    let _ = self.tx.send(ev).await;
  }
}

// In setup
let (bus, mut rx) = StrictBus::new(1024);
app.manage(bus.clone());
// Spawn a background aggregator that drains rx and batches to the UI (as before).

tauri::async_runtime::spawn(async move {
  let mut buf = Vec::with_capacity(256);
  let mut ticker = tokio::time::interval(Duration::from_millis(16));
  loop {
    tokio::select! {
      _ = ticker.tick() => { if !buf.is_empty() { let _ = window.emit("chat:delta", serde_json::json!({"events": &buf })); buf.clear(); } }
      Some(ev) = rx.recv() => { buf.push(ev); if buf.len() >= 256 { let _ = window.emit("chat:delta", serde_json::json!({"events": &buf })); buf.clear(); } }
      else => break,
    }
  }
});

// In a Tauri command, never await on .publish directly if you're on the hot path.
#[tauri::command]
async fn deliver_all(bus: State<'_, StrictBus>, room_id: String, text: String) -> Result<(), String> {
  let ev = ChatEvent { room_id, text, ts_ms: 0 };
  // Offload the potential await to a background task
  let b = bus.clone();
  tauri::async_runtime::spawn(async move { b.publish(ev).await; });
  Ok(())
}

This keeps delivery strict (no drops) by slowing producers when the UI is slow. That’s often fine for network-bound producers; for CPU-heavy tokenization, make sure you yield so the runtime doesn’t starve.

Tuning knobs that actually matter

• Channel capacity: for broadcast, start at 1–4k. For strict mpsc, 256–1k usually works because batching drains quickly.
• Batch window: 8–16 ms lines up with 60 Hz and keeps the UI quick. If payloads are large, 33 ms can help.
• Max batch size: 128–512. Oversizing just taxes JSON serialization and JS parsing.
• Per-room fairness: one channel per room with a fair round-robin in the emitter when serving multiple chats.
• Drop metrics: expose dropped_total via a Tauri command or periodic event so you can adjust sizing with data.

Failure modes and guardrails

• Deadlocks: don’t await send() on a bounded mpsc inside a path that ties up the same runtime workers. Spawn instead.
• Emitter reentrancy: emitting per message without batching burns time on JSON work.
• Shutdown: when all senders drop, channels close; exit the emitter cleanly on Closed.
• Observability: add tracing around batching and emitting; publish a heartbeat with drops and batch sizes.

TL;DR recipe

• Use tokio::broadcast::channel for UI-facing streams to bound memory with drop-oldest behavior.
• Batch to frame time or a max size before calling window.emit.
• If you can’t lose data, use a bounded mpsc and backpressure producers off the UI path.
• Measure drops and tune capacity and batch windows so floods don’t sink the app.