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 cf59a0935712e3fdcd178246aa2196bf7ac3726d Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 13 18:26:07 2026 +0200 fix(sdk): keep inactive routing APIs private The operation driver is not active yet, so its routing targets should not form part of the public common API. Keep them inside the driver and hide unavailable recovery methods. Document lost recovery state after session reset and the auto-commit lane exemption. --- core/common/src/lib.rs | 1 - core/common/src/traits/binary_transport.rs | 21 --------------------- core/common/src/traits/client.rs | 6 ++++++ core/sdk/src/driver.rs | 22 +++++++++++++++++++--- core/sdk/src/driver/route_cache.rs | 4 +++- core/sdk/src/session_owner/lanes.rs | 3 ++- 6 files changed, 30 insertions(+), 27 deletions(-) diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index 56bbabfa1..71e11bf16 100644 --- a/core/common/src/lib.rs +++ b/core/common/src/lib.rs @@ -81,7 +81,6 @@ pub use iggy_binary_protocol::responses::messages::{ }; pub use traits::binary_client::BinaryClient; pub use traits::binary_transport::BinaryTransport; -pub use traits::binary_transport::{PartitionTarget, RequestTarget}; pub use traits::binary_transport::{VsrSessionControl, VsrSessionSealed, decode_session_binding}; pub use traits::client::Client; pub use traits::cluster_client::ClusterClient; diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs index ac8ffcb44..3ee42f044 100644 --- a/core/common/src/traits/binary_transport.rs +++ b/core/common/src/traits/binary_transport.rs @@ -33,27 +33,6 @@ use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::responses::users::LoginRegisterResponse; use std::sync::Arc; -/// Where a binary client sends one request. -#[derive(Debug, Clone, Copy)] -pub enum RequestTarget<'a> { - /// The metadata primary. - Metadata, - /// The primary of one partition. - Partition(PartitionTarget<'a>), -} - -/// A partition request as its caller spelled the partition. -#[derive(Debug, Clone, Copy)] -pub struct PartitionTarget<'a> { - pub stream_id: &'a Identifier, - pub topic_id: &'a Identifier, - pub partition_id: u32, - /// The context of the poll reply that the request continues. `None` - /// takes the context of the route. Every retry keeps the context that - /// the first attempt sent. - pub context: Option<PartitionContext>, -} - #[async_trait] pub trait BinaryTransport { /// Gets the state of the client. diff --git a/core/common/src/traits/client.rs b/core/common/src/traits/client.rs index c7ab5a85a..a93a8addd 100644 --- a/core/common/src/traits/client.rs +++ b/core/common/src/traits/client.rs @@ -90,6 +90,11 @@ pub trait Client: /// operation stays retained. An operation that is not retained gives /// `OperationNotRetained(op)`. A client that retains no operations /// returns `FeatureUnavailable`. + /// + /// A session reset drops retained operations without settling them. + /// `OperationNotRetained(op)` after a reset means the outcome is still + /// unknown and can no longer be recovered through this client. + #[doc(hidden)] async fn resolve_operation(&self, _op: u64, _deadline: Instant) -> Result<(), IggyError> { Err(IggyError::FeatureUnavailable) } @@ -98,6 +103,7 @@ pub trait Client: /// of `op` stays unknown. An operation that is not retained gives /// `OperationNotRetained(op)`. A client that retains no operations /// returns `FeatureUnavailable`. + #[doc(hidden)] async fn abandon_operation(&self, _op: u64) -> Result<(), IggyError> { Err(IggyError::FeatureUnavailable) } diff --git a/core/sdk/src/driver.rs b/core/sdk/src/driver.rs index e66edcb2d..e58c21c7b 100644 --- a/core/sdk/src/driver.rs +++ b/core/sdk/src/driver.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! The operation driver sends every request of a binary client. One deadline +//! The operation driver coordinates a request across replica links. One deadline //! bounds the lane wait, the lookups, the attempts and the pauses of a call. //! A frame that can have taken effect is never frozen again, so a retry never //! applies a request twice. @@ -38,7 +38,7 @@ use iggy_binary_protocol::requests::messages::GetPartitionRouteRequest; use iggy_binary_protocol::responses::messages::PartitionRouteResponse; use iggy_binary_protocol::responses::users::LoginRegisterResponse; use iggy_binary_protocol::{WireDecode, WireEncode}; -use iggy_common::{ClientState, IggyError, RequestTarget}; +use iggy_common::{ClientState, Identifier, IggyError}; use route_cache::RouteCache; use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use tokio::time::{Instant, timeout_at}; @@ -52,6 +52,23 @@ pub(crate) use poll_router::{ }; pub(crate) use route_cache::PartitionKey; +/// Where a binary client sends one request. +#[derive(Debug, Clone, Copy)] +pub(crate) enum RequestTarget<'a> { + Metadata, + Partition(PartitionTarget<'a>), +} + +/// A partition request as its caller spelled the partition. +#[derive(Debug, Clone, Copy)] +pub(crate) struct PartitionTarget<'a> { + pub(crate) stream_id: &'a Identifier, + pub(crate) topic_id: &'a Identifier, + pub(crate) partition_id: u32, + /// `None` captures the route context. Retries retain the first context sent. + pub(crate) context: Option<PartitionContext>, +} + /// Sends the requests of one client over its replica links. pub(crate) struct OperationDriver { session: Arc<SessionOwner>, @@ -681,7 +698,6 @@ mod tests { CREATE_STREAM_CODE, GET_CLUSTER_METADATA_CODE, GET_STREAMS_CODE, POLL_MESSAGES_CODE, STORE_CONSUMER_OFFSET_CODE, }; - use iggy_common::{Identifier, PartitionTarget}; use std::iter::repeat_n; use std::time::Duration; use tokio::time::{sleep, timeout}; diff --git a/core/sdk/src/driver/route_cache.rs b/core/sdk/src/driver/route_cache.rs index a47ccec54..d70ed2854 100644 --- a/core/sdk/src/driver/route_cache.rs +++ b/core/sdk/src/driver/route_cache.rs @@ -23,9 +23,11 @@ use iggy_binary_protocol::requests::messages::GetPartitionRouteRequest; use iggy_binary_protocol::responses::messages::PartitionRouteResponse; use iggy_common::wire_conversions::identifier_to_wire; -use iggy_common::{Identifier, IggyError, PartitionTarget}; +use iggy_common::{Identifier, IggyError}; use std::collections::HashMap; +use super::PartitionTarget; + /// The most routes that one client keeps. A new route beyond it evicts the /// oldest lookup. pub(crate) const MAX_CACHED_ROUTES: usize = 4096; diff --git a/core/sdk/src/session_owner/lanes.rs b/core/sdk/src/session_owner/lanes.rs index cc49c9f2f..5e7a65ea7 100644 --- a/core/sdk/src/session_owner/lanes.rs +++ b/core/sdk/src/session_owner/lanes.rs @@ -48,7 +48,8 @@ pub(crate) enum LaneKey { impl LaneKey { /// The lane of `code`. A write to a partition takes the lane of the /// incarnation that its context carries. Reads, Register and Logout take - /// no lane. + /// no lane. This includes auto-commit polls: their offset effect does not + /// acquire a client mutation lane. pub(crate) fn of(code: u32, context: &PartitionContext, partition_id: u32) -> Option<Self> { match operation_for_code(code) { Operation::NonReplicated | Operation::Register | Operation::Logout => None,
