[Rust] Add persistent gRPC transport

rust/NEXT_CHANGELOG.md

@@ -47,6 +47,10 @@
 
 ### Internal Changes
 
+- Added the feature-gated persistent gRPC transport, durable wire offsets,
+  resume-watermark reconciliation after a lost acknowledgment, and validation
+  for setup responses, acknowledgment bounds, and offset overflow.
+
 ### Breaking Changes
 
 ### Deprecations

rust/sdk/Cargo.toml

@@ -77,6 +77,8 @@ internal-arrow-c-data = [
 ]
 # Zero-copy protobuf parser.
 zeroparser = ["dep:self_cell", "dep:prost-build"]
+# Persistent streams with exactly-once semantics (EoS); API is in development.
+eos = []
 testing = ["dep:futures"]
 # Test-only deterministic seams (barriers/notifies) for the Arrow stream. Zero footprint
 # unless enabled; never enabled by FFI/production builds.

rust/sdk/src/landing_zone.rs

@@ -85,6 +85,28 @@ impl<T: Clone> LandingZone<T> {
         all_items
     }
 
+    /// Removes a prefix of observed items while `should_remove` returns true.
+    ///
+    /// Observed items stay in FIFO order. Stops at the first item that should
+    /// remain. Releases the corresponding backpressure permits.
+    pub fn remove_observed_prefix(&self, mut should_remove: impl FnMut(&T) -> bool) -> Vec<T> {
+        let mut state = self.state.lock().expect("Lock poisoned");
+        let mut permits = self.permits.lock().expect("Lock poisoned");
+        let mut removed = Vec::new();
+        while let Some(front) = state.observed_items.front() {
+            if !should_remove(front) {
+                break;
+            }
+            let item = state
+                .observed_items
+                .pop_front()
+                .expect("front existed before pop");
+            permits.pop_front();
+            removed.push(item);
+        }
+        removed
+    }
+
     /// Adds an item to the queue.
     ///
     /// This method will block if the maximum number of inflight requests has been reached,
@@ -282,6 +304,34 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn test_remove_observed_prefix_preserves_fifo_suffix() {
+        let lz = LandingZone::new(4);
+        for value in 1..=4 {
+            lz.add(value).await;
+        }
+        for _ in 0..3 {
+            lz.observe().await;
+        }
+
+        assert_eq!(lz.remove_observed_prefix(|value| *value <= 2), vec![1, 2]);
+        assert_eq!(lz.reset_observe(), 1);
+        assert_eq!(lz.observe().await, 3);
+        assert_eq!(lz.observe().await, 4);
+    }
+
+    #[tokio::test]
+    async fn test_remove_observed_prefix_never_removes_unsent_items() {
+        let lz = LandingZone::new(2);
+        lz.add(1).await;
+        lz.add(2).await;
+        lz.observe().await;
+
+        assert_eq!(lz.remove_observed_prefix(|_| true), vec![1]);
+        assert_eq!(lz.len(), 1);
+        assert_eq!(lz.observe().await, 2);
+    }
+
     #[tokio::test]
     async fn test_remove_all() {
         let lz = Arc::new(LandingZone::new(10));

rust/sdk/src/record_types.rs

@@ -20,6 +20,8 @@ use crate::databricks::zerobus::{
 };
 use crate::OffsetId;
 
+use crate::databricks::zerobus::persistent_stream_request::Payload as PersistentRequestPayload;
+
 /// A type alias for a protobuf-encoded record.
 pub type ProtoEncodedRecord = Vec<u8>;
 
@@ -257,6 +259,21 @@ impl EncodedBatch {
         }
     }
 
+    pub(crate) fn into_persistent_request_payload(
+        self,
+        offset_id: OffsetId,
+    ) -> PersistentRequestPayload {
+        match self.into_request_payload(offset_id) {
+            RequestPayload::IngestRecord(record) => PersistentRequestPayload::IngestRecord(record),
+            RequestPayload::IngestRecordBatch(batch) => {
+                PersistentRequestPayload::IngestRecordBatch(batch)
+            }
+            RequestPayload::CreateStream(_) => {
+                unreachable!("encoded batches only produce ingest payloads")
+            }
+        }
+    }
+
     /// Returns the number of records in this batch.
     pub fn get_record_count(&self) -> usize {
         match self {
@@ -778,6 +795,20 @@ mod tests {
                 _ => panic!("Expected IngestRecordBatch payload"),
             }
         }
+
+        #[test]
+        fn test_into_persistent_request_payload() {
+            let record = r#"{"id": 1}"#.to_string();
+            let batch = EncodedBatch::Json(smallvec![record.clone()]);
+
+            match batch.into_persistent_request_payload(42) {
+                PersistentRequestPayload::IngestRecord(req) => {
+                    assert_eq!(req.offset_id, Some(42));
+                    assert_eq!(req.record, Some(IngestRequestRecord::JsonRecord(record)));
+                }
+                _ => panic!("Expected persistent IngestRecord payload"),
+            }
+        }
     }
 
     mod encoded_batch_iter {