feat(server): add deferred PollMessages wait timeout#3605
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #3605 +/- ##
============================================
- Coverage 74.10% 73.99% -0.11%
Complexity 937 937
============================================
Files 1258 1265 +7
Lines 131485 133155 +1670
Branches 107354 108836 +1482
============================================
+ Hits 97435 98528 +1093
- Misses 30963 31462 +499
- Partials 3087 3165 +78
🚀 New features to boost your workflow:
|
arunsingh
changed the title
arunsingh
force-pushed
the
feat/poll-messages-wait-timeout
branch
from
bbfb304 to
fbd4e0f
Compare
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
review priorities, most important first: (1) user-facing API ergonomics, (2) tests, (3) implementation in core/server-ng, (4) implementation in core/server - core/server is getting replaced by server-ng in about a month, so comments on it matter mostly as the blueprint for the server-ng port, not as long-lived code.
findings that don't map to changed lines:
- server-ng ignores the new
wait_timeout_usentirely:decode_poll_requestincore/server-ng/src/dispatch.rsbuildsPollingArgs::new(strategy, count, auto_commit)and drops the decoded timeout; there is no wait/waiter logic anywhere in that crate. a client callingpoll_messages_with_timeoutwith a non-zero timeout against server-ng gets an immediate empty response - no wait, no error, nothing detectable. since server-ng is about to become the server, shipping the wait path only in core/server means the feature effectively vanishes on the switch. minimum: server-ng should reject non-zero timeouts loudly; better: port the wait path as an immediate follow-up. cg_vsr.rsno longer exists on current master, so this branch needs a rebase and the new vsr test needs a new home; the modify/delete conflict is guaranteed.- edge on the append path (
core/server/src/shard/system/messages.rsaround 392-433, unchanged lines): the offset advances (data becomes readable) before the thresholdpersist_frozen_batches_to_disk().await?; if that persist errors, the request handler returns beforewake_poll_waiters, so a waiter sleeps out its full timeout despite readable data. self-heals on the next poll - worth a comment or waking before the persist block. - pre-existing, just amplified by long polls: auto-commit commits the offset before the response is sent, so a disconnect between commit and delivery skips that batch for the consumer (at-most-once). worth documenting on
auto_commitnow that polls can be parked for seconds.
also, please remove the markdown file scripts/poll-wait-timeout-comparison.md - it has way too strong LLM smell for my liking ;)
| .await | ||
| .unwrap(); | ||
|
|
||
| assert_eq!(polled.partition_id, 1); |
There was a problem hiding this comment.
this assert fails under --features vsr: cargo test -p integration --features vsr poll_messages_wait_timeout_consumer_group panics right here with left: 0, right: 1. the chain: the poll passes Some(0) with a group consumer; client-side vsr group resolution only kicks in when partition_id is None, so this goes out as an explicit partition-0 read; server-ng serves exactly that partition (no owned-partition scan, no wait) while the message sits on partition 1. CI never runs the integration suite with the vsr feature, so this stays green there and only fails locally. either cfg-gate run_consumer_group_checks out of vsr like the sibling checks, or implement the owned-partition scan in server-ng first.
separate but related: the non-vsr run of this scenario passes via the pre-wait re-check (data already present before the poll), so even the green path never exercises a real wait-then-wake for consumer groups - a produce-after-delay variant like the one in run_wake_after_append_checks would cover it.
| .await?; | ||
|
|
||
| shard.metrics.increment_messages(messages_count as u64); | ||
| shard.wake_poll_waiters(&namespace); |
There was a problem hiding this comment.
wake_poll_waiters grabs the global std::sync::Mutex around PollWaiterRegistry on every append (same at line 423), even when nobody waits. on the thread-per-core runtime that's a blocking mutex shared by all shard threads on the hottest path - contention or the cache-line bounce stalls the whole core, not just one task. cheap fix: an AtomicUsize live-waiter count checked before locking, skip when zero; the existing register-then-repoll sequence already closes the race, so no wake can be lost.
| .poll_messages( | ||
| client_id, | ||
| topic, | ||
| consumer.clone(), |
There was a problem hiding this comment.
consumer.clone() here runs on every poll, before the zero-timeout gate - Consumer carries an Identifier backed by Vec<u8>, so that's a heap alloc per poll even for plain immediate polls, plus one more clone per namespace in poll_wait_namespaces (line 165). IggyShard::poll_messages only borrows the consumer internally and both callers are in this file, so taking &Consumer removes all of them.
| let partition_id = req.partition_id; | ||
| let count = req.count; | ||
| let auto_commit = req.auto_commit; | ||
| let wait_timeout = Duration::from_micros(req.wait_timeout_us); |
There was a problem hiding this comment.
no server-side cap on the client-supplied timeout - u64::MAX micros (~584k years) is accepted and there's no config to bound it, so clients can park connections in the wait state indefinitely. related: once a namespace hits MAX_WAITERS_PER_NAMESPACE (1024), register returns None and the next poller silently degrades to an immediate empty response instead of waiting. a configurable max wait (clamped, not rejected) plus a debug log on the cap path would cover both.
| batch = next_batch; | ||
| } | ||
|
|
||
| if batch.is_empty() && !waiters.is_empty() { |
There was a problem hiding this comment.
one wake buys exactly one re-poll: if a group peer drains the partition between wake and re-poll, this returns empty with most of the timeout budget unused and the client goes back to busy-looping - the thing this feature exists to kill. also, matches!(..., Ok(true)) means a wake permit landing exactly at the deadline can lose to the timer and skip the final re-poll. looping wait+re-poll until the deadline fixes both.
| auto_commit: bool, | ||
| wait_timeout: Duration, | ||
| ) -> Result<PolledMessages, IggyError> { | ||
| if wait_timeout.is_zero() { |
There was a problem hiding this comment.
mutual delegation trap: this default delegates zero-timeout to self.poll_messages, while IggyClient and ClientWrapper implement poll_messages by calling poll_messages_with_timeout(Duration::ZERO). it only terminates because those impls also override poll_messages_with_timeout - remove one of those overrides later and it compiles fine, then stack-overflows at runtime. worth a doc line on the trait: implementors must override at least one of the pair with a real implementation.
| .await; | ||
| } | ||
|
|
||
| Err(IggyError::FeatureUnavailable) |
There was a problem hiding this comment.
non-zero timeout on a transport without an override surfaces as bare FeatureUnavailable ("Feature is unavailable") - nothing points at the wait timeout as the culprit. HTTP users hit this first; a doc note on the method (and ideally the transport in the error path) would save them the head-scratch.
|
|
||
| /// Poll wait timeout in human readable format, e.g. "10ms", "1s" | ||
| #[arg(long, default_value_t = IggyDuration::from_str(DEFAULT_POLL_WAIT_TIMEOUT).unwrap(), value_parser = IggyDuration::from_str)] | ||
| pub poll_wait_timeout: IggyDuration, |
There was a problem hiding this comment.
IggyDuration::from_str maps 0, unlimited, disabled and none all to zero - so --poll-wait-timeout unlimited reads like 'wait forever' but silently turns deferred polling off, and since the conflict guards check is_zero() it sails past those too. doc that 0s disables; rejecting the word forms for this flag would be even better.
| BUSY_LOOP_BATCHES=200 | ||
| fi | ||
|
|
||
| wait_timeouts=("0s" "10ms" "100ms" "1s") |
There was a problem hiding this comment.
this comparison structurally can't see the one always-on cost the PR adds: wake_poll_waiters runs on the produce path unconditionally, so it costs the same in the 0s and deferred arms and cancels out of every comparison here - and all suites run a single producer, so the cross-shard mutex contention never shows up either. measuring the append-path overhead needs a multi-producer/multi-stream run against a master-built server, not 0s vs deferred on the patched binary.
| local command="$3" | ||
| local log_file | ||
| local resource_log | ||
| local sampler_pid="" |
There was a problem hiding this comment.
sampler_pid is local to run_suite, so the on_interrupt/EXIT traps can't reach it - ctrl-c mid-suite leaks the sampler as a forever pgrep+sleep loop. hoist it to a global and kill it in the trap.
2 participants
Summary
Closes #3470.
Adds deferred polling to
PollMessageswith a timeout. Timeout0preserves existing immediate behavior. Non-zero timeout waits until messages are readable or the timeout expires.Design
u64timeout in microsecondscore/serverimplementation is kept small for upcomingserver-ngportOut of scope
Tests
Local checks
Commands run:
Full
cargo testis not claimed as passing locally. Remaining isolated unrelated failures:cli::system::test_snapshot_cmd::should_be_successfulserver::scenarios::log_rotation_scenario::log_rotation_should_be_valid::config_unlimited_archives_expectsAI usage
Used AI assistance for design planning and review and local verification were reviewed by the author.