fix: add backpressure and cancellation to the topic writer pipeline

[sshaplygin]

Problem

The topic writer pipeline had no backpressure anywhere along its length:

• Queue::add_message accepted messages unconditionally.
• The batch channel feeding the compression worker was mpsc::unbounded_channel.
• The channel carrying compressed requests back out was unbounded too.

A producer writing faster than the network drains grows all three without limit, so the failure mode is memory exhaustion rather than a slowed-down caller.

Separately, the compression worker's two loops only exited when their channels closed. Nothing observed the writer's CancellationToken, so a worker parked on queue.submit or on tx.send could outlive a cancelled writer.

Change

Bound the queue. Queue gains a max_buffered_messages cap (16 000 by default, counting queued plus in-flight-to-server messages via a new MessageQueue::len). At capacity, add_message waits on a message_removed_or_closed notifier instead of accepting. Acknowledgements release capacity; close_for_new_messages and notify_reception_tickets wake every waiter via notify_waiters so nobody is stranded when the queue shuts down. The wait re-checks is_open_for_new_messages, so a caller blocked on a full queue that then closes gets an error rather than a hang.

Bound the channels. Both the batch and compressed-request channels become mpsc::channel, sized available_parallelism() * OUTPUT_BACKLOG_PER_TASK. Because a bounded send can now block, write_messages_loop selects it against the cancellation token.

Make compression cancellable. Every await in both worker loops — receive, codec_selector.step, queue.submit, and the output send — is selected against the cancellation token, so cancelling the writer stops the worker instead of leaving it parked.

Tests

• add_message_waits_for_capacity_until_an_ack_releases_a_slot fills a queue of capacity 1, asserts a second add_message does not complete, then acknowledges the first and asserts the second is admitted.
• cancellation_stops_scheduler_waiting_for_a_worker_slot uses a HoldingExecutor that never runs the work it is handed, so the scheduler blocks on the single worker slot; cancelling must still unwind the task within 50 ms.

Note for reviewers

16 000 is a default with no configuration knob attached. If it should be reachable through WriterOptions, or aligned with an existing limit, that is a straightforward follow-up. Related: #565 (TopicWriter write MaxInFlight) and #563 (cancellation safety) look adjacent to this work.

Verification

cargo fmt --check
cargo clippy --workspace --all-targets --no-deps --exclude=ydb-grpc -- -D warnings
cargo test --workspace     # 216 passed, 88 ignored

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 98.80240% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 86.94%. Comparing base (a6d7911) to head (5498db7).

Files with missing lines	Patch %	Lines
...src/client_topic/compression/compression_worker.rs	98.86%	1 Missing ⚠️
ydb/src/client_topic/topicwriter/queue.rs	98.24%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #591      +/-   ##
==========================================
+ Coverage   86.91%   86.94%   +0.02%     
==========================================
  Files         198      198              
  Lines       19492    19635     +143     
==========================================
+ Hits        16941    17071     +130     
- Misses       2551     2564      +13     

Flag	Coverage Δ
rust-1.88.0	86.94% <98.80%> (+0.02%)	⬆️
rust-1.96.1	87.24% <99.39%> (+0.08%)	⬆️
tests	86.94% <98.80%> (+0.02%)	⬆️
ubuntu	86.94% <98.80%> (+0.02%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.