Workspacify

server/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "server"
+edition = "2024"
+
+[dependencies]
+configuration = { workspace = true }
+error-stack = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }

server/src/lib.rs

@@ -0,0 +1,27 @@
+mod socket;
+mod state;
+
+use configuration::server::ServerConfiguration;
+use error_stack::ResultExt;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum TopLevelError {
+    #[error("Configuration error")]
+    ConfigurationError,
+    #[error("Failure running UNIX socket handling task")]
+    SocketHandlerFailure,
+}
+
+pub async fn run(mut config: ServerConfiguration) -> error_stack::Result<(), TopLevelError> {
+    let _server_state = state::ServerState::default();
+
+    let listeners = config
+        .generate_listener_sockets()
+        .await
+        .change_context(TopLevelError::ConfigurationError)?;
+
+    for (_name, _listener) in listeners.into_iter() {}
+
+    Ok(())
+}

server/src/socket.rs

@@ -0,0 +1,78 @@
+use crate::TopLevelError;
+use error_stack::ResultExt;
+use tokio::net::{UnixListener, UnixStream};
+use tracing::Instrument;
+
+pub struct SocketServer {
+    name: String,
+    path: String,
+    num_sessions_run: u64,
+    listener: UnixListener,
+}
+
+#[allow(unused)]
+impl SocketServer {
+    /// Create a new server that will handle inputs from the client program.
+    ///
+    /// This function will just generate the function required, without starting the
+    /// underlying task. To start the task, use [`SocketServer::start`], although that
+    /// method will take ownership of the object.
+    pub fn new(name: String, listener: UnixListener) -> Self {
+        let path = listener
+            .local_addr()
+            .map(|x| x.as_pathname().map(|p| format!("{}", p.display())))
+            .unwrap_or_else(|_| None)
+            .unwrap_or_else(|| "<unknown>".to_string());
+
+        tracing::trace!(%name, %path, "Creating new socket listener");
+
+        SocketServer {
+            name,
+            path,
+            num_sessions_run: 0,
+            listener,
+        }
+    }
+
+    /// Start running the service, returning a handle that will pass on an error if
+    /// one occurs in the core of this task.
+    ///
+    /// Typically, errors shouldn't happen in the core task, as all it does is listen
+    /// for new connections and then spawn other tasks based on them. If errors occur
+    /// there, the core task should be unaffected.
+    pub async fn start(mut self) -> error_stack::Result<(), TopLevelError> {
+        loop {
+            let (stream, addr) = self
+                .listener
+                .accept()
+                .await
+                .change_context(TopLevelError::SocketHandlerFailure)?;
+            let remote_addr = addr
+                .as_pathname()
+                .map(|x| x.display())
+                .map(|x| format!("{}", x))
+                .unwrap_or("<unknown>".to_string());
+
+            let span = tracing::debug_span!(
+                "unix socket handler",
+                socket_name = %self.name,
+                socket_path = %self.path,
+                session_no = %self.num_sessions_run,
+                %remote_addr,
+            );
+
+            self.num_sessions_run += 1;
+
+            tokio::task::spawn(Self::run_session(stream).instrument(span))
+                .await
+                .change_context(TopLevelError::SocketHandlerFailure)?;
+        }
+    }
+
+    /// Run a session.
+    ///
+    /// This is here because it's convenient, not because it shares state (obviously,
+    /// given the type signature). But it's somewhat logically associated with this type,
+    /// so it seems reasonable to make it an associated function.
+    async fn run_session(handle: UnixStream) {}
+}

server/src/state.rs

@@ -0,0 +1,31 @@
+use crate::TopLevelError;
+use tokio::task::JoinSet;
+
+#[derive(Default)]
+pub struct ServerState {
+    /// The set of top level tasks that are currently running.
+    top_level_tasks: JoinSet<Result<(), TopLevelError>>,
+}
+
+impl ServerState {
+    /// Block until all the current top level tasks have closed.
+    ///
+    /// This will log any errors found in these tasks to the current log,
+    /// if there is one, but will otherwise drop them.
+    #[allow(unused)]
+    pub async fn shutdown(&mut self) {
+        while let Some(next) = self.top_level_tasks.join_next_with_id().await {
+            match next {
+                Err(e) => tracing::error!(id = %e.id(), "Failed to attach to top-level task"),
+
+                Ok((id, Err(e))) => {
+                    tracing::error!(%id, "Top-level server error: {}", e);
+                }
+
+                Ok((id, Ok(()))) => {
+                    tracing::debug!(%id, "Cleanly closed server task.");
+                }
+            }
+        }
+    }
+}