On-Device Twitch IRC in Rust: Reconnect State Machine + Backoff

Why run Twitch IRC on-device?

Ingest chat on the device and skip the middlebox and the serverless line item. For desktop apps, stream overlays, or edge agents driving local LLMs, a single self-contained binary with no cloud dependency is the right shape and keeps ops simple. Rust fits: small footprint, predictable resource use, and tight control over reconnect policy and timing.

How do you build a robust Twitch IRC reconnect loop in Rust?

Ship a small, explicit state machine around tokio-tungstenite. Connect, authenticate, join, then pump reads and writes with proper ping/pong handling. On errors or Twitch RECONNECT, move to Backoff with jitter, clamp to a maximum, and try again. Keep state pure and transitions obvious. Isolate I/O. Make it cancellation-safe with tokio::select!.

States and transitions:

• Disconnected → Connecting → Authenticating → Joining → Connected
• Any error/close/RECONNECT → Backoff(delay) → Connecting
• Shutdown can interrupt at any point

Each state should do only what it owns. Don’t bury retries inside helpers.

Dependencies and project setup

# Cargo.toml
[package]
name = "twitch-irc-on-device"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
futures-util = "0.3"
rand = "0.8"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
# TLS via rustls webpki roots; works cross-platform
tokio-tungstenite = { version = "0.21", features = ["rustls-tls-webpki-roots"] }

Twitch IRC handshake in 4 frames

On a fresh socket, send these four in order:

• CAP REQ :twitch.tv/tags twitch.tv/commands twitch.tv/membership
• PASS oauth:YOUR_TOKEN
• NICK your_nick
• JOIN #channel

Twitch then sends 001..004 numeric messages, followed by ROOMSTATE/NAMES, and later PRIVMSG for chat lines. When you receive PING :tmi.twitch.tv, reply with PONG :tmi.twitch.tv.

Minimal command cheat sheet

From	Message	Action
Srv	PING :tmi.twitch.tv	Send PONG :tmi.twitch.tv
Srv	RECONNECT	Close immediately, reconnect
You	CAP REQ :twitch.tv/...	Enable tags/commands/membership
You	PASS oauth:token	Auth via OAuth token
You	NICK nick	Set nick (login)
You	JOIN #channel	Subscribe to a channel

What’s the best exponential backoff with jitter for WebSockets?

Use exponential backoff with jitter. Full Jitter or Decorrelated Jitter both avoid synchronized spikes; for this client, Decorrelated Jitter works well. Start small (around 250 ms), cap at a ceiling (60–120 s), randomize each attempt, and reset after a period of stable connectivity.

Below is a decorrelated-jitter backoff utility that stays deterministic and testable.

// backoff.rs
use rand::{rngs::StdRng, Rng, SeedableRng};

#[derive(Debug, Clone)]
pub struct JitterBackoff {
    base: f64,       // seconds
    cap: f64,        // seconds
    factor: f64,     // multiplicative factor (typically 3.0)
    prev: f64,       // previous delay
    rng: StdRng,
}

impl JitterBackoff {
    pub fn new(base_ms: u64, cap_ms: u64, seed: u64) -> Self {
        Self {
            base: base_ms as f64 / 1000.0,
            cap: cap_ms as f64 / 1000.0,
            factor: 3.0,
            prev: 0.0,
            rng: StdRng::seed_from_u64(seed),
        }
    }

    // Decorrelated Jitter: next = min(cap, rand(base, prev * factor))
    pub fn next_delay_secs(&mut self) -> f64 {
        let low = self.base;
        let high = (self.prev * self.factor).max(self.base);
        let next = self.rng.gen_range(low..=high).min(self.cap);
        self.prev = next;
        next
    }

    pub fn reset(&mut self) { self.prev = 0.0; }
}

• base=0.25s, cap=60s is a good starting point for chat.
• Reset backoff after you’ve been Connected for, say, 10–30s.

End-to-end: a resilient Twitch IRC client state machine

Below is a single-task runner that:

• Connects to wss://irc-ws.chat.twitch.tv:443
• Authenticates and joins one channel
• Handles PING/PONG and server RECONNECT
• Reconnects with decorrelated-jitter backoff
• Emits lines downstream via a channel (stubbed as println! here)

// main.rs
use futures_util::{SinkExt, StreamExt};
use tokio::{select, time::{sleep, Duration}};
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_tungstenite::connect_async;
use tracing::{error, info, warn};

