[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.

I removed the (sink, mode) double match. OutboundConnection now keeps the matching sender and request stream together, and open_rpc constructs the corresponding sink.

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?

I renamed to ingest_failed

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.

Yep, I updated anyway

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?

Added EncodedBatch::into_persistent_request_payload, matching the ephemeral conversion helper.

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.

Yep, removed

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: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?

Added a comment documenting the remaining hole: if create succeeds server-side but its response is lost, retrying may create an orphan because no client-supplied ID exists yet. Resume remains idempotent.

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.

Added validation before touching the landing zone

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>

teodordelibasic-db

Shape looks good, let's just leave a couple of TODO comments for non-blocking hardening work for some edge cases.

teodordelibasic-db · 2026-08-28T13:07:04Z

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


nit: Period at the end.

teodordelibasic-db · 2026-08-28T13:12:22Z

        // Mutated only on the persistent recovery path (flip create → resume);
        // that mutation is `eos`-gated, so `mut` is otherwise unused.
-        #[cfg_attr(not(feature = "eos"), allow(unused_mut))] mut kind: GrpcConnectionMode,
+        mut kind: GrpcConnectionMode,


Why did we remove the feature gating for mutability, also is the comment above now stale?

teodordelibasic-db · 2026-08-28T13:36:11Z

+            // remaining observed records for resend. A first-process resume has
+            // an empty landing zone and only seeds the receiver watermark.
+            let mut initial_last_acked_offset: OffsetId = -1;
+            if let Some(watermark) = last_committed_offset {


Let's just leave a TODO immediately before this watermark handling so the missing validation is not lost? It should record that automatic resume must validate the normalized watermark against the previous ACK and highest successfully sent wire offset before reconciling, including None as the no-commit position -1.

// TODO: Validate the resumed watermark against retained ACK and sent-offset // bounds before reconciling; treat None as -1 and fail closed if out of range.

teodordelibasic-db · 2026-08-28T13:39:07Z

-                                        ));
-                                    let _ = server_error_tx.send(Some(error.clone()));
-                                    return Err(error);
+                            landing_zone.observed_count(),


Let's leave a TODO at the observed_count() argument noting that ACK bounds need the highest successfully sent wire offset rather than a landing-zone count. observed_count() counts records even though a batch has one wire offset, and counting observed batches alone is also insufficient because observe runs before the send succeeds.

// TODO: Bound ACKs by the highest successfully sent wire offset, not landing-zone // counts; batches use one offset and observation precedes a successful send.

teodordelibasic-db · 2026-08-28T13:39:34Z

+        // since the server never re-acks offsets it committed in a prior session.
+        // 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 {


Let's leave a TODO before this fallible post-initialization block noting that the supervisor and optional callback task must be cleaned up if watermark validation fails. Dropping their JoinHandles detaches them rather than cancelling them.

// TODO: Validate the resume watermark before spawning tasks, or cancel and abort // the spawned tasks on post-initialization errors so their handles are not detached.

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>

teodordelibasic-db approved these changes Aug 28, 2026

View reviewed changes

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!

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!

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!

teodordelibasic-db left a comment

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