Skip to content

[Rust] Test persistent streams - #764

Merged
elenagaljak-db merged 4 commits into
mainfrom
stack/eos-tests
Oct 16, 2026
Merged

[Rust] Test persistent streams#764
elenagaljak-db merged 4 commits into
mainfrom
stack/eos-tests

Conversation

@elenagaljak-db

@elenagaljak-db elenagaljak-db commented Aug 20, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

What changes are proposed in this pull request?

This PR adds stateful mock-server support and integration coverage for persistent streams.

Test coverage

The mock keeps durable state across connections:

  • Create mints a unique stream_id.
  • Acknowledged offsets stay stored by stream_id after a disconnect.
  • Resume returns the last committed offset.
  • Unknown stream IDs return NOT_FOUND.
  • Ingestion uses the persistent response envelope.

persistent_stream_tests.rs has 10 tests. They cover create, ingest, flush and resume with continuous offsets; resume with no committed records; batches that use one offset; offsets that stay separate per stream_id; unknown stream IDs; a create response without a stream_id; an unexpected open response; lost-ack reconciliation from the resume watermark, with and without recovery; and a resume with a mismatched record type.

Review follow-ups from #763

  • [Rust] Name shared gRPC stream parts: prepare_grpc() returns a crate-private GrpcStreamParts struct instead of a four-element tuple, and ZerobusClient and Channel are imported at the top of the file.
  • [Rust] Move ephemeral builder entry points: build() moves to stream_builder/ephemeral.rs, and multiplexed() and multiplexed_ack_callback() move to multiplexed.rs. mod.rs keeps the builder state and the setters that PersistentStreamBuilder also uses. The Arrow Flight parts stay in mod.rs until the Arrow mux work in [Rust] Add Arrow mux API #866 lands.

Both commits are moves and renames only. They change no public API and no behavior. The builder has the same 59 methods with the same visibility before and after.
The test infrastructure is recorded in rust/NEXT_CHANGELOG.md.

Stack

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

How is this tested?

  • cargo test -p tests --test persistent_stream_tests — 2 tests passed
  • cargo test -p databricks-zerobus-ingest-sdk --features eos --lib — 156 tests passed
  • cargo fmt --all -- --check
  • git diff --check

This was referenced Aug 20, 2026
Merged
Merged
Merged
@elenagaljak-db elenagaljak-db mentioned this pull request Aug 20, 2026
Open
@elenagaljak-db
elenagaljak-db force-pushed the stack/eos-tests branch 2 times, most recently from 15972c6 to 8fb5685 Compare August 20, 2026 14:43
@elenagaljak-db
elenagaljak-db force-pushed the stack/eos-tests branch 2 times, most recently from f59fa66 to b1e0fec Compare August 26, 2026 15:56
@elenagaljak-db
elenagaljak-db force-pushed the stack/eos-public-api branch 2 times, most recently from 0ecfe35 to c202abf Compare August 28, 2026 16:41
@elenagaljak-db
elenagaljak-db force-pushed the stack/eos-tests branch 2 times, most recently from 1b7a882 to d34bcf2 Compare August 28, 2026 17:27
Base automatically changed from stack/eos-public-api to main October 5, 2026 08:15
@elenagaljak-db elenagaljak-db changed the title [Rust] Test persistent stream resume [Rust] Test persistent streams Oct 6, 2026
danilonajkov-db
danilonajkov-db previously approved these changes Oct 8, 2026
View reviewed changes
Comment thread rust/sdk/src/builder/stream_builder/ephemeral.rs Outdated
} = self.prepare_grpc().await?;
let stream =
ZerobusStream::new_stream(channel, table_properties, headers_provider, config).await?;
crate::client_warnings::record_stream_creation(stream.table_properties.table_name.as_str());

@danilonajkov-db danilonajkov-db Oct 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we import this at the top

@elenagaljak-db elenagaljak-db Oct 16, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yep, fixed

Comment thread rust/sdk/src/builder/stream_builder/mod.rs
//! shared setters. The ephemeral terminals live in their own files:
//! `ephemeral.rs` holds `build()`, and `multiplexed.rs` holds `multiplexed()`
//! and `multiplexed_ack_callback()`. Persistent streams do not support
//! multiplexing or Arrow Flight. The Arrow Flight setters and `build_arrow()`

@danilonajkov-db danilonajkov-db Oct 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Persistent streams do not support multiplexing or Arrow Flight.

is it a plan to add them?

@elenagaljak-db elenagaljak-db Oct 16, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, there is plan for arrow. And for multiplexing I'm not sure how it would work since client does want to have some control probably over which data it's sending to which stream, also client needs to be aware of stream ids. There is probably some way to go about designing this but lets leave it for a follow up, after Private Preview

elenagaljak-db and others added 4 commits October 16, 2026 10:49
@elenagaljak-db
[Rust] Test persistent stream resume
0e18f03 
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
@elenagaljak-db Isaac
[Rust] Name shared gRPC stream parts
590cd4d 
Replace the tuple that prepare_grpc returns with a crate-private
GrpcStreamParts struct, and import ZerobusClient and Channel instead of
using fully qualified paths. Addresses review nits on #763.

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
@elenagaljak-db Isaac
[Rust] Move ephemeral builder entry points
cc6ce21 
Move StreamBuilder::build into stream_builder/ephemeral.rs, and move
multiplexed() and multiplexed_ack_callback() into multiplexed.rs, so
mod.rs keeps the builder state and the setters that the persistent
builder also uses. No public API or behavior change. The Arrow Flight
setters and build_arrow() stay in mod.rs until the Arrow mux work in

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
@elenagaljak-db Isaac
[Rust] Import client warning helpers
ddb476d 
Import record_stream_creation and warn_payload_limit_ignored_for_arrow
at the top of the builder files instead of calling them through the
crate::client_warnings path. Addresses a review comment on #764.

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
@elenagaljak-db
elenagaljak-db added this pull request to the merge queue Oct 16, 2026
Merged via the queue into main with commit eb15995 Oct 16, 2026
50 of 51 checks passed
@elenagaljak-db
elenagaljak-db deleted the stack/eos-tests branch October 16, 2026 11:44
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@teodordelibasic-db teodordelibasic-db teodordelibasic-db approved these changes
@danilonajkov-db danilonajkov-db Awaiting requested review from danilonajkov-db

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@elenagaljak-db @teodordelibasic-db @danilonajkov-db