Add real SRT relay integration tests, CI, and route persistence

- Integration tests in crates/srt-io/tests/relay.rs drive the full
  input -> crosspoint -> output path with real srt-tokio client sockets
  as the encoder/decoder, including a live re-route mid-stream over an
  already-established SRT connection. This is meaningfully stronger
  verification than the earlier lsof/curl-level checks.
- GitHub Actions CI: fmt --check, clippy -D warnings, test, on every
  push/PR to main.
- Optional routing persistence (Phase 3 roadmap item): an output's
  in-memory route is written to a JSON state file on change and
  reloaded on startup, overriding the config default_source. Disabled
  by default (opt in via [state] in the TOML config) so existing
  in-memory-only behavior is unchanged unless configured. Verified with
  a real process restart, not just unit tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: allan sargeant

.github/workflows/ci.yml

@@ -0,0 +1,26 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: rustfmt, clippy
+      - uses: Swatinem/rust-cache@v2
+      - name: fmt
+        run: cargo fmt --all -- --check
+      - name: clippy
+        run: cargo clippy --all-targets -- -D warnings
+      - name: test
+        run: cargo test --all

Cargo.lock

@@ -11,6 +11,13 @@
 # Local web UI + REST API for the crosspoint grid.
 bind = "0.0.0.0:8080"
 
+[state]
+# Optional. If present, routing changes made via the web UI/API are
+# persisted here and reloaded on startup (overriding each output's
+# default_source below). Remove this section to keep routing in-memory
+# only, reset to default_source on every restart.
+path = "state/routes.json"
+
 [[inputs]]
 id = "cam1"
 mode = "listener"

config/example.toml

@@ -14,6 +14,7 @@ srt-io = { path = "../srt-io" }
 crosspoint-web = { path = "../web" }
 tokio = { version = "1", features = ["full"] }
 serde = { version = "1", features = ["derive"] }
+serde_json = "1"
 toml = "0.8"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

crates/router/Cargo.toml

@@ -1,3 +1,5 @@
+use std::path::PathBuf;
+
 use serde::Deserialize;
 use srt_io::Endpoint;
 
@@ -8,6 +10,16 @@ pub struct Config {
     pub inputs: Vec<InputConfig>,
     #[serde(default)]
     pub outputs: Vec<OutputConfig>,
+    /// If present, routing changes are persisted to disk and reloaded on
+    /// startup (overriding each output's `default_source`). Omit to keep
+    /// routing in-memory only, reset to `default_source` on every restart.
+    #[serde(default)]
+    pub state: Option<StateConfig>,
+}
+
+#[derive(Deserialize)]
+pub struct StateConfig {
+    pub path: PathBuf,
 }
 
 #[derive(Deserialize)]

crates/router/src/config.rs

@@ -1,5 +1,7 @@
 mod config;
+mod state;
 
+use std::collections::HashMap;
 use std::net::SocketAddr;
 
 use anyhow::{Context, Result};
@@ -34,16 +36,40 @@ async fn main() -> Result<()> {
         tracing::info!(id = %input.id, "starting SRT input");
         srt_io::spawn_input(input.id, input.endpoint, crosspoint.clone());
     }
+
+    let persisted_routes: HashMap<String, String> = match &config.state {
+        Some(state_cfg) => {
+            let routes = state::load_routes(&state_cfg.path);
+            if !routes.is_empty() {
+                tracing::info!(
+                    path = %state_cfg.path.display(),
+                    count = routes.len(),
+                    "loaded persisted routing state"
+                );
+            }
+            routes
+        }
+        None => HashMap::new(),
+    };
+
     for output in config.outputs {
-        tracing::info!(id = %output.id, "starting SRT output");
+        let initial_source = persisted_routes
+            .get(&output.id)
+            .cloned()
+            .unwrap_or(output.default_source);
+        tracing::info!(id = %output.id, source = %initial_source, "starting SRT output");
         srt_io::spawn_output(
             output.id,
             output.endpoint,
-            output.default_source,
+            initial_source,
             crosspoint.clone(),
         );
     }
 
