[Ingress] ingress-client crate

- `ingress-client` implements the runtime layer that receives ingress traffic, fans it out to the correct partition, and tracks completion. It exposes:
  - `Ingress`, enforces inflight budgets, and resolves partition IDs before sending work downstream.
  - The session subsystem that batches `IngestRecords`, retries connections, and reports commit status to callers.
- `ingress-core` only ingests records and notify the caller once the record is "committed" to bifrost by the PP. This makes it useful to implement kafka ingress and other external ingestion

Author: muhamadazmy

Cargo.lock

@@ -85,6 +85,7 @@ restate-types = { path = "crates/types" }
 restate-utoipa = { path = "crates/utoipa" }
 restate-wal-protocol = { path = "crates/wal-protocol" }
 restate-worker = { path = "crates/worker" }
+restate-ingress-client = { path = "crates/ingress-client" }
 
 # this workspace-hack package is overridden by a patch below to use workspace-hack subdir when building in this repo
 # outside this repo, the crates.io restate-workspace-hack (an empty package) will be used instead

Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "restate-ingress-client"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+publish = false
+
+[dependencies]
+arc-swap = { workspace = true }
+dashmap = { workspace = true }
+futures = { workspace = true }
+pin-project-lite = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+tokio-util = { workspace = true }
+tracing = { workspace = true }
+
+restate-workspace-hack = { workspace = true }
+restate-core = { workspace = true }
+restate-types = { workspace = true }
+
+[dev-dependencies]
+restate-core = { workspace = true, features = ["test-util"] }
+test-log = { workspace = true }
+bytes = { workspace = true }
+googletest = { workspace = true }

crates/ingress-client/Cargo.toml

@@ -0,0 +1,93 @@
+use tokio::time::{Sleep, sleep};
+use tokio_stream::adapters::Fuse;
+use tokio_stream::{Stream, StreamExt};
+
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{Context, Poll, ready};
+use pin_project_lite::pin_project;
+use std::time::Duration;
+
+// This file is a copy from `tokio_stream` until PR https://github.com/tokio-rs/tokio/pull/7715 is released
+
+pin_project! {
+    /// Stream returned by the [`chunks_timeout`](super::StreamExt::chunks_timeout) method.
+    #[must_use = "streams do nothing unless polled"]
+    #[derive(Debug)]
+    pub struct ChunksTimeout<S: Stream> {
+        #[pin]
+        stream: Fuse<S>,
+        #[pin]
+        deadline: Option<Sleep>,
+        duration: Duration,
+        items: Vec<S::Item>,
+        cap: usize, // https://github.com/rust-lang/futures-rs/issues/1475
+    }
+}
+
+impl<S: Stream> ChunksTimeout<S> {
+    pub fn new(stream: S, max_size: usize, duration: Duration) -> Self {
+        ChunksTimeout {
+            stream: stream.fuse(),
+            deadline: None,
+            duration,
+            items: Vec::with_capacity(max_size),
+            cap: max_size,
+        }
+    }
+    /// Drains the buffered items, returning them without waiting for the timeout or capacity limit.
+    pub fn into_remainder(mut self: Pin<&mut Self>) -> Vec<S::Item> {
+        let me = self.as_mut().project();
+        std::mem::take(me.items)
+    }
+}
+
+impl<S: Stream> Stream for ChunksTimeout<S> {
+    type Item = Vec<S::Item>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut me = self.as_mut().project();
+        loop {
+            match me.stream.as_mut().poll_next(cx) {
+                Poll::Pending => break,
+                Poll::Ready(Some(item)) => {
+                    if me.items.is_empty() {
+                        me.deadline.set(Some(sleep(*me.duration)));
+                        me.items.reserve_exact(*me.cap);
+                    }
+                    me.items.push(item);
+                    if me.items.len() >= *me.cap {
+                        return Poll::Ready(Some(std::mem::take(me.items)));
+                    }
+                }
+                Poll::Ready(None) => {
+                    // Returning Some here is only correct because we fuse the inner stream.
+                    let last = if me.items.is_empty() {
+                        None
+                    } else {
+                        Some(std::mem::take(me.items))
+                    };
+
+                    return Poll::Ready(last);
+                }
+            }
+        }
+
+        if !me.items.is_empty() {
+            if let Some(deadline) = me.deadline.as_pin_mut() {
+                ready!(deadline.poll(cx));
+            }
+            return Poll::Ready(Some(std::mem::take(me.items)));
+        }
+
+        Poll::Pending
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
+        let (lower, upper) = self.stream.size_hint();
+        let lower = (lower / self.cap).saturating_add(chunk_len);
+        let upper = upper.and_then(|x| x.checked_add(chunk_len));
+        (lower, upper)
+    }
+}