[Rust] Add persistent gRPC transport by elenagaljak-db · Pull Request #762 · databricks/zerobus-sdk

elenagaljak-db · 2026-08-20T08:48:28Z

What changes are proposed in this pull request?

This PR implements the internal gRPC transport and recovery machinery for persistent streams. It deliberately does not expose a new public SDK type; that surface is added in #763.

The sender, receiver, connection, and supervisor tasks are shared between ephemeral and persistent streams through a typed transport seam. Persistent streams use the dedicated RPC while ephemeral behavior remains unchanged.

Persistent behavior includes:

Create once, then resume the same stream_id after recovery.
Stable logical offsets on the persistent wire protocol.
New ingestion beginning at last_committed_offset + 1 after process-level resume.
Landing-zone reconciliation for the lost-ack race where the server committed records before the client disconnected.
Local completion of reconciled waiters and callbacks, with only offsets above the watermark resent.
Existing backpressure, flush, close, callback, credential-refresh, and initial-retry behavior shared across transports.

The implementation remains behind eos. Its internal transport/recovery addition is recorded in rust/NEXT_CHANGELOG.md.

Stack

[Rust] Add persistent stream proto #761 — protobuf contract
[Rust] Add persistent gRPC transport #762 — gRPC transport and recovery engine
[Rust] Expose persistent stream API #763 — public Rust API
[Rust] Test persistent stream resume #764 — stateful mock and integration tests
[Rust] Document persistent streams #765 — documentation and runnable example

How is this tested?

cargo check -p databricks-zerobus-ingest-sdk --features eos
cargo test -p databricks-zerobus-ingest-sdk --features eos --lib — 156 tests passed
End-to-end persistent coverage is added in [Rust] Test persistent stream resume #764.

teodordelibasic-db · 2026-08-25T09:53:48Z

 ]
 # Zero-copy protobuf parser.
 zeroparser = ["dep:self_cell", "dep:prost-build"]
+# Persistent (Eos) streams; in development, API is unstable


nit: (exactly-once semantics / EoS) just to be clearer maybe.

teodordelibasic-db · 2026-08-25T09:54:37Z

+    ///
+    /// Used by the persistent-stream resume path to reconcile the retained tail
+    /// against the server's committed watermark: it removes the prefix of
+    /// records the server has already durably stored (offset ≤ resume watermark)


nit: Let's use <= maybe. 😄

teodordelibasic-db · 2026-08-25T10:14:17Z

+            // `make_outbound`, so a mismatch is unreachable. Only present when
+            // more than one variant exists (i.e. `eos` is enabled).
+            #[cfg(feature = "eos")]
+            _ => Err(Self::open_failed()),


Hm can the OutboundSink and TransportKind enums be merged somehow so we don't have to do this double matching? I don't have a specific idea in mind.

teodordelibasic-db · 2026-08-25T10:18:18Z