mod backoff;
use backoff::JitterBackoff;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Phase { Disconnected, Connecting, Authenticating, Joining, Connected, Backoff, Shutdown }

struct Config {
    token: String,   // oauth:XXXXXXXXX (include prefix)
    nick: String,    // your login
    channel: String, // e.g. #somechannel
}

struct Client {
    cfg: Config,
    boff: JitterBackoff,
}

impl Client {
    fn new(cfg: Config) -> Self {
        Self { cfg, boff: JitterBackoff::new(250, 60_000, 0xBEEF) }
    }

    async fn run(mut self, mut shutdown: tokio::sync::watch::Receiver<bool>) -> anyhow::Result<()> {
        tracing_subscriber::fmt()
            .with_env_filter("info")
            .with_target(false)
            .init();

        let mut phase = Phase::Disconnected;
        let mut stable_connected_since: Option<std::time::Instant> = None;

        loop {
            if *shutdown.borrow() { phase = Phase::Shutdown; }

            match phase {
                Phase::Shutdown => {
                    info!("shutdown requested");
                    return Ok(());
                }
                Phase::Disconnected | Phase::Backoff => {
                    let delay = if phase == Phase::Backoff { self.boff.next_delay_secs() } else { 0.0 };
                    if delay > 0.0 { 
                        info!(delay_ms = (delay*1000.0) as u64, "backing off");
                        select! {
                            _ = sleep(Duration::from_secs_f64(delay)) => {},
                            _ = shutdown.changed() => continue,
                        }
                    }
                    phase = Phase::Connecting;
                }
                Phase::Connecting => {
                    info!("connecting wss://irc-ws.chat.twitch.tv:443");
                    let fut = connect_async("wss://irc-ws.chat.twitch.tv:443");
                    let (ws_stream, _resp) = match fut.await {
                        Ok(ok) => ok,
                        Err(e) => { error!(?e, "connect failed"); phase = Phase::Backoff; continue; }
                    };

                    let (mut sink, mut stream) = ws_stream.split();
                    phase = Phase::Authenticating;

                    // Handshake
                    if let Err(e) = sink.send(Message::Text("CAP REQ :twitch.tv/tags twitch.tv/commands twitch.tv/membership\r\n".into())).await { error!(?e, "cap req failed"); phase = Phase::Backoff; continue; }
                    if let Err(e) = sink.send(Message::Text(format!("PASS {}\r\n", self.cfg.token))).await { error!(?e, "pass failed"); phase = Phase::Backoff; continue; }
                    if let Err(e) = sink.send(Message::Text(format!("NICK {}\r\n", self.cfg.nick))).await { error!(?e, "nick failed"); phase = Phase::Backoff; continue; }

                    phase = Phase::Joining;
                    if let Err(e) = sink.send(Message::Text(format!("JOIN {}\r\n", self.cfg.channel))).await { error!(?e, "join failed"); phase = Phase::Backoff; continue; }

                    phase = Phase::Connected;
                    stable_connected_since = Some(std::time::Instant::now());
                    self.boff.reset();

                    // Connected pump loop
                    let mut write = sink;
                    loop {
                        select! {
                            // Handle shutdown
                            _ = shutdown.changed() => { warn!("shutdown while connected"); phase = Phase::Shutdown; break; }

                            // Read incoming frames
                            msg = stream.next() => {
                                match msg {
                                    None => { warn!("ws closed by server"); phase = Phase::Backoff; break; }
                                    Some(Err(e)) => { error!(?e, "ws read error"); phase = Phase::Backoff; break; }
                                    Some(Ok(Message::Text(text))) => {
                                        for line in text.split("\r\n").filter(|l| !l.is_empty()) {
                                            if let Some(new_phase) = Self::handle_line(line, &mut write).await? { phase = new_phase; break; }
                                        }
                                        if phase != Phase::Connected { break; }
                                    }
                                    Some(Ok(Message::Ping(payload))) => {
                                        let _ = write.send(Message::Pong(payload)).await;
                                    }
                                    Some(Ok(Message::Close(frame))) => {
                                        warn!(?frame, "close frame"); phase = Phase::Backoff; break;
                                    }
                                    _ => {}
                                }
                            }
                        }

                        // If we've been stable for >30s, reset backoff entirely (helps after captive portals)
                        if let Some(since) = stable_connected_since {
                            if since.elapsed() > Duration::from_secs(30) { self.boff.reset(); stable_connected_since = None; }
                        }
                    }
                }
                _ => unreachable!(),
            }
        }
    }

