This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch peer_shard_read_stale_state in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9d2e8ee05b2362a72f4ab16a0c17056f406c6bca Author: Grzegorz Koszyk <[email protected]> AuthorDate: Tue Sep 1 14:51:44 2026 +0200 fix(cluster): serve metadata reads at or above the client's own writes --- .../org/apache/iggy/bdd/BasicMessagingSteps.java | 8 +- bdd/rust/tests/steps/streams.rs | 31 +- core/metadata/src/impls/metadata.rs | 59 ++++ core/server/src/bootstrap.rs | 19 +- core/server/src/dispatch.rs | 137 +++++++- core/server/src/http.rs | 2 +- core/server/src/http/error.rs | 26 +- core/server/src/http/extractor.rs | 18 +- core/server/src/http/reads.rs | 210 +++++++++++- core/server/src/http/reply.rs | 2 +- core/server/src/http/session.rs | 118 +++++-- core/server/src/http/state.rs | 40 ++- core/server/src/http/submit.rs | 14 +- core/server/src/session_manager.rs | 89 ++++- core/simulator/src/client.rs | 47 ++- core/simulator/src/lib.rs | 376 ++++++++++++++++++++- core/simulator/src/replica.rs | 11 +- 17 files changed, 1134 insertions(+), 73 deletions(-) diff --git a/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java b/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java index 0fc05c28c..c5439f201 100644 --- a/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java +++ b/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java @@ -142,8 +142,14 @@ public class BasicMessagingSteps { @Then("getting the stream by its numeric ID should return no stream") public void getStreamReturnsNoStream() { + // The assertion is "not the stream we deleted", not "nothing at this id": + // the server hands out the lowest free stream id, so once these scenarios + // run concurrently against one server a fresh create can legitimately + // occupy the deleted stream's id. Optional<StreamDetails> stream = getClient().streams().getStream(context.lastStreamId); - assertTrue(stream.isEmpty(), "Deleted stream should not be returned"); + assertTrue( + stream.isEmpty() || !stream.get().name().equals(context.lastStreamName), + "Deleted stream should not be returned"); } @When("I create a topic with name {string} in stream {int} with {int} partitions") diff --git a/bdd/rust/tests/steps/streams.rs b/bdd/rust/tests/steps/streams.rs index ba1248f3f..f94ef4027 100644 --- a/bdd/rust/tests/steps/streams.rs +++ b/bdd/rust/tests/steps/streams.rs @@ -18,11 +18,6 @@ use crate::common::global_context::GlobalContext; use cucumber::{given, then, when}; use iggy::prelude::{Identifier, StreamClient, StreamUpdateOptions}; -use std::time::Duration; -use tokio::time::{Instant, sleep}; - -const METADATA_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(2); -const METADATA_CONVERGENCE_POLL: Duration = Duration::from_millis(10); #[given("I have no streams in the system")] pub async fn given_no_streams(world: &mut GlobalContext) { @@ -128,18 +123,20 @@ pub async fn when_delete_stream_by_numeric_id(world: &mut GlobalContext) { #[then("getting the stream by its numeric ID should return no stream")] pub async fn then_get_stream_returns_no_stream(world: &mut GlobalContext) { - let deadline = Instant::now() + METADATA_CONVERGENCE_TIMEOUT; - loop { - get_stream_by_numeric_id(world).await; - if world.last_stream_name.is_none() { - return; - } - assert!( - Instant::now() < deadline, - "Deleted stream should not be returned after {METADATA_CONVERGENCE_TIMEOUT:?}" - ); - sleep(METADATA_CONVERGENCE_POLL).await; - } + // Read before the get overwrites it. The assertion is "not the stream we + // deleted", not "nothing at this id": `IdSlab::insert` hands out the lowest + // free key and these scenarios share one server, so a concurrent create can + // legitimately occupy the deleted stream's id. + let deleted = world + .last_stream_name + .clone() + .expect("Stream should have been created"); + get_stream_by_numeric_id(world).await; + assert_ne!( + world.last_stream_name.as_ref(), + Some(&deleted), + "Deleted stream should not be returned" + ); } async fn create_stream(world: &mut GlobalContext, stream_name: &str) { diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 6ff355e47..e0d7e5775 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -66,6 +66,8 @@ use std::cell::{Cell, RefCell}; use std::mem::size_of; use std::path::Path; use std::rc::Rc; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use tracing::{debug, error, info, warn}; fn freeze_client_reply( @@ -769,6 +771,21 @@ pub struct IggyMetadata<C, J, S, M, SB = PingPongSuperblock> { /// whole snapshot on shard 0's pump, and hands each requester its own /// multi-MB copy. transfer_offer_cache: RefCell<Option<Rc<StateTransferOffer>>>, + /// Highest metadata op whose apply has been PUBLISHED on this node, shared + /// by every shard. + /// + /// `consensus.commit_min()` answers the same question but exists only on + /// shard 0, so a read served by a peer shard has no way to tell whether the + /// node caught up to an op its client already saw committed. One + /// process-wide atomic does, for one `Acquire` load on the read fast path. + /// + /// Written `Release` right after each apply's `publish()`, read `Acquire`; + /// observing `>= op` therefore happens-after that publish, so a following + /// left-right `enter()` is guaranteed to see the op. `fetch_max` rather + /// than `store` because three writers move it -- the commit loop, the + /// recovery seed, and a state-transfer install -- and only monotonicity + /// makes their order irrelevant. + applied_frontier: Arc<AtomicU64>, } impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> @@ -808,11 +825,41 @@ where commit_notifier: RefCell::new(None), client_table_frontier: Cell::new(0), transfer_offer_cache: RefCell::new(None), + applied_frontier: Arc::new(AtomicU64::new(0)), } } } impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> { + /// Share one process-wide applied frontier with every other shard. + /// + /// Consumed at construction rather than swapped in later: a shard that + /// served a read against its own private cell would gate on a number that + /// never moves. Shard 0 mints the cell in bootstrap, before any shard is + /// built, and hands each shard a clone. + #[must_use] + pub fn with_applied_frontier(mut self, applied_frontier: Arc<AtomicU64>) -> Self { + self.applied_frontier = applied_frontier; + self + } + + /// Highest metadata op this NODE has applied and published, readable on + /// every shard. Reads gate on it so a client cannot be served state older + /// than a write it already saw acked. + #[must_use] + pub fn applied_frontier(&self) -> u64 { + self.applied_frontier.load(Ordering::Acquire) + } + + /// Publish `op` as applied. Monotone, so a lower value is a no-op. + /// + /// Must run AFTER the apply's `publish()` and, on the commit path, in the + /// same await-free region as `advance_commit_min`: a reader that sees the + /// frontier must be guaranteed to see the op's effects. + pub fn advance_applied_frontier(&self, op: u64) { + self.applied_frontier.fetch_max(op, Ordering::Release); + } + /// Slot capacity of the LIVE client table, i.e. the largest transferred /// table this replica can absorb. /// @@ -1839,6 +1886,7 @@ where if snapshot_seq > consensus.sequencer().current_sequence() { consensus.sequencer().set_sequence(snapshot_seq); } + self.advance_applied_frontier(snapshot_seq); } // Before the superblock write, so the durable record carries the frontier // this transfer just established rather than the pre-transfer one. @@ -2827,6 +2875,10 @@ where reply }; consensus.advance_commit_min(prepare_header.op); + // Paired with the counter bump, and before the reply leaves: a + // client that holds this reply may re-home onto any shard and read, + // and the read gate admits it only once the frontier covers the op. + self.advance_applied_frontier(prepare_header.op); emit_sim_event(SimEventKind::OperationCommitted, &event); // Fire subscriber BEFORE wire send. Slot already updated @@ -3574,6 +3626,7 @@ where prepare, ); consensus.advance_commit_min(op); + self.advance_applied_frontier(op); debug!("commit_journal: committed op={op}"); } } @@ -5672,6 +5725,12 @@ mod tests { journal_handle.header(1).is_some() && journal_handle.header(2).is_some(), "ops at or below the floor stay for the walk and tail repair" ); + assert_eq!( + md.applied_frontier(), + SNAPSHOT_SEQ, + "the snapshot IS ops up to its sequence applied, so the read gate has \ + to admit reads at the floor the install jumped to" + ); } #[compio::test] diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 4320c3bc5..5066c147b 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -802,6 +802,10 @@ pub fn bootstrap( // Shared metadata-group view: written by shard 0's publisher task, read by // every shard's cluster-metadata roster so leader marking works off-shard. let metadata_view = Arc::new(AtomicU64::new(crate::cluster_meta::METADATA_VIEW_UNKNOWN)); + // Shared applied-metadata frontier: shard 0's commit path advances it, every + // shard's read gate reads it. Minted here, before any shard exists, because + // a shard holding a private cell would gate reads on a number nothing moves. + let metadata_applied_frontier = Arc::new(AtomicU64::new(0)); // Every shard's metric handles, minted before the threads spawn: each // shard bumps its own entry, and shard 0's HTTP scrape endpoint registers // the whole set (counters are Arc-backed, so cross-thread reads see the @@ -842,6 +846,7 @@ pub fn bootstrap( }; let metadata_view_for_shard = Arc::clone(&metadata_view); + let applied_frontier_for_shard = Arc::clone(&metadata_applied_frontier); let shard_metrics_for_shard = shard_metrics_all.clone(); let handle = match thread::Builder::new() .name(format!("shard-{shard_id}")) @@ -860,6 +865,7 @@ pub fn bootstrap( barrier_for_shard, owner_table_for_shard, metadata_view_for_shard, + applied_frontier_for_shard, shard_metrics_for_shard, ) }) { @@ -925,6 +931,7 @@ fn run_shard_thread( barrier: BootstrapBarrier, owner_table: Arc<ReplicaOwnerTable>, metadata_view: Arc<AtomicU64>, + metadata_applied_frontier: Arc<AtomicU64>, shard_metrics_all: Vec<ShardMetrics>, ) -> Result<(), ServerError> { // Armed for the whole thread body: a post-spawn error `?` or a panic @@ -968,6 +975,7 @@ fn run_shard_thread( barrier, owner_table, metadata_view, + metadata_applied_frontier, shard_metrics_all, )) .await @@ -997,6 +1005,7 @@ async fn shard_main( barrier: BootstrapBarrier, owner_table: Arc<ReplicaOwnerTable>, metadata_view: Arc<AtomicU64>, + metadata_applied_frontier: Arc<AtomicU64>, shard_metrics_all: Vec<ShardMetrics>, ) -> Result<(), ServerError> { let topology = resolve_tcp_topology(config, replica_id)?; @@ -1148,7 +1157,15 @@ async fn shard_main( superblock_for_metadata, mux_stm, Some(PathBuf::from(&config.system.path)), - ); + ) + .with_applied_frontier(metadata_applied_frontier); + // Recovery already replayed the committed WAL prefix into the state + // machine, so the frontier resumes where the commit walk will rather than + // at zero -- otherwise every read on a rebooted node parks until its + // deadline. No-op on peer shards, which share shard 0's cell. + if let Some(consensus) = metadata.consensus.as_ref() { + metadata.advance_applied_frontier(consensus.commit_min()); + } // Size the VSR client table before listeners bind and any client registers. // Must precede the recovered-table install below: the setter rebuilds the // table from scratch, so running it afterwards would drop every resumed diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index 0a494f524..8a3a51e68 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -37,6 +37,7 @@ use crate::dispatch::authz::{ authorize_default_read, authorize_partition_op, authorize_partition_read, authorize_uid, send_deny_reply, send_non_replicated_deny, send_unbound_deny_reply, }; +use crate::http::reply::transient_code; use crate::login_register::LoginRegisterError; use crate::pat::maybe_rewrite_pat_request; use crate::responses::{ @@ -60,10 +61,10 @@ use consensus::{ }; use iggy_binary_protocol::PrepareHeader; use iggy_binary_protocol::codes::{ - GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, - GET_ME_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_SNAPSHOT_FILE_CODE, GET_STATS_CODE, - LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE, POLL_MESSAGES_CODE, - SYNC_CONSUMER_GROUP_CODE, + DESCRIBE_OPTIONS_CODE, GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CLUSTER_METADATA_CODE, + GET_CONSUMER_OFFSET_CODE, GET_ME_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_SNAPSHOT_FILE_CODE, + GET_STATS_CODE, LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE, + POLL_MESSAGES_CODE, SYNC_CONSUMER_GROUP_CODE, }; use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy; @@ -92,8 +93,9 @@ use iggy_binary_protocol::{ AckLevel, ClientVersionInfo, Command, ConsensusHeader, EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome, ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome, ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, KIND_CONSUMER_GROUP, - MAX_PARTITIONS_PER_REQUEST, Operation, ProtocolVersion, RequestHeader, RoutedRequestHeader, - WireDecode, WireEncode, WireIdentifier, WireOptions, is_protocol_compatible, + MAX_PARTITIONS_PER_REQUEST, Operation, ProtocolVersion, ReplyHeader, RequestHeader, + RoutedRequestHeader, WireDecode, WireEncode, WireIdentifier, WireOptions, + is_protocol_compatible, }; use iggy_common::{ IggyByteSize, IggyError, MaxTopicSize, PollingStrategy, SnapshotCompression, @@ -1356,6 +1358,13 @@ async fn handle_client_request<B, MJ, S, SB>( // shard 0 can't route by the consensus client id (no home-shard bits). match submit_client_request_on_owner(shard, request).await { Some(reply) => { + // Recorded before the reply reaches the socket, so a read the client + // sends the instant it decodes this frame already sees the mark. + if let Some(commit) = committed_reply_commit(&reply) { + sessions + .borrow_mut() + .record_metadata_watermark(transport_client_id, commit); + } // The raw PAT token never enters consensus (it is non-deterministic // and secret), so the committed reply body is empty. Substitute the // raw-token response here, on the minting client's home shard, using @@ -1607,6 +1616,107 @@ pub(crate) async fn dispatch_partition_request<B, MJ, S, SB>( shard.dispatch(request.into_generic()); } +/// Poll cadence while a read waits for this node's applied metadata frontier. +/// The consensus tick, so a node one commit behind resumes on the next commit +/// broadcast rather than a tick later. +const READ_FRONTIER_POLL: Duration = Duration::from_millis(10); + +/// Polls a held read is given before it fails retryable: 3s at the cadence +/// above. Long enough to ride out a view change, far below the SDK's 30s +/// request budget, inside which it replays the same id on the same connection. +const READ_FRONTIER_MAX_POLLS: u32 = 300; + +/// Hold a local metadata read until this node has applied everything the +/// connection was told committed. +/// +/// A committed reply hands the client an op number; answering its next read +/// from a state machine below that op contradicts the frame the client is +/// holding. The lag is real on a node whose commit walk trails the client's +/// epoch -- a backup that forwarded the client's register binds a committed +/// session while its own `commit_journal` is still behind it (see +/// [`crate::auth`]) -- so the gate is not about peer shards: every shard of a +/// node reads one shared frontier and gates identically. +/// +/// Fast path is a single `Acquire` load and no await, which is what keeps an +/// uncontended read shared-nothing. Otherwise it polls the bus timer (virtual +/// under the simulator, wall clock in production) from inside the +/// per-connection drain task, so only this connection waits. Expiry fails +/// loud and retryable rather than serving state the client already saw +/// replaced; the log carries both numbers so a frontier that stopped moving +/// is visible instead of showing up as a hang. +#[allow(clippy::future_not_send)] +async fn await_metadata_read_frontier<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, + sessions: &Rc<RefCell<SessionManager>>, + transport_client_id: u128, +) -> Result<(), IggyError> +where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + // Own statement: the borrow has to be released before the poll below, which + // awaits on the same task the session manager's mutators run on. + let watermark = sessions.borrow().metadata_watermark(transport_client_id); + let metadata = shard.plane.metadata(); + if metadata.applied_frontier() >= watermark { + return Ok(()); + } + for _ in 0..READ_FRONTIER_MAX_POLLS { + shard.bus.sleep(READ_FRONTIER_POLL).await; + if metadata.applied_frontier() >= watermark { + return Ok(()); + } + } + warn!( + frontier = metadata.applied_frontier(), + watermark, "metadata read frontier unreached past deadline; failing the read retryable" + ); + Err(IggyError::TransientNotCommitted) +} + +/// The commit position a COMMITTED metadata reply carries, or `None` when the +/// frame promises the client nothing. +/// +/// Three frames arrive on this path and only one is a promise. An eviction is +/// an `EvictionHeader` whose bytes would cast cleanly as a reply, so the +/// command is checked first (same guard as [`build_raw_pat_reply`]). A +/// pre-consensus rejection stamps the primary's `commit_max`, an op the caller +/// was never told committed and, on a backup-homed caller, one the read gate +/// would then wait for. A committed business rejection (duplicate name, bad +/// expiry) DID commit and counts. +/// +/// Shared with the HTTP write path, which grades the same three frames off the +/// same submit entry point (`submit_client_request_on_owner`); one classifier +/// is what keeps the two planes' watermarks meaning the same thing. +pub(crate) fn committed_reply_commit(reply: &Message<GenericHeader>) -> Option<u64> { + if reply.header().command != Command::Reply || transient_code(reply).is_some() { + return None; + } + let header = reply.as_slice().get(..size_of::<ReplyHeader>())?; + bytemuck::checked::try_from_bytes::<ReplyHeader>(header) + .ok() + .map(|header| header.commit) +} + +/// Whether `code`'s answer comes from the metadata state machine, and so must +/// not be served below the caller's watermark. +/// +/// The two exclusions only look like metadata reads: `DescribeOptions` decodes +/// a static catalog, and `GetClusterMetadata` answers from the configured +/// roster plus the consensus view. Holding either buys no consistency, and the +/// roster read is on the SDK's leader-discovery path, where the wait would be +/// real. +/// +/// Shared with the HTTP read path, which gates the identical set of command +/// codes through `build_non_replicated_response`: two lists would drift, and a +/// code dropped from one plane's list is a silent stale read on that plane. +pub(crate) const fn read_needs_metadata_frontier(code: u32) -> bool { + !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE) +} + #[allow(clippy::future_not_send, clippy::too_many_lines)] async fn handle_non_replicated_request<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, @@ -1655,6 +1765,13 @@ async fn handle_non_replicated_request<B, MJ, S, SB>( handle_get_me(shard, sessions, transport_client_id, &request).await; } GET_PERSONAL_ACCESS_TOKENS_CODE => { + if let Err(error) = + await_metadata_read_frontier(shard, sessions, transport_client_id).await + { + send_non_replicated_deny(shard, &request, transport_client_id, error.as_code()) + .await; + return; + } handle_get_personal_access_tokens(shard, sessions, transport_client_id, &request).await; } GET_CLIENTS_CODE => { @@ -1739,6 +1856,14 @@ async fn handle_non_replicated_request<B, MJ, S, SB>( handle_sync_consumer_group(shard, transport_client_id, &request).await; } _ => { + if read_needs_metadata_frontier(code) + && let Err(error) = + await_metadata_read_frontier(shard, sessions, transport_client_id).await + { + send_non_replicated_deny(shard, &request, transport_client_id, error.as_code()) + .await; + return; + } let roster = sessions.borrow().cluster_roster(); let client_ip = client_address.map(|address| address.ip()); if client_ip.is_none() { diff --git a/core/server/src/http.rs b/core/server/src/http.rs index ed670883d..291b39d49 100644 --- a/core/server/src/http.rs +++ b/core/server/src/http.rs @@ -31,7 +31,7 @@ mod jwks; mod jwt; mod metrics; mod reads; -mod reply; +pub mod reply; mod session; mod state; mod submit; diff --git a/core/server/src/http/error.rs b/core/server/src/http/error.rs index db3edcc5a..0ee842227 100644 --- a/core/server/src/http/error.rs +++ b/core/server/src/http/error.rs @@ -471,6 +471,14 @@ pub(in crate::http) enum ReadError { /// [`service_unavailable`] body, retryable once the cluster re-commits the /// suffix. RecoveryIncomplete, + /// A metadata read waited out its budget with this node's applied frontier + /// still below the op the caller was told committed. Fail-closed on the + /// same retryable 503 as [`Self::RecoveryIncomplete`]: the two are the same + /// hazard (serving state the caller already saw replaced) reached from + /// different directions, and 503 is what the binary transports' equivalent + /// refusal (`TransientNotCommitted`) already renders as, so an SDK that + /// speaks both sees one answer. Never a 2xx with stale state. + MetadataFrontierUnreached, /// A partition read (poll / consumer-offset) got no reply from the owning /// shard within the mesh budget. 504 like a produce timeout: the outcome is /// unknown (the abandoned read may still be running), so the caller retries. @@ -486,7 +494,7 @@ impl IntoResponse for ReadError { Self::NotFound => CustomError::ResourceNotFound.into_response(), Self::NotPrimary => not_primary_response(), Self::RedirectToPrimary(location) => primary_redirect_response(&location), - Self::RecoveryIncomplete => service_unavailable(), + Self::RecoveryIncomplete | Self::MetadataFrontierUnreached => service_unavailable(), Self::Timeout => gateway_timeout_response( "partition_read_timeout", "the partition owner did not answer the read in time; retry", @@ -816,4 +824,20 @@ mod tests { not_primary.headers().get(RETRY_AFTER) ); } + + // An unreached read frontier must never degrade into a 2xx carrying stale + // state, and must not read as terminal either: it is the same retryable 503 + // the recovery barrier's expiry renders, so an SDK retries rather than + // surfacing the read as failed. + #[test] + fn metadata_frontier_unreached_renders_the_same_retryable_503_as_the_barrier() { + let frontier = ReadError::MetadataFrontierUnreached.into_response(); + let recovery = ReadError::RecoveryIncomplete.into_response(); + assert_eq!(frontier.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(frontier.status(), recovery.status()); + assert_eq!( + frontier.headers().get(RETRY_AFTER), + Some(&HeaderValue::from(RETRY_AFTER_SECONDS)) + ); + } } diff --git a/core/server/src/http/extractor.rs b/core/server/src/http/extractor.rs index e818a4e69..a2239a745 100644 --- a/core/server/src/http/extractor.rs +++ b/core/server/src/http/extractor.rs @@ -94,6 +94,14 @@ impl FromRequestParts<HttpState> for Authenticated { /// [`resolve_credential`] chokepoint, so a JWT and a PAT are honored identically. pub struct Identity { pub user_id: u32, + /// Session-table key of the presenting credential (`jwt:{jti}` / + /// `pat:{sha}`), the SAME key [`Authenticated`] resolves its session under. + /// + /// A read mints no session, so this is the only join back to what this + /// credential's writes committed: the read gate looks its metadata + /// watermark up by this key. Carried rather than resolved again in the + /// handler because [`resolve_credential`] already computed it. + pub session_key: String, /// Original request path + query (e.g. `/streams?consistency=linearizable`), /// captured so a linearizable read that reaches a follower can build the /// `Location` for its 307 redirect to the primary. Empty only when the URI @@ -116,15 +124,18 @@ impl FromRequestParts<HttpState> for Identity { ) -> Result<Self, Self::Rejection> { let bearer = bearer_token(&parts.headers)?; - // Verify only. The session key and expiry `resolve_credential` also - // returns feed the write path's session table; a read discards them. + // Verify only: no session is minted or Registered. The key is kept + // (the read gate resolves this credential's metadata watermark under + // it) while the expiry is discarded - nothing here installs a table + // entry to expire. // The verify is `!Send` (a trusted-issuer JWT may await a JWKS fetch), // so bridge it with `SendWrapper` - sound only because compio pins this // future to shard 0's single thread, the only thread the JWKS client // ever runs on (mirrors legacy `HttpSafeShard`). It holds no `RefCell` // borrow or `DashMap` guard across the `.await`, so a sibling task // scheduled on this thread meanwhile never observes a borrowed cell. - let (_key, user_id, _expiry) = SendWrapper::new(resolve_credential(state, bearer)).await?; + let (session_key, user_id, _expiry) = + SendWrapper::new(resolve_credential(state, bearer)).await?; let path_and_query = parts .uri .path_and_query() @@ -142,6 +153,7 @@ impl FromRequestParts<HttpState> for Identity { } Ok(Self { user_id, + session_key, path_and_query, client_ip, }) diff --git a/core/server/src/http/reads.rs b/core/server/src/http/reads.rs index 08e743f47..2a2830cac 100644 --- a/core/server/src/http/reads.rs +++ b/core/server/src/http/reads.rs @@ -15,11 +15,14 @@ // specific language governing permissions and limitations // under the License. -//! Read-path gates: the shared per-op RBAC + consistency check, the local -//! metadata-STM read entry, and the wire/domain identifier resolvers the read -//! and data-plane routes ground their scopes through. +//! Read-path gates: the shared per-op RBAC + consistency check, the two waits +//! a local read serves behind (the post-restart recovery barrier and the +//! per-credential read-your-writes frontier), the local metadata-STM read +//! entry, and the wire/domain identifier resolvers the read and data-plane +//! routes ground their scopes through. use crate::bootstrap::ServerShard; +use crate::dispatch::read_needs_metadata_frontier; use bytes::Bytes; use consensus::MetadataHandle; use iggy_binary_protocol::WireIdentifier; @@ -94,7 +97,15 @@ pub(in crate::http) async fn read_local( rule: impl FnOnce(&Permissioner, u32) -> Result<(), IggyError>, ) -> Result<Bytes, ReadError> { await_recovery_barrier(&state.shard).await?; + // Ahead of the frontier wait on purpose. `authorize_read` renders the + // linearizable follower redirect, which must answer 307 immediately - + // parking first would delay a request this node is not going to serve at + // all - and an authorization denial is terminal, so holding the connection + // for it buys nothing. authorize_read(state, identity, consistency, rule)?; + if read_needs_metadata_frontier(code) { + await_metadata_read_frontier(state, identity).await?; + } let clients_count = if code == GET_STATS_CODE { u32::try_from(SendWrapper::new(state.shard.list_all_clients()).await.len()) .unwrap_or(u32::MAX) @@ -117,6 +128,87 @@ pub(in crate::http) async fn read_local( } } +/// Poll cadence while a metadata read waits for this node's applied frontier. +/// The recovery barrier's cadence and the binary read gate's: what the wait is +/// usually short of is a single commit broadcast. +const READ_FRONTIER_POLL: std::time::Duration = std::time::Duration::from_millis(10); + +/// Polls one held read is given before it fails retryable: 3s at the cadence +/// above. Matches the binary read gate's budget so both planes give up at the +/// same point, and stays far below the 30s the control-plane write path already +/// spends replaying a transient frame. +const READ_FRONTIER_MAX_POLLS: u32 = 300; + +/// Hold a local metadata read until this node has applied everything the +/// presenting credential was told committed. +/// +/// A committed control-plane reply hands the caller an op number; answering its +/// next read from a state machine below that op contradicts the response it is +/// holding. The lag is real on a node that is not the metadata primary: a +/// healthy backup FORWARDS a `Register` to the primary +/// (`dispatch::submit_register_local_or_forward`) and binds the committed epoch +/// while its own commit walk is still behind it - and a cluster without shared +/// bearer key material runs with HTTP forwarding off, so control-plane writes +/// stay on that backup instead of being relayed. +/// +/// Adjacent to `?consistency=linearizable`, not in competition with it. That +/// asks for the freshest CLUSTER state and is answered by leaving this node +/// (307 to the primary), which [`authorize_read`] decides before this wait and +/// which this wait never sees. This gate makes an UNQUALIFIED read +/// read-your-writes for its own credential, at no redirect and no consensus +/// round trip. +/// +/// Scope is this node's own view. A credential whose write this node relayed +/// over HTTP, or that wrote through a different node entirely, left no +/// watermark here; closing that needs the serving primary's commit op to reach +/// the reading node, which nothing in the response carries today. +async fn await_metadata_read_frontier( + state: &HttpInner, + identity: &Identity, +) -> Result<(), ReadError> { + let metadata = state.shard.plane.metadata(); + hold_for_frontier( + || metadata.applied_frontier(), + state.metadata_watermark(&identity.session_key), + READ_FRONTIER_MAX_POLLS, + ) + .await +} + +/// Poll `frontier` for `watermark` up to `max_polls` times, then give up +/// retryable. Split from its call site so the fast path, the park, the catch-up +/// and the expiry are all testable without a live shard - the same reason +/// [`barrier_state`] is split out below. +/// +/// A caller with nothing to read back has `watermark == 0`, which the first +/// comparison satisfies: one `Acquire` load, no await, no allocation. Expiry is +/// loud and carries both numbers, so a frontier that stopped moving is visible +/// instead of showing up as latency. +async fn hold_for_frontier( + frontier: impl Fn() -> u64, + watermark: u64, + max_polls: u32, +) -> Result<(), ReadError> { + if frontier() >= watermark { + return Ok(()); + } + for _ in 0..max_polls { + // `compio::time::sleep` like the recovery barrier below: this listener + // is pinned to shard 0's compio thread, which has no blocking pool to + // hand a wait to. Only this request parks. + compio::time::sleep(READ_FRONTIER_POLL).await; + if frontier() >= watermark { + return Ok(()); + } + } + tracing::warn!( + frontier = frontier(), + watermark, + "metadata read frontier unreached past deadline; failing read with retryable 503" + ); + Err(ReadError::MetadataFrontierUnreached) +} + /// One recovery-barrier check's outcome, factored out of [`await_recovery_barrier`] /// so the expiry decision is unit-testable without a runtime: the loop reads the /// clock and injects whether the deadline has passed. @@ -290,7 +382,117 @@ pub(in crate::http) fn authorize_data_plane( #[cfg(test)] mod tests { - use super::{BarrierWait, barrier_state}; + use super::{ + BarrierWait, READ_FRONTIER_MAX_POLLS, ReadError, barrier_state, hold_for_frontier, + read_needs_metadata_frontier, + }; + use iggy_binary_protocol::codes::{ + DESCRIBE_OPTIONS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUPS_CODE, + GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, + GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE, + }; + use std::cell::Cell; + + /// A caller with nothing to read back (`watermark == 0`) and one whose + /// watermark this node has already applied are the whole steady state, and + /// neither may cost a park: exactly one load, no await. A gate that polled + /// here would put 10ms on every REST read in the cluster. + #[compio::test] + async fn given_a_frontier_at_the_watermark_when_gating_should_serve_without_parking() { + for (frontier_value, watermark) in [(0, 0), (7, 7), (9, 7)] { + let loads = Cell::new(0u32); + let outcome = hold_for_frontier( + || { + loads.set(loads.get() + 1); + frontier_value + }, + watermark, + READ_FRONTIER_MAX_POLLS, + ) + .await; + assert!( + outcome.is_ok(), + "frontier {frontier_value} covers {watermark}" + ); + assert_eq!( + loads.get(), + 1, + "frontier {frontier_value} covers {watermark}, so the read must not poll" + ); + } + } + + /// The gate's whole point: a read whose credential was told op 9 committed + /// is held, not answered, while this node is still at op 4 - and it is + /// answered as soon as the node catches up, rather than being failed. + #[compio::test] + async fn given_a_frontier_behind_the_watermark_when_gating_should_hold_until_it_catches_up() { + const CATCH_UP_AFTER: u32 = 2; + let polls = Cell::new(0u32); + let outcome = hold_for_frontier( + || { + let seen = polls.get(); + polls.set(seen + 1); + if seen >= CATCH_UP_AFTER { 9 } else { 4 } + }, + 9, + 8, + ) + .await; + + assert!( + outcome.is_ok(), + "the read must be served once the node caught up" + ); + assert!( + polls.get() > CATCH_UP_AFTER, + "the read was answered off the lagging frontier after {} loads", + polls.get() + ); + } + + /// A node can legitimately never catch up (a durably lagging replica), so + /// the park is bounded - and the exit is a retryable refusal, never the + /// stale answer. `MetadataFrontierUnreached` renders the shared 503 (see + /// `error.rs`). + #[compio::test] + async fn given_a_frontier_that_never_catches_up_when_gating_should_fail_retryable() { + let outcome = hold_for_frontier(|| 4, 9, 3).await; + assert!( + matches!(outcome, Err(ReadError::MetadataFrontierUnreached)), + "an unreached frontier must refuse the read, not serve it" + ); + } + + /// The HTTP read routes share the binary dispatch's exclusion list, so this + /// pins what that list means for the codes HTTP actually serves: every + /// entity read is gated, and the static option catalog is not. Forking the + /// list per plane is what this is here to catch. + #[test] + fn given_the_http_read_codes_when_classified_should_gate_all_but_the_static_catalog() { + for code in [ + GET_STREAMS_CODE, + GET_STREAM_CODE, + GET_TOPICS_CODE, + GET_TOPIC_CODE, + GET_USERS_CODE, + GET_USER_CODE, + GET_CONSUMER_GROUPS_CODE, + GET_PERSONAL_ACCESS_TOKENS_CODE, + GET_STATS_CODE, + ] { + assert!( + read_needs_metadata_frontier(code), + "code {code} answers from the metadata STM and must be gated" + ); + } + for code in [DESCRIBE_OPTIONS_CODE, GET_CLUSTER_METADATA_CODE] { + assert!( + !read_needs_metadata_frontier(code), + "code {code} answers from a static catalog or the roster; holding it buys nothing" + ); + } + } #[test] fn barrier_state_ready_when_no_barrier_armed() { diff --git a/core/server/src/http/reply.rs b/core/server/src/http/reply.rs index c17f5ce78..e0eeaa6a0 100644 --- a/core/server/src/http/reply.rs +++ b/core/server/src/http/reply.rs @@ -156,7 +156,7 @@ pub(in crate::http) fn committed_payload( /// entered the pipeline and is safe to re-issue anywhere, while /// `TransientNotCommitted` may still commit and only a same-session same-id /// replay is safe. -pub(in crate::http) fn transient_code(reply: &Message<GenericHeader>) -> Option<IggyError> { +pub fn transient_code(reply: &Message<GenericHeader>) -> Option<IggyError> { match result_code(reply_body(reply)) { Some(code) if code == IggyError::TransientNotCommitted.as_code() => { Some(IggyError::TransientNotCommitted) diff --git a/core/server/src/http/session.rs b/core/server/src/http/session.rs index aa1db608c..fe9e21ac2 100644 --- a/core/server/src/http/session.rs +++ b/core/server/src/http/session.rs @@ -119,9 +119,50 @@ pub(in crate::http) struct HttpSession { /// [`MAX_IN_FLIGHT_WRITES_PER_SESSION`]. Only [`InFlightWriteGuard`] /// touches it, so every admission is paired with exactly one release. pub(in crate::http) in_flight_writes: Cell<u32>, + /// Highest metadata op this credential has been told committed. + /// + /// Seeded from [`Self::session`] (the `Register` commit op, which floors + /// every metadata op that register could have observed) and raised by every + /// committed control-plane reply on this session. The read gate holds a + /// local read until the node's applied frontier covers it, so a caller + /// cannot be served state older than a write it already saw acked. + /// + /// Per-credential rather than per-request because that is the unit a + /// bearer's requests share; a caller presenting a fresh credential + /// re-seeds from the session it registers. A plain `Cell` suffices on + /// single-threaded shard 0, and it is never read across an `.await`. + pub(in crate::http) metadata_watermark: Cell<u64>, } impl HttpSession { + /// Build the session a completed `Register` establishes: the identity + /// fields come from the caller, every counter starts where the write and + /// read paths expect it to. + /// + /// The seeds are here rather than at the mint site because + /// `metadata_watermark` starting at `session` is a correctness invariant + /// (see the field), and one that no fixture can be trusted to restate. + pub(in crate::http) fn registered( + key: String, + client_id: u128, + session: u64, + user_id: u32, + expiry: u64, + ) -> Self { + Self { + key, + client_id, + session, + user_id, + expiry, + gate: Mutex::new(FIRST_REQUEST_ID), + data_request: Cell::new(FIRST_REQUEST_ID), + registry_token: Cell::new(None), + in_flight_writes: Cell::new(0), + metadata_watermark: Cell::new(session), + } + } + /// Mint the next data-plane request id. Also consumed by the `?ack=none` /// path, which installs no slot: sharing one counter keeps a shed reply's /// id from ever colliding with a live awaited slot on this session. @@ -130,6 +171,19 @@ impl HttpSession { self.data_request.set(id + 1); id } + + /// Raise this session's metadata watermark to `commit`. Monotone, so a + /// reply that lands out of order (concurrent requests on one credential + /// are legal) cannot lower it. + /// + /// Only COMMITTED metadata replies belong here. A pre-consensus rejection + /// stamps the primary's `commit_max`, an op this caller was never promised + /// and, on a backup, one its own reads would then wait for; partition-plane + /// replies carry a different group's commit position entirely. + pub(in crate::http) fn record_metadata_watermark(&self, commit: u64) { + self.metadata_watermark + .set(self.metadata_watermark.get().max(commit)); + } } /// Serializes first-use VSR registration per credential key so a herd of @@ -264,17 +318,13 @@ mod tests { /// construction plus the live cancellation smoke, not faked here. #[compio::test] async fn detached_task_advances_gate_and_ignores_dead_receiver() { - let session = Rc::new(HttpSession { - key: "jwt:test".to_owned(), - client_id: 7, - session: 1, - user_id: DEFAULT_ROOT_USER_ID, - expiry: u64::MAX, - gate: Mutex::new(FIRST_REQUEST_ID), - data_request: Cell::new(FIRST_REQUEST_ID), - registry_token: Cell::new(None), - in_flight_writes: Cell::new(0), - }); + let session = Rc::new(HttpSession::registered( + "jwt:test".to_owned(), + 7, + 1, + DEFAULT_ROOT_USER_ID, + u64::MAX, + )); let (result_slot, committed) = oneshot::channel::<u64>(); // The handler future dies (client disconnect) before the task runs. drop(committed); @@ -292,21 +342,24 @@ mod tests { assert_eq!(*session.gate.lock().await, FIRST_REQUEST_ID + 1); } + /// Register commit op every fixture binds. Non-zero on purpose: the read + /// gate's watermark seeds from it, so a fixture at zero could not tell a + /// seeded session from an unseeded one. + const FIXTURE_EPOCH: u64 = 1; + /// `InstanceToken` has no public constructor, so fixtures carry no reply /// target; the token-teardown branch of the sweep/forget helpers is - /// exercised via their `Option` path, not fabricated here. + /// exercised via their `Option` path, not fabricated here. Built through + /// [`HttpSession::registered`], the same constructor the live mint uses, so + /// a fixture can never drift from the seeds production ships. fn fake_session(key: &str, client_id: u128, expiry: u64) -> Rc<HttpSession> { - Rc::new(HttpSession { - key: key.to_owned(), + Rc::new(HttpSession::registered( + key.to_owned(), client_id, - session: 1, - user_id: DEFAULT_ROOT_USER_ID, + FIXTURE_EPOCH, + DEFAULT_ROOT_USER_ID, expiry, - gate: Mutex::new(FIRST_REQUEST_ID), - data_request: Cell::new(FIRST_REQUEST_ID), - registry_token: Cell::new(None), - in_flight_writes: Cell::new(0), - }) + )) } // The barrier is what makes a herd of concurrent first-requests for one @@ -417,4 +470,27 @@ mod tests { "the pointer fence spares the re-registered session" ); } + + /// The mark is a floor the read gate waits for, so nothing may lower it: + /// two concurrent requests on one credential can have their committed + /// replies land out of order, and the later-but-lower reply must not undo + /// the earlier-but-higher one. The seed is the register's own commit op + /// (`session`), which floors every op that register could have observed. + #[test] + fn given_out_of_order_replies_when_recording_should_keep_the_watermark_monotone() { + let session = fake_session("jwt:a", 1, u64::MAX); + assert_eq!( + session.metadata_watermark.get(), + session.session, + "a fresh session starts at its register commit op" + ); + + session.record_metadata_watermark(50); + session.record_metadata_watermark(7); + assert_eq!( + session.metadata_watermark.get(), + 50, + "a lower commit must not lower the mark" + ); + } } diff --git a/core/server/src/http/state.rs b/core/server/src/http/state.rs index 8ea1bb486..4e0dd2ee9 100644 --- a/core/server/src/http/state.rs +++ b/core/server/src/http/state.rs @@ -34,7 +34,6 @@ use iggy_common::{ClusterMetadata, IggyTimestamp}; use message_bus::InstanceToken; use metadata::MetadataSubmitError; use send_wrapper::SendWrapper; -use tokio::sync::Mutex; use tracing::warn; use crate::bootstrap::ServerShard; @@ -45,8 +44,8 @@ use crate::http::forward::ForwardState; use crate::http::jwt::JwtManager; use crate::http::metrics::HttpMetrics; use crate::http::session::{ - BarrierEntry, FIRST_REQUEST_ID, FRESH_ENTRY_WATERMARK, HttpSession, RegistrationBarrier, - forget_if_same, live_entry, sweep_expired, + BarrierEntry, FRESH_ENTRY_WATERMARK, HttpSession, RegistrationBarrier, forget_if_same, + live_entry, sweep_expired, }; /// Response header carrying the current VSR view number. Stamped by @@ -210,6 +209,25 @@ impl HttpInner { } } + /// Highest metadata op the credential behind `key` has been told + /// committed, or `0` when this node has told it none - an unknown key, so + /// no write of its ever ran here and there is nothing to read back. + /// + /// Deliberately NOT expiry-filtered, unlike [`Self::live_session`]: the + /// number is a consistency floor, not a capability, and the request that + /// consults it has already re-verified the bearer. Dropping the floor + /// because a swept-but-still-present entry aged out would reintroduce the + /// stale read for exactly the callers still holding a committed reply. + /// + /// Confines the shared `RefCell` borrow to this call, so it can never span + /// the read gate's `.await`. + pub(in crate::http) fn metadata_watermark(&self, key: &str) -> u64 { + self.sessions + .borrow() + .get(key) + .map_or(0, |session| session.metadata_watermark.get()) + } + /// Clone the live (non-expired) entry for `key`, if present. Confines the /// shared `RefCell` borrow to this call so it can never span an `.await`. fn live_session(&self, key: &str, now_secs: u64) -> Option<Rc<HttpSession>> { @@ -341,17 +359,19 @@ impl HttpInner { ); return Err(AuthError::SessionIdTaken); } - Ok(Rc::new(HttpSession { + // `bound.epoch` also seeds the read gate's watermark: a HEALTHY BACKUP + // forwards the register to the primary (see + // `submit_register_local_or_forward`), so this node can hand back an + // epoch its own commit walk has not reached, and the caller's first + // read would otherwise be served from state older than the register it + // is holding. + Ok(Rc::new(HttpSession::registered( key, client_id, - session: bound.epoch, + bound.epoch, user_id, expiry, - gate: Mutex::new(FIRST_REQUEST_ID), - data_request: Cell::new(FIRST_REQUEST_ID), - registry_token: Cell::new(None), - in_flight_writes: Cell::new(0), - })) + ))) } /// Drop the session table entry for `session`, but only if it is still the diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs index ca763d836..4c3b0ffef 100644 --- a/core/server/src/http/submit.rs +++ b/core/server/src/http/submit.rs @@ -35,8 +35,8 @@ use tracing::warn; use crate::bootstrap::ServerShard; use crate::dispatch::{ - dispatch_partition_request, resolve_delete_segments_truncate, submit_client_request_on_owner, - submit_logout_on_owner, + committed_reply_commit, dispatch_partition_request, resolve_delete_segments_truncate, + submit_client_request_on_owner, submit_logout_on_owner, }; use crate::http::admission::admit_partition_write; use crate::http::error::{PartitionWriteError, WriteError}; @@ -115,6 +115,16 @@ pub(in crate::http) async fn submit_committed( compio::runtime::spawn(async move { let result = submit_gated(&shard, &task_session, operation, max_tokens_per_user, &body).await; + // Recorded here rather than after the await below, for the same reason + // the submit is detached: a caller that disconnected mid-write still + // committed the op, and its next request on this credential must not be + // served state older than what committed. Ordered before the wake, so a + // read issued the instant the response lands already sees the mark. + if let Ok((_, reply, _)) = &result + && let Some(commit) = committed_reply_commit(reply) + { + task_session.record_metadata_watermark(commit); + } // A failed send means the handler died mid-await; the submit itself // already completed, which is the invariant that matters. let _ = result_slot.send(result); diff --git a/core/server/src/session_manager.rs b/core/server/src/session_manager.rs index 84635a8b4..cdba5367d 100644 --- a/core/server/src/session_manager.rs +++ b/core/server/src/session_manager.rs @@ -79,6 +79,18 @@ pub struct Connection { pub last_heartbeat: Instant, /// Recorded at login; `None` until the connection authenticates. pub sdk: Option<ClientSdkInfo>, + /// Highest metadata op this connection has been told committed. + /// + /// Seeded from the bound session (the register's own commit op, which + /// floors everything the client committed before it re-homed) and raised by + /// every committed reply relayed on this connection. The read gate holds a + /// local read until the node's applied frontier covers it, so a client + /// cannot be served state older than a write it already saw acked. + /// + /// Per-connection rather than per-client: the number only has to cover what + /// THIS socket was told, and a client that reconnects re-seeds from the + /// session it binds. + pub metadata_watermark: u64, } /// Bridges transport connections to consensus sessions. @@ -141,6 +153,7 @@ impl SessionManager { state: ConnectionState::Connected, last_heartbeat: Instant::now(), sdk: None, + metadata_watermark: 0, }); } @@ -268,15 +281,47 @@ impl SessionManager { } // Now mutate the target connection. - self.connections.get_mut(&connection_id).unwrap().state = ConnectionState::Bound { + let bound = self + .connections + .get_mut(&connection_id) + .expect("bind_session: connection validated above, single-threaded"); + bound.state = ConnectionState::Bound { user_id, client_id, session, }; + // The session IS the register's commit op, so it floors every metadata + // op this client saw committed before it re-homed here. Without the + // seed a re-homed connection reads at zero and the gate admits the + // pre-write state its own last write already replaced. + bound.metadata_watermark = bound.metadata_watermark.max(session); self.client_to_connection.insert(client_id, connection_id); Ok(()) } + /// Raise this connection's metadata watermark to `commit`. Monotone, so a + /// late or out-of-order reply cannot lower it; no-op for an unknown + /// connection. + /// + /// Only committed replies belong here. A pre-consensus rejection stamps the + /// primary's `commit_max`, which is an op this connection was never + /// promised and, on a backup-homed connection, one it would then wait for. + pub fn record_metadata_watermark(&mut self, connection_id: u128, commit: u64) { + if let Some(conn) = self.connections.get_mut(&connection_id) { + conn.metadata_watermark = conn.metadata_watermark.max(commit); + } + } + + /// The highest metadata op this connection was told committed, or `0` when + /// it was told none (an unknown or still-unbound connection, which has no + /// write to read back). + #[must_use] + pub fn metadata_watermark(&self, connection_id: u128) -> u64 { + self.connections + .get(&connection_id) + .map_or(0, |conn| conn.metadata_watermark) + } + /// Look up the consensus session for a connection. /// /// Returns `(client_id, session)` if the connection is `Bound`, `None` otherwise. @@ -603,4 +648,46 @@ mod tests { "a second disconnect has nothing left to release" ); } + + /// The bind seed is what makes a re-homed connection safe: the register's + /// commit op floors every metadata op the client committed elsewhere, so + /// the read gate cannot admit the pre-write state on a node that has not + /// caught up. A recorder that could lower the mark would undo it. + #[test] + fn given_a_bound_connection_when_replies_arrive_should_keep_the_watermark_monotone() { + let mut mgr = SessionManager::new(); + let conn = 1; + mgr.ensure_connection(conn, addr(5200), ClientTransportKind::Tcp); + assert_eq!( + mgr.metadata_watermark(conn), + 0, + "an unbound connection was promised nothing" + ); + + mgr.login(conn, 3).unwrap(); + mgr.bind_session(conn, 100, 42).unwrap(); + assert_eq!( + mgr.metadata_watermark(conn), + 42, + "the bound session is the register's commit op and floors the mark" + ); + + mgr.record_metadata_watermark(conn, 50); + mgr.record_metadata_watermark(conn, 7); + assert_eq!( + mgr.metadata_watermark(conn), + 50, + "a lower commit must not lower the mark" + ); + } + + /// An unknown connection is not an error: the disconnect callback can win + /// the race against a reply relay, and a gate reading `0` then serves the + /// read instead of parking a socket that is already gone. + #[test] + fn given_an_unknown_connection_when_recording_a_watermark_should_be_inert() { + let mut mgr = SessionManager::new(); + mgr.record_metadata_watermark(9, 5); + assert_eq!(mgr.metadata_watermark(9), 0); + } } diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs index 1b1501aee..191e523c5 100644 --- a/core/simulator/src/client.rs +++ b/core/simulator/src/client.rs @@ -16,7 +16,7 @@ // under the License. use bytes::{Bytes, BytesMut}; -use iggy_binary_protocol::codes::POLL_MESSAGES_CODE; +use iggy_binary_protocol::codes::{GET_STREAM_CODE, POLL_MESSAGES_CODE}; use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::requests::consumer_groups::{ CreateConsumerGroupRequest, DeleteConsumerGroupRequest, @@ -36,7 +36,8 @@ use iggy_binary_protocol::requests::personal_access_tokens::{ }; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::requests::streams::{ - CreateStreamRequest, DeleteStreamRequest, PurgeStreamRequest, UpdateStreamRequest, + CreateStreamRequest, DeleteStreamRequest, GetStreamRequest, PurgeStreamRequest, + UpdateStreamRequest, }; use iggy_binary_protocol::requests::topics::{ CreateTopicRequest, DeleteTopicRequest, PurgeTopicRequest, UpdateTopicRequest, @@ -648,6 +649,48 @@ impl SimClient { .expect("poll request must be valid") } + /// Build a `GET_STREAM` read for `name`. + /// + /// A `NonReplicated` metadata read, so it is answered from whichever + /// replica's state machine the request lands on rather than routed to the + /// primary: the command code sits in the header's `reserved` prefix, the + /// group is the metadata sentinel, and the request id echoes the counter + /// without advancing it (matching [`Self::poll_messages`] and the SDK). + /// Requires a bound session, since the read is auth-gated. + /// + /// # Panics + /// Panics if `name` is not a valid wire name or the request buffer is + /// invalid. + #[allow(clippy::cast_possible_truncation)] + pub fn get_stream(&self, name: &str) -> Message<RoutedRequestHeader> { + let body = GetStreamRequest { + stream_id: WireIdentifier::named(name).expect("stream name must be valid"), + } + .to_bytes(); + + let header_size = std::mem::size_of::<RoutedRequestHeader>(); + let total_size = header_size + body.len(); + let mut reserved = [0u8; 52]; + reserved[..4].copy_from_slice(&GET_STREAM_CODE.to_le_bytes()); + let header = RoutedRequestHeader { + command: iggy_binary_protocol::Command::Request, + operation: Operation::NonReplicated, + size: total_size as u32, + client: self.client_id, + session: self.session_id(), + request: self.request_counter.get(), + reserved, + group: METADATA_GROUP, + ..Default::default() + }; + + let mut buffer = Vec::with_capacity(total_size); + buffer.extend_from_slice(bytemuck::bytes_of(&header)); + buffer.extend_from_slice(&body); + Message::try_from(Owned::<4096>::copy_from_slice(&buffer)) + .expect("get stream request must be valid") + } + /// Store offset with explicit `AckLevel`. `NoAck` takes the primary's /// fast path (no replication); `Quorum` goes through VSR. /// diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 7b722eb9b..93a28213b 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -57,7 +57,7 @@ use std::collections::{HashMap, HashSet}; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::rc::Rc; use std::sync::Arc; -use std::sync::atomic::AtomicBool; +use std::sync::atomic::{AtomicBool, AtomicU64}; /// Poll budget per [`DetExecutor::run_until_stalled`]. Pumps are event-driven, so /// hitting it means a task is spin-waking: a bug, panicked with the seed. @@ -395,6 +395,10 @@ impl Simulator { // reader-mode mirror from it and reads committed metadata through the // shared handle. Built in index order, so shard 0's bundle exists first. let mut metadata_bundle: Option<replica::SimMetadataBundle> = None; + // One applied-metadata frontier per REPLICA, shared by its shards, + // as the server bootstrap mints one per process. Volatile: a restart + // below builds a fresh cell, matching a rebooted node. + let metadata_applied_frontier = Arc::new(AtomicU64::new(0)); for shard_idx in 0..shards_per_replica { let inbox = inboxes[usize::from(shard_idx)] .take() @@ -429,6 +433,7 @@ impl Simulator { (shard_idx == 0).then(|| replica_data_dir.clone()).flatten(), // Fresh boot: `init_partition` seeds later, before any workload. &[], + Arc::clone(&metadata_applied_frontier), ); if shard_idx == 0 { metadata_bundle = Some( @@ -1066,6 +1071,7 @@ impl Simulator { /// # Panics /// If the replica is not crashed, or its shard count does not fit `u16`; mesh /// construction caps it. + #[allow(clippy::too_many_lines)] pub fn replica_restart(&mut self, replica_index: u8) { assert!( self.crashed.contains(&replica_index), @@ -1120,6 +1126,7 @@ impl Simulator { let mut stop_txs = Vec::with_capacity(usize::from(shards_per_replica)); let mut pump_tasks = Vec::with_capacity(usize::from(shards_per_replica)); let mut metadata_bundle: Option<replica::SimMetadataBundle> = None; + let metadata_applied_frontier = Arc::new(AtomicU64::new(0)); for shard_idx in 0..shards_per_replica { let inbox = inboxes[usize::from(shard_idx)] .take() @@ -1151,6 +1158,7 @@ impl Simulator { metadata_incarnation, (shard_idx == 0).then(|| replica_data_dir.clone()).flatten(), &seed_namespaces, + Arc::clone(&metadata_applied_frontier), ); if shard_idx == 0 { metadata_bundle = @@ -4824,3 +4832,369 @@ mod repair_frontier_tests { ); } } + +#[cfg(test)] +mod metadata_read_frontier_tests { + //! A client that committed a metadata write and then re-homed onto a + //! lagging backup must never be served the pre-write state. + //! + //! The window is not peer shards on one node: a committed reply is only + //! produced after `gated_apply` published, and every shard reads the same + //! left-right buffers. It is a node whose commit walk trails the epoch the + //! client already holds. Register forwarding is the supported way to get + //! there -- a backup verifies the credentials itself, forwards only the + //! consensus proposal, and binds the committed session while its own + //! `commit_journal` is still behind that op (see `server::auth`). + //! + //! Blocking replication INTO one backup while the quorum commits without + //! it produces the lag deterministically, and the login still completes + //! because `ForwardRegister` / `ForwardRegisterResult` are left flowing. + //! + //! One shard per replica for the end-to-end read, deliberately. The harness + //! homes each inbound client packet on a seeded-random shard while every + //! shard owns its own `SessionManager`, so on a multi-shard replica a bound + //! session cannot reliably receive its own follow-up requests -- and the lag + //! under test is the node's, not a shard's. + //! + //! The second test is the other half: peer shards are not the WINDOW, but + //! they are how a peer-homed read learns the node's position at all, and + //! sharing one frontier cell across a replica's shards is what makes that + //! work. It runs multi-shard and drives the cell directly, so it needs no + //! session and dodges the homing problem entirely. + + use super::*; + use crate::client::SimClient; + use iggy_binary_protocol::responses::streams::get_stream::GetStreamResponse; + use iggy_binary_protocol::{Command, RoutedRequestHeader, WireDecode}; + + /// Replica 0 leads the metadata plane at view 0, so this one is a backup + /// for the whole run and is the node the client re-homes onto. + const LAGGING: u8 = 1; + + /// Steps the read is given while the backup is still cut off. A server that + /// answers a metadata read from an unconverged state answers within a + /// couple of these; the gate must hold the read past all of them. + /// + /// Well under the gate's own poll budget, so expiry cannot masquerade as a + /// held read. + const STALE_WINDOW_STEPS: u32 = 50; + + /// Steps allowed for repair to reach the backup and the held read to answer + /// once replication is restored. + const CONVERGE_STEPS: u32 = 2_000; + + /// The frames that would let the backup learn the committed writes. Journal + /// repair and `StartView` adoption are cut with the same knife as live + /// replication: any one of them left open closes the lag this exercises. + const REPLICATION_FRAMES: [Command; 5] = [ + Command::Prepare, + Command::Commit, + Command::RepairPrepare, + Command::RepairDone, + Command::StartView, + ]; + + /// A stream deleted before the client re-homed must not come back on the + /// backup that has not applied the delete yet. + #[test] + fn given_backup_behind_the_client_epoch_when_reading_a_deleted_stream_should_not_serve_it() { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + + let replica_count: u8 = 3; + let client_id: u128 = 1; + let stream_name = "read-your-writes"; + let network_opts = packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed: 0x1A7E_0F31, + ..packet::PacketSimulatorOptions::default() + }; + let mut sim = Simulator::with_shards_shell( + usize::from(replica_count), + 1, + std::iter::once(client_id), + network_opts, + ); + + let client = SimClient::new(client_id); + sim.shell_login(&client); + + // The create lands on every replica: the backup has to HOLD the stream + // for the read below to be able to serve a stale one. + let created = commit_write(&mut sim, client_id, 0, client.create_stream(stream_name)); + step_until_applied(&mut sim, LAGGING, created); + + // Cut replication into the backup, so the delete commits on the quorum + // formed by the primary and the remaining replica and never reaches it. + // Journal repair and `StartView` adoption go with it: any one of them + // left open closes the lag this exercises. + set_replication(&mut sim, LAGGING, false); + + let deleted = commit_write(&mut sim, client_id, 0, client.delete_stream(stream_name)); + assert!( + deleted > created, + "the delete must commit above the create, else the read below cannot \ + distinguish the two states" + ); + + // Re-home the same client onto the backup. The login is forwarded, so + // the session it binds IS a committed op above the delete while the + // backup's own applied frontier is still below it. + sim.shell_login_via(&client, LAGGING); + assert!( + sim.network.delivered_any(Command::ForwardRegister), + "no ForwardRegister crossed the wire: the backup answered the login \ + itself, so the client never re-homed" + ); + let lagging_commit = metadata_commit(&sim, usize::from(LAGGING)); + assert!( + (created..deleted).contains(&lagging_commit), + "the backup applied up to op {lagging_commit}, outside the window \ + [{created}, {deleted}) this test needs: it must hold the create and \ + miss the delete" + ); + assert_eq!( + read_stream_name_on(&sim, LAGGING, stream_name), + Some(stream_name.to_string()), + "the backup no longer holds the deleted stream, so a read cannot \ + serve a stale one and the assertions below prove nothing" + ); + + let read = client.get_stream(stream_name); + let request_id = read.header().request; + sim.submit_request(client_id, LAGGING, read.into_generic()); + + // Phase 1: still cut off. Any answer here is served from state that + // predates the delete the client already saw committed. + let mut early = None; + for _ in 0..STALE_WINDOW_STEPS { + if let Some(reply) = sim + .step() + .into_iter() + .find(|reply| reply.header().request == request_id) + { + early = Some(reply); + break; + } + } + if let Some(reply) = early { + panic!( + "the backup answered a metadata read while its applied frontier \ + ({}) was below the client's committed epoch ({deleted}): status={}, \ + stream={:?}", + metadata_commit(&sim, usize::from(LAGGING)), + reply.header().status, + read_stream_name(&reply), + ); + } + + // Phase 2: restore replication. The held read must answer from the + // converged state, which no longer holds the stream. + set_replication(&mut sim, LAGGING, true); + for _ in 0..CONVERGE_STEPS { + if let Some(reply) = sim + .step() + .into_iter() + .find(|reply| reply.header().request == request_id) + { + assert_eq!( + reply.header().status, + 0, + "the read failed instead of answering once the backup converged" + ); + assert_eq!( + read_stream_name(&reply), + None, + "the converged backup still serves the deleted stream" + ); + return; + } + } + panic!( + "no answer to the held metadata read within {CONVERGE_STEPS} steps of \ + restored replication; backup applied frontier {}, client epoch {deleted}", + metadata_commit(&sim, usize::from(LAGGING)), + ); + } + + /// Shards per replica for the sharing test below. Two is the whole + /// population that matters: shard 0 and one peer. + const SHARED_FRONTIER_SHARDS: u16 = 2; + + /// Advance applied to shard 0, chosen above whatever recovery seeded so a + /// peer reading the pre-advance value cannot pass by accident. + const SHARED_FRONTIER_ADVANCE: u64 = 7; + + /// A peer shard owns no metadata consensus, so `commit_min` -- the number + /// the pre-existing read barrier gates on -- does not exist there at all. + /// The applied frontier is the only thing its read gate can consult, and it + /// arrives by being ONE cell per process rather than one per shard. + /// + /// Nothing else in the system observes that. A private cell per shard still + /// compiles, still serves every read, and its only symptom is that each + /// metadata read homed on a peer shard parks for the whole deadline and + /// then fails retryable -- a latency cliff behind a warning, not an error. + /// So the invariant is asserted directly rather than through a request: a + /// peer must see an advance it did not make. + #[test] + fn given_a_peer_shard_when_shard_zero_advances_the_frontier_should_observe_it() { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + + let replica_count: u8 = 3; + let network_opts = packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed: 0x5EED_5A11, + ..packet::PacketSimulatorOptions::default() + }; + let sim = Simulator::with_shards( + usize::from(replica_count), + SHARED_FRONTIER_SHARDS, + std::iter::once(1u128), + network_opts, + ); + + let shards = &sim.replicas[0].shards; + assert_eq!( + shards.len(), + usize::from(SHARED_FRONTIER_SHARDS), + "the replica did not build the peer shard this test needs" + ); + for (shard_idx, shard) in shards.iter().enumerate().skip(1) { + assert!( + shard.plane.metadata().consensus.is_none(), + "shard {shard_idx} owns consensus, so it is not the peer whose \ + only source for the frontier is shard 0's cell" + ); + } + + let advanced = shards[0].plane.metadata().applied_frontier() + SHARED_FRONTIER_ADVANCE; + shards[0] + .plane + .metadata() + .advance_applied_frontier(advanced); + + for (shard_idx, shard) in shards.iter().enumerate() { + assert_eq!( + shard.plane.metadata().applied_frontier(), + advanced, + "shard {shard_idx} did not observe shard 0's advance: the \ + applied-frontier cell is per shard, not per process, so every \ + metadata read homed here parks until its deadline" + ); + } + } + + /// Open or close every replication route from the primary into `replica`. + fn set_replication(sim: &mut Simulator, replica: u8, open: bool) { + for frame in REPLICATION_FRAMES { + let filter = sim + .network + .link_filter_mut(ProcessId::Replica(0), ProcessId::Replica(replica)); + if open { + filter.insert(frame); + } else { + filter.remove(frame); + } + } + } + + /// Step until `replica` has applied `op`, so the state the read sees is the + /// one this test set up rather than whatever the last reply happened to + /// leave behind. + fn step_until_applied(sim: &mut Simulator, replica: u8, op: u64) { + for _ in 0..SETUP_TOTAL_STEPS { + if metadata_commit(sim, usize::from(replica)) >= op { + return; + } + sim.step(); + } + panic!( + "replica {replica} never applied op {op} (stuck at {})", + metadata_commit(sim, usize::from(replica)), + ); + } + + /// Read `name` straight out of a replica's committed metadata, bypassing + /// the read path under test. + fn read_stream_name_on(sim: &Simulator, replica: u8, name: &str) -> Option<String> { + sim.replicas[usize::from(replica)].shards[0] + .plane + .metadata() + .mux_stm + .streams() + .read(|inner| { + inner + .items + .iter() + .find(|(_, stream)| &*stream.name == name) + .map(|(_, stream)| stream.name.to_string()) + }) + } + + /// Submit one replicated metadata write to `target`, step until its reply + /// lands, and return the op it committed at. + fn commit_write( + sim: &mut Simulator, + client_id: u128, + target: u8, + request: Message<RoutedRequestHeader>, + ) -> u64 { + let request_id = request.header().request; + sim.submit_request(client_id, target, request.into_generic()); + for _ in 0..SETUP_TOTAL_STEPS { + if let Some(reply) = sim + .step() + .into_iter() + .find(|reply| reply.header().request == request_id) + { + assert_eq!( + reply.header().status, + 0, + "metadata write {request_id} was refused" + ); + assert!( + !setup_reply_is_transient(&reply), + "metadata write {request_id} was rejected in transit, so it never \ + committed and cannot anchor the read below" + ); + return reply.header().commit; + } + } + panic!("metadata write {request_id} never committed"); + } + + /// The stream name a `GetStream` reply carries, or `None` for the + /// empty-body not-found answer. + fn read_stream_name(reply: &Message<ReplyHeader>) -> Option<String> { + let body = reply + .as_slice() + .get(size_of::<ReplyHeader>()..reply.header().size as usize) + .unwrap_or_default(); + if body.is_empty() { + return None; + } + let (response, _) = GetStreamResponse::decode(body) + .expect("a non-empty GetStream reply must decode as GetStreamResponse"); + Some(response.stream.name.as_str().to_string()) + } + + /// Committed metadata op on a replica's shard 0. + fn metadata_commit(sim: &Simulator, replica_idx: usize) -> u64 { + sim.replicas[replica_idx].shards[0] + .plane + .metadata() + .consensus + .as_ref() + .expect("shard 0 owns metadata consensus") + .commit_min() + } +} diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs index 638a61dce..9aa65c363 100644 --- a/core/simulator/src/replica.rs +++ b/core/simulator/src/replica.rs @@ -39,6 +39,7 @@ use shard::shards_table::PapayaShardsTable; use std::cell::RefCell; use std::rc::Rc; use std::sync::Arc; +use std::sync::atomic::AtomicU64; // TODO: Make configurable const CLUSTER_ID: u128 = 1; @@ -156,6 +157,7 @@ pub fn new_shard( incarnation: u128, data_dir: Option<std::path::PathBuf>, seed_namespaces: &[(server_common::sharding::IggyNamespace, u32)], + applied_frontier: Arc<AtomicU64>, ) -> (Rc<Replica>, Option<SimMetadataBundle>) { // Metadata is single-writer, mirroring the server bootstrap. Shard 0 owns // the only writable STM; every peer shard rebuilds a reader-mode mirror from @@ -303,7 +305,8 @@ pub fn new_shard( superblock, mux, data_dir, - ); + ) + .with_applied_frontier(applied_frontier); // Both halves are load-bearing: the pairing keeps a later view-change superblock // write from regressing to `(0, 0)`, and the folded table is the floor the replayed @@ -367,6 +370,12 @@ pub fn new_shard( ); } } + // Same seed the server bootstrap does after its own replay: the frontier + // resumes where the commit walk will, so a read on a rebuilt replica does + // not park until its deadline. No-op on peer shards, which share the cell. + if let Some(consensus) = metadata.consensus.as_ref() { + metadata.advance_applied_frontier(consensus.commit_min()); + } // Mint the peers' read-side bundle AFTER reconstruction so it reflects the // recovered state. Shard 0 only; peers pass it back in as `reader_bundle`. let metadata_bundle = (shard_idx == 0).then(|| metadata.mux_stm.factory_bundle());
