spetz commented on code in PR #4169:
URL: https://github.com/apache/iggy/pull/4169#discussion_r4008781632
##########
core/consensus/src/client_table.rs:
##########
@@ -1653,6 +1674,27 @@ impl ClientTable {
self.slots[slot_idx].as_ref().map(|entry| entry.epoch)
}
+ /// Attach only after the caller has authenticated `user_id` and waited
+ /// for the local metadata frontier to cover the requested session.
+ /// The registered user owns the session, matching authenticated login
Review Comment:
A doc comment now states that the registered user owns the session and that
client ids and epochs are identifiers rather than authentication secrets. The
check is unchanged, so a second connection authenticated as the same user can
still adopt another group member's identity and take its partitions. The sweep
validator called this resolution out in advance: documenting it records the
behavior as intended rather than closing it. The request doc at
`core/binary_protocol/src/requests/system/attach_consumer_session.rs:25` is
untouched and still reads "The server verifies ownership and the exact parent
epoch", so the two documents now disagree about the same guarantee
##########
core/server/config.toml:
##########
@@ -993,12 +997,16 @@ offset_reservation_lease = 65536
# served from the ring before falling back to bulk sync, at the cost of pinned
# memory per partition. Must be > 0 and <= 65536. Single-replica partitions
# retain nothing regardless.
-evicted_ring_capacity = 4096
+evicted_ring_capacity = 65536
# Byte ceiling for the evicted ring per partition; whichever ring cap (this or
# evicted_ring_capacity) trips first evicts. Bounds the ring memory a burst of
# large batches can pin. Must be > 0 and <= "256 MiB".
-evicted_ring_bytes_max = "16 MiB"
+# PER PARTITION and per replica, so the node-wide ceiling is this times the
partitions it
+# hosts, pinned as anonymous memory beside the same bytes in page cache.
Raising it on a
+# node with many partitions is a large memory decision;
partition_repair_ring_bytes on the
+# metrics endpoint reports what the rings on each shard actually hold.
+evicted_ring_bytes_max = "64 MiB"
Review Comment:
`evicted_ring_bytes_max` is still 64 MiB per partition per replica with no
node-wide budget. The `BTreeMap` change removed the CPU cost of the larger ring
but not its memory ceiling.
##########
core/partitions/src/persistence.rs:
##########
@@ -1140,49 +1155,69 @@ impl<S: DurableStorage> PartitionPersistence<S> {
break;
}
if epoch == self.epoch.get() && !self.retired.get() {
- let mut references = self.segment_references.borrow_mut();
- if rebuild_references {
- references.clear();
- references.extend(journal.written_segment_references(0));
- } else if let Some(from_op) =
self.written_head.get().checked_add(1) {
-
references.extend(journal.written_segment_references(from_op));
- }
- drop(references);
- self.disk_bytes.set(journal.size_bytes());
- self.retained_bytes.set(journal.retained_bytes());
- self.segment_checkpoint.set(journal.segment_checkpoint());
- let advanced = journal.durable_op() != self.durable_head.get()
- || journal.checkpoint_op() != self.checkpoint.get()
- || journal.certified_log_view() !=
self.certified_log_view.get()
- || (journal.segment_checkpoint().is_some()
- && journal.head() != self.written_head.get());
- self.certified_log_view.set(journal.certified_log_view());
- if self
- .requested_log_view
- .get()
- .is_some_and(|(view, _, _)| Some(view) ==
self.certified_log_view.get())
- {
- self.requested_log_view.set(None);
- }
- self.written_head.set(journal.head());
- self.durable_head.set(journal.durable_op());
- if journal.checkpoint_op() > self.checkpoint.get() {
- self.accepted
- .borrow_mut()
- .checkpoint(journal.checkpoint_op());
- }
- self.checkpoint.set(journal.checkpoint_op());
- self.checkpoint_checksum.set(journal.checkpoint_checksum());
- self.purge_generation.set(journal.purge_marker().0);
- self.purge_floor.set(journal.purge_marker().1);
- if advanced {
- self.notify();
- }
+ self.publish_mutation(journal, rebuild_references);
+ }
+ // Reclaim after publishing the mutation when the queue is empty.
+ // Appends arriving during reclaim still wait for its unlinks and
+ // directory barrier.
+ // A partition that never drains would then never reclaim, so force
+ // a pass every `RECLAIM_MUTATIONS_MAX` mutations and accept that
+ // one group's latency.
+ mutations_since_reclaim += 1;
+ let idle = self.queue.borrow().is_empty();
+ if idle || mutations_since_reclaim >= RECLAIM_MUTATIONS_MAX {
Review Comment:
At the legal 1 MiB minimum segment size a checkpoint queues roughly 64
links, so 64 mutations enqueue about 520 obsolete paths against the 16 a pass
removes.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+const MAX_CACHED_ROUTES: usize = 4096;
+const MAX_DATA_CONNECTIONS: usize = 256;
+const POLL_TIMEOUT: Duration = Duration::from_secs(30);
+const ROUTING_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+const ROUTING_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(1);
+pub(crate) const ROSTER_READ_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub(crate) const fn is_poll_routing_code(code: u32) -> bool {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
+ self.next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ let mut routes = self
+ .routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
+ }
+
+ pub(crate) fn remember_credentials(&self, credentials: Credentials,
user_id: u32) {
+ *self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) =
Some((credentials, user_id));
+ }
+
+ pub(crate) fn forget_credentials(&self) {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ }
+
+ pub(crate) fn credentials(&self) -> Option<Credentials> {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .as_ref()
+ .map(|(credentials, _)| credentials.clone())
+ }
+
+ pub(crate) fn refresh_password(&self, user: &Identifier, new_password:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, password), user_id))
=
+ credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *password = SecretString::from(new_password.to_owned());
+ }
+ }
+
+ pub(crate) fn refresh_username(&self, user: &Identifier, new_username:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, _), user_id)) =
credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *username = new_username.to_owned();
+ }
+ }
+}
+
+impl<T: PollTransport> PollRouter<T> {
+ pub(crate) async fn poll(
+ &self,
+ coordinator: &T,
+ request: &PollMessagesRequest,
+ ) -> Result<Bytes, IggyError> {
+ if !request.auto_commit || !self.is_clustered(coordinator).await? {
+ return coordinator
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await;
+ }
+ let payload = request.to_bytes();
+ let parameters_size = request.strategy.encoded_size() +
size_of::<u32>() + size_of::<u8>();
+ let key = (
+ GET_POLL_ROUTING_CODE,
+ payload.slice(..payload.len() - parameters_size),
+ );
+ self.send_routed(coordinator, POLL_MESSAGES_ON_PRIMARY_CODE, key,
payload)
+ .await
+ }
+
+ pub(crate) async fn write_offset(
+ &self,
+ coordinator: &T,
+ code: u32,
+ payload: Bytes,
+ ) -> Result<Bytes, IggyError> {
+ if !self.is_clustered(coordinator).await? {
+ return coordinator.send_raw_with_response(code, payload).await;
+ }
+ let (_, route_size) =
+ GetConsumerOffsetRequest::decode(&payload).map_err(|_|
IggyError::InvalidCommand)?;
+ let key = (
+ GET_CONSUMER_OFFSET_ROUTING_CODE,
+ payload.slice(..route_size),
+ );
+ self.send_routed(coordinator, code, key, payload).await
+ }
+
+ pub(crate) async fn is_clustered(&self, coordinator: &T) -> Result<bool,
IggyError> {
+ if self.roster_size.load(Ordering::Acquire) == 0 {
+ let response = timeout(
+ ROSTER_READ_TIMEOUT,
+ coordinator.send_poll_control(GET_CLUSTER_METADATA_CODE,
Bytes::new()),
+ )
+ .await
+ .map_err(|_| IggyError::TransientNotAccepted)??;
+ let metadata = ClusterMetadataResponse::decode_from(&response)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ if metadata.nodes.is_empty() {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ self.roster_size
+ .store(metadata.nodes.len(), Ordering::Release);
+ }
+ Ok(self.roster_size.load(Ordering::Acquire) > 1)
+ }
+
+ async fn send_routed(
+ &self,
+ coordinator: &T,
+ code: u32,
+ key: RouteKey,
+ payload: Bytes,
+ ) -> Result<Bytes, IggyError> {
+ let now = Instant::now();
+ let deadline = now + POLL_TIMEOUT;
+ let result = timeout_at(deadline, async {
+ let heartbeat_due = {
+ let mut next = self
+ .next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let due = next.is_some_and(|next| now >= next);
+ if next.is_none() || due {
+ *next = Some(now +
coordinator.get_heartbeat_interval().get_duration());
+ }
+ due
+ };
+ if heartbeat_due {
+ coordinator
+ .send_poll_control(PING_CODE, Bytes::new())
+ .await?;
+ }
+ let mut retry_interval = ROUTING_RETRY_INTERVAL;
+ loop {
+ let result = self.poll_once(coordinator, code, &key,
&payload).await;
+ if !matches!(result, Err(IggyError::TransientNotAccepted)) {
+ return result;
+ }
+ self.routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(&key);
+ if Instant::now() + retry_interval >= deadline {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ sleep(retry_interval).await;
+ retry_interval = (retry_interval *
2).min(ROUTING_RETRY_MAX_INTERVAL);
Review Comment:
The retry interval now doubles from 50 ms to a 1 second ceiling, which cuts
a refused poll from roughly 600 coordinator round trips over its 30 second
budget to roughly 30. The route is still removed on every
`TransientNotAccepted` retry at `core/sdk/src/poll_routing.rs:292`, so a
refusal unrelated to leadership still re-resolves a route that did not change.
##########
core/server/config.toml:
##########
@@ -726,7 +726,8 @@ pin_cores = true
# ~ the prepare queue depth of the planes the shard hosts ([metadata] on
# shard 0, [partition] elsewhere) times replica_count times directions.
# Both depths are tunable, so raising either raises the capacity needed here.
-inbox_capacity = 1024
+# Slots are allocated at startup per shard, even before traffic arrives.
+inbox_capacity = 65536
Review Comment:
`inbox_capacity` is still 65536 preallocated slots per shard, against a
comment whose own sizing rule does not produce that number.
##########
core/configs/src/server_config/partition.rs:
##########
@@ -115,7 +115,7 @@ fn default_offset_reservation_lease() -> NonZeroU32 {
}
/// Mirrors `partitions::EVICTED_RING_CAPACITY`.
-pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 4096;
+pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 65536;
Review Comment:
`DEFAULT_EVICTED_RING_CAPACITY` still equals `MAX_EVICTED_RING_CAPACITY`.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+const MAX_CACHED_ROUTES: usize = 4096;
+const MAX_DATA_CONNECTIONS: usize = 256;
+const POLL_TIMEOUT: Duration = Duration::from_secs(30);
+const ROUTING_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+const ROUTING_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(1);
+pub(crate) const ROSTER_READ_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub(crate) const fn is_poll_routing_code(code: u32) -> bool {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
+ self.next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ let mut routes = self
+ .routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
+ }
+
+ pub(crate) fn remember_credentials(&self, credentials: Credentials,
user_id: u32) {
+ *self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) =
Some((credentials, user_id));
+ }
+
+ pub(crate) fn forget_credentials(&self) {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ }
+
+ pub(crate) fn credentials(&self) -> Option<Credentials> {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .as_ref()
+ .map(|(credentials, _)| credentials.clone())
+ }
+
+ pub(crate) fn refresh_password(&self, user: &Identifier, new_password:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, password), user_id))
=
+ credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *password = SecretString::from(new_password.to_owned());
+ }
+ }
+
+ pub(crate) fn refresh_username(&self, user: &Identifier, new_username:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, _), user_id)) =
credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *username = new_username.to_owned();
+ }
+ }
+}
+
+impl<T: PollTransport> PollRouter<T> {
+ pub(crate) async fn poll(
+ &self,
+ coordinator: &T,
+ request: &PollMessagesRequest,
+ ) -> Result<Bytes, IggyError> {
+ if !request.auto_commit || !self.is_clustered(coordinator).await? {
+ return coordinator
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await;
+ }
+ let payload = request.to_bytes();
+ let parameters_size = request.strategy.encoded_size() +
size_of::<u32>() + size_of::<u8>();
+ let key = (
+ GET_POLL_ROUTING_CODE,
+ payload.slice(..payload.len() - parameters_size),
+ );
+ self.send_routed(coordinator, POLL_MESSAGES_ON_PRIMARY_CODE, key,
payload)
+ .await
+ }
Review Comment:
The shared wire prefix is still unasserted.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+const MAX_CACHED_ROUTES: usize = 4096;
+const MAX_DATA_CONNECTIONS: usize = 256;
+const POLL_TIMEOUT: Duration = Duration::from_secs(30);
+const ROUTING_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+const ROUTING_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(1);
+pub(crate) const ROSTER_READ_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub(crate) const fn is_poll_routing_code(code: u32) -> bool {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
+ self.next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ let mut routes = self
+ .routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
+ }
+
+ pub(crate) fn remember_credentials(&self, credentials: Credentials,
user_id: u32) {
+ *self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) =
Some((credentials, user_id));
+ }
+
+ pub(crate) fn forget_credentials(&self) {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ }
+
+ pub(crate) fn credentials(&self) -> Option<Credentials> {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .as_ref()
+ .map(|(credentials, _)| credentials.clone())
+ }
+
+ pub(crate) fn refresh_password(&self, user: &Identifier, new_password:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, password), user_id))
=
+ credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *password = SecretString::from(new_password.to_owned());
+ }
+ }
+
+ pub(crate) fn refresh_username(&self, user: &Identifier, new_username:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, _), user_id)) =
credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *username = new_username.to_owned();
+ }
+ }
+}
+
+impl<T: PollTransport> PollRouter<T> {
+ pub(crate) async fn poll(
+ &self,
+ coordinator: &T,
+ request: &PollMessagesRequest,
+ ) -> Result<Bytes, IggyError> {
+ if !request.auto_commit || !self.is_clustered(coordinator).await? {
+ return coordinator
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await;
+ }
+ let payload = request.to_bytes();
+ let parameters_size = request.strategy.encoded_size() +
size_of::<u32>() + size_of::<u8>();
+ let key = (
+ GET_POLL_ROUTING_CODE,
+ payload.slice(..payload.len() - parameters_size),
+ );
+ self.send_routed(coordinator, POLL_MESSAGES_ON_PRIMARY_CODE, key,
payload)
+ .await
+ }
+
+ pub(crate) async fn write_offset(
+ &self,
+ coordinator: &T,
+ code: u32,
+ payload: Bytes,
+ ) -> Result<Bytes, IggyError> {
+ if !self.is_clustered(coordinator).await? {
+ return coordinator.send_raw_with_response(code, payload).await;
+ }
+ let (_, route_size) =
+ GetConsumerOffsetRequest::decode(&payload).map_err(|_|
IggyError::InvalidCommand)?;
+ let key = (
+ GET_CONSUMER_OFFSET_ROUTING_CODE,
+ payload.slice(..route_size),
+ );
+ self.send_routed(coordinator, code, key, payload).await
+ }
+
+ pub(crate) async fn is_clustered(&self, coordinator: &T) -> Result<bool,
IggyError> {
+ if self.roster_size.load(Ordering::Acquire) == 0 {
+ let response = timeout(
+ ROSTER_READ_TIMEOUT,
+ coordinator.send_poll_control(GET_CLUSTER_METADATA_CODE,
Bytes::new()),
+ )
+ .await
+ .map_err(|_| IggyError::TransientNotAccepted)??;
+ let metadata = ClusterMetadataResponse::decode_from(&response)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ if metadata.nodes.is_empty() {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ self.roster_size
+ .store(metadata.nodes.len(), Ordering::Release);
+ }
+ Ok(self.roster_size.load(Ordering::Acquire) > 1)
+ }
+
+ async fn send_routed(
+ &self,
+ coordinator: &T,
+ code: u32,
+ key: RouteKey,
+ payload: Bytes,
+ ) -> Result<Bytes, IggyError> {
+ let now = Instant::now();
+ let deadline = now + POLL_TIMEOUT;
+ let result = timeout_at(deadline, async {
+ let heartbeat_due = {
+ let mut next = self
+ .next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let due = next.is_some_and(|next| now >= next);
+ if next.is_none() || due {
+ *next = Some(now +
coordinator.get_heartbeat_interval().get_duration());
+ }
+ due
+ };
+ if heartbeat_due {
+ coordinator
+ .send_poll_control(PING_CODE, Bytes::new())
+ .await?;
+ }
+ let mut retry_interval = ROUTING_RETRY_INTERVAL;
+ loop {
+ let result = self.poll_once(coordinator, code, &key,
&payload).await;
+ if !matches!(result, Err(IggyError::TransientNotAccepted)) {
+ return result;
+ }
+ self.routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(&key);
+ if Instant::now() + retry_interval >= deadline {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ sleep(retry_interval).await;
+ retry_interval = (retry_interval *
2).min(ROUTING_RETRY_MAX_INTERVAL);
+ }
+ })
+ .await;
+ match result {
+ Ok(result) => result,
+ Err(_) => {
+ self.routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(&key);
+ Err(IggyError::TransientNotCommitted)
+ }
+ }
+ }
+
+ async fn poll_once(
+ &self,
+ coordinator: &T,
+ code: u32,
+ key: &RouteKey,
+ payload: &Bytes,
+ ) -> Result<Bytes, IggyError> {
+ let route = self.route(coordinator, key, payload).await?;
+ let slot = {
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ self.validate_route(&route)?;
+ if let Some(slot) = connections.get(&route.endpoint) {
+ Arc::clone(slot)
+ } else {
+ if connections.len() >= MAX_DATA_CONNECTIONS {
+ error!(endpoint = route.endpoint, protocol = %T::PROTOCOL,
limit = MAX_DATA_CONNECTIONS, "primary poll connection pool is full");
+ return Err(IggyError::InvalidConfiguration);
+ }
+ let slot = Arc::default();
+ connections.insert(route.endpoint.clone(), Arc::clone(&slot));
+ slot
+ }
+ };
+ let mut connection = slot.lock().await;
Review Comment:
One connection slot per endpoint, still locked across the attach and the
poll, so a client gets at most one concurrent auto-commit poll per node.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+const MAX_CACHED_ROUTES: usize = 4096;
+const MAX_DATA_CONNECTIONS: usize = 256;
+const POLL_TIMEOUT: Duration = Duration::from_secs(30);
+const ROUTING_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+const ROUTING_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(1);
+pub(crate) const ROSTER_READ_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub(crate) const fn is_poll_routing_code(code: u32) -> bool {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
Review Comment:
`clear_session` still leaves `roster_size` set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
