feat: add request response behaviour for stack leader communication

.vscode/settings.json

@@ -1,22 +1,23 @@
 {
-    "rust-analyzer.check.command": "clippy",
-    "rust-analyzer.check.extraArgs": [
-        "--",
-        "-D",
-        "warnings",
-        "-W",
-        "clippy::pedantic",
-        "-W",
-        "clippy::nursery",
-        "-W",
-        "clippy::style",
-        "-W",
-        "clippy::complexity",
-        "-W",
-        "clippy::perf",
-        "-W",
-        "clippy::suspicious",
-        "-W",
-        "clippy::correctness"
-    ]
+  "rust-analyzer.check.command": "clippy",
+  "rust-analyzer.check.extraArgs": [
+    "--",
+    "-D",
+    "warnings",
+    "-W",
+    "clippy::pedantic",
+    "-W",
+    "clippy::nursery",
+    "-W",
+    "clippy::style",
+    "-W",
+    "clippy::complexity",
+    "-W",
+    "clippy::perf",
+    "-W",
+    "clippy::suspicious",
+    "-W",
+    "clippy::correctness"
+  ],
+  "idf.pythonInstallPath": "/usr/local/bin/python3"
 }

Cargo.toml

@@ -19,6 +19,7 @@ license = "Apache-2.0"
 
 [workspace.dependencies]
 aes-gcm = "0.10.3"
+async-trait = "0.1.86"
 anyhow = "1.0.97"
 atoma-confidential = { path = "./atoma-confidential" }
 atoma-daemon = { path = "./atoma-daemon" }

atoma-bin/atoma_node.rs

@@ -196,7 +196,9 @@ async fn main() -> Result<()> {
     let p2p_node_service_handle = spawn_with_shutdown(
         async move {
             let p2p_node =
-                AtomaP2pNode::start(config.p2p, Arc::new(keystore), p2p_event_sender, false)?;
+                AtomaP2pNode::start(config.p2p, Arc::new(keystore), p2p_event_sender, false)
+                    .await
+                    .map_err(|e| anyhow::anyhow!("Failed to start P2P node: {}", e))?;
             let pinned_future = Box::pin(p2p_node.run(p2p_node_service_shutdown_receiver));
             pinned_future.await
         },

atoma-p2p-tester/src/main.rs

@@ -62,10 +62,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Create and start the P2P node
     let keystore: FileBasedKeystore =
         FileBasedKeystore::new(&PathBuf::from(&config.sui.sui_keystore_path()))
-            .context("Failed to create keystore")?;
+            .with_context(|| "Failed to create keystore")?;
 
     let node = AtomaP2pNode::start(config.p2p, Arc::new(keystore), atoma_p2p_sender, false)
-        .context("Failed to start P2P node")?;
+        .await
+        .with_context(|| "Failed to start P2P node")?;
 
     let atoma_p2p_node_handle =
         spawn_with_shutdown(node.run(shutdown_receiver.clone()), shutdown_sender.clone());

atoma-p2p/Cargo.toml

@@ -5,6 +5,8 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+anyhow = { workspace = true }
+async-trait = { workspace = true }
 blake3 = { workspace = true }
 bytes = { workspace = true }
 ciborium = { workspace = true }
@@ -20,12 +22,14 @@ libp2p = { workspace = true, features = [
 	"mdns",
 	"kad",
 	"macros",
+	"request-response",
 	"quic",
 	"tcp",
 	"yamux",
 	"noise",
 	"metrics",
 	"rsa",
+	"cbor",
 ] }
 fastcrypto = { workspace = true }
 flume = { workspace = true }
@@ -39,6 +43,7 @@ sui-keys = { workspace = true }
 sui-sdk = { workspace = true }
 serde_json = { workspace = true }
 sysinfo = { workspace = true }
+sqlx = { workspace = true, features = ["runtime-tokio", "postgres"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }

atoma-p2p/src/config.rs

@@ -47,6 +47,9 @@ pub struct AtomaP2pNodeConfig {
     /// of the form (`serving_engine`, `metrics_endpoint`)
     /// (e.g. `"meta-llama/Llama-3.2-3B-Instruct" => ("vllm", "http://chat-completions:8000/metrics")`)
     pub metrics_endpoints: HashMap<String, (String, String)>,
+
+    /// Database connection URL for `PostgreSQL`
+    pub database_url: String,
 }
 
 impl AtomaP2pNodeConfig {

atoma-p2p/src/errors.rs

@@ -62,4 +62,14 @@ pub enum AtomaP2pNodeError {
     PublishError(String),
     #[error("DNS resolver error: `{0}`")]
     DnsError(#[from] std::io::Error),
+    #[error("Failed to connect to database: `{0}`")]
+    DatabaseConnectionError(#[from] sqlx::Error),
+    #[error("External error: `{0}`")]
+    External(String),
+}
+
+impl From<anyhow::Error> for AtomaP2pNodeError {
+    fn from(err: anyhow::Error) -> Self {
+        Self::External(err.to_string())
+    }
 }