+        }
+    }
+
+    fn send_failed() -> ZerobusError {


nit: The naming matrix of these two methods and their helper error methods are a bit mixed up I'd say.

send_open - open_failed.
send_ingest - send_failed, I'd say the second one here should be ingest_failed. I get the sentiment behind send_failed, we are sending a batch, but WDYT?

teodordelibasic-db · 2026-08-25T10:19:22Z

+    }
+
+    fn send_failed() -> ZerobusError {
+        ZerobusError::StreamClosedError(tonic::Status::internal("Failed to send record"))


nit: Probably pre-existing, but "Failed to send batch" is more suitable.

teodordelibasic-db · 2026-08-25T10:19:50Z

+            }
+            #[cfg(feature = "eos")]
+            OutboundSink::Persistent(tx) => {
+                let payload = ingest_payload_to_persistent(batch.into_request_payload(offset_id));


Can we have a batch.into_persistent_request_payload or something like that?

teodordelibasic-db · 2026-08-25T10:22:06Z

+
+/// The inbound half of a stream: wraps the concrete tonic response stream and
+/// yields normalized `InboundMessage`s.
+pub(super) enum InboundStream {


Similar question as https://github.com/databricks/zerobus-sdk/pull/762/changes#r3851952011 if it makes sense, although here the sentiment is not as strong since we have no double matching. In general looks good regarding division of responsibilities, but I feel a bit we have two many enums where one variant is Ephemeral and the other is Persistent.

teodordelibasic-db · 2026-08-25T10:37:12Z

            let resent_records = landing_zone_recovery.observed_count();
            let resent_batches = landing_zone_recovery.reset_observe();
+
+            // Resume alignment (persistent streams). The server reports how far


We should remove the Decision 6 part and in general comment can probably be shorter.

teodordelibasic-db · 2026-08-25T10:39:03Z

+pub(super) struct StreamConnection {
+    pub(super) sink: OutboundSink,
+    pub(super) inbound: InboundStream,
+    pub(super) stream_id: String,


These two last fields are kind of duplicated with StreamInitInfo.

teodordelibasic-db · 2026-08-25T11:40:08Z

+                }
+            }
+            #[cfg(feature = "eos")]
+            InboundStream::Persistent(s) => {


Mismatched setup responses currently succeed. Both CreateStreamResponse and ResumeStreamResponse are accepted regardless of whether the client requested a create or resume. The fallback in connection.rs:169 then lets a resume adopt the ID from an unexpected create response, while a create can accept an unexpected resume response without receiving a stream ID.

Could we preserve the response variant and validate it against the requested operation, for example:

enum Opened { Created { stream_id: String }, Resumed { last_committed_offset: Option<i64> }, }

The create path should only accept Opened::Created, and the resume path should only accept Opened::Resumed.

teodordelibasic-db · 2026-08-25T11:44:23Z

            // the one-shot limit; reconnect keeps its existing timeout-wrapped path unchanged.
            let is_initial = initial_stream_creation;
            let attempt = AtomicUsize::new(0);
            let create_attempt = || {


Maybe we should also note the idempotency hole here when create persistent stream fails but client gets no response from server?

teodordelibasic-db · 2026-08-25T11:45:54Z

+                        };
+                        let mut last_logical_acked_offset = -2;
+                        let mut map = oneshot_map.lock().await;
+                        for _offset_to_ack in (last_acked_offset + 1)..=durability_ack_up_to_offset


Let's reject an ACK lower than last_acked_offset or higher than the greatest sent wire offset before removing anything from landing_zone and close the stream on either protocol violation. We added such checks recently to the Arrow Flight SDK. It's an edge case for a malformed server, but worth just filling all the gaps.

teodordelibasic-db · 2026-08-25T11:59:43Z

+        // Safe because the constructor returns before any user ingest, so no real
+        // ack can race this initial value.
+        if let Some(watermark) = init_info.last_committed_offset {
+            logical_offset_id_generator.set_next(watermark + 1);


Let's use checked arithmetic here. A peer can return i64::MAX for this int64 field, at which point watermark + 1 panics when overflow checks are enabled and can wrap to a negative offset otherwise.

let next_offset = watermark.checked_add(1).ok_or_else(|| { ZerobusError::UnexpectedStreamResponseError( "Persistent stream offset space is exhausted".to_string(), ) })?; logical_offset_id_generator.set_next(next_offset);

Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>

This was referenced Aug 20, 2026

[Rust] Add persistent stream proto #761

Merged

[Rust] Test persistent stream resume #764

Open

[Rust] Expose persistent stream API #763

Open

elenagaljak-db force-pushed the stack/eos-grpc-core branch from 98b4072 to 9dcabe6 Compare August 20, 2026 09:27

elenagaljak-db force-pushed the stack/eos-proto branch from 4ff9401 to fd8c70d Compare August 20, 2026 09:27

elenagaljak-db mentioned this pull request Aug 20, 2026

[Rust] Document persistent streams #765

Open

elenagaljak-db force-pushed the stack/eos-grpc-core branch from 9dcabe6 to 18a643f Compare August 20, 2026 10:47

elenagaljak-db force-pushed the stack/eos-proto branch 2 times, most recently from 1ffcd2e to 82a81bf Compare August 20, 2026 13:09

elenagaljak-db force-pushed the stack/eos-grpc-core branch from 18a643f to abf5a92 Compare August 20, 2026 13:09

Base automatically changed from stack/eos-proto to main August 24, 2026 13:43

elenagaljak-db requested a review from a team August 24, 2026 14:57

elenagaljak-db force-pushed the stack/eos-grpc-core branch from abf5a92 to 12c4957 Compare August 24, 2026 16:40

teodordelibasic-db reviewed Aug 25, 2026

View reviewed changes

elenagaljak-db added 3 commits August 25, 2026 13:07

[Rust] Add persistent gRPC transport

06127af

Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>

[Rust] Address transport review

d04da07

Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>

[Rust] Validate persistent protocol

137f398

Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>

elenagaljak-db force-pushed the stack/eos-grpc-core branch from a574b4e to 137f398 Compare August 25, 2026 13:48

elenagaljak-db added 2 commits August 25, 2026 16:51

[Rust] Clarify persistent recovery

848e0ff

Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>

[Rust] Simplify persistent recovery

a02b336

Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>

Navigation Menu

Uh oh!

[Rust] Add persistent gRPC transport - #762

[Rust] Add persistent gRPC transport#762
elenagaljak-db wants to merge 5 commits into
mainfrom
stack/eos-grpc-core

elenagaljak-db commented Aug 20, 2026 •

edited

Loading

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

teodordelibasic-db Aug 25, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

elenagaljak-db commented Aug 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes are proposed in this pull request?

Stack

How is this tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

elenagaljak-db commented Aug 20, 2026 •

edited

Loading