[Rust] Multiplexed stream ingest followups

[danilonajkov-db]

What changes are proposed in this pull request?

This fixes the MultiplexedStream poison/admission race and makes capacity waits bounded.

The new ingest flow is:

1. Reserve real landing-zone capacity on the selected sub-stream.
2. Enter the sub-stream's private ingest-ordering section.
3. Acquire the mux admission read lock and re-check mux state.
4. Re-check sub-stream state.
5. Allocate the offset and enqueue using the reserved capacity.

The poison path acquires the mux admission write lock before flushing sub-streams, so no record can be admitted between the final closed check and the poison flush.

ZerobusStream retains ownership of its internal ordering lock through enqueue_reserved_admitted; the mux supplies only the admission check. Capacity reservations are opaque and cannot expose the underlying semaphore implementation.

Capacity waits are bounded by the existing flush_timeout_ms option. If acknowledgments never free capacity, ingest_record / ingest_records return ZerobusError::ConnectionTimeout instead of waiting indefinitely. A capacity timeout does not poison an otherwise healthy mux.

How is this tested?

• Regression test for concurrent ingests waiting for capacity while a sub-stream failure poisons the mux.
• Regression test for a permanently full sub-stream returning ConnectionTimeout after the configured timeout.
• Full multiplexed stream integration suite: 23 tests passed.
• Rust SDK unit suite: 141 tests passed.
• Doc tests passed.
• cargo fmt --all --check
• cargo clippy -p databricks-zerobus-ingest-sdk --all-targets -- -D warnings

[danilonajkov-db]

@elenagaljak-db thanks for the review, i've decided to move the ack callback to another PR to not mix contexts here. I've pasted your comments regarding the ack callback on that PR and resolved the ones here

[teodordelibasic-db]

Not sure if we discussed this before.

Since we hold this lock across join_all(...flush()), an ingest already parked at admission.read().await in enqueue_reserved_admitted cannot acquire the read lock until the whole flush finishes, so it stays blocked for that window before it sees is_closed and errors, and because it also holds sync_mutex, other ingests on that sub-stream queue behind it.

The write lock maybe only needs to be held long enough to drain readers that are mid-enqueue, not for the whole flush. Could we take and immediately drop it as a barrier after setting is_closed, then run the flush and signal_shutdown() unlocked? Any reader that acquires the read lock afterward fails check_closed() anyway, so nothing new is admitted.

[danilonajkov-db]

make sense to not block ingests if stream is mid close/failure and return error. Fixed

[teodordelibasic-db]

Do you think it would make sense to have a separate knob for this timeout other than the flush one?

[danilonajkov-db]

I added a seperate timeout variable, but didn't make it configureable for the clients. My thinking is that in the future we want the mux stream automatically try the next stream if the first one is full, so it will become an implementation detail the customer doesn't need to worry about. WDYT?

[danilonajkov-db]

ill wait for #523 to be merged so I can fix the tests