README: document multi-viewer + bandwidth pre-flight

Updates the status section to move multi-viewer out of "not yet
working", adds a Configuration section pointing at the new TOML config
at ~/.config/pixelpass/config.toml, and a Multi-viewer section
covering the lazy-sticky lifecycle, the --max-viewers cap, the
bandwidth-bitrate tradeoff, and how to fit more viewers by dropping
--bitrate. Known-limitations section gains "late joiners see ~2 s of
garbage" (expected behavior) and drops the now-stale "single viewer
per host" line.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Cargo.lock

@@ -3074,6 +3074,7 @@ dependencies = [
  "anyhow",
  "arboard",
  "ashpd",
+ "chrono",
  "clap",
  "dialoguer",
  "directories",
@@ -3086,8 +3087,10 @@ dependencies = [
  "thiserror 2.0.18",
  "tokio",
  "tokio-util",
+ "toml",
  "tracing",
  "tracing-subscriber",
+ "ureq",
  "uuid",
  "x11rb",
 ]
@@ -4448,6 +4451,34 @@ version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
 
+[[package]]
+name = "ureq"
+version = "3.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0"
+dependencies = [
+ "base64",
+ "log",
+ "percent-encoding",
+ "rustls",
+ "rustls-pki-types",
+ "ureq-proto",
+ "utf8-zero",
+ "webpki-roots",
+]
+
+[[package]]
+name = "ureq-proto"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c"
+dependencies = [
+ "base64",
+ "http",
+ "httparse",
+ "log",
+]
+
 [[package]]
 name = "url"
 version = "2.5.8"
@@ -4461,6 +4492,12 @@ dependencies = [
  "serde_derive",
 ]
 
+[[package]]
+name = "utf8-zero"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
+
 [[package]]
 name = "utf8_iter"
 version = "1.0.4"

Cargo.toml

@@ -30,6 +30,9 @@ uuid = { version = "1", features = ["v4"] }
 iroh-tickets = "1.0.0-rc.0"
 dialoguer = { version = "0.12", default-features = false }
 arboard = { version = "3", default-features = false, features = ["wayland-data-control"] }
+ureq = { version = "3", default-features = false, features = ["rustls"] }
+toml = "1"
+chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
 
 [profile.release]
 lto = "thin"

README.md

@@ -22,11 +22,15 @@ Working:
 - iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified
 - Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker
 - Headless mode for scripts (`pixelpass <ticket>`)
+- Multi-viewer fanout (default 2, configurable via `--max-viewers`;
+  shared gst pipeline, one broadcast channel per host)
+- First-run upstream bandwidth pre-flight, persisted to
+  `~/.config/pixelpass/config.toml` and used to auto-size the default
+  viewer cap
 
 Not yet working:
 - X11 capture (stubbed, returns an error)
 - Per-app audio routing (`--app <name>` is a flag stub)
-- Multi-viewer (single viewer per host by design right now)
 - `--repair` (PipeWire orphan cleanup) is a stub
 
 ## Quick start
@@ -39,7 +43,14 @@ pixelpass
 
 On the host machine: pick "Host", share a monitor via the portal dialog,
 the ticket lands on your clipboard. Send it to your viewer however you
-like (chat, email, paste in a note).
+like (chat, email, paste in a note). The same ticket works for multiple
+viewers up to your `--max-viewers` cap.
+
+The very first host launch offers a one-time upstream bandwidth test
+(~5 s, ~5 MB to Cloudflare's open speed-test endpoint) so it can pick
+a sensible default for the viewer cap. You can skip it and a
+conservative default (2 viewers) is used; re-run it later with
+`pixelpass --reconfigure`.
 
 On the viewer machine: run `pixelpass`, pick "View", paste the ticket,
 pick mpv or VLC. The player launches detached and the stream starts.
@@ -154,6 +165,42 @@ relay path otherwise. Both have been verified end-to-end.
 - **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on
   the host's hardware.
 
+## Configuration
+
+`pixelpass` keeps a small TOML config at `~/.config/pixelpass/config.toml`
+(or the XDG equivalent). Right now it only stores the result of the
+bandwidth pre-flight:
+
+```toml
+[bandwidth]
+status = "measured"            # measured | skipped | failed | unmeasured
+upstream_mbps = 8.78           # safe estimate (raw * 0.8)
+measured_at = "2026-05-21T20:41:16Z"
+```
+
+- `pixelpass --reconfigure` re-runs the test (e.g. after an ISP change).
+- Deleting the file resets pixelpass to first-run state.
+- Skip is sticky — once you skip the test, pixelpass won't ask again
+  unless you reconfigure.
+
+## Multi-viewer
+
+One gst capture pipeline fans out to N concurrent viewers via a
+`tokio::sync::broadcast` channel. The same ticket is reusable: as long
+as a viewer is connected, capture stays alive; when the last one
+leaves, the pipeline tears down and the portal stops streaming. A new
+viewer connecting after that re-triggers the portal dialog.
+
+Capacity is bounded by upstream bandwidth (each viewer is its own
+encrypted egress). The default cap comes from the bandwidth pre-flight
+result; `--max-viewers <N>` overrides it. When the cap is hit,
+additional connections are politely refused with a "host is full"
+message and the host keeps running.
+
+For more viewers, drop the per-viewer bitrate: e.g. `pixelpass
+--bitrate 2500 --max-viewers 4` fits four 2.5 Mbps streams in roughly
+12 Mbps of upstream.
+
 ## Known limitations and gotchas
 
 - **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family
@@ -164,9 +211,8 @@ relay path otherwise. Both have been verified end-to-end.
 - **Audio echo** if the host plays the stream through speakers and
   captures system audio — expected, the mic / monitor picks up the
   playback. Headphones bypass it.
-- **Single viewer per host** by design right now. Restarting the player
-  against the same URL fails with "connection refused"; restart the
-  viewer too.
+- **Late joiners see ~2 s of garbage** before the next keyframe lets
+  their decoder lock. Expected behavior, not a bug.
 - **VAAPI driver must be package-tracked**, not an orphaned `.so` on
   disk. mpv's `--hwdec=auto` silently falls back to software decode
   otherwise, which then chokes on a low-power viewer.

src/cli.rs

@@ -45,6 +45,13 @@ pub struct Cli {
     #[arg(long)]
     pub low_latency: bool,
 
+    /// Maximum number of concurrent viewers. Additional connections are
+    /// politely refused with a "host full" message. Defaults to the
+    /// connection-aware recommendation from the bandwidth pre-flight if
+    /// available, otherwise 2.
+    #[arg(long)]
+    pub max_viewers: Option<u32>,
+
     // ── viewer options ────────────────────────────────────────────────
     /// Local TCP port for the viewer to expose (default: random).
     #[arg(long, default_value_t = 0)]
@@ -58,6 +65,12 @@ pub struct Cli {
     /// Clean up orphaned PipeWire state from a crashed host run, then exit.
     #[arg(long)]
     pub repair: bool,
+
+    /// Re-run the bandwidth pre-flight test, save the result, then exit.
+    /// Use this if your connection has changed (new ISP, moved house, etc.)
+    /// or if the previously saved test result is stale.
+    #[arg(long)]
+    pub reconfigure: bool,
 }
 
 #[derive(ValueEnum, Clone, Copy, Debug)]
@@ -76,6 +89,7 @@ pub struct HostOpts {
     pub framerate: u32,
     pub no_hwencode: bool,
     pub low_latency: bool,
+    pub max_viewers: Option<u32>,
     pub interactive: bool,
 }
 
@@ -96,6 +110,7 @@ impl Cli {
             framerate: self.framerate,
             no_hwencode: self.no_hwencode,
             low_latency: self.low_latency,
+            max_viewers: self.max_viewers,
             interactive,
         }
     }

src/common/bandwidth.rs

@@ -0,0 +1,73 @@
+//! One-shot upstream bandwidth measurement against Cloudflare's open
+//! speed-test endpoint. POST a fixed payload, time it, derive Mbps.
+//!
+//! Run via `tokio::task::spawn_blocking` from async contexts — ureq is a
+//! blocking client and we don't want to wedge the tokio runtime during
+//! the test.
+
+use anyhow::{Context, Result};
+use std::time::{Duration, Instant};
+
+const ENDPOINT: &str = "https://speed.cloudflare.com/__up";
+const PAYLOAD_BYTES: usize = 5 * 1024 * 1024; // 5 MiB
+const HTTP_TIMEOUT: Duration = Duration::from_secs(30);
+/// Multiplier applied to the raw measurement. TCP slow-start, ramp-up, and
+/// real-world contention all mean a one-shot upstream test slightly
+/// overestimates sustainable throughput; clamp to 80% for headroom.
+const SAFETY_FACTOR: f64 = 0.80;
+
+/// Result of a successful measurement.
+#[derive(Debug, Clone)]
+pub struct Measurement {
+    /// Raw measured throughput in megabits per second.
+    pub raw_mbps: f64,
+    /// `raw_mbps * SAFETY_FACTOR` — the value to use when sizing things.
+    pub safe_mbps: f64,
+    /// How long the upload took.
+    pub elapsed: Duration,
+}
+
+/// Blocking upload-speed test. Call from a `spawn_blocking` task.
+pub fn measure_upstream_blocking() -> Result<Measurement> {
+    let payload = vec![0u8; PAYLOAD_BYTES];
+    let agent = ureq::Agent::config_builder()
+        .timeout_global(Some(HTTP_TIMEOUT))
+        .build()
+        .new_agent();
+
+    let start = Instant::now();
+    let response = agent
+        .post(ENDPOINT)
+        .content_type("application/octet-stream")
+        .send(&payload[..])
+        .context("upload request to Cloudflare failed")?;
+    let elapsed = start.elapsed();
+
+    let status = response.status();
+    if !status.is_success() {
+        anyhow::bail!("Cloudflare returned HTTP {status}");
+    }
+
+    let bits = (PAYLOAD_BYTES as f64) * 8.0;
+    let seconds = elapsed.as_secs_f64().max(0.001);
+    let raw_mbps = bits / seconds / 1_000_000.0;
+    let safe_mbps = raw_mbps * SAFETY_FACTOR;
+
+    Ok(Measurement {
+        raw_mbps,
+        safe_mbps,
+        elapsed,
+    })
+}
+
+/// Convert a safe-upstream Mbps figure plus the host's per-viewer bitrate
+/// (kbps for video, ignoring audio + protocol overhead which we account for
+/// via SAFETY_FACTOR) into a recommended viewer count. Floors to at least 1.
+pub fn recommended_max_viewers(safe_mbps: f64, bitrate_kbps: u32) -> u32 {
+    let per_viewer_mbps = (bitrate_kbps as f64) / 1000.0;
+    if per_viewer_mbps <= 0.0 {
+        return 1;
+    }
+    let n = (safe_mbps / per_viewer_mbps).floor();
+    if n < 1.0 { 1 } else { n as u32 }
+}

src/common/config.rs

@@ -0,0 +1,101 @@
+//! Persistent user-level config at `~/.config/pixelpass/config.toml`.
+//!
+//! Right now this only tracks the bandwidth pre-flight result. Future
+//! preferences (default player, default bitrate, etc.) can hang off the
+//! same file under their own `[section]`.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
+use directories::ProjectDirs;
+use serde::{Deserialize, Serialize};
+use std::fs;
+use std::io::Write;
+use std::path::PathBuf;
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct Config {
+    #[serde(default)]
+    pub bandwidth: BandwidthEntry,
+}
+
+/// Result of the first-run upstream measurement.
+///
+/// `status = "unmeasured"` means we've never asked the user — show the
+/// first-run dialog. `"measured"` means we have a number. `"skipped"`
+/// means the user opted out (sticky — don't ask again). `"failed"`
+/// means the last attempt errored and we should ask the user on next
+/// interactive launch whether to retry.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct BandwidthEntry {
+    #[serde(default = "default_status")]
+    pub status: BandwidthStatus,
+    #[serde(default)]
+    pub upstream_mbps: Option<f64>,
+    #[serde(default)]
+    pub measured_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum BandwidthStatus {
+    Unmeasured,
+    Measured,
+    Skipped,
+    Failed,
+}
+
+impl Default for BandwidthStatus {
+    fn default() -> Self {
+        Self::Unmeasured
+    }
+}
+
+fn default_status() -> BandwidthStatus {
+    BandwidthStatus::Unmeasured
+}
+
+/// Returns `~/.config/pixelpass/config.toml` (or the XDG equivalent on other
+/// platforms). The parent directory is created lazily by [`save`].
+pub fn config_path() -> Result<PathBuf> {
+    let dirs = ProjectDirs::from("", "", "pixelpass")
+        .context("could not locate a config directory for pixelpass")?;
+    Ok(dirs.config_dir().join("config.toml"))
+}
+
+/// Returns the loaded config, or a `Default` instance if the file doesn't
+/// exist yet. Bubble up parse errors so we don't silently overwrite a
+/// hand-edited config the user is debugging.
+pub fn load() -> Result<Config> {
+    let path = config_path()?;
+    match fs::read_to_string(&path) {
+        Ok(s) => toml::from_str::<Config>(&s)
+            .with_context(|| format!("failed to parse {}", path.display())),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Config::default()),
+        Err(e) => Err(e).with_context(|| format!("failed to read {}", path.display())),
+    }
+}
+
+/// Atomic write via tempfile-in-same-dir + rename.
+pub fn save(cfg: &Config) -> Result<()> {
+    let path = config_path()?;
+    let parent = path
+        .parent()
+        .context("config path has no parent directory")?;
+    fs::create_dir_all(parent)
+        .with_context(|| format!("failed to create {}", parent.display()))?;
+
+    let serialized =
+        toml::to_string_pretty(cfg).context("failed to serialize config to TOML")?;
+
+    let tmp = parent.join(format!(".config.toml.tmp.{}", std::process::id()));
+    {
+        let mut f = fs::File::create(&tmp)
+            .with_context(|| format!("failed to create {}", tmp.display()))?;
+        f.write_all(serialized.as_bytes())
+            .with_context(|| format!("failed to write {}", tmp.display()))?;
+        f.sync_all().ok();
+    }
+    fs::rename(&tmp, &path)
+        .with_context(|| format!("failed to rename {} -> {}", tmp.display(), path.display()))?;
+    Ok(())
+}

src/common/mod.rs

@@ -1,4 +1,6 @@
 pub mod alpn;
+pub mod bandwidth;
+pub mod config;
 pub mod deps;
 pub mod display;
 pub mod process;

src/host/mod.rs

@@ -5,10 +5,28 @@ use anyhow::{Result, bail};
 use iroh::Endpoint;
 use iroh::endpoint::{Connection, presets};
 use iroh_tickets::endpoint::EndpointTicket;
+use std::time::Duration;
+use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::CancellationToken;
 
 use crate::cli::HostOpts;
-use crate::common::{alpn::ALPN, deps, display::DisplayServer, signal};
+use crate::common::{
+    alpn::ALPN, bandwidth, config, config::BandwidthStatus, deps, display::DisplayServer, signal,
+    tunnel,
+};
+
+use self::capture::CaptureHandle;
+
+/// Messages from per-viewer tasks to the capture supervisor.
+enum SupervisorMsg {
+    /// A new viewer wants in. Supervisor replies with the local capture
+    /// HTTP port to connect to, or an error string if the host is full or
+    /// capture spawn failed.
+    AddViewer(oneshot::Sender<Result<u16, String>>),
+    /// A viewer's session ended. Supervisor decrements the count and tears
+    /// down capture if it just hit zero.
+    RemoveViewer,
+}
 
 pub async fn run(opts: HostOpts) -> Result<()> {
     let display = DisplayServer::resolve(opts.display_server);
@@ -21,6 +39,11 @@ pub async fn run(opts: HostOpts) -> Result<()> {
         );
     }
 
+    let resolution = resolve_max_viewers(&opts);
+    if resolution.value == 0 {
+        bail!("--max-viewers must be at least 1");
+    }
+
     let cancel = signal::install_ctrl_c();
 
     let endpoint = Endpoint::builder(presets::N0)
@@ -31,73 +54,177 @@ pub async fn run(opts: HostOpts) -> Result<()> {
     let addr = endpoint.addr();
     let ticket = EndpointTicket::new(addr);
     let clipboard_ok = opts.interactive && copy_to_clipboard(&ticket.to_string());
-    print_host_banner(&ticket, display, &opts, clipboard_ok);
+    print_host_banner(&ticket, display, &opts, &resolution, clipboard_ok);
 
-    let result = accept_loop(&endpoint, display, &opts, cancel.clone()).await;
+    let (sup_tx, sup_rx) = mpsc::channel::<SupervisorMsg>(16);
+    let supervisor = tokio::spawn(supervise(opts.clone(), display, resolution.value, sup_rx));
+
+    accept_loop(&endpoint, sup_tx.clone(), cancel.clone()).await;
+
+    drop(sup_tx);
+    let _ = supervisor.await;
 
     endpoint.close().await;
-    result
+    Ok(())
 }
 
 async fn accept_loop(
     endpoint: &Endpoint,
-    display: DisplayServer,
-    opts: &HostOpts,
+    sup_tx: mpsc::Sender<SupervisorMsg>,
     cancel: CancellationToken,
-) -> Result<()> {
-    tokio::select! {
-        _ = cancel.cancelled() => {
-            tracing::info!("cancellation requested before any peer connected");
-            Ok(())
-        }
-        accepted = endpoint.accept() => {
-            let Some(incoming) = accepted else {
-                bail!("endpoint stopped accepting connections");
-            };
-            let conn = incoming.await?;
-            let remote = conn.remote_id();
-            tracing::info!(%remote, "peer connected");
-            eprintln!("\n[pixelpass] peer connected: {remote}\n");
-            handle_peer(conn, display, opts, cancel).await
+) {
+    loop {
+        tokio::select! {
+            _ = cancel.cancelled() => {
+                tracing::info!("cancellation requested — closing accept loop");
+                return;
+            }
+            accepted = endpoint.accept() => {
+                let Some(incoming) = accepted else {
+                    tracing::info!("endpoint stopped accepting connections");
+                    return;
+                };
+                let conn = match incoming.await {
+                    Ok(c) => c,
+                    Err(e) => {
+                        tracing::warn!("incoming connection failed: {e:#}");
+                        continue;
+                    }
+                };
+                let sup_tx = sup_tx.clone();
+                let cancel = cancel.clone();
+                tokio::spawn(handle_peer(conn, sup_tx, cancel));
+            }
         }
     }
 }
 
 async fn handle_peer(
     conn: Connection,
-    display: DisplayServer,
-    opts: &HostOpts,
+    sup_tx: mpsc::Sender<SupervisorMsg>,
     cancel: CancellationToken,
-) -> Result<()> {
-    let (quic_send, quic_recv) = conn.accept_bi().await?;
+) {
+    let remote = conn.remote_id();
 
-    let capture_handle = capture::spawn(display, opts).await?;
-    let port = capture_handle.local_port();
-    let tcp = wayland::connect_to_capture(port, std::time::Duration::from_secs(5)).await?;
+    let (reply_tx, reply_rx) = oneshot::channel();
+    if sup_tx.send(SupervisorMsg::AddViewer(reply_tx)).await.is_err() {
+        tracing::warn!(%remote, "supervisor channel closed; dropping peer");
+        return;
+    }
+    let port = match reply_rx.await {
+        Ok(Ok(p)) => p,
+        Ok(Err(reason)) => {
+            tracing::warn!(%remote, %reason, "refusing viewer");
+            eprintln!("[pixelpass] refusing viewer {remote}: {reason}");
+            return;
+        }
+        Err(_) => {
+            tracing::warn!(%remote, "supervisor reply dropped; dropping peer");
+            return;
+        }
+    };
 
-    let bridge = crate::common::tunnel::bridge(quic_send, quic_recv, tcp);
+    let (quic_send, quic_recv) = match conn.accept_bi().await {
+        Ok(s) => s,
+        Err(e) => {
+            tracing::warn!(%remote, "accept_bi failed: {e:#}");
+            let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
+            return;
+        }
+    };
 
+    eprintln!("[pixelpass] viewer connected: {remote}");
+
+    let tcp = match wayland::connect_to_capture(port, Duration::from_secs(5)).await {
+        Ok(t) => t,
+        Err(e) => {
+            tracing::warn!(%remote, "connect_to_capture failed: {e:#}");
+            let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
+            return;
+        }
+    };
+
+    let bridge = tunnel::bridge(quic_send, quic_recv, tcp);
     tokio::select! {
         res = bridge => {
             if let Err(e) = res {
-                tracing::warn!("bridge ended with error: {e:#}");
+                tracing::warn!(%remote, "bridge ended with error: {e:#}");
             } else {
-                tracing::info!("bridge closed cleanly");
+                tracing::info!(%remote, "bridge closed cleanly");
             }
         }
         _ = cancel.cancelled() => {
-            tracing::info!("cancellation requested during stream");
+            tracing::info!(%remote, "cancellation during stream");
         }
     }
 
-    capture_handle.shutdown().await;
-    Ok(())
+    eprintln!("[pixelpass] viewer disconnected: {remote}");
+    let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
+}
+
+/// Owns the single shared CaptureHandle and the active viewer count. Spawns
+/// capture lazily on the first AddViewer; tears it down when the count drops
+/// back to zero. Enforces the max-viewers cap by refusing AddViewer when
+/// the count is already at the cap.
+async fn supervise(
+    opts: HostOpts,
+    display: DisplayServer,
+    max_viewers: u32,
+    mut rx: mpsc::Receiver<SupervisorMsg>,
+) {
+    let mut handle: Option<CaptureHandle> = None;
+    let mut count: u32 = 0;
+
+    while let Some(msg) = rx.recv().await {
+        match msg {
+            SupervisorMsg::AddViewer(reply) => {
+                if count >= max_viewers {
+                    let _ = reply.send(Err(format!(
+                        "host is full ({count} of {max_viewers} viewers connected)"
+                    )));
+                    continue;
+                }
+
+                if handle.is_none() {
+                    tracing::info!("first viewer arriving — spawning capture");
+                    match capture::spawn(display, &opts).await {
+                        Ok(h) => handle = Some(h),
+                        Err(e) => {
+                            let _ = reply.send(Err(format!("capture spawn failed: {e:#}")));
+                            continue;
+                        }
+                    }
+                }
+
+                let port = handle.as_ref().expect("handle was just set").local_port();
+                count += 1;
+                let _ = reply.send(Ok(port));
+                tracing::info!(active = count, cap = max_viewers, "viewer joined");
+            }
+            SupervisorMsg::RemoveViewer => {
+                count = count.saturating_sub(1);
+                tracing::info!(active = count, cap = max_viewers, "viewer left");
+                if count == 0
+                    && let Some(h) = handle.take()
+                {
+                    tracing::info!("last viewer left — tearing down capture");
+                    h.shutdown().await;
+                }
+            }
+        }
+    }
+
+    if let Some(h) = handle.take() {
+        tracing::info!("host shutdown — tearing down capture");
+        h.shutdown().await;
+    }
 }
 
 fn print_host_banner(
     ticket: &EndpointTicket,
     display: DisplayServer,
     opts: &HostOpts,
+    resolution: &MaxViewersResolution,
     clipboard_ok: bool,
 ) {
     eprintln!();
@@ -106,23 +233,79 @@ fn print_host_banner(
     eprintln!("│  capture        : {}", capture_summary(opts));
     eprintln!("│  bitrate / fps  : {} kbps @ {} fps", opts.bitrate, opts.framerate);
     eprintln!("│  hw encode      : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" });
+    eprintln!("│  max viewers    : {} ({})", resolution.value, resolution.source.label());
     eprintln!("│");
     if clipboard_ok {
         eprintln!("│  Your share code has been copied to your clipboard.");
-        eprintln!("│  Send it to your viewer. (If clipboard didn't work, the");
+        eprintln!("│  Send it to your viewer(s). (If clipboard didn't work, the");
         eprintln!("│  code is also shown below for manual copy.)");
     } else {
-        eprintln!("│  Share this ticket with your viewer:");
+        eprintln!("│  Share this ticket with your viewer(s):");
     }
     eprintln!("│");
     eprintln!("│      pixelpass {ticket}");
     eprintln!("│");
-    eprintln!("│  Capture will not start until the viewer connects.");
-    eprintln!("│  Press Ctrl+C to stop.");
+    eprintln!("│  Capture starts when the first viewer connects, runs while");
+    eprintln!("│  any viewer is connected, and tears down when the last one");
+    eprintln!("│  leaves. Press Ctrl+C to stop the host entirely.");
     eprintln!("└────────────────────────────────────────────────────────────");
     eprintln!();
 }
 
+/// How we arrived at the final viewer cap. Surfaced in the banner so the
+/// user can tell at a glance whether the number is what they specified,
+/// what their measured upstream supports, or just the fallback default.
+struct MaxViewersResolution {
+    value: u32,
+    source: MaxViewersSource,
+}
+
+enum MaxViewersSource {
+    /// User passed --max-viewers explicitly.
+    UserFlag,
+    /// Derived from the saved bandwidth measurement.
+    BandwidthMeasurement { safe_mbps: f64 },
+    /// No flag, no measurement — falling back.
+    DefaultFallback,
+}
+
+impl MaxViewersSource {
+    fn label(&self) -> String {
+        match self {
+            MaxViewersSource::UserFlag => "user-specified".to_string(),
+            MaxViewersSource::BandwidthMeasurement { safe_mbps } => {
+                format!("auto: {safe_mbps:.1} Mbps measured upstream")
+            }
+            MaxViewersSource::DefaultFallback => {
+                "default — run `pixelpass --reconfigure` for a connection-aware value".to_string()
+            }
+        }
+    }
+}
+
+fn resolve_max_viewers(opts: &HostOpts) -> MaxViewersResolution {
+    if let Some(n) = opts.max_viewers {
+        return MaxViewersResolution {
+            value: n,
+            source: MaxViewersSource::UserFlag,
+        };
+    }
+    if let Ok(cfg) = config::load()
+        && cfg.bandwidth.status == BandwidthStatus::Measured
+        && let Some(upstream) = cfg.bandwidth.upstream_mbps
+    {
+        let n = bandwidth::recommended_max_viewers(upstream, opts.bitrate);
+        return MaxViewersResolution {
+            value: n,
+            source: MaxViewersSource::BandwidthMeasurement { safe_mbps: upstream },
+        };
+    }
+    MaxViewersResolution {
+        value: 2,
+        source: MaxViewersSource::DefaultFallback,
+    }
+}
+
 fn copy_to_clipboard(text: &str) -> bool {
     match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) {
         Ok(()) => true,

src/host/wayland.rs

@@ -1,7 +1,10 @@
 //! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch
 //! pipewiresrc → MPEG-TS on gst stdout → in-process HTTP server bound on a
-//! random localhost port. The host bridge TCP-connects to that server and
-//! pumps bytes to QUIC.
+//! random localhost port. One gst child feeds a tokio::sync::broadcast channel;
+//! the HTTP listener accepts multiple connections and each one drains its own
+//! fresh broadcast::Receiver — so a single capture pipeline fans out to N
+//! concurrent viewers. Slow consumers see Lagged and skip ahead; the MPEG-TS
+//! stream resyncs at the next keyframe.
 
 use anyhow::{Context, Result, bail};
 use ashpd::{
@@ -16,18 +19,30 @@ use nix::sys::signal::{Signal, kill};
 use nix::unistd::{Pid, close};
 use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
 use std::process::Stdio;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::process::{Child, ChildStdout, Command};
+use tokio::sync::broadcast;
 use tokio::task::JoinHandle;
 use tokio::time::{Instant, sleep, timeout};
 
 use crate::cli::HostOpts;
 
+/// Broadcast-channel capacity in chunks. Each chunk is up to 64 KiB from gst
+/// stdout, so 16 chunks ≈ 1 MiB ≈ ~2 s of buffered jitter at the default
+/// 4 Mbps bitrate. A viewer that falls behind by more than this gets Lagged
+/// and skips ahead — MPEG-TS resyncs at the next keyframe.
+const FANOUT_CAPACITY: usize = 16;
+
+/// Size of each chunk read from gst stdout.
+const READ_CHUNK: usize = 64 * 1024;
+
 pub struct CaptureHandle {
     port: u16,
     gst: Option<Child>,
+    reader: Option<JoinHandle<()>>,
     server: Option<JoinHandle<()>>,
 }
 
@@ -37,8 +52,8 @@ impl CaptureHandle {
     }
 
     /// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then
-    /// abort the HTTP server task. Call this before dropping; Drop only fires
-    /// the kill backstop.
+    /// abort the reader + accept-loop tasks. Call this before dropping; Drop
+    /// only fires the kill backstop.
     pub async fn shutdown(mut self) {
         if let Some(child) = self.gst.as_mut()
             && let Some(pid) = child.id()
@@ -49,6 +64,9 @@ impl CaptureHandle {
             let _ = timeout(Duration::from_millis(1000), child.wait()).await;
             let _ = child.start_kill();
         }
+        if let Some(task) = self.reader.take() {
+            task.abort();
+        }
         if let Some(task) = self.server.take() {
             task.abort();
         }
@@ -60,6 +78,9 @@ impl Drop for CaptureHandle {
         if let Some(child) = self.gst.as_mut() {
             let _ = child.start_kill();
         }
+        if let Some(task) = self.reader.as_ref() {
+            task.abort();
+        }
         if let Some(task) = self.server.as_ref() {
             task.abort();
         }
@@ -199,28 +220,68 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
         .take()
         .context("gst-launch-1.0 stdout pipe unavailable")?;
 
-    // 4. Spawn the HTTP server task. It owns the listener + gst stdout: it
-    //    accepts one client (the host's bridge socket via connect_to_capture),
-    //    drains the HTTP request, writes a fixed MPEG-TS response, then
-    //    copies gst stdout to the socket forever.
-    let server = tokio::spawn(serve_capture(listener, gst_stdout));
+    // 4. Set up the broadcast fanout. The reader task pumps gst stdout chunks
+    //    into the channel; the accept-loop task spawns one sender task per
+    //    accepted TCP connection, each draining a fresh broadcast::Receiver.
+    let (tx, _) = broadcast::channel::<Arc<Vec<u8>>>(FANOUT_CAPACITY);
+
+    let reader = tokio::spawn(pump_gst_to_broadcast(gst_stdout, tx.clone()));
+    let server = tokio::spawn(run_accept_loop(listener, tx));
 
     Ok(CaptureHandle {
         port,
         gst: Some(gst),
+        reader: Some(reader),
         server: Some(server),
     })
 }
 
-async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
-    let mut sock = match listener.accept().await {
-        Ok((s, _)) => s,
-        Err(e) => {
-            tracing::warn!("capture HTTP accept failed: {e}");
-            return;
+/// Reads gst's stdout in chunks and broadcasts each to all current subscribers.
+/// `broadcast::send` returns Err when there are no receivers; we ignore it and
+/// keep reading so gst doesn't backpressure waiting for a viewer.
+async fn pump_gst_to_broadcast(
+    mut gst_stdout: ChildStdout,
+    tx: broadcast::Sender<Arc<Vec<u8>>>,
+) {
+    let mut buf = vec![0u8; READ_CHUNK];
+    loop {
+        match gst_stdout.read(&mut buf).await {
+            Ok(0) => {
+                tracing::info!("gst stdout EOF — fanout reader exiting");
+                return;
+            }
+            Ok(n) => {
+                let chunk = Arc::new(buf[..n].to_vec());
+                let _ = tx.send(chunk);
+            }
+            Err(e) => {
+                tracing::warn!("gst stdout read error: {e}");
+                return;
+            }
         }
-    };
+    }
+}
 
+/// Accepts TCP connections on the local capture port forever. Each accepted
+/// connection becomes its own viewer-serving task with a private receiver.
+async fn run_accept_loop(listener: TcpListener, tx: broadcast::Sender<Arc<Vec<u8>>>) {
+    loop {
+        let sock = match listener.accept().await {
+            Ok((s, _)) => s,
+            Err(e) => {
+                tracing::warn!("capture HTTP accept failed: {e}");
+                return;
+            }
+        };
+        let rx = tx.subscribe();
+        tokio::spawn(serve_one_viewer(sock, rx));
+    }
+}
+
+/// Drains the HTTP request, writes a fixed 200 OK, then pumps broadcast
+/// chunks to the socket until the channel closes or the socket errors out.
+/// On Lagged (slow consumer), skip ahead — MPEG-TS recovers at next keyframe.
+async fn serve_one_viewer(mut sock: TcpStream, mut rx: broadcast::Receiver<Arc<Vec<u8>>>) {
     if !drain_http_request(&mut sock).await {
         return;
     }
@@ -234,7 +295,23 @@ async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
         return;
     }
 
-    let _ = tokio::io::copy(&mut gst_stdout, &mut sock).await;
+    loop {
+        match rx.recv().await {
+            Ok(chunk) => {
+                if sock.write_all(&chunk).await.is_err() {
+                    return;
+                }
+            }
+            Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                tracing::warn!(
+                    skipped,
+                    "viewer fanout lagged — MPEG-TS will resync at next keyframe"
+                );
+                continue;
+            }
+            Err(broadcast::error::RecvError::Closed) => return,
+        }
+    }
 }
 
 async fn drain_http_request(sock: &mut TcpStream) -> bool {

src/interactive.rs

@@ -4,6 +4,7 @@ use iroh_tickets::endpoint::EndpointTicket;
 use std::str::FromStr;
 
 use crate::cli::Cli;
+use crate::common::{bandwidth, config};
 use crate::{host, viewer};
 
 pub async fn run(cli: Cli) -> Result<()> {
@@ -20,7 +21,10 @@ pub async fn run(cli: Cli) -> Result<()> {
         .interact()?;
 
     match choice {
-        0 => host::run(cli.into_host_opts(true)).await,
+        0 => {
+            preflight_if_needed(&theme).await;
+            host::run(cli.into_host_opts(true)).await
+        }
         _ => {
             let ticket = prompt_ticket(&theme)?;
             viewer::run(ticket, cli.into_viewer_opts(true)).await
@@ -28,6 +32,138 @@ pub async fn run(cli: Cli) -> Result<()> {
     }
 }
 
+/// `pixelpass --reconfigure` entry point: unconditionally re-run the
+/// bandwidth pre-flight test, save the result, and return. Used to
+/// refresh a stale measurement (e.g. user moved house, changed ISP).
+pub async fn run_reconfigure() -> Result<()> {
+    eprintln!();
+    eprintln!("Re-running bandwidth pre-flight test…");
+    let mut cfg = config::load().unwrap_or_default();
+    run_bandwidth_test(&mut cfg).await;
+    Ok(())
+}
+
+/// First-run pre-flight gate. Called once, when the user picks "Host" in
+/// the interactive menu. Behavior by saved status:
+/// - Unmeasured (first ever launch): explain + offer Run / Skip
+/// - Failed (previous attempt errored): offer Retry / give-up-and-skip
+/// - Measured or Skipped: silent — never re-prompts
+async fn preflight_if_needed(theme: &ColorfulTheme) {
+    let mut cfg = config::load().unwrap_or_default();
+    match cfg.bandwidth.status {
+        config::BandwidthStatus::Measured | config::BandwidthStatus::Skipped => return,
+        config::BandwidthStatus::Unmeasured => {
+            eprintln!();
+            eprintln!("First-time setup");
+            eprintln!("────────────────");
+            eprintln!("PixelPass can measure your upload speed to recommend a safe");
+            eprintln!("default for how many viewers your connection can handle.");
+            eprintln!("The test takes about 5 seconds and uploads ~5 MB to");
+            eprintln!("Cloudflare's open speed-test endpoint.");
+            eprintln!();
+            eprintln!("If you skip, a conservative default (2 viewers) is used.");
+            eprintln!("You can run the test later with `pixelpass --reconfigure`.");
+            eprintln!();
+
+            let Ok(choice) = Select::with_theme(theme)
+                .with_prompt("What would you like to do?")
+                .items(&[
+                    "Run the bandwidth test (recommended)",
+                    "Skip — use the conservative default",
+                ])
+                .default(0)
+                .interact()
+            else {
+                return;
+            };
+
+            if choice == 1 {
+                cfg.bandwidth = config::BandwidthEntry {
+                    status: config::BandwidthStatus::Skipped,
+                    upstream_mbps: None,
+                    measured_at: None,
+                };
+                let _ = config::save(&cfg);
+                eprintln!("Pre-flight skipped.");
+                return;
+            }
+            run_bandwidth_test(&mut cfg).await;
+        }
+        config::BandwidthStatus::Failed => {
+            eprintln!();
+            let Ok(choice) = Select::with_theme(theme)
+                .with_prompt("Last bandwidth test failed. Try again?")
+                .items(&[
+                    "Yes — retry now",
+                    "No — use the conservative default",
+                ])
+                .default(0)
+                .interact()
+            else {
+                return;
+            };
+
+            if choice == 1 {
+                cfg.bandwidth = config::BandwidthEntry {
+                    status: config::BandwidthStatus::Skipped,
+                    upstream_mbps: None,
+                    measured_at: None,
+                };
+                let _ = config::save(&cfg);
+                eprintln!("OK — using the conservative default.");
+                return;
+            }
+            run_bandwidth_test(&mut cfg).await;
+        }
+    }
+}
+
+async fn run_bandwidth_test(cfg: &mut config::Config) {
+    eprintln!();
+    eprintln!("Measuring upstream…");
+
+    let result = tokio::task::spawn_blocking(bandwidth::measure_upstream_blocking).await;
+    let measurement = match result {
+        Ok(Ok(m)) => m,
+        Ok(Err(e)) => {
+            eprintln!("Test failed: {e:#}");
+            eprintln!("Marking as failed — you'll be asked again on next launch.");
+            cfg.bandwidth = config::BandwidthEntry {
+                status: config::BandwidthStatus::Failed,
+                upstream_mbps: None,
+                measured_at: None,
+            };
+            let _ = config::save(cfg);
+            return;
+        }
+        Err(join_err) => {
+            eprintln!("Test task panicked: {join_err}");
+            cfg.bandwidth = config::BandwidthEntry {
+                status: config::BandwidthStatus::Failed,
+                upstream_mbps: None,
+                measured_at: None,
+            };
+            let _ = config::save(cfg);
+            return;
+        }
+    };
+
+    eprintln!(
+        "Measured {:.2} Mbps up (safe estimate {:.2} Mbps, took {:.1}s).",
+        measurement.raw_mbps,
+        measurement.safe_mbps,
+        measurement.elapsed.as_secs_f64()
+    );
+    cfg.bandwidth = config::BandwidthEntry {
+        status: config::BandwidthStatus::Measured,
+        upstream_mbps: Some(measurement.safe_mbps),
+        measured_at: Some(chrono::Utc::now()),
+    };
+    if let Err(e) = config::save(cfg) {
+        eprintln!("Warning: failed to save result: {e:#}");
+    }
+}
+
 fn print_welcome() {
     eprintln!();
     eprintln!("Welcome to PixelPass.");

src/main.rs

@@ -20,6 +20,10 @@ async fn main() -> Result<()> {
         return repair::run().await;
     }
 
+    if cli.reconfigure {
+        return interactive::run_reconfigure().await;
+    }
+
     match cli.ticket.as_deref() {
         Some(s) => {
             let ticket: EndpointTicket = s.parse().map_err(|e| {