This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/partition-routing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3f7e6ac809f5dcdfcf087a32e01d23bc16658b37 Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 13 22:33:33 2026 +0200 fix(server): refine consumer progress dispatch Queued conflicts repeatedly decoded bodies and resolved namespaces. Pure offset/timestamp polls also waited on unrelated stored progress. Cache keys after barriers at the applied metadata frontier and reserve progress lanes for stored-offset dependencies. Clarify catch-up and expiry limits, and prove held owner reads do not block PING. --- core/server/src/dispatch/mod.rs | 313 ++++++++++++++++++++++++++++------ core/server/src/dispatch/partition.rs | 117 ++++++++++++- core/server/src/http/forward.rs | 5 +- 3 files changed, 374 insertions(+), 61 deletions(-) diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index 6e81a5ca6..b4776fefe 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -87,7 +87,39 @@ use std::sync::Arc; use std::task::{Poll, Waker}; use tracing::{debug, warn}; -type ClientRequestQueues = Rc<RefCell<AHashMap<u128, VecDeque<Message<GenericHeader>>>>>; +type ClientRequestQueues = Rc<RefCell<AHashMap<u128, VecDeque<QueuedClientRequest>>>>; + +struct QueuedClientRequest { + message: Message<GenericHeader>, + // A queued metadata barrier can change the namespace resolved by this frame. + progress: Option<(u64, Option<ConsumerProgress>)>, +} + +impl From<Message<GenericHeader>> for QueuedClientRequest { + fn from(message: Message<GenericHeader>) -> Self { + Self { + message, + progress: None, + } + } +} + +impl QueuedClientRequest { + fn progress_key( + &mut self, + metadata_frontier: u64, + resolve: impl FnOnce(&Message<GenericHeader>) -> Option<ConsumerProgress>, + ) -> Option<ConsumerProgress> { + if let Some((frontier, key)) = self.progress + && frontier == metadata_frontier + { + return key; + } + let key = resolve(&self.message); + self.progress = Some((metadata_frontier, key)); + key + } +} /// Requests one client may have queued behind a request this shard has not /// answered yet. @@ -140,7 +172,7 @@ fn enqueue_client_request<B, MJ, S, SB>( deny_overflowing_client_request(shard_handle, client_id, message); return; } - queue.push_back(message); + queue.push_back(message.into()); } let running = active.borrow().get(&client_id).cloned(); if let Some(drain) = running { @@ -259,10 +291,14 @@ impl Drop for ActiveDrainSlot { /// prologue of each. Login, logout, `BIND_SESSION` and partition admission /// finish on the drain. Metadata commits run in tasks, but later non-PING /// frames wait in the queue until the metadata lane releases, before they -/// capture session state or resolve a namespace. PING can pass that queue. -/// Polls and explicit offset writes share a lane per resolved namespace and -/// consumer until their reply. Independent requests can pass a queued conflict, -/// but cannot pass a queued session or metadata barrier. A PING still waits +/// capture session state or resolve a namespace. This orders dispatch, not local +/// metadata catch-up: partition writes still resolve from this shard's STM +/// without waiting for the connection watermark. PING can pass that queue. +/// Stored-offset reads, Next/auto-commit polls and explicit offset writes share +/// a lane per resolved namespace and consumer until their reply or wait expiry. +/// Expiry releases only this connection's guard, not an admitted mutation, which +/// may still commit after later progress. Independent requests can pass a queued +/// conflict, but cannot pass a queued session or metadata barrier. A PING still waits /// behind a frame already suspended in admission or binding. /// /// `capacity` caps those tasks at `message_bus.client_queue_capacity`, the @@ -404,19 +440,26 @@ async fn drain_client_requests<B, MJ, S, SB>( SB: SuperblockStore + 'static, { loop { - let next = pop_next_client_request(&queues, client_id, |message| { - if is_ping(message) { + let metadata_frontier = shard.plane.metadata().applied_frontier().get(); + let next = pop_next_client_request(&queues, client_id, |queued| { + if is_ping(&queued.message) { return true; } if drain.metadata_write.get() { return false; } - progress_key(&shard, message).is_none_or(|key| { - drain.in_flight.get() < drain.capacity && !drain.progress.borrow().contains(&key) - }) + queued + .progress_key(metadata_frontier, |message| progress_key(&shard, message)) + .is_none_or(|key| { + drain.in_flight.get() < drain.capacity + && !drain.progress.borrow().contains(&key) + }) }); - if let Some(message) = next { - let progress = progress_key(&shard, &message).map(|key| drain.admit_progress(key)); + if let Some(queued) = next { + let progress = queued + .progress + .and_then(|(_, key)| key) + .map(|key| drain.admit_progress(key)); handle_client_request( &shard, &sessions, @@ -424,7 +467,7 @@ async fn drain_client_requests<B, MJ, S, SB>( max_tokens_per_user, drain, client_id, - message, + queued.message, progress, ) .await; @@ -439,8 +482,8 @@ async fn drain_client_requests<B, MJ, S, SB>( fn pop_next_client_request( queues: &ClientRequestQueues, client_id: u128, - ready: impl Fn(&Message<GenericHeader>) -> bool, -) -> Option<Message<GenericHeader>> { + mut ready: impl FnMut(&mut QueuedClientRequest) -> bool, +) -> Option<QueuedClientRequest> { // Freed as soon as it drains empty. Keeping it would cost the connection // one `VecDeque` (at its peak depth, since `pop_front` never gives // capacity back), and the connection-lost hook cannot be the sole owner: @@ -451,28 +494,30 @@ fn pop_next_client_request( let queue = queues.get_mut(&client_id)?; let mut waiting = false; let mut barrier = false; - let index = queue.iter().position(|message| { - if is_ping(message) { + let index = queue.iter_mut().position(|queued| { + if is_ping(&queued.message) { return true; } if barrier { return false; } - let orders_followers = message - .try_as_typed::<RequestHeader>() - .map_or(true, |request| { - let header = request.header(); - if header.operation == Operation::NonReplicated { - header.reserved[..4] == BIND_SESSION_CODE.to_le_bytes() - } else { - !header.operation.is_partition() - } - }); + let orders_followers = + queued + .message + .try_as_typed::<RequestHeader>() + .map_or(true, |request| { + let header = request.header(); + if header.operation == Operation::NonReplicated { + header.reserved[..4] == BIND_SESSION_CODE.to_le_bytes() + } else { + !header.operation.is_partition() + } + }); if waiting && orders_followers { barrier = true; return false; } - if ready(message) { + if ready(queued) { return true; } waiting = true; @@ -1111,16 +1156,19 @@ mod tests { use crate::cluster_meta::ClusterRoster; use crate::dispatch::host::ServerHost; use crate::dispatch::test_support::{ - FIRST_BOOT, SpyBus, TestMux, TestShard, UNDECLARED_OPERATION, prepare_message, - register_reply, request_message, test_shard, + FIRST_BOOT, SpyBus, TestMux, TestShard, UNDECLARED_OPERATION, create_one_partition, + prepare_message, register_reply, request_message, test_shard, }; use consensus::Sequencer; use consensus::client_table::bind_verifier; use iggy_binary_protocol::Command; use iggy_binary_protocol::codes::{ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, GET_ME_CODE, POLL_MESSAGES_CODE, + POLL_MESSAGES_ON_PRIMARY_CODE, }; + use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy; use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; + use iggy_binary_protocol::requests::messages::PollMessagesRequest; use iggy_binary_protocol::requests::streams::CreateStreamRequest; use iggy_binary_protocol::requests::system::BindSessionRequest; use iggy_binary_protocol::requests::users::{ @@ -1139,13 +1187,13 @@ mod tests { use metadata::stm::StateMachine; use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig}; use server_common::MESSAGE_ALIGN; - use server_common::sharding::ShardId; + use server_common::sharding::{PartitionLocation, ShardId}; use shard::metrics::ShardMetrics; - use shard::shards_table::PapayaShardsTable; + use shard::shards_table::{PapayaShardsTable, ShardsTable}; use shard::{ ConnectedClientInfo, LifecycleFrame, ListClientsReply, MetadataSubmit, NoopHost, - PartitionConsensusConfig, ReplicaTopology, ShardFrame, ShardHost, ShardIdentity, - shard_channel, + PartitionConsensusConfig, PartitionRead, PartitionReadReply, ReplicaTopology, ShardFrame, + ShardHost, ShardIdentity, shard_channel, }; use std::mem::size_of; use std::sync::atomic::AtomicBool; @@ -2352,6 +2400,151 @@ mod tests { assert_eq!(reply_status(&replies[1].1), 0); } + #[compio::test] + async fn given_a_held_partition_poll_when_ping_follows_should_answer_ping_before_owner_reply() { + const POLL_REQUEST: u64 = 1; + const PING_REQUEST: u64 = 2; + for code in [POLL_MESSAGES_CODE, POLL_MESSAGES_ON_PRIMARY_CODE] { + let bus = SpyBus::default(); + let mut shard = test_shard(&bus, 0, 1, FIRST_BOOT); + let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1); + shard.attach_senders(vec![sender]); + let shard = Rc::new(shard); + let namespace = create_one_partition(&shard); + shard + .shards_table() + .insert(namespace, PartitionLocation::new(ShardId::new(0), 0)); + commit_root_session(&shard); + let shard_handle = Rc::new(RefCell::new(Some(Rc::downgrade(&shard)))); + let host = ServerHost::new(&bus, &shard_handle, Arc::new(ServerConfig::default()), 1); + bind_root_session(&shard, host.sessions()).await; + let body = PollMessagesRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(1)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + strategy: WirePollingStrategy::offset(0), + count: 1, + auto_commit: false, + } + .to_bytes(); + let mut poll = wire_request( + Operation::NonReplicated, + ROOT_CLIENT, + ROOT_SESSION, + POLL_REQUEST, + &body, + ); + let header = bytemuck::checked::from_bytes_mut::<RequestHeader>( + &mut poll.as_mut_slice()[..size_of::<RequestHeader>()], + ); + header.reserved[..4].copy_from_slice(&code.to_le_bytes()); + header.partition_incarnation = shard + .plane + .metadata() + .mux_stm + .streams() + .created_revision_for_namespace(namespace) + .unwrap(); + host.on_client_request(ROOT_TRANSPORT, poll.into_generic()); + settle().await; + let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead { + read: PartitionRead::PollOnPrimary { .. }, + reply, + .. + }) = owner_inbox + .try_recv() + .expect("the poll must reach its partition owner") + else { + panic!("the owner must receive a primary-gated poll"); + }; + assert!(bus.client_replies.borrow().is_empty()); + + host.on_client_request( + ROOT_TRANSPORT, + root_read(PING_CODE, PING_REQUEST).into_generic(), + ); + await_replies(&bus, 1).await; + { + let replies = bus.client_replies.borrow(); + assert_eq!( + replies.len(), + 1, + "PING must not wait for partition I/O on code {code}" + ); + assert_eq!(reply_request(&replies[0].1), PING_REQUEST); + } + reply + .try_send(PartitionReadReply::Rejected( + IggyError::TransientNotAccepted, + )) + .unwrap(); + await_replies(&bus, 2).await; + let replies = bus.client_replies.borrow(); + assert_eq!(replies.len(), 2); + assert_eq!(reply_request(&replies[1].1), POLL_REQUEST); + } + } + + #[compio::test] + async fn given_queued_progress_when_metadata_changes_should_resolve_once_per_frontier() { + const CREATED_FRONTIER: u64 = ROOT_SESSION + 1; + let (_, shard, _) = root_host(ServerConfig::default()); + let body = StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(1)), + stream_id: WireIdentifier::named("stream").unwrap(), + topic_id: WireIdentifier::named("topic").unwrap(), + partition_id: Some(0), + offset: 1, + ack: AckLevel::Quorum, + } + .to_bytes(); + let mut queued = QueuedClientRequest::from( + wire_request( + Operation::StoreConsumerOffset, + ROOT_CLIENT, + ROOT_SESSION, + 1, + &body, + ) + .into_generic(), + ); + assert!( + queued.progress.is_none(), + "enqueue must not resolve ahead of a metadata barrier" + ); + let resolutions = Cell::new(0); + let resolve = |message: &Message<GenericHeader>| { + resolutions.set(resolutions.get() + 1); + progress_key(&shard, message) + }; + for _ in 0..3 { + assert_eq!(queued.progress_key(ROOT_SESSION, resolve), None); + } + assert_eq!( + resolutions.get(), + 1, + "wakes must reuse a resolved missing key too" + ); + let namespace = create_one_partition(&shard); + shard + .plane + .metadata() + .applied_frontier() + .advance(CREATED_FRONTIER); + for _ in 0..3 { + assert_eq!( + queued.progress_key(shard.plane.metadata().applied_frontier().get(), resolve), + Some((namespace.inner(), iggy_binary_protocol::KIND_CONSUMER, 1)) + ); + } + assert_eq!( + resolutions.get(), + 2, + "new metadata must invalidate the cached namespace" + ); + } + #[compio::test] async fn given_pending_metadata_when_read_and_ping_queue_should_answer_ping_then_wake_read() { const READ_REQUEST: u64 = 1; @@ -2363,10 +2556,13 @@ mod tests { let metadata = drain.admit(Admission::MetadataLane).await; let queues = Rc::new(RefCell::new(AHashMap::from_iter([( ROOT_TRANSPORT, - VecDeque::from([ - held_get_me(READ_REQUEST, 0), - root_read(PING_CODE, PING_REQUEST).into_generic(), - ]), + VecDeque::from( + [ + held_get_me(READ_REQUEST, 0), + root_read(PING_CODE, PING_REQUEST).into_generic(), + ] + .map(QueuedClientRequest::from), + ), )]))); let draining = drain_client_requests( shard, @@ -2413,22 +2609,26 @@ mod tests { }; let queues = Rc::new(RefCell::new(AHashMap::from_iter([( ROOT_TRANSPORT, - VecDeque::from([ - wire_request( - Operation::StoreConsumerOffset, - ROOT_CLIENT, - ROOT_SESSION, - BLOCKED, - &[], - ) - .into_generic(), - barrier, - held_get_me(READ, 0), - root_read(PING_CODE, PING).into_generic(), - ]), + VecDeque::from( + [ + wire_request( + Operation::StoreConsumerOffset, + ROOT_CLIENT, + ROOT_SESSION, + BLOCKED, + &[], + ) + .into_generic(), + barrier, + held_get_me(READ, 0), + root_read(PING_CODE, PING).into_generic(), + ] + .map(QueuedClientRequest::from), + ), )]))); - let ready = |message: &Message<GenericHeader>| { - message + let ready = |queued: &mut QueuedClientRequest| { + queued + .message .try_as_typed::<RequestHeader>() .unwrap() .header() @@ -2436,12 +2636,13 @@ mod tests { != BLOCKED }; let ping = pop_next_client_request(&queues, ROOT_TRANSPORT, ready).unwrap(); - assert!(is_ping(&ping)); + assert!(is_ping(&ping.message)); assert!(pop_next_client_request(&queues, ROOT_TRANSPORT, ready).is_none()); for expected in [BLOCKED, BARRIER, READ] { let message = pop_next_client_request(&queues, ROOT_TRANSPORT, |_| true).unwrap(); assert_eq!( message + .message .try_as_typed::<RequestHeader>() .unwrap() .header() @@ -2887,7 +3088,7 @@ mod tests { .borrow_mut() .entry(CLIENT) .or_default() - .push_back(non_replicated_request(CLIENT, PING_CODE)); + .push_back(non_replicated_request(CLIENT, PING_CODE).into()); assert!(pop_next_client_request(&queues, CLIENT, |_| true).is_some()); assert!( diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index 08e99c4e2..b9ea3b553 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -350,7 +350,9 @@ async fn relay_partition_reply<B, MJ, S, SB>( let Some(reply) = shard.await_partition_submit(ticket).await else { // Abandoned or expired. Deliberately silent: the outcome is // unknown, and a synthesized failure could contradict a write that - // commits moments later. The client's read-timeout is the recovery. + // commits moments later. Releasing the progress guard does not + // cancel that write or fence it against later progress. + // The client's read-timeout is the recovery. return; }; if let Some(kind) = consumer_kind @@ -379,7 +381,7 @@ async fn relay_partition_reply<B, MJ, S, SB>( }); } -/// The offset key shared by polls and explicit progress writes on a connection. +/// The offset key shared by stored-offset reads and progress writes on a connection. /// Failed resolution is left to the normal handler's authorization and error path. pub(super) fn progress_key<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, @@ -405,6 +407,9 @@ where == POLL_MESSAGES_ON_PRIMARY_CODE.to_le_bytes() => { let poll = PollMessagesRequest::decode_from(request.body()).ok()?; + if !poll.auto_commit && poll.strategy.kind != WirePollingStrategy::next().kind { + return None; + } GetConsumerOffsetRequest { consumer: poll.consumer, stream_id: poll.stream_id, @@ -1831,6 +1836,114 @@ mod tests { assert_eq!(metadata.applied_frontier().get(), LATER_FRONTIER); } + #[compio::test] + async fn given_poll_strategies_when_resolving_progress_should_only_order_stored_offset_dependencies() + { + const CONSUMER: u32 = 1; + let bus = SpyBus::default(); + let shard = Rc::new(test_shard(&bus, 0, 1, 1)); + let namespace = create_one_partition(&shard); + let expected = Some((namespace.inner(), KIND_CONSUMER, CONSUMER as usize)); + for code in [POLL_MESSAGES_CODE, POLL_MESSAGES_ON_PRIMARY_CODE] { + for (strategy, reads_stored) in [ + (WirePollingStrategy::offset(0), false), + (WirePollingStrategy::timestamp(1), false), + (WirePollingStrategy::first(), false), + (WirePollingStrategy::last(), false), + (WirePollingStrategy::next(), true), + ] { + for auto_commit in [false, true] { + let body = PollMessagesRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + strategy: strategy.clone(), + count: 1, + auto_commit, + } + .to_bytes(); + let mut request = request_message(Operation::NonReplicated, 1, 1, 1, &body); + bytemuck::checked::from_bytes_mut::<RequestHeader>( + &mut request.as_mut_slice()[..size_of::<RequestHeader>()], + ) + .reserved[..4] + .copy_from_slice(&code.to_le_bytes()); + assert_eq!( + progress_key(&shard, &request.into_generic()), + if reads_stored || auto_commit { + expected + } else { + None + }, + "code {code}, strategy {strategy:?}, auto_commit {auto_commit}", + ); + } + } + } + let requests = [ + request_message( + Operation::StoreConsumerOffset, + 1, + 1, + 1, + &StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + offset: 1, + ack: iggy_binary_protocol::AckLevel::Quorum, + } + .to_bytes(), + ), + request_message( + Operation::DeleteConsumerOffset, + 1, + 1, + 2, + &DeleteConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + ack: iggy_binary_protocol::AckLevel::Quorum, + } + .to_bytes(), + ), + ]; + for request in requests { + assert_eq!( + progress_key(&shard, &request.into_generic()), + expected, + "explicit writes must retain the same progress lane" + ); + } + let mut read = request_message( + Operation::NonReplicated, + 1, + 1, + 3, + &GetConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(CONSUMER)), + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: Some(0), + } + .to_bytes(), + ); + bytemuck::checked::from_bytes_mut::<RequestHeader>( + &mut read.as_mut_slice()[..size_of::<RequestHeader>()], + ) + .reserved[..4] + .copy_from_slice(&GET_CONSUMER_OFFSET_CODE.to_le_bytes()); + assert_eq!( + progress_key(&shard, &read.into_generic()), + expected, + "stored-offset reads must still wait for explicit progress" + ); + } + #[compio::test] async fn given_invalid_partition_writes_when_resolving_should_answer_typed_error_codes() { const VSR_CLIENT: u128 = 1; diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs index beae5ffc0..6046bd511 100644 --- a/core/server/src/http/forward.rs +++ b/core/server/src/http/forward.rs @@ -48,9 +48,8 @@ //! primary's pipeline: a connect-phase failure, and for a metadata relay a //! 503 whose body carries the `TransientNotAccepted` or `NotPrimary` code or //! a 307 from a stale target, for a partition relay a 503 carrying -//! `NotPrimary`. Every other outcome - including a 503 carrying -//! `TransientNotCommitted`, whose op -//! may still commit - is relayed as-is, because re-issuing it under a fresh +//! `NotPrimary`. Every other outcome, including `TransientNotCommitted`, is +//! relayed as-is: its op may still commit, and re-issuing it under a fresh //! session would defeat the consensus dedup and double-apply the op. //! - A partition read changes no state, so a transport failure after its send //! is retried too: the resend cannot apply anything twice.
