feat(net): add threaded tokio runtime bridge for connections

crates/net/src/lib.rs

@@ -12,6 +12,12 @@ pub mod error;
 pub mod handshake;
 pub mod runtime;
 
+pub use runtime::{ConnectOutcome, NetworkServer, ServerEvent, connect_in_background};
+
+/// Default UDP port the server binds and the client connects to when none is configured.
+// TODO: make the bind address and port configurable through server/client configuration.
+pub const DEFAULT_PORT: u16 = 25565;
+
 #[cfg(test)]
 #[path = "tests/handshake.rs"]
 mod handshake_tests;

crates/net/src/runtime.rs

@@ -1,5 +1,261 @@
 // SPDX-License-Identifier: AGPL-3.0-only
 
 //! Threaded `tokio` runtime bridge between the async network and the synchronous simulation.
-//!
-//! Owns the runtime thread and the channels that carry inbound and outbound messages across the async-sync boundary, keeping the async runtime confined to this crate. Populated in a later concept; currently a placeholder.
+
+use std::net::SocketAddr;
+use std::thread;
+
+use shared::protocol::{ClientHello, HandshakeAck};
+use tracing::{info, warn};
+
+use crate::endpoint::{client_endpoint, server_endpoint};
+use crate::error::NetError;
+use crate::handshake::{ServerConnection, accept_connection, connect};
+
+/// Channel receiver delivering the outcome of a background client connect: the negotiated [`HandshakeAck`] on success, or a human-readable error string on failure.
+pub type ConnectOutcome = crossbeam_channel::Receiver<Result<HandshakeAck, String>>;
+
+/// An event surfaced by the network thread to the synchronous server loop.
+#[derive(Debug)]
+pub enum ServerEvent {
+    /// A client completed the Synvael handshake. Carries the stable per-session id and the `ClientHello` it presented.
+    ClientConnected {
+        /// Stable identifier assigned to this session for the lifetime of the connection.
+        id: u64,
+        /// The identity and build parameters the client advertised.
+        hello: ClientHello,
+    },
+    /// A previously connected client's session ended.
+    ClientDisconnected {
+        /// Identifier of the session that ended, matching the earlier [`ServerEvent::ClientConnected`].
+        id: u64,
+        /// Human-readable description of why the connection closed.
+        reason: String,
+    },
+}
+
+/// Handle to the background networking thread and its owned `tokio` runtime.
+#[derive(Debug)]
+pub struct NetworkServer {
+    /// Events produced by the accept loop, drained by the synchronous simulation thread.
+    events: crossbeam_channel::Receiver<ServerEvent>,
+    /// Shutdown signal. Dropping this sender resolves the accept loop's receiver and breaks the loop.
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    /// Join handle for the network thread, awaited on drop for an orderly shutdown.
+    thread: Option<thread::JoinHandle<()>>,
+}
+
+impl NetworkServer {
+    /// Spawns a dedicated network thread, builds a current-thread `tokio` runtime on it, binds a QUIC server endpoint on `bind`, and runs the accept loop.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`NetError::Io`] if the runtime cannot be built, the network thread cannot be spawned, or the thread exits before reporting a bound address, and any error from [`server_endpoint`] if the endpoint cannot be constructed or bound.
+    pub fn spawn(
+        bind: SocketAddr,
+        server_build: String,
+        tick_rate_hint: u16,
+    ) -> Result<(Self, SocketAddr), NetError> {
+        let (events_tx, events_rx) = crossbeam_channel::unbounded();
+        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
+        // Reports the bound address (or the error that prevented binding) back to this thread, so `spawn` can surface it synchronously.
+        let (ready_tx, ready_rx) = crossbeam_channel::bounded::<Result<SocketAddr, NetError>>(1);
+
+        let thread = thread::Builder::new()
+            .name("net-server".to_owned())
+            .spawn(move || {
+                let runtime = match tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                {
+                    Ok(runtime) => runtime,
+                    Err(error) => {
+                        let _ = ready_tx.send(Err(NetError::Io(error)));
+                        return;
+                    }
+                };
+
+                runtime.block_on(async move {
+                    // The endpoint is constructed inside the runtime context because `quinn` binds its driver task onto the current runtime.
+                    let endpoint = match server_endpoint(bind) {
+                        Ok(endpoint) => endpoint,
+                        Err(error) => {
+                            let _ = ready_tx.send(Err(error));
+                            return;
+                        }
+                    };
+                    let local_addr = match endpoint.local_addr() {
+                        Ok(addr) => addr,
+                        Err(error) => {
+                            let _ = ready_tx.send(Err(NetError::Io(error)));
+                            return;
+                        }
+                    };
+                    // If the caller has already given up, there is nothing to serve.
+                    if ready_tx.send(Ok(local_addr)).is_err() {
+                        return;
+                    }
+
+                    accept_loop(
+                        endpoint,
+                        events_tx,
+                        shutdown_rx,
+                        server_build,
+                        tick_rate_hint,
+                    )
+                    .await;
+                });
+            })
+            .map_err(NetError::Io)?;
+
+        let local_addr = match ready_rx.recv() {
+            Ok(Ok(addr)) => addr,
+            Ok(Err(error)) => return Err(error),
+            Err(_) => {
+                return Err(NetError::Io(std::io::Error::other(
+                    "network thread exited before reporting a bound address",
+                )));
+            }
+        };
+
+        Ok((
+            Self {
+                events: events_rx,
+                shutdown: Some(shutdown_tx),
+                thread: Some(thread),
+            },
+            local_addr,
+        ))
+    }
+
+    /// Drains every [`ServerEvent`] currently queued, without blocking.
+    pub fn poll_events(&self) -> impl Iterator<Item = ServerEvent> + '_ {
+        self.events.try_iter()
+    }
+}
+
+impl Drop for NetworkServer {
+    fn drop(&mut self) {
+        // Dropping the sender resolves the accept loop's shutdown receiver, breaking the loop and letting `block_on` return so the thread unwinds and the runtime is dropped.
+        self.shutdown.take();
+        if let Some(thread) = self.thread.take() {
+            let _ = thread.join();
+        }
+    }
+}
+
+/// Accepts incoming QUIC connections until the endpoint stops yielding them or a shutdown is signalled, spawning one handshake task per connection.
+async fn accept_loop(
+    endpoint: quinn::Endpoint,
+    events: crossbeam_channel::Sender<ServerEvent>,
+    mut shutdown: tokio::sync::oneshot::Receiver<()>,
+    server_build: String,
+    tick_rate_hint: u16,
+) {
+    // Session ids are handed out sequentially; the accept loop is the sole assigner, so a plain counter suffices.
+    let mut next_id: u64 = 0;
+    loop {
+        tokio::select! {
+            incoming = endpoint.accept() => {
+                let Some(incoming) = incoming else { break };
+                let id = next_id;
+                next_id += 1;
+                tokio::spawn(handle_connection(
+                    incoming,
+                    id,
+                    events.clone(),
+                    server_build.clone(),
+                    tick_rate_hint,
+                ));
+            }
+            // Resolves when the `NetworkServer` handle is dropped (sender gone) or an explicit signal is sent.
+            _ = &mut shutdown => break,
+        }
+    }
+    info!("network accept loop shutting down");
+}
+
+/// Performs the handshake for one incoming connection and, on success, reports connect and disconnect events for its session.
+async fn handle_connection(
+    incoming: quinn::Incoming,
+    id: u64,
+    events: crossbeam_channel::Sender<ServerEvent>,
+    server_build: String,
+    tick_rate_hint: u16,
+) {
+    match accept_connection(incoming, server_build, tick_rate_hint).await {
+        Ok(ServerConnection {
+            connection, hello, ..
+        }) => {
+            // If the receiver is gone the server is shutting down; drop the connection silently.
+            if events
+                .send(ServerEvent::ClientConnected { id, hello })
+                .is_err()
+            {
+                return;
+            }
+            let reason = connection.closed().await;
+            let _ = events.send(ServerEvent::ClientDisconnected {
+                id,
+                reason: reason.to_string(),
+            });
+        }
+        Err(error) => {
+            warn!(%error, id, "connection handshake failed");
+        }
+    }
+}
+
+/// Runs a one-shot connect and Synvael handshake against `server_addr` on a background `tokio` thread, reporting the outcome to the returned receiver.
+#[must_use]
+pub fn connect_in_background(server_addr: SocketAddr, hello: ClientHello) -> ConnectOutcome {
+    let (outcome_tx, outcome_rx) = crossbeam_channel::bounded(1);
+    // Retained so a failure to spawn the thread can still be reported to the caller.
+    let spawn_err_tx = outcome_tx.clone();
+
+    let spawned = thread::Builder::new()
+        .name("net-client".to_owned())
+        .spawn(move || {
+            let runtime = match tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+            {
+                Ok(runtime) => runtime,
+                Err(error) => {
+                    let _ = outcome_tx.send(Err(format!("failed to build tokio runtime: {error}")));
+                    return;
+                }
+            };
+
+            runtime.block_on(async move {
+                let endpoint = match client_endpoint() {
+                    Ok(endpoint) => endpoint,
+                    Err(error) => {
+                        let _ = outcome_tx.send(Err(error.to_string()));
+                        return;
+                    }
+                };
+
+                // The server name matches the self-signed certificate's subject; the current verifier accepts any certificate regardless.
+                match connect(&endpoint, server_addr, "localhost", hello).await {
+                    Ok(connected) => {
+                        if outcome_tx.send(Ok(connected.ack.clone())).is_err() {
+                            return;
+                        }
+                        // Keep the connection alive on the network thread until the server closes it. A full client session pump is a later concept.
+                        let reason = connected.connection.closed().await;
+                        info!(%reason, "server connection closed");
+                    }
+                    Err(error) => {
+                        let _ = outcome_tx.send(Err(error.to_string()));
+                    }
+                }
+            });
+        });
+
+    if let Err(error) = spawned {
+        let _ = spawn_err_tx.send(Err(format!("failed to spawn network thread: {error}")));
+    }
+
+    outcome_rx
+}