+    if let Some(state_cfg) = config.state {
+        state::spawn_persistence(state_cfg.path, crosspoint.clone());
+    }
+
     let bind: SocketAddr = config
         .web
         .bind

crates/router/src/main.rs

@@ -0,0 +1,79 @@
+//! Optional persistence of the crosspoint's routing table across restarts.
+//! Deliberately polling-based rather than a change-hook on
+//! `crosspoint-core`: it keeps the core engine free of any
+//! persistence-specific API, the same tradeoff the web UI already makes
+//! with its own 1s state poll.
+
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use crosspoint_core::Crosspoint;
+use tokio::time::sleep;
+
+const PERSIST_POLL_INTERVAL: Duration = Duration::from_millis(500);
+
+/// Load a previously persisted output -> source routing table. Any problem
+/// reading or parsing the file (missing, corrupt, unreadable) is logged and
+/// treated as "nothing persisted yet" rather than failing startup — a
+/// router should still come up on its config defaults if its state file is
+/// bad, not refuse to start.
+pub fn load_routes(path: &Path) -> HashMap<String, String> {
+    match std::fs::read_to_string(path) {
+        Ok(raw) => serde_json::from_str(&raw).unwrap_or_else(|err| {
+            tracing::warn!(path = %path.display(), %err, "state file exists but failed to parse, ignoring");
+            HashMap::new()
+        }),
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => HashMap::new(),
+        Err(err) => {
+            tracing::warn!(path = %path.display(), %err, "failed to read state file, ignoring");
+            HashMap::new()
+        }
+    }
+}
+
+fn save_routes(path: &Path, routes: &HashMap<String, String>) {
+    let body = match serde_json::to_string_pretty(routes) {
+        Ok(body) => body,
+        Err(err) => {
+            tracing::warn!(%err, "failed to serialize routing state");
+            return;
+        }
+    };
+    if let Some(parent) = path.parent() {
+        if !parent.as_os_str().is_empty() {
+            if let Err(err) = std::fs::create_dir_all(parent) {
+                tracing::warn!(parent = %parent.display(), %err, "failed to create state directory");
+                return;
+            }
+        }
+    }
+    // Write-then-rename so a crash mid-write can never leave a truncated
+    // state file behind for the next startup to trip over.
+    let tmp_path = path.with_extension("json.tmp");
+    if let Err(err) = std::fs::write(&tmp_path, body) {
+        tracing::warn!(path = %tmp_path.display(), %err, "failed to write state file");
+        return;
+    }
+    if let Err(err) = std::fs::rename(&tmp_path, path) {
+        tracing::warn!(path = %path.display(), %err, "failed to move state file into place");
+    }
+}
+
+/// Spawn a task that watches the crosspoint's routing table and persists it
+/// to `path` whenever it changes. Runs until the process exits.
+pub fn spawn_persistence(path: PathBuf, crosspoint: Arc<Crosspoint>) {
+    tokio::spawn(async move {
+        let mut last = crosspoint.routes();
+        loop {
+            sleep(PERSIST_POLL_INTERVAL).await;
+            let current = crosspoint.routes();
+            if current != last {
+                tracing::debug!(path = %path.display(), "routing changed, persisting");
+                save_routes(&path, &current);
+                last = current;
+            }
+        }
+    });
+}

crates/router/src/state.rs