    // Returns Some(new_phase) to break connection loop transitions (e.g., Backoff on RECONNECT)
    async fn handle_line(line: &str, write: &mut (impl SinkExt<Message, Error=tokio_tungstenite::tungstenite::Error> + Unpin)) -> anyhow::Result<Option<Phase>> {
        // PING
        if line.starts_with("PING ") {
            let payload = line.trim_start_matches("PING ");
            write.send(Message::Text(format!("PONG {}\r\n", payload))).await?;
            return Ok(None);
        }

        // Server asks us to reconnect without penalty
        if line.contains(" RECONNECT") {
            warn!("server requested RECONNECT");
            return Ok(Some(Phase::Backoff));
        }

        // Basic chat printout: look for PRIVMSG
        if let Some(idx) = line.find(" PRIVMSG ") {
            // Example: @tags;more :nick!user@host PRIVMSG #chan :hello
            let payload = line[idx+9..].to_string();
            if let Some(colon) = payload.find(" :") {
                let channel = &payload[..colon].trim();
                let message = &payload[colon+2..];
                println!("{} | {}", channel, message);
            }
        }

        Ok(None)
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let token = std::env::var("TWITCH_OAUTH").expect("TWITCH_OAUTH=oauth:... env var required");
    let nick = std::env::var("TWITCH_NICK").expect("TWITCH_NICK required");
    let channel = std::env::var("TWITCH_CHANNEL").expect("TWITCH_CHANNEL like #shroud required");

    let cfg = Config { token, nick, channel };
    let client = Client::new(cfg);

    let (tx, rx) = tokio::sync::watch::channel(false);

    // Graceful Ctrl+C
    tokio::spawn(async move {
        let _ = tokio::signal::ctrl_c().await;
        let _ = tx.send(true);
    });

    client.run(rx).await
}

Notes:

• Keep the oauth: prefix in TWITCH_OAUTH (for example, oauth:abcd...). Generate it through Twitch OAuth flows or developer token utilities.
• The stream and sink are split to manage backpressure. For high-volume, multi-channel ingestion, prefer a bounded mpsc for the writer and a dedicated pump.

How do you avoid rate limits and message loss without a server?

Batch command writes and enforce a small token bucket on the writer; Twitch allows modest command rates. Never block reads. Consume frames promptly, push them to an internal queue, and let the consumer apply backpressure. On reconnect, send only the minimal handshake; do not replay historical JOINs or messages.

Quick token bucket sketch for commands:

struct Bucket { cap: u32, tokens: u32, refill_per_sec: f32, last: std::time::Instant }
impl Bucket {
    fn new(cap: u32, refill_per_sec: f32) -> Self { Self { cap, tokens: cap, refill_per_sec, last: std::time::Instant::now() } }
    fn allow(&mut self) -> bool {
        let now = std::time::Instant::now();
        let dt = now.duration_since(self.last).as_secs_f32();
        self.last = now;
        let add = (dt * self.refill_per_sec) as u32;
        self.tokens = (self.tokens + add).min(self.cap);
        if self.tokens > 0 { self.tokens -= 1; true } else { false }
    }
}

Production hardening tips

• Parse tags. Twitch tags include user-id, room-id, and emote spans; do a zero-copy parse with nom or a small hand-rolled splitter.
• TLS roots. The rustls webpki roots feature avoids cross-platform surprises from OS stores.
• RECONNECT fast-path. Treat it like a successful connection (reset backoff), then add a small jitter (10–500 ms) to spread clients.
• Telemetry. Log close codes, RTT (PING→PONG delta), backoff stage, and session duration.
• Testing. Simulate a flaky network by killing the socket, ensure you drop to Backoff, and verify you reach Connected again within a bounded wall time.

Where to go from here?

• Multi-channel: track JOIN targets and re-issue on reconnect, rate-limiting JOIN bursts.
• Structured output: map PRIVMSG into a typed event and forward to an internal queue, LLM, or overlay.
• Mobile: this pattern runs on Android/iOS with rustls; gate network transitions with OS reachability signals to pause and resume the state machine.