numinnex commented on code in PR #4506: URL: https://github.com/apache/iggy/pull/4506#discussion_r4260704827
########## core/sdk/src/session_owner/lanes.rs: ########## @@ -0,0 +1,605 @@ +// 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. + +//! A lane admits one mutation at a time, in arrival order. There is one lane +//! for metadata and one for each partition history. A mutation whose outcome +//! is unknown stays retained in its lane, and the lane admits nothing else +//! until a resolve or an abandon frees it. So a retained operation is always +//! the newest operation of its lane, and its replay can never age out. + +use super::SessionOwner; +use crate::driver::PartitionKey; +use crate::links::Frame; +use crate::vsr::encode_request_header; +use bytes::Bytes; +use iggy_binary_protocol::consensus::{Operation, operation_for_code}; +use iggy_binary_protocol::primitives::partition_history::PartitionContext; +use iggy_common::IggyError; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, OwnedMutexGuard}; +use tokio::time::{Instant, timeout_at}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum LaneKey { + Metadata, + /// The incarnation names one history of the partition for every spelling + /// of its stream and topic. + Partition { + incarnation: u64, + partition_id: u32, + }, +} + +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. + 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, + operation if operation.is_partition() => Some(Self::Partition { + incarnation: context.incarnation, + partition_id, + }), + _ => Some(Self::Metadata), + } + } +} + +pub(super) type Lanes = HashMap<LaneKey, Lane>; + +#[derive(Debug, Default)] +pub(super) struct Lane { + gate: Arc<Mutex<()>>, + /// The tickets that wait for the gate, hold it or resume. The lane leaves + /// the table when none is left and nothing is retained. + tickets: usize, + retained: Option<RetainedOperation>, +} + +impl Lane { + fn holds(&self, op: u64) -> bool { + self.retained + .as_ref() + .is_some_and(|retained| retained.op() == op) + } +} + +/// A frozen mutation. Every attempt and every replay sends the same frame. +#[derive(Debug, Clone)] +pub(crate) struct RetainedOperation { + pub(crate) code: u32, + pub(crate) frame: Frame, + /// The partition of a partition write, `None` for any other request. A + /// replay looks up its route by this key. + pub(crate) partition: Option<PartitionKey>, +} + +impl RetainedOperation { + pub(crate) fn op(&self) -> u64 { + self.frame.header.request + } +} + +/// The right of one call to send in its lane. Dropping the ticket frees the +/// lane, except that a frame which left the client without a definitive +/// outcome stays retained. +#[derive(Debug)] +pub(crate) struct LaneTicket { + owner: Arc<SessionOwner>, + /// `None` for a call that takes no lane. + key: Option<LaneKey>, + guard: Option<OwnedMutexGuard<()>>, + operation: Option<RetainedOperation>, + fence: u64, + /// A resume replays an operation that its lane still retains. + resumed: bool, + sent: bool, + done: bool, +} + +impl SessionOwner { + /// Waits for the lane of `key` in arrival order. A lane that retains an + /// operation fails at once with `LaneBlocked`, and a lane still busy at + /// `deadline` fails with `TransientNotAccepted`. Neither failure takes a + /// request id. + pub(crate) async fn enter_lane( + self: &Arc<Self>, + key: Option<LaneKey>, + deadline: Instant, + ) -> Result<LaneTicket, IggyError> { + let Some(key) = key else { + return Ok(LaneTicket::new(Arc::clone(self), None, false)); + }; + let (mut ticket, gate) = { + let mut lanes = self.lanes(); + let lane = lanes.entry(key).or_default(); Review Comment: nit: `entry(key).or_default()` allocates a new `Arc<Mutex<()>>` for every uncontended mutation because the lane is removed when idle; `route_cache.rs:203` also scans up to 4096 entries per insert at capacity. Cold today (driver unused). ########## core/sdk/src/driver/route_cache.rs: ########## @@ -0,0 +1,379 @@ +// 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. + +//! The routes of one client. A route has no age limit. It goes stale after +//! a refusal or a failure, and a stale route keeps its view as the lowest +//! view that its next lookup accepts. So a lookup never goes back to an +//! older primary of the same partition history. + +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 std::collections::HashMap; + +/// The most routes that one client keeps. A new route beyond it evicts the +/// oldest lookup. +pub(crate) const MAX_CACHED_ROUTES: usize = 4096; + +/// A partition as its caller spelled it. A stream or topic named by number +/// and the same one named by name are two keys. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct PartitionKey { + pub(crate) stream_id: Identifier, + pub(crate) topic_id: Identifier, + pub(crate) partition_id: u32, +} + +impl From<&PartitionTarget<'_>> for PartitionKey { + fn from(target: &PartitionTarget<'_>) -> Self { + Self { + stream_id: target.stream_id.clone(), + topic_id: target.topic_id.clone(), + partition_id: target.partition_id, + } + } +} + +impl TryFrom<&PartitionKey> for GetPartitionRouteRequest { + type Error = IggyError; + + fn try_from(key: &PartitionKey) -> Result<Self, Self::Error> { + Ok(Self { + stream_id: identifier_to_wire(&key.stream_id)?, + topic_id: identifier_to_wire(&key.topic_id)?, + partition_id: key.partition_id, + }) + } +} + +#[derive(Debug)] +struct Entry { + route: PartitionRouteResponse, + /// The session fence at the lookup. A route serves only that session. + fence: u64, + stale: bool, + /// The lowest view that the next lookup accepts. + floor: u32, + /// The replica that the next lookup asks first. The ids after it follow. + ask_first: Option<u8>, + /// Orders the entries for eviction. + lookup: u64, +} + +#[derive(Debug, Default)] +pub(crate) struct RouteCache { + entries: HashMap<PartitionKey, Entry>, + lookups: u64, +} + +impl RouteCache { + /// The route of `key`, unless it is stale, from another session, or + /// older than the metadata `watermark` that the session saw. + pub(crate) fn usable( + &self, + key: &PartitionKey, + fence: u64, + watermark: u64, + ) -> Option<PartitionRouteResponse> { + self.entries Review Comment: nit: `usable` requires `route.metadata_op >= watermark`, and every metadata reply the session observes raises the watermark, so one metadata write invalidates every cached route and each partition is looked up again. Fencing only on commits that touched the route's topic would avoid the extra round trips once the driver is live. ########## core/common/src/traits/binary_transport.rs: ########## @@ -33,6 +33,27 @@ 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. Review Comment: **warning**: `RequestTarget` / `PartitionTarget` (and `Client::resolve_operation` / `abandon_operation` in `client.rs:84`) become stable public API with only the unused driver behind them (`driver.rs` is `allow(dead_code)`), every shipped client answers `FeatureUnavailable`, and `PartitionTarget` has pub fields without `#[non_exhaustive]`. Suggest `pub(crate)` in the sdk for the targets and `#[doc(hidden)]` on the two methods until the driver is wired. ########## core/common/src/traits/client.rs: ########## @@ -79,4 +80,25 @@ pub trait Client: /// /// HTTP keeps no connection, so it publishes no events. async fn subscribe_events(&self) -> Receiver<DiagnosticEvent>; + + /// Replays the retained operation `op` until it commits or `deadline` + /// passes. `op` is the request id that `OperationUncertain` or + /// `LaneBlocked` carries. + /// + /// `Ok(())` means that the operation committed exactly once. After + /// `OperationUncertain(op)` the outcome is still unknown, and the Review Comment: nit: after a session reset (logout or reconnect under a new session) retained operations are dropped, so a later `resolve_operation(op)` returns `OperationNotRetained` while the outcome is still unknown. Worth saying here that after a reset it means "unknown and unrecoverable", not "settled". ########## core/binary_protocol/src/consensus/header.rs: ########## @@ -49,6 +49,8 @@ //! headers are sealed on neither side, so SDKs are untouched. Review Comment: nit: the compatibility note stops at command 30 and doesn't mention `SessionRetirementProgress` (31), the new `ForwardCompletion` (32) or the `partition_incarnation` carved from reserved bytes in 11 control headers, and "SDKs are untouched" no longer holds for this PR. Related: the PR breaks the client wire, the replica wire and the Rust SDK API, so the title wants a `!` and the `breaking:client-to-replica-protocol`, `breaking:replica-to-replica-protocol` and `breaking:rust-sdk` labels. ########## core/binary_protocol/src/requests/messages/get_partition_route.rs: ########## @@ -20,16 +20,17 @@ use bytes::{BufMut, BytesMut}; use crate::codec::{WireDecode, WireEncode, read_u32_le}; use crate::{WireError, WireIdentifier}; -/// Partition context discovery authorized by send permission, without exposing topic details. -/// The response is a `PartitionContext` captured before a new send is admitted. +/// Route lookup for one partition, authorized by send or poll permission on its Review Comment: nit: "without exposing topic details" isn't accurate: the reply carries `partitions_count`, the view and the primary replica, and poll permission alone is enough to ask. ########## core/shard/src/lib.rs: ########## @@ -4511,17 +4581,38 @@ where } } + fn partition_incarnation_matches( + &self, + partition: &IggyPartition<B, SB>, + partition_incarnation: u64, + frame: &'static str, + ) -> bool { + if partition_incarnation == partition.created_revision() { + return true; + } + tracing::debug!( Review Comment: nit: incarnation-mismatch drops are logged only at `debug!`, so an operator who skipped the all-at-once upgrade just sees a partition group that can't elect. A rate-limited counter (labelled for incarnation 0) would make it visible without flooding logs. ########## core/shard/src/lib.rs: ########## @@ -4841,10 +4934,14 @@ where && let Some(from_op) = partition_view_divergence_from(self.id, partition, pending.as_ref()) { + partition.consensus().settle_view_log_proof(); Review Comment: nit: the view-log proof is settled here (and at :6083) before the deferred truncate runs. If `truncate_resident_from` errors after `transition.take()`, `fail_io_work(.., false)` can return without fatal and `commit_journal` is no longer fenced over the unproven entries. Latent today; settling in the `ViewApplied` arm after a successful truncate keeps one owner of the proof bit. ########## core/partitions/src/io.rs: ########## @@ -1295,7 +1331,14 @@ impl OffsetIoJob { impl OffsetDeleteIoJob { #[allow(clippy::future_not_send)] pub(crate) async fn execute(self) -> Result<bool, IggyError> { - delete_persisted_offset(&self.path).await + let removed = delete_persisted_offset(&self.path).await?; + // After the unlink only as defense in depth. WAL replay of the delete, + // boot dropping a marker beside any offset file and the transfer bound + // `emptied_at <= commit_op` carry the crash safety. + if let Some((marker, op)) = &self.emptied_marker { Review Comment: nit: the marker is described as defense in depth, but a `replace_file` failure after the authoritative unlink propagates and goes fatal. The comment (also at :1269) overclaims too: WAL replay of the delete never rewrites the marker. Suggest warn-and-drop plus a corrected comment. ########## core/partitions/src/iggy_partition.rs: ########## @@ -9919,6 +10172,8 @@ where self.publish_consumer_offset_effect(*effect); self.apply_consumer_offset_commit(effect.pending); } + self.durable_consumer_offsets Review Comment: nit: `emptied_at` is written back unconditionally from the snapshot taken in `select_commit`; a no-ack store that publishes in between gets overwritten with the stale value (safe, only refuses more, but memory, disk and offer diverge). Write it only when this run's plan set `emptied_at`. ########## core/metadata/src/stm/lifecycle.rs: ########## @@ -616,7 +592,7 @@ impl StreamsSnapshot { if let Some(pending) = assignment.pending && ((pending.owner.client_id == 0) != (pending.owner.session == 0) || pending.owner.generation == 0 - || pending.owner.generation > group.generation + || pending.owner.generation != pending.metadata_op Review Comment: nit: owner generation now means the metadata op and validation requires it, but `SNAPSHOT_FORMAT_VERSION` in `stm/snapshot.rs` stays 11. Old data dirs are already refused by the storage format bump, so this is hygiene: bump to 12 with a doc line so a mismatched snapshot fails as `UnsupportedFormatVersion` rather than `InvalidState`. ########## core/common/src/types/cluster/node.rs: ########## @@ -23,6 +23,10 @@ use std::fmt::Display; #[derive(Debug, Deserialize, Serialize, Clone)] pub struct ClusterNode { + /// The node's configured VSR replica id. Key nodes by it: the position in Review Comment: nit: the doc says to key nodes by `replica_id`, but the C++ FFI `ClusterNode` drops it (`foreign/cpp/src/type_conversion.rs:296`) and the CLI `get_cluster_metadata` table doesn't print it. It is also a required serde field without `#[serde(default)]`, so a new Rust HTTP client can't read an older server's JSON. -- 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]
