From f063afac84748e88e811c04f9f78641b36a83665 Mon Sep 17 00:00:00 2001 From: Arun Singh Date: Fri, 21 Aug 2026 22:01:41 +0530 Subject: [PATCH 1/2] Add deferred poll timeout support Signed-off-by: Arun Singh --- .../src/requests/messages/poll_messages.rs | 104 +++++++++++++++--- .../src/traits/binary_impls/messages.rs | 62 ++++++++++- core/common/src/traits/message_client.rs | 33 ++++++ .../tests/server/poll_semantics_vsr.rs | 43 ++++++++ .../client_wrappers/binary_message_client.rs | 40 ++++++- core/sdk/src/clients/binary_message.rs | 28 ++++- core/server/src/dispatch.rs | 18 +++ core/server/src/http/wire.rs | 2 + core/simulator/src/client.rs | 1 + 9 files changed, 307 insertions(+), 24 deletions(-) diff --git a/core/binary_protocol/src/requests/messages/poll_messages.rs b/core/binary_protocol/src/requests/messages/poll_messages.rs index 4c6413cfc8..680a987263 100644 --- a/core/binary_protocol/src/requests/messages/poll_messages.rs +++ b/core/binary_protocol/src/requests/messages/poll_messages.rs @@ -17,7 +17,7 @@ use crate::WireError; use crate::WireIdentifier; -use crate::codec::{WireDecode, WireEncode, read_u8, read_u32_le}; +use crate::codec::{WireDecode, WireEncode, read_u8, read_u32_le, read_u64_le}; use crate::primitives::consumer::WireConsumer; use crate::primitives::polling_strategy::WirePollingStrategy; use bytes::{BufMut, BytesMut}; @@ -27,7 +27,7 @@ use bytes::{BufMut, BytesMut}; /// Wire format: /// ```text /// [consumer][stream_id][topic_id][partition_flag:1][partition_id:4 LE] -/// [strategy:9][count:4 LE][auto_commit:1] +/// [strategy:9][count:4 LE][auto_commit:1][wait_timeout_us:8 LE] /// ``` /// /// `partition_id` encoding: a u8 flag (1=Some, 0=None) followed by 4 bytes @@ -41,6 +41,7 @@ pub struct PollMessagesRequest { pub strategy: WirePollingStrategy, pub count: u32, pub auto_commit: bool, + pub wait_timeout_us: u64, } const PARTITION_FLAG_SIZE: usize = 1; @@ -48,6 +49,7 @@ const PARTITION_VALUE_SIZE: usize = 4; const STRATEGY_SIZE: usize = 9; const COUNT_SIZE: usize = 4; const AUTO_COMMIT_SIZE: usize = 1; +const WAIT_TIMEOUT_SIZE: usize = 8; impl WireEncode for PollMessagesRequest { fn encoded_size(&self) -> usize { @@ -59,6 +61,7 @@ impl WireEncode for PollMessagesRequest { + STRATEGY_SIZE + COUNT_SIZE + AUTO_COMMIT_SIZE + + WAIT_TIMEOUT_SIZE } fn encode(&self, buf: &mut BytesMut) { @@ -75,6 +78,7 @@ impl WireEncode for PollMessagesRequest { self.strategy.encode(buf); buf.put_u32_le(self.count); buf.put_u8(u8::from(self.auto_commit)); + buf.put_u64_le(self.wait_timeout_us); } } @@ -104,6 +108,13 @@ impl WireDecode for PollMessagesRequest { pos += 4; let auto_commit = read_u8(buf, pos)? != 0; pos += 1; + let wait_timeout_us = if buf.len() == pos { + 0 + } else { + let wait_timeout_us = read_u64_le(buf, pos)?; + pos += WAIT_TIMEOUT_SIZE; + wait_timeout_us + }; Ok(( Self { @@ -114,6 +125,7 @@ impl WireDecode for PollMessagesRequest { strategy, count, auto_commit, + wait_timeout_us, }, pos, )) @@ -124,9 +136,8 @@ impl WireDecode for PollMessagesRequest { mod tests { use super::*; - #[test] - fn roundtrip_with_partition() { - let req = PollMessagesRequest { + fn request(wait_timeout_us: u64) -> PollMessagesRequest { + PollMessagesRequest { consumer: WireConsumer::consumer(WireIdentifier::numeric(1)), stream_id: WireIdentifier::numeric(10), topic_id: WireIdentifier::numeric(20), @@ -134,7 +145,22 @@ mod tests { strategy: WirePollingStrategy::offset(100), count: 50, auto_commit: true, - }; + wait_timeout_us, + } + } + + #[test] + fn roundtrip_with_zero_wait_timeout() { + let req = request(0); + let bytes = req.to_bytes(); + let (decoded, consumed) = PollMessagesRequest::decode(&bytes).unwrap(); + assert_eq!(consumed, bytes.len()); + assert_eq!(decoded, req); + } + + #[test] + fn roundtrip_with_non_zero_wait_timeout() { + let req = request(250_000); let bytes = req.to_bytes(); let (decoded, consumed) = PollMessagesRequest::decode(&bytes).unwrap(); assert_eq!(consumed, bytes.len()); @@ -151,6 +177,7 @@ mod tests { strategy: WirePollingStrategy::first(), count: 10, auto_commit: false, + wait_timeout_us: 0, }; let bytes = req.to_bytes(); let (decoded, consumed) = PollMessagesRequest::decode(&bytes).unwrap(); @@ -168,6 +195,7 @@ mod tests { strategy: WirePollingStrategy::offset(0), count: 1, auto_commit: false, + wait_timeout_us: 1, }; let bytes = req.to_bytes(); let (decoded, consumed) = PollMessagesRequest::decode(&bytes).unwrap(); @@ -175,6 +203,54 @@ mod tests { assert_eq!(decoded, req); } + #[test] + fn legacy_request_without_wait_timeout_decodes_as_zero() { + let req = request(0); + let bytes = req.to_bytes(); + let legacy_len = bytes.len() - WAIT_TIMEOUT_SIZE; + let (decoded, consumed) = PollMessagesRequest::decode(&bytes[..legacy_len]).unwrap(); + + assert_eq!(consumed, legacy_len); + assert_eq!(decoded.wait_timeout_us, 0); + assert_eq!(decoded, req); + } + + #[test] + fn partial_trailing_wait_timeout_returns_error() { + let req = request(0); + let bytes = req.to_bytes(); + let legacy_len = bytes.len() - WAIT_TIMEOUT_SIZE; + + for timeout_bytes in 1..WAIT_TIMEOUT_SIZE { + assert!( + PollMessagesRequest::decode(&bytes[..legacy_len + timeout_bytes]).is_err(), + "expected error with {timeout_bytes} trailing timeout bytes" + ); + } + } + + #[test] + fn encoded_size_includes_wait_timeout() { + let req = request(42_000); + let bytes = req.to_bytes(); + let legacy_size = req.consumer.encoded_size() + + req.stream_id.encoded_size() + + req.topic_id.encoded_size() + + PARTITION_FLAG_SIZE + + PARTITION_VALUE_SIZE + + STRATEGY_SIZE + + COUNT_SIZE + + AUTO_COMMIT_SIZE; + let wait_timeout_bytes = req.wait_timeout_us.to_le_bytes(); + + assert_eq!(req.encoded_size(), legacy_size + WAIT_TIMEOUT_SIZE); + assert_eq!(bytes.len(), req.encoded_size()); + assert_eq!( + &bytes[legacy_size..legacy_size + WAIT_TIMEOUT_SIZE], + wait_timeout_bytes.as_slice() + ); + } + #[test] fn partition_none_encodes_zero_bytes() { let req = PollMessagesRequest { @@ -185,6 +261,7 @@ mod tests { strategy: WirePollingStrategy::first(), count: 1, auto_commit: false, + wait_timeout_us: 0, }; let bytes = req.to_bytes(); // After consumer(7) + stream_id(6) + topic_id(6) = offset 19 @@ -199,18 +276,11 @@ mod tests { } #[test] - fn truncated_returns_error() { - let req = PollMessagesRequest { - consumer: WireConsumer::consumer(WireIdentifier::numeric(1)), - stream_id: WireIdentifier::numeric(1), - topic_id: WireIdentifier::numeric(1), - partition_id: Some(1), - strategy: WirePollingStrategy::offset(0), - count: 1, - auto_commit: false, - }; + fn truncated_required_fields_return_error() { + let req = request(0); let bytes = req.to_bytes(); - for i in 0..bytes.len() { + let legacy_len = bytes.len() - WAIT_TIMEOUT_SIZE; + for i in 0..legacy_len { assert!( PollMessagesRequest::decode(&bytes[..i]).is_err(), "expected error for truncation at byte {i}" diff --git a/core/common/src/traits/binary_impls/messages.rs b/core/common/src/traits/binary_impls/messages.rs index bf720a1fe8..f44a9a6020 100644 --- a/core/common/src/traits/binary_impls/messages.rs +++ b/core/common/src/traits/binary_impls/messages.rs @@ -37,11 +37,19 @@ use iggy_binary_protocol::requests::messages::{ FlushUnsavedBufferRequest, PollMessagesRequest, RawMessage, SendMessagesEncoder, }; use iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse; +use std::time::Duration; /// Max attempts to resolve a fenced consumer-group poll: one re-sync after the /// coordinator rejects a stale assignment, then retry once. const GROUP_POLL_MAX_ATTEMPTS: usize = 2; +fn duration_to_wait_timeout_us(wait_timeout: Duration) -> Result { + wait_timeout + .as_micros() + .try_into() + .map_err(|_| IggyError::InvalidNumberValue) +} + fn group_cache_key(stream_id: &Identifier, topic_id: &Identifier, group_id: &Identifier) -> String { format!("{stream_id}|{topic_id}|{group_id}") } @@ -176,6 +184,7 @@ async fn poll_group_messages( strategy: &PollingStrategy, count: u32, auto_commit: bool, + wait_timeout_us: u64, ) -> Result { let key = group_cache_key(stream_id, topic_id, &consumer.id); if !client.consumer_group_state().has_assignment(&key) { @@ -205,6 +214,7 @@ async fn poll_group_messages( strategy: polling_strategy_to_wire(strategy), count, auto_commit, + wait_timeout_us, }; match client .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes()) @@ -283,8 +293,33 @@ impl MessageClient for B { strategy: &PollingStrategy, count: u32, auto_commit: bool, + ) -> Result { + self.poll_messages_with_timeout( + stream_id, + topic_id, + partition_id, + consumer, + strategy, + count, + auto_commit, + Duration::ZERO, + ) + .await + } + + async fn poll_messages_with_timeout( + &self, + stream_id: &Identifier, + topic_id: &Identifier, + partition_id: Option, + consumer: &Consumer, + strategy: &PollingStrategy, + count: u32, + auto_commit: bool, + wait_timeout: Duration, ) -> Result { fail_if_not_authenticated(self).await?; + let wait_timeout_us = duration_to_wait_timeout_us(wait_timeout)?; // VSR: a consumer-group poll without an explicit partition is resolved // client-side from the member's cached assignment (the broker routes // explicit partitions only). @@ -297,6 +332,7 @@ impl MessageClient for B { strategy, count, auto_commit, + wait_timeout_us, ) .await; } @@ -308,6 +344,7 @@ impl MessageClient for B { strategy: polling_strategy_to_wire(strategy), count, auto_commit, + wait_timeout_us, }; let response = self .send_raw_with_response(POLL_MESSAGES_CODE, req.to_bytes()) @@ -402,9 +439,12 @@ impl MessageClient for B { #[cfg(test)] mod tests { - use super::{committed_send_confirmations, decode_send_confirmations}; + use super::{ + committed_send_confirmations, decode_send_confirmations, duration_to_wait_timeout_us, + }; use crate::{IggyError, SendMessagesConfirmationResponse, SendMessagesResponse}; use iggy_binary_protocol::codec::WireEncode; + use std::time::Duration; fn response() -> SendMessagesResponse { SendMessagesResponse { @@ -494,4 +534,24 @@ mod tests { ); } } + + #[test] + fn wait_timeout_uses_microseconds() { + assert_eq!( + duration_to_wait_timeout_us(Duration::from_millis(25)).unwrap(), + 25_000 + ); + assert_eq!( + duration_to_wait_timeout_us(Duration::from_nanos(999)).unwrap(), + 0 + ); + } + + #[test] + fn wait_timeout_overflow_is_rejected() { + assert_eq!( + duration_to_wait_timeout_us(Duration::new(u64::MAX, 0)), + Err(IggyError::InvalidNumberValue) + ); + } } diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs index 0c5fa11e4b..c43b45c647 100644 --- a/core/common/src/traits/message_client.rs +++ b/core/common/src/traits/message_client.rs @@ -20,6 +20,7 @@ use crate::{ SendMessagesResponse, }; use async_trait::async_trait; +use std::time::Duration; /// This trait defines the methods to interact with the messaging module. #[async_trait] @@ -41,6 +42,38 @@ pub trait MessageClient { auto_commit: bool, ) -> Result; + /// Poll messages and wait up to `wait_timeout` when no messages are + /// immediately available. A zero timeout preserves immediate polling. + /// Transports without deferred-poll support return `FeatureUnavailable`. + #[allow(clippy::too_many_arguments)] + async fn poll_messages_with_timeout( + &self, + stream_id: &Identifier, + topic_id: &Identifier, + partition_id: Option, + consumer: &Consumer, + strategy: &PollingStrategy, + count: u32, + auto_commit: bool, + wait_timeout: Duration, + ) -> Result { + if wait_timeout.is_zero() { + return self + .poll_messages( + stream_id, + topic_id, + partition_id, + consumer, + strategy, + count, + auto_commit, + ) + .await; + } + + Err(IggyError::FeatureUnavailable) + } + /// Send messages using specified partitioning strategy to the given stream and topic by unique IDs or names. /// /// Authentication is required, and the permission to send the messages. diff --git a/core/integration/tests/server/poll_semantics_vsr.rs b/core/integration/tests/server/poll_semantics_vsr.rs index a3a90f06d0..f10f35c658 100644 --- a/core/integration/tests/server/poll_semantics_vsr.rs +++ b/core/integration/tests/server/poll_semantics_vsr.rs @@ -88,6 +88,49 @@ async fn given_missing_partition_when_polling_should_reject_partition_not_found( assert_eq!(valid.messages.len(), 0, "empty topic polls empty"); } +#[iggy_harness(test_client_transport = [Tcp])] +async fn given_non_zero_wait_timeout_when_polling_should_reject_feature_unavailable( + harness: &TestHarness, +) { + let client = harness.tcp_root_client().await.expect("tcp root client"); + client + .create_stream("poll-timeout-stream") + .await + .expect("create stream"); + let stream_id = Identifier::from_str_value("poll-timeout-stream").expect("stream identifier"); + client + .create_topic( + &stream_id, + "poll-timeout-topic", + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); + let topic_id = Identifier::from_str_value("poll-timeout-topic").expect("topic identifier"); + + let result = client + .poll_messages_with_timeout( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + 1, + false, + Duration::from_secs(1), + ) + .await; + + assert!( + matches!(&result, Err(error) if error.as_code() == IggyError::FeatureUnavailable.as_code()), + "the active server must reject unsupported deferred waits, got {result:?}" + ); +} + #[iggy_harness( test_client_transport = [Tcp] )] diff --git a/core/sdk/src/client_wrappers/binary_message_client.rs b/core/sdk/src/client_wrappers/binary_message_client.rs index eb3eed94f9..4b0bbe5c9a 100644 --- a/core/sdk/src/client_wrappers/binary_message_client.rs +++ b/core/sdk/src/client_wrappers/binary_message_client.rs @@ -22,6 +22,7 @@ use iggy_common::{ Consumer, Identifier, IggyError, IggyMessage, Partitioning, PolledMessages, PollingStrategy, SendMessagesResponse, }; +use std::time::Duration; #[async_trait] impl MessageClient for ClientWrapper { @@ -34,11 +35,35 @@ impl MessageClient for ClientWrapper { strategy: &PollingStrategy, count: u32, auto_commit: bool, + ) -> Result { + self.poll_messages_with_timeout( + stream_id, + topic_id, + partition_id, + consumer, + strategy, + count, + auto_commit, + Duration::ZERO, + ) + .await + } + + async fn poll_messages_with_timeout( + &self, + stream_id: &Identifier, + topic_id: &Identifier, + partition_id: Option, + consumer: &Consumer, + strategy: &PollingStrategy, + count: u32, + auto_commit: bool, + wait_timeout: Duration, ) -> Result { match self { ClientWrapper::Iggy(client) => { client - .poll_messages( + .poll_messages_with_timeout( stream_id, topic_id, partition_id, @@ -46,12 +71,13 @@ impl MessageClient for ClientWrapper { strategy, count, auto_commit, + wait_timeout, ) .await } ClientWrapper::Http(client) => { client - .poll_messages( + .poll_messages_with_timeout( stream_id, topic_id, partition_id, @@ -59,12 +85,13 @@ impl MessageClient for ClientWrapper { strategy, count, auto_commit, + wait_timeout, ) .await } ClientWrapper::Tcp(client) => { client - .poll_messages( + .poll_messages_with_timeout( stream_id, topic_id, partition_id, @@ -72,12 +99,13 @@ impl MessageClient for ClientWrapper { strategy, count, auto_commit, + wait_timeout, ) .await } ClientWrapper::Quic(client) => { client - .poll_messages( + .poll_messages_with_timeout( stream_id, topic_id, partition_id, @@ -85,12 +113,13 @@ impl MessageClient for ClientWrapper { strategy, count, auto_commit, + wait_timeout, ) .await } ClientWrapper::WebSocket(client) => { client - .poll_messages( + .poll_messages_with_timeout( stream_id, topic_id, partition_id, @@ -98,6 +127,7 @@ impl MessageClient for ClientWrapper { strategy, count, auto_commit, + wait_timeout, ) .await } diff --git a/core/sdk/src/clients/binary_message.rs b/core/sdk/src/clients/binary_message.rs index db722a0066..57d7301915 100644 --- a/core/sdk/src/clients/binary_message.rs +++ b/core/sdk/src/clients/binary_message.rs @@ -24,6 +24,7 @@ use iggy_common::{ Consumer, Identifier, IggyError, IggyMessage, Partitioning, PolledMessages, PollingStrategy, SendMessagesResponse, }; +use std::time::Duration; #[async_trait] impl MessageClient for IggyClient { @@ -36,6 +37,30 @@ impl MessageClient for IggyClient { strategy: &PollingStrategy, count: u32, auto_commit: bool, + ) -> Result { + self.poll_messages_with_timeout( + stream_id, + topic_id, + partition_id, + consumer, + strategy, + count, + auto_commit, + Duration::ZERO, + ) + .await + } + + async fn poll_messages_with_timeout( + &self, + stream_id: &Identifier, + topic_id: &Identifier, + partition_id: Option, + consumer: &Consumer, + strategy: &PollingStrategy, + count: u32, + auto_commit: bool, + wait_timeout: Duration, ) -> Result { if count == 0 { return Err(IggyError::InvalidMessagesCount); @@ -45,7 +70,7 @@ impl MessageClient for IggyClient { .client .read() .await - .poll_messages( + .poll_messages_with_timeout( stream_id, topic_id, partition_id, @@ -53,6 +78,7 @@ impl MessageClient for IggyClient { strategy, count, auto_commit, + wait_timeout, ) .await?; diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index 5dab8e2fd5..f75ffc2a11 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -2146,6 +2146,21 @@ async fn handle_poll_messages( .await; return; }; + if wire.wait_timeout_us != 0 { + warn!( + transport_client_id, + wait_timeout_us = wire.wait_timeout_us, + "deferred poll waits are not available on the active server" + ); + send_non_replicated_deny( + shard, + request, + transport_client_id, + IggyError::FeatureUnavailable.as_code(), + ) + .await; + return; + } // Gate on (stream, topic) before touching the partition plane. A resolution // miss falls through to the resolve path below (empty-poll / not-found); a // denial replies status!=0 with an empty body, distinct from the empty-poll @@ -2491,6 +2506,9 @@ where S: 'static, SB: SuperblockStore + 'static, { + if wire.wait_timeout_us != 0 { + return Err(IggyError::FeatureUnavailable); + } let strategy = polling_strategy_from_wire(&wire.strategy)?; let args = PollingArgs::new(strategy, wire.count, wire.auto_commit); diff --git a/core/server/src/http/wire.rs b/core/server/src/http/wire.rs index 20ada0c332..7ec4322dec 100644 --- a/core/server/src/http/wire.rs +++ b/core/server/src/http/wire.rs @@ -118,6 +118,7 @@ pub(in crate::http) fn poll_wire_request( }, count: query.count, auto_commit: query.auto_commit, + wait_timeout_us: 0, }) } @@ -422,6 +423,7 @@ mod tests { assert_eq!(wire.strategy.value, 0); assert_eq!(wire.count, 25); assert!(wire.auto_commit); + assert_eq!(wire.wait_timeout_us, 0); } #[test] diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs index 3e7742ed1e..e5dcd368fb 100644 --- a/core/simulator/src/client.rs +++ b/core/simulator/src/client.rs @@ -575,6 +575,7 @@ impl SimClient { strategy: WirePollingStrategy::first(), count, auto_commit: false, + wait_timeout_us: 0, } .to_bytes(); From d86e92b22e1672789e319ed5e97fcd08aceeeec9 Mon Sep 17 00:00:00 2001 From: Arun Singh Date: Fri, 21 Aug 2026 23:40:33 +0530 Subject: [PATCH 2/2] Fix Clippy failures in deferred poll support Signed-off-by: Arun Singh --- .../src/traits/binary_impls/messages.rs | 33 +++++++++++----- core/server/src/dispatch.rs | 39 +++++++++++++------ 2 files changed, 50 insertions(+), 22 deletions(-) diff --git a/core/common/src/traits/binary_impls/messages.rs b/core/common/src/traits/binary_impls/messages.rs index f44a9a6020..3f609a2a32 100644 --- a/core/common/src/traits/binary_impls/messages.rs +++ b/core/common/src/traits/binary_impls/messages.rs @@ -43,6 +43,14 @@ use std::time::Duration; /// coordinator rejects a stale assignment, then retry once. const GROUP_POLL_MAX_ATTEMPTS: usize = 2; +struct PollGroupOptions<'a> { + consumer: &'a Consumer, + strategy: &'a PollingStrategy, + count: u32, + auto_commit: bool, + wait_timeout_us: u64, +} + fn duration_to_wait_timeout_us(wait_timeout: Duration) -> Result { wait_timeout .as_micros() @@ -180,12 +188,15 @@ async fn poll_group_messages( client: &B, stream_id: &Identifier, topic_id: &Identifier, - consumer: &Consumer, - strategy: &PollingStrategy, - count: u32, - auto_commit: bool, - wait_timeout_us: u64, + options: PollGroupOptions<'_>, ) -> Result { + let PollGroupOptions { + consumer, + strategy, + count, + auto_commit, + wait_timeout_us, + } = options; let key = group_cache_key(stream_id, topic_id, &consumer.id); if !client.consumer_group_state().has_assignment(&key) { sync_group_assignment(client, stream_id, topic_id, &consumer.id).await?; @@ -328,11 +339,13 @@ impl MessageClient for B { self, stream_id, topic_id, - consumer, - strategy, - count, - auto_commit, - wait_timeout_us, + PollGroupOptions { + consumer, + strategy, + count, + auto_commit, + wait_timeout_us, + }, ) .await; } diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index f75ffc2a11..da0c9f6a18 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -2115,6 +2115,32 @@ async fn evict_stale_client( } } +/// Reject a non-zero wait timeout until active-server deferred waits are implemented. +async fn reject_deferred_poll( + shard: &Rc>, + request: &Message, + transport_client_id: u128, + wait_timeout_us: u64, +) where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + warn!( + transport_client_id, + wait_timeout_us, "deferred poll waits are not available on the active server" + ); + send_non_replicated_deny( + shard, + request, + transport_client_id, + IggyError::FeatureUnavailable.as_code(), + ) + .await; +} + /// Serve `poll_messages`: resolve the partition namespace, run the read on /// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode /// the stored batches into the legacy wire `PolledMessages` body. @@ -2147,18 +2173,7 @@ async fn handle_poll_messages( return; }; if wire.wait_timeout_us != 0 { - warn!( - transport_client_id, - wait_timeout_us = wire.wait_timeout_us, - "deferred poll waits are not available on the active server" - ); - send_non_replicated_deny( - shard, - request, - transport_client_id, - IggyError::FeatureUnavailable.as_code(), - ) - .await; + reject_deferred_poll(shard, request, transport_client_id, wire.wait_timeout_us).await; return; } // Gate on (stream, topic) before touching the partition plane. A resolution