feat: M3 seed — the lumen/1 native protocol: QUIC control plane + reference client (Phase 5)

The first end-to-end run of lumen's own protocol, past the GameStream compatibility layer. - lumen-core/src/quic.rs (behind the `quic` feature): the lumen/1 handshake — Hello/Welcome/ Start as length-prefixed LE binary on one QUIC bi-stream. Welcome carries the COMPLETE data-plane Config: mode, FEC scheme incl. GF(2^16) Leopard (inexpressible in GameStream), shard sizing, AES-GCM key + per-direction salt, data UDP port. Plus quinn endpoint helpers (self-signed server; accepts-any client — pinning lands with the trust model) and framed async IO. Round-trip unit-tested. - lumen-host m3-host: serves one lumen/1 session — QUIC handshake, then a NATIVE thread (no async on the frame path — design invariant) streams deterministic 64KB test frames through the hardened M1 Session over UdpTransport. - lumen-client-rs: from scaffold to working reference client — connects, negotiates, brings up the client Session over UDP, reassembles + FEC-recovers + byte-verifies every frame. VALIDATED END-TO-END on localhost: 300/300 frames verified, 0 mismatches, through QUIC-negotiated GF(2^16) FEC + AES-GCM over real UDP sockets. M4 (decode+present) builds on this exact client skeleton. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-09 23:33:40 +00:00
parent 1eeb35a723
commit de3123038f
10 changed files with 621 additions and 14 deletions
@@ -9,6 +9,9 @@ authors.workspace = true
 repository.workspace = true

 [dependencies]
-lumen-core = { path = "../lumen-core" }
+lumen-core = { path = "../lumen-core", features = ["quic"] }
+quinn = "0.11"
+tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "macros"] }
+anyhow = "1"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
@@ -1,9 +1,15 @@
-//! `lumen-client-rs` — the reference client (plan M4). Exists to exercise the `lumen/1`
-//! (P2) transport: `lumen_core` pulls reassembled, FEC-recovered access units; decode via
-//! VAAPI; present via wgpu/Vulkan aligned to client vsync (frame pacing, plan §7).
+//! `lumen-client-rs` — the reference client. M3 seed mode: speak `lumen/1` (QUIC control
+//! plane) to a lumen host, bring up the client `lumen_core::Session` over UDP, reassemble +
+//! FEC-recover the host's deterministic test frames, and verify them byte-exactly. (M4 adds
+//! VAAPI decode + wgpu present on this same skeleton.)
 //!
-//! Status: scaffold. The client side of `lumen_core` ([`lumen_core::Session::poll_frame`])
-//! is already complete and tested; this binary wires it to a real decoder + presenter.
+//! Usage: `lumen-client-rs [--connect HOST:PORT]` (default `127.0.0.1:9777`).
+
+use anyhow::{anyhow, Context, Result};
+use lumen_core::config::Role;
+use lumen_core::quic::{endpoint, io, Hello, Start, Welcome};
+use lumen_core::transport::UdpTransport;
+use lumen_core::{LumenError, Session};

 fn main() {
    tracing_subscriber::fmt()
@@ -11,12 +17,125 @@ fn main() {
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
        )
        .init();
+    let addr = std::env::args()
+        .skip_while(|a| a != "--connect")
+        .nth(1)
+        .unwrap_or_else(|| "127.0.0.1:9777".into());
+    if let Err(e) = run(&addr) {
+        tracing::error!("{e:#}");
+        std::process::exit(1);
+    }
+}
+
+fn run(addr: &str) -> Result<()> {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()?;
+    rt.block_on(session(addr))
+}
+
+async fn session(addr: &str) -> Result<()> {
+    let remote: std::net::SocketAddr = addr.parse().context("--connect host:port")?;
+    let ep = endpoint::client_insecure().map_err(|e| anyhow!("QUIC client endpoint: {e}"))?;
+    let conn = ep
+        .connect(remote, "lumen")
+        .context("connect")?
+        .await
+        .context("QUIC handshake")?;
+    tracing::info!(%remote, "lumen/1 connected");
+    let (mut send, mut recv) = conn.open_bi().await.context("open control stream")?;
+
+    io::write_msg(
+        &mut send,
+        &Hello {
+            abi_version: lumen_core::ABI_VERSION,
+        }
+        .encode(),
+    )
+    .await?;
+    let welcome = Welcome::decode(&io::read_msg(&mut recv).await?)
+        .map_err(|e| anyhow!("Welcome decode: {e:?}"))?;
+    tracing::info!(
+        mode = ?welcome.mode,
+        fec = ?welcome.fec,
+        encrypt = welcome.encrypt,
+        frames = welcome.frames,
+        "session offer"
+    );
+
+    // Reserve our data-plane port, then tell the host to start.
+    let probe = std::net::UdpSocket::bind("0.0.0.0:0")?;
+    let udp_port = probe.local_addr()?.port();
+    drop(probe);
+    io::write_msg(
+        &mut send,
+        &Start {
+            client_udp_port: udp_port,
+        }
+        .encode(),
+    )
+    .await?;
+
+    let host_udp = std::net::SocketAddr::new(remote.ip(), welcome.udp_port);
+    let cfg = welcome.session_config(Role::Client);
+    let expected = welcome.frames;
+
+    // Data plane on a blocking thread (native threads only on the frame path).
+    let verified = tokio::task::spawn_blocking(move || -> Result<u32> {
+        let transport =
+            UdpTransport::connect(&format!("0.0.0.0:{udp_port}"), &host_udp.to_string())
+                .context("bind data plane")?;
+        let mut session =
+            Session::new(cfg, Box::new(transport)).map_err(|e| anyhow!("client session: {e:?}"))?;
+        let mut ok = 0u32;
+        let mut mismatched = 0u32;
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+        let mut last_rx = std::time::Instant::now();
+        while ok + mismatched < expected && std::time::Instant::now() < deadline {
+            match session.poll_frame() {
+                Ok(frame) => {
+                    last_rx = std::time::Instant::now();
+                    let idx = u32::from_le_bytes(frame.data[0..4].try_into().unwrap());
+                    if frame.data == test_frame(idx, frame.data.len()) {
+                        ok += 1;
+                    } else {
+                        mismatched += 1;
+                        tracing::warn!(idx, "frame content mismatch");
+                    }
+                }
+                Err(LumenError::NoFrame) => {
+                    if last_rx.elapsed() > std::time::Duration::from_secs(5) {
+                        break; // stream went quiet
+                    }
+                    std::thread::sleep(std::time::Duration::from_micros(500));
+                }
+                Err(e) => return Err(anyhow!("poll_frame: {e:?}")),
+            }
+        }
+        tracing::info!(ok, mismatched, expected, "verification complete");
+        anyhow::ensure!(mismatched == 0, "{mismatched} corrupted frames");
+        anyhow::ensure!(ok == expected, "received {ok}/{expected} frames");
+        Ok(ok)
+    })
+    .await??;

    tracing::info!(
-        "lumen-client-rs scaffold (lumen_core ABI v{})",
-        lumen_core::ABI_VERSION
-    );
-    tracing::info!(
-        "intended flow: lumen_core::Session(client) over UDP → poll_frame → VAAPI decode → wgpu present"
+        verified,
+        "lumen/1 session PASSED — GF(2^16) FEC + AES-GCM over real UDP, QUIC-negotiated"
    );
+    conn.close(0u32.into(), b"done");
+    Ok(())
+}
+
+/// The host's deterministic test frame (mirror of `lumen-host::m3::test_frame`).
+fn test_frame(idx: u32, len: usize) -> Vec<u8> {
+    let mut d = vec![0u8; len];
+    if len >= 4 {
+        d[0..4].copy_from_slice(&idx.to_le_bytes());
+    }
+    for (i, b) in d.iter_mut().enumerate().skip(4) {
+        *b = (idx as u8).wrapping_add(i as u8);
+    }
+    d
 }