hubcio commented on code in PR #4024:
URL: https://github.com/apache/iggy/pull/4024#discussion_r3927021323
##########
core/server/src/http/submit.rs:
##########
@@ -106,13 +105,32 @@ pub(in crate::http) async fn submit_committed(
let (result_slot, committed) = oneshot::channel();
let shard = Rc::clone(&state.shard);
let task_session = Rc::clone(session);
+ let watermarks = Rc::clone(&state.metadata_watermarks);
let body = body.to_vec();
let max_tokens_per_user = state.max_tokens_per_user;
// Detached so a client disconnect cannot abandon the gate mid-submit;
// the write runs to completion regardless of handler liveness.
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 as this user 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.
+ //
+ // This node's own view only: on a follower with HTTP forwarding ON the
Review Comment:
critical: with forwarding on, the follower relays the write to the primary,
so this task never runs and no floor lands. unqualified GETs stay local, so
`POST /streams` then `GET /streams` here is the stale read this PR exists to
stop.
##########
core/server/src/http/reads.rs:
##########
@@ -91,10 +97,26 @@ pub(in crate::http) async fn read_local(
consistency: Consistency,
code: u32,
body: &[u8],
- rule: impl FnOnce(&Permissioner, u32) -> Result<(), IggyError>,
+ rule: impl Fn(&Permissioner, u32) -> Result<(), IggyError>,
) -> Result<Bytes, ReadError> {
await_recovery_barrier(&state.shard).await?;
Review Comment:
critical: `get_clients`, `get_client`, `get_snapshot`, `poll_messages` and
`get_consumer_offset` call `authorize_read` directly and never reach here, so
they get neither the barrier nor the frontier gate. their binary twins are
gated. route them through `read_local`.
##########
core/metadata/src/applied_frontier.rs:
##########
@@ -0,0 +1,300 @@
+// 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 node-wide applied metadata frontier and the wait a read parks on.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Mutex;
+use std::sync::PoisonError;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::task::{Context, Poll, Waker};
+
+/// Highest metadata op whose apply has been PUBLISHED on this node, shared by
+/// every shard, plus the wakers of the reads waiting for it to reach them.
+///
+/// `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 cell
+/// does, for one `Acquire` load on the read fast path.
+///
+/// The op is `Release`-written right after each apply's `publish()` and
+/// `Acquire`-read; 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.
+///
+/// A `std::sync::Mutex` guards the waiter list, not a `tokio` one: it is taken
+/// and dropped inside [`Self::advance`] and inside one `poll`, never across an
+/// `.await`, and it has to be `Sync` because the writer is shard 0's thread
+/// while the sleepers are on every shard.
+#[derive(Debug, Default)]
+pub struct AppliedFrontier {
+ op: AtomicU64,
+ waiters: Mutex<Waiters>,
+}
+
+/// Registered waits, keyed by an id so a re-poll can refresh its own waker and
+/// a dropped wait (an HTTP client that disconnected mid-read) can remove it.
+#[derive(Debug, Default)]
+struct Waiters {
+ next_id: u64,
+ entries: Vec<Waiter>,
+}
+
+#[derive(Debug)]
+struct Waiter {
+ id: u64,
+ target: u64,
+ waker: Waker,
+}
+
+impl AppliedFrontier {
+ /// Highest metadata op this NODE has applied and published.
+ #[must_use]
+ pub fn get(&self) -> u64 {
+ self.op.load(Ordering::Acquire)
+ }
+
+ /// Publish `op` as applied and wake every read waiting at or below it.
+ /// Monotone, so a lower value is a no-op and wakes nobody.
+ ///
+ /// 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(&self, op: u64) {
+ if self.op.fetch_max(op, Ordering::Release) >= op {
+ return;
+ }
+ let mut waiters =
self.waiters.lock().unwrap_or_else(PoisonError::into_inner);
+ waiters.entries.retain(|waiter| {
+ if waiter.target > op {
+ return true;
+ }
+ waiter.waker.wake_by_ref();
Review Comment:
warning: `wake_by_ref` runs inside `retain` with the mutex held, on the
commit path. every woken reader then blocks on that same lock in `deregister`.
collect the wakers, drop the guard, then wake.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -250,7 +459,13 @@ pub(in crate::dispatch) async fn
handle_non_replicated_request<B, MJ, S, SB>(
}
SYNC_CONSUMER_GROUP_CODE => {
Review Comment:
warning: the poll and consumer-offset arms above skip the gate, but
`authorize_partition_read` and `resolve_poll_request` both read the metadata
STM. create a topic then poll on a lagging node and you get a silent empty poll
or a spurious 403.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +128,187 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// Budget one held read is given before it fails retryable, in consensus
+/// ticks: six commit-broadcast intervals.
+///
+/// Sized for a node merely behind on its commit walk, NOT for a view change --
+/// detecting one costs `heartbeat_timeout` and escalating it another
+/// `view_change_status_timeout`, and `recovery_barrier_deadline` budgets at
+/// least 15s for the same event, so a read that waits out an election is a
read
+/// the caller should retry elsewhere. Far below the SDK's 30s request budget,
+/// inside which it replays the same id on the same connection.
+///
+/// In ticks rather than a bare duration because that is the unit the thing
+/// being waited for moves in, and the unit the simulator steps in.
+#[allow(clippy::cast_possible_truncation)]
+pub const READ_FRONTIER_BUDGET_TICKS: u32 = 6 *
TimeoutManager::COMMIT_MESSAGE_TICKS as u32;
Review Comment:
critical: `COMMIT_MESSAGE_TICKS` is only the compile-time default -
`set_commit_message_ticks` overrides it at runtime from `[cluster]
commit_broadcast_interval`. at 2s this budget is 1.5 broadcasts, not six, so a
merely lagging backup gets spurious 503s. derive it from the configured ticks.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +128,187 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// Budget one held read is given before it fails retryable, in consensus
+/// ticks: six commit-broadcast intervals.
+///
+/// Sized for a node merely behind on its commit walk, NOT for a view change --
+/// detecting one costs `heartbeat_timeout` and escalating it another
+/// `view_change_status_timeout`, and `recovery_barrier_deadline` budgets at
+/// least 15s for the same event, so a read that waits out an election is a
read
+/// the caller should retry elsewhere. Far below the SDK's 30s request budget,
+/// inside which it replays the same id on the same connection.
+///
+/// In ticks rather than a bare duration because that is the unit the thing
+/// being waited for moves in, and the unit the simulator steps in.
+#[allow(clippy::cast_possible_truncation)]
+pub const READ_FRONTIER_BUDGET_TICKS: u32 = 6 *
TimeoutManager::COMMIT_MESSAGE_TICKS as u32;
+
+/// The same budget as a duration, for the timers the two planes measure it
+/// with.
+pub const READ_FRONTIER_BUDGET: Duration =
+ match TICK_INTERVAL.checked_mul(READ_FRONTIER_BUDGET_TICKS) {
+ Some(budget) => budget,
+ None => panic!("the read frontier budget must fit a Duration"),
+ };
+
+/// Whether `code`'s answer comes from the metadata state machine, and so must
+/// not be served below the caller's watermark.
+///
+/// The two named 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. A code this build does not know is excluded
+/// too: its only outcome is `InvalidCommand`, and parking a terminal error for
+/// the whole budget serves nobody.
+///
+/// A deny-list otherwise, so a read code added later is gated by default: the
+/// failure mode of forgetting to add one is a wait, while forgetting to add it
+/// to an allow-list is a silent stale read.
+///
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
+ !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE)
+ && lookup_command(code).is_some()
+}
+
+/// Whether a frontier wait actually parked.
+///
+/// The caller's authorization resolved its scope off the pre-wait state
+/// machine, and a wait that parked is one where that state machine moved, so
+/// only the parked outcome forces the gate to run again.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FrontierWait {
+ /// The frontier already covered the watermark: no await ran.
+ Ready,
+ /// The read parked, and the frontier caught up while it waited.
+ CaughtUp,
+}
+
+/// The frontier never reached the watermark inside the budget. Each plane
+/// renders it in its own error currency.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FrontierUnreached;
+
+/// Hold a read until `frontier` covers `watermark`, or until `budget` expires.
+///
+/// Event-driven: the wait is woken by the commit that advances the frontier
+/// (see [`AppliedFrontier::advance`]), so a read resumes on the commit it was
+/// short of rather than on a poll that happens to land after it. The caller
+/// supplies the budget as a future because the two read planes measure time
+/// differently -- the shard bus timer, which is virtual under the simulator,
+/// against `compio::time` on the HTTP listener.
+///
+/// Shared by both planes because one wait with two copies is one wait with a
+/// drift vector, and split from its callers so the fast path, the park, the
+/// wake and the expiry are all testable without a live shard.
+///
+/// A caller with nothing to read back has `watermark == 0`, which the first
+/// comparison satisfies: one `Acquire` load, no registration, no await. Expiry
+/// is loud and carries both numbers, so a frontier that stopped moving is
+/// visible instead of showing up as latency.
+pub async fn hold_for_frontier(
+ frontier: &AppliedFrontier,
+ watermark: u64,
+ budget: impl Future<Output = ()>,
+) -> Result<FrontierWait, FrontierUnreached> {
+ if frontier.get() >= watermark {
+ return Ok(FrontierWait::Ready);
+ }
+ let reached = pin!(frontier.reached(watermark));
+ let budget = pin!(budget);
+ match select(reached, budget).await {
+ Either::Left(((), _)) => Ok(FrontierWait::CaughtUp),
+ Either::Right(((), _)) => {
+ warn!(
+ frontier = frontier.get(),
+ watermark,
+ budget = ?READ_FRONTIER_BUDGET,
+ "metadata read frontier unreached inside the budget; failing
the read retryable"
+ );
+ Err(FrontierUnreached)
+ }
+ }
+}
+
+/// 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::dispatch::session_ops`]) -- 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. A park costs this connection more than the
+/// read itself: the per-connection drain loop serves one frame at a time, so
Review Comment:
warning: the per-client queue in `dispatch/mod.rs` is an uncapped `VecDeque`
drained one frame at a time, so a park here lets that client's pipelined frames
pile up unbounded for the whole budget. cap it, or cut the binary budget.
##########
core/metadata/src/applied_frontier.rs:
##########
@@ -0,0 +1,300 @@
+// 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 node-wide applied metadata frontier and the wait a read parks on.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Mutex;
+use std::sync::PoisonError;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::task::{Context, Poll, Waker};
+
+/// Highest metadata op whose apply has been PUBLISHED on this node, shared by
+/// every shard, plus the wakers of the reads waiting for it to reach them.
+///
+/// `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 cell
+/// does, for one `Acquire` load on the read fast path.
+///
+/// The op is `Release`-written right after each apply's `publish()` and
+/// `Acquire`-read; 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.
+///
+/// A `std::sync::Mutex` guards the waiter list, not a `tokio` one: it is taken
+/// and dropped inside [`Self::advance`] and inside one `poll`, never across an
+/// `.await`, and it has to be `Sync` because the writer is shard 0's thread
+/// while the sleepers are on every shard.
+#[derive(Debug, Default)]
+pub struct AppliedFrontier {
+ op: AtomicU64,
+ waiters: Mutex<Waiters>,
+}
+
+/// Registered waits, keyed by an id so a re-poll can refresh its own waker and
+/// a dropped wait (an HTTP client that disconnected mid-read) can remove it.
+#[derive(Debug, Default)]
+struct Waiters {
+ next_id: u64,
+ entries: Vec<Waiter>,
+}
+
+#[derive(Debug)]
+struct Waiter {
+ id: u64,
+ target: u64,
+ waker: Waker,
+}
+
+impl AppliedFrontier {
+ /// Highest metadata op this NODE has applied and published.
+ #[must_use]
+ pub fn get(&self) -> u64 {
+ self.op.load(Ordering::Acquire)
+ }
+
+ /// Publish `op` as applied and wake every read waiting at or below it.
+ /// Monotone, so a lower value is a no-op and wakes nobody.
+ ///
+ /// 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(&self, op: u64) {
+ if self.op.fetch_max(op, Ordering::Release) >= op {
+ return;
+ }
+ let mut waiters =
self.waiters.lock().unwrap_or_else(PoisonError::into_inner);
Review Comment:
nit: takes the mutex on every advancing commit even when nothing is parked,
which is the normal case. guard it with a relaxed waiter count.
##########
core/simulator/src/lib.rs:
##########
@@ -5097,3 +5105,377 @@ 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::dispatch::session_ops`).
+ //!
+ //! 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, and what is left of that budget is [`CONVERGE_STEPS`].
+ const STALE_WINDOW_STEPS: u32 = 50;
+
+ /// Steps left for repair to reach the backup and the held read to answer
+ /// once replication is restored.
+ ///
+ /// Derived, not chosen: one `sim.step()` advances the virtual clock by one
+ /// consensus tick, the unit the gate's budget is denominated in, so phase
1
+ /// spends `STALE_WINDOW_STEPS` of that budget and what remains is the
whole
+ /// window the read can still be answered in. A larger number would just
+ /// spin past an expiry the status assertion below already caught.
+ const CONVERGE_STEPS: u32 = server::METADATA_READ_FRONTIER_BUDGET_TICKS -
STALE_WINDOW_STEPS;
Review Comment:
nit: `STALE_WINDOW_STEPS + CONVERGE_STEPS` is exactly the budget, so repair
one tick slow flips this to the expiry path and the status assert fails
confusingly. leave margin and cover expiry in its own case.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +128,187 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// Budget one held read is given before it fails retryable, in consensus
+/// ticks: six commit-broadcast intervals.
+///
+/// Sized for a node merely behind on its commit walk, NOT for a view change --
+/// detecting one costs `heartbeat_timeout` and escalating it another
+/// `view_change_status_timeout`, and `recovery_barrier_deadline` budgets at
+/// least 15s for the same event, so a read that waits out an election is a
read
+/// the caller should retry elsewhere. Far below the SDK's 30s request budget,
+/// inside which it replays the same id on the same connection.
+///
+/// In ticks rather than a bare duration because that is the unit the thing
+/// being waited for moves in, and the unit the simulator steps in.
+#[allow(clippy::cast_possible_truncation)]
+pub const READ_FRONTIER_BUDGET_TICKS: u32 = 6 *
TimeoutManager::COMMIT_MESSAGE_TICKS as u32;
+
+/// The same budget as a duration, for the timers the two planes measure it
+/// with.
+pub const READ_FRONTIER_BUDGET: Duration =
+ match TICK_INTERVAL.checked_mul(READ_FRONTIER_BUDGET_TICKS) {
+ Some(budget) => budget,
+ None => panic!("the read frontier budget must fit a Duration"),
+ };
+
+/// Whether `code`'s answer comes from the metadata state machine, and so must
+/// not be served below the caller's watermark.
+///
+/// The two named 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. A code this build does not know is excluded
+/// too: its only outcome is `InvalidCommand`, and parking a terminal error for
+/// the whole budget serves nobody.
+///
+/// A deny-list otherwise, so a read code added later is gated by default: the
Review Comment:
nit: the deny-list only covers codes reaching the `_` arm - every named arm
opts in by hand, and three don't. drop the claim or move the gate to one call
at the top of the match.
##########
core/integration/tests/server/http_read_your_writes.rs:
##########
@@ -0,0 +1,137 @@
+// 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.
+
+//! Read-your-writes over the REST listener, end to end: an unqualified read
+//! must never answer below the metadata op the same caller was already told
+//! committed.
+//!
+//! The window is a node that HANDED OUT a committed op it has not applied yet.
+//! On the REST plane that node is a follower running a `Register`: the session
+//! its first authenticated request mints is forwarded to the primary
+//! (`dispatch::submit_register_local_or_forward`), so the follower answers
with
+//! an epoch its own commit walk can still be behind. Everything the metadata
+//! group committed below that epoch is therefore state the caller has been
+//! promised and this follower may not have applied.
+//!
+//! Each round seeds a stream through the primary over TCP, then authenticates
+//! on the follower, which binds an epoch above that create. The follower's
next
+//! read is the assertion: it must not answer from before the stream existed.
+//! Logout is the authenticated request that binds it, deliberately - it tears
+//! the session entry down again, so the floor the read waits on has to outlive
+//! the session that established it.
+//!
+//! The seeding runs over TCP rather than the primary's own REST listener so
the
+//! only HTTP sessions in play are the follower's: a second long-lived REST
+//! session would be competing for VSR client ids with the fresh register each
+//! round mints, which is a different subject.
+//!
+//! The suite asserts the GUARANTEE, not the mechanism: whether the gate parked
+//! is invisible from outside, and on a fast local cluster the follower often
Review Comment:
warning: the header above concedes the follower usually applies within the
tick, so this passes with the gate deleted. it pins the wiring, not the
guarantee - the binary plane has a deterministic simulator proof, this one has
none.
##########
core/server/src/responses.rs:
##########
@@ -1571,23 +1610,14 @@ pub fn build_raw_pat_reply(
bytemuck::checked::try_from_bytes::<ReplyHeader>(&committed.as_slice()[..header_len])
Review Comment:
warning: `[..header_len]` panics on a frame shorter than `ReplyHeader`,
where the twin in `dispatch/submit.rs` uses `.get()` and warns. use a checked
slice.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +128,187 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// Budget one held read is given before it fails retryable, in consensus
+/// ticks: six commit-broadcast intervals.
+///
+/// Sized for a node merely behind on its commit walk, NOT for a view change --
+/// detecting one costs `heartbeat_timeout` and escalating it another
+/// `view_change_status_timeout`, and `recovery_barrier_deadline` budgets at
+/// least 15s for the same event, so a read that waits out an election is a
read
+/// the caller should retry elsewhere. Far below the SDK's 30s request budget,
+/// inside which it replays the same id on the same connection.
+///
+/// In ticks rather than a bare duration because that is the unit the thing
+/// being waited for moves in, and the unit the simulator steps in.
+#[allow(clippy::cast_possible_truncation)]
+pub const READ_FRONTIER_BUDGET_TICKS: u32 = 6 *
TimeoutManager::COMMIT_MESSAGE_TICKS as u32;
+
+/// The same budget as a duration, for the timers the two planes measure it
+/// with.
+pub const READ_FRONTIER_BUDGET: Duration =
+ match TICK_INTERVAL.checked_mul(READ_FRONTIER_BUDGET_TICKS) {
+ Some(budget) => budget,
+ None => panic!("the read frontier budget must fit a Duration"),
+ };
+
+/// Whether `code`'s answer comes from the metadata state machine, and so must
+/// not be served below the caller's watermark.
+///
+/// The two named 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. A code this build does not know is excluded
+/// too: its only outcome is `InvalidCommand`, and parking a terminal error for
+/// the whole budget serves nobody.
+///
+/// A deny-list otherwise, so a read code added later is gated by default: the
+/// failure mode of forgetting to add one is a wait, while forgetting to add it
+/// to an allow-list is a silent stale read.
+///
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
+ !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE)
+ && lookup_command(code).is_some()
+}
+
+/// Whether a frontier wait actually parked.
+///
+/// The caller's authorization resolved its scope off the pre-wait state
+/// machine, and a wait that parked is one where that state machine moved, so
+/// only the parked outcome forces the gate to run again.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FrontierWait {
+ /// The frontier already covered the watermark: no await ran.
+ Ready,
+ /// The read parked, and the frontier caught up while it waited.
+ CaughtUp,
+}
+
+/// The frontier never reached the watermark inside the budget. Each plane
+/// renders it in its own error currency.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FrontierUnreached;
+
+/// Hold a read until `frontier` covers `watermark`, or until `budget` expires.
+///
+/// Event-driven: the wait is woken by the commit that advances the frontier
+/// (see [`AppliedFrontier::advance`]), so a read resumes on the commit it was
+/// short of rather than on a poll that happens to land after it. The caller
+/// supplies the budget as a future because the two read planes measure time
+/// differently -- the shard bus timer, which is virtual under the simulator,
+/// against `compio::time` on the HTTP listener.
+///
+/// Shared by both planes because one wait with two copies is one wait with a
+/// drift vector, and split from its callers so the fast path, the park, the
+/// wake and the expiry are all testable without a live shard.
+///
+/// A caller with nothing to read back has `watermark == 0`, which the first
+/// comparison satisfies: one `Acquire` load, no registration, no await. Expiry
+/// is loud and carries both numbers, so a frontier that stopped moving is
+/// visible instead of showing up as latency.
+pub async fn hold_for_frontier(
+ frontier: &AppliedFrontier,
+ watermark: u64,
+ budget: impl Future<Output = ()>,
+) -> Result<FrontierWait, FrontierUnreached> {
+ if frontier.get() >= watermark {
+ return Ok(FrontierWait::Ready);
+ }
+ let reached = pin!(frontier.reached(watermark));
+ let budget = pin!(budget);
+ match select(reached, budget).await {
+ Either::Left(((), _)) => Ok(FrontierWait::CaughtUp),
+ Either::Right(((), _)) => {
+ warn!(
+ frontier = frontier.get(),
+ watermark,
+ budget = ?READ_FRONTIER_BUDGET,
Review Comment:
nit: one `warn!` per expired read is a warn per client per budget window on
a lagging backup - a flood in exactly the case it's meant to surface. drop to
`debug!` plus a counter.
##########
core/server/src/dispatch/submit.rs:
##########
@@ -192,3 +196,143 @@ where
});
rx.recv().await.ok().flatten()
}
+
+/// The commit position a SUCCESSFULLY COMMITTED metadata reply carries, or
+/// `None` when the frame promises the caller nothing.
+///
+/// Only a success promises. Every other frame on this path stamps `commit`
with
+/// the primary's `commit_max`, an op the caller was never told committed and,
+/// on a backup-homed caller, one its own reads would then wait for:
+///
+/// - 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 request-level denial names itself in `ReplyHeader.status`, the channel
+/// the SDK peeks before body decode (see `build_deny_reply`);
+/// - a transient rejection did not commit and will be replayed;
+/// - a TERMINAL pre-consensus rejection (`PreflightOutcome::Reject`, e.g. a
+/// fenced session) is a result section carrying a non-transient code, which
+/// is byte-identical to a COMMITTED business rejection (duplicate name, bad
+/// expiry). Neither is separable here, and neither has to be: a rejection
+/// mutated nothing, so the caller has nothing to read back from it, and
+/// grading both as no-promise is the only reading that cannot make a read
+/// wait for an op that never committed.
+///
+/// Every reply on this path is result-framed (`Operation::is_result_framed`
+/// covers every metadata op; the partition plane grades through
+/// `classify_partition_reply` instead), so a missing result section is a
+/// malformed frame, not a bare payload.
+///
+/// Shared with the HTTP write path, which grades the same 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 fn committed_reply_commit(reply: &Message<GenericHeader>) -> Option<u64> {
Review Comment:
simplification: this and `build_raw_pat_reply` run the same three checks on
the same frames and have to stay in step. extract one `committed_reply_header`
helper both call.
##########
core/server/src/session_manager.rs:
##########
@@ -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);
Review Comment:
nit: `metadata_watermark` survives the demotion above and isn't cleared on
`login`, so a different user on the same socket inherits the old floor. only
over-waits, never serves stale, but worth resetting.
--
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]