@@ -0,0 +1,138 @@
+//! Integration tests that exercise the full input -> crosspoint -> output
+//! relay path over *real* SRT connections (real handshake, real protocol
+//! traffic via srt-tokio acting as the test's "encoder"/"decoder" clients)
+//! rather than just asserting a UDP socket is bound. This is the strongest
+//! local verification available without real third-party SRT hardware.
+
+use std::time::{Duration, Instant};
+
+use bytes::Bytes;
+use crosspoint_core::Crosspoint;
+use futures::SinkExt;
+use srt_io::{spawn_input, spawn_output, Endpoint};
+use srt_tokio::SrtSocket;
+use tokio::time::{sleep, timeout};
+use tokio_stream::StreamExt;
+
+const RECV_TIMEOUT: Duration = Duration::from_secs(5);
+/// Buffer for connection setup / task scheduling on localhost before we
+/// start relying on ordering between "peer connected" and "relay task
+/// subscribed to its source" — see the module doc on why this matters.
+const SETTLE: Duration = Duration::from_millis(400);
+
+async fn call(addr: &str) -> SrtSocket {
+    SrtSocket::builder()
+        .call(addr, None)
+        .await
+        .unwrap_or_else(|e| panic!("failed to connect to {addr}: {e}"))
+}
+
+#[tokio::test]
+async fn relays_bytes_end_to_end_over_real_srt() {
+    let crosspoint = Crosspoint::new();
+
+    spawn_input(
+        "test-source".into(),
+        Endpoint::Listener {
+            bind: "127.0.0.1:18601".into(),
+        },
+        crosspoint.clone(),
+    );
+    spawn_output(
+        "test-output".into(),
+        Endpoint::Listener {
+            bind: "127.0.0.1:18602".into(),
+        },
+        "test-source".into(),
+        crosspoint.clone(),
+    );
+    sleep(SETTLE).await;
+
+    // Real SRT clients standing in for an encoder and a decoder, exactly as
+    // a real one would connect to this router.
+    let mut encoder = call("127.0.0.1:18601").await;
+    let mut decoder = call("127.0.0.1:18602").await;
+    sleep(SETTLE).await;
+
+    let payload = Bytes::from_static(b"hello from a real srt sender");
+    encoder
+        .send((Instant::now(), payload.clone()))
+        .await
+        .expect("encoder send");
+
+    let (_t, received) = timeout(RECV_TIMEOUT, decoder.try_next())
+        .await
+        .expect("timed out waiting for the relayed payload")
+        .expect("decoder read error")
+        .expect("decoder stream ended before receiving anything");
+
+    assert_eq!(received, payload);
+}
+
+#[tokio::test]
+async fn output_switches_source_live_over_real_srt() {
+    let crosspoint = Crosspoint::new();
+
+    spawn_input(
+        "source-a".into(),
+        Endpoint::Listener {
+            bind: "127.0.0.1:18611".into(),
+        },
+        crosspoint.clone(),
+    );
+    spawn_input(
+        "source-b".into(),
+        Endpoint::Listener {
+            bind: "127.0.0.1:18612".into(),
+        },
+        crosspoint.clone(),
+    );
+    spawn_output(
+        "test-output".into(),
+        Endpoint::Listener {
+            bind: "127.0.0.1:18613".into(),
+        },
+        "source-a".into(),
+        crosspoint.clone(),
+    );
+    sleep(SETTLE).await;
+
+    let mut encoder_a = call("127.0.0.1:18611").await;
+    let mut encoder_b = call("127.0.0.1:18612").await;
+    let mut decoder = call("127.0.0.1:18613").await;
+    sleep(SETTLE).await;
+
+    // Routed to source-a by default: a payload from A arrives, one from B
+    // (sent but never selected) does not.
+    encoder_a
+        .send((Instant::now(), Bytes::from_static(b"from-a-1")))
+        .await
+        .unwrap();
+    encoder_b
+        .send((Instant::now(), Bytes::from_static(b"from-b-ignored")))
+        .await
+        .unwrap();
+
+    let (_t, first) = timeout(RECV_TIMEOUT, decoder.try_next())
+        .await
+        .expect("timed out waiting for first payload")
+        .expect("decoder read error")
+        .expect("decoder stream ended early");
+    assert_eq!(first, Bytes::from_static(b"from-a-1"));
+
+    // Live re-route to source-b, same output connection, no reconnect.
+    assert!(crosspoint.route("test-output", "source-b"));
+    sleep(SETTLE).await;
+
+    encoder_b
+        .send((Instant::now(), Bytes::from_static(b"from-b-1")))
+        .await
+        .unwrap();
+
+    let (_t, second) = timeout(RECV_TIMEOUT, decoder.try_next())
+        .await
+        .expect("timed out waiting for second payload")
+        .expect("decoder read error")
+        .expect("decoder stream ended early");
+    assert_eq!(second, Bytes::from_static(b"from-b-1"));
+}