Ingress dispatcher is concurrent and integrated with bifrost

- IngressDispatcher is not a coroutine anymore
- Handles inputs concurrently
- Writes to bifrost and receives responses from network

Cargo.toml

@@ -87,6 +87,7 @@ bytes-utils = "0.1.3"
 bytestring = { version = "1.2", features = ["serde"] }
 chrono = { version = "0.4.31", default-features = false, features = ["clock"] }
 criterion = "0.5"
+dashmap = { version = "5.5.3" }
 datafusion = { version = "35.0.0" }
 datafusion-expr = { version = "35.0.0" }
 derive_builder = "0.12.0"

crates/bifrost/src/loglets/memory_loglet.rs

@@ -18,7 +18,7 @@ use async_trait::async_trait;
 use restate_types::logs::{Payload, SequenceNumber};
 use tokio::sync::oneshot::{Receiver, Sender};
 use tokio::sync::Mutex as AsyncMutex;
-use tracing::info;
+use tracing::{debug, info};
 
 use crate::loglet::{Loglet, LogletBase, LogletOffset, LogletProvider};
 use crate::metadata::LogletParams;
@@ -208,7 +208,7 @@ impl LogletBase for MemoryLoglet {
     async fn append(&self, payload: Payload) -> Result<LogletOffset, Error> {
         let mut log = self.log.lock().unwrap();
         let offset = self.index_to_offset(log.len());
-        info!(
+        debug!(
             "Appending record to in-memory loglet {:?} at offset {}",
             self.params, offset,
         );

crates/core/src/lib.rs

@@ -22,4 +22,4 @@ pub use task_center_types::*;
 mod test_env;
 
 #[cfg(any(test, feature = "test-util"))]
-pub use test_env::{create_mock_nodes_config, TestCoreEnv, TestCoreEnvBuilder};
+pub use test_env::{create_mock_nodes_config, MockNetworkSender, TestCoreEnv, TestCoreEnvBuilder};

crates/core/src/task_center.rs

@@ -319,7 +319,7 @@ impl TaskCenter {
         future: F,
     ) -> O
     where
-        F: Future<Output = O> + Send,
+        F: Future<Output = O>,
     {
         let cancel_token = CancellationToken::new();
         let id = TaskId::from(NEXT_TASK_ID.fetch_add(1, Ordering::SeqCst));

crates/core/src/task_center_types.rs

@@ -62,6 +62,10 @@ pub enum TaskKind {
     #[strum(props(OnCancel = "abort"))]
     MetadataBackgroundSync,
     RpcServer,
+    /// A type for ingress until we start enforcing timeouts for inflight requests. This enables us
+    /// to shutdown cleanly without waiting indefinitely.
+    #[strum(props(OnCancel = "abort"))]
+    IngressServer,
     RoleRunner,
     SystemService,
     Ingress,

crates/ingress-dispatcher/Cargo.toml

@@ -14,17 +14,17 @@ mocks = ["dep:restate-test-util"]
 [dependencies]
 restate-bifrost = { workspace = true }
 restate-core = { workspace = true }
-restate-futures-util = { workspace = true }
+restate-node-protocol = { workspace = true }
 restate-pb = { workspace = true, features = ["restate-types"] }
 restate-schema-api = { workspace = true, features = ["subscription"] }
 restate-test-util = { workspace = true, optional = true }
 restate-types = { workspace = true }
 restate-wal-protocol = { workspace = true }
 
 anyhow = { workspace = true }
-assert2 = { workspace = true }
 bytes = { workspace = true }
 bytestring = { workspace = true }
+dashmap = { workspace = true }
 drain = { workspace = true }
 prost = { workspace = true }
 thiserror = { workspace = true }