core: add configuration for clustering

Cargo.toml

@@ -41,8 +41,9 @@ publish = false
 [dependencies]
 anyhow.workspace = true
 figment = { version = "0.10.8", features = ["env", "toml"] } # configuration files
-hyper = { version = "1.0.0-rc.3,<1.0.0-rc.4", features = ["server", "http1"] } # http server
+hyper = { version = "1.0.1", features = ["server", "http1"] } # http server
-http-body-util = "0.1.0-rc.3"
+http-body-util = "0.1.0"
+hyper-util = { version = "0.1", features = ["server", "http1", "tokio"] }
 serde.workspace = true
 serde_json = "1.0.93"
 tokio.workspace = true

config.toml

@@ -12,3 +12,9 @@ key = "./certs/xmpp.key"
 
 [storage]
 db_path = "db.sqlite"
+
+[clustering]
+self_id = 1
+listen_on = "0.0.0.0:23732"
+advertised_address = "127.0.0.1:23732"
+bootstrap_via = ["0.0.0.0:23733"]

crates/lavina-core/src/clustering/mod.rs

@@ -0,0 +1,34 @@
+use std::net::SocketAddr;
+
+use anyhow::Result;
+use serde::Deserialize;
+
+#[derive(Deserialize, Debug, Clone)]
+pub struct ClusteringConfig {
+    /// The unique identifier of the node in the cluster, needed to distinguish it form other nodes.
+    pub self_id: u32,
+    /// The local address where this node listens incoming for node-to-node communication.
+    pub listen_on: SocketAddr,
+    /// This address is advertised via a gossip mechanism and used by other nodes in the cluster to communicate with this node.
+    ///
+    /// This is useful when a node is behind a NAT or a proxy, where the publicly accessible IP
+    /// address can differ from the local IP address.
+    pub advertised_address: SocketAddr,
+    /// A list of network addresses with the node's initial set of peers in the cluster.
+    /// These nodes will be contacted first in order to discover additional nodes and to initialize the local state.
+    pub bootstrap_via: Box<[SocketAddr]>,
+}
+
+/// Cluster registry is a service which handles all connections to other nodes of the cluster.
+///
+/// It maintains the local view of the cluster state, including the set of nodes,
+/// the status of each node and the shard allocation metadata.
+struct ClusterRegistry {}
+
+impl ClusterRegistry {
+    /// Creates a new cluster registry and launches all necessary daemon fibers.
+    pub async fn launch(config: ClusteringConfig) -> Result<ClusterRegistry> {
+        todo!()
+    }
+}
+

crates/lavina-core/src/lib.rs

@@ -4,5 +4,6 @@ pub mod prelude;
 pub mod repo;
 pub mod room;
 pub mod terminator;
+pub mod clustering;
 
 mod table;

src/http.rs

@@ -7,6 +7,7 @@ use hyper::body::Bytes;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Method, Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
 use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
 use serde::{Deserialize, Serialize};
 use tokio::net::TcpListener;
@@ -52,6 +53,7 @@ async fn main_loop(
             _ = &mut termination => break,
             result = listener.accept() => {
                 let (stream, _) = result?;
+                let stream = TokioIo::new(stream);
                 let metrics = metrics.clone();
                 let rooms = rooms.clone();
                 let storage = storage.clone();

src/main.rs

@@ -20,6 +20,7 @@ struct ServerConfig {
     irc: projection_irc::ServerConfig,
     xmpp: projection_xmpp::ServerConfig,
     storage: lavina_core::repo::StorageConfig,
+    clustering: lavina_core::clustering::ClusteringConfig,
 }
 
 #[derive(Parser)]
@@ -48,6 +49,7 @@ async fn main() -> Result<()> {
         irc: irc_config,
         xmpp: xmpp_config,
         storage: storage_config,
+        clustering: clustering_config,
     } = config;
     let mut metrics = MetricsRegistry::new();
     let storage = Storage::open(storage_config).await?;