hubcio commented on code in PR #4024:
URL: https://github.com/apache/iggy/pull/4024#discussion_r3925553734
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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;
+
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
Review Comment:
warning: this predicate is only consulted in the `_` arm, so `GET_ME`,
`GET_CLIENTS`, `GET_CLIENT` and `SYNC_CONSUMER_GROUP` read the metadata state
machine ungated. gate those arms individually - hoisting above the match would
park `PING` and `POLL_MESSAGES`, since this is a deny-list.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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;
+
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
+ !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE)
+}
+
+/// 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. 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)
Review Comment:
warning: every SDK replays `TransientNotCommitted` on the same connection
and never fails over, so a durably lagging replica burns the client's full 30s
read timeout and then errors. use `TransientNotAccepted` - a read never entered
a pipeline, and it's the code that drives the roster walk.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -254,6 +339,14 @@ pub(in crate::dispatch) 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)
Review Comment:
warning: this parks before authorization and before decode, so an
unauthorized read or an unknown command code holds the connection for the full
budget before its terminal error. move the gate after `authorize_default_read`
- the HTTP side already does this and says why.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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;
Review Comment:
warning: 3s can't ride out a view change - detection needs
`heartbeat_timeout` (5s) and escalation another `view_change_status_timeout`
(5s), and `recovery_barrier_deadline` budgets at least 15s for the same event.
derive the deadline from those timeouts, or restate this as ~6 commit broadcast
intervals.
##########
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,
Review Comment:
warning: `POST /users/refresh-token` mints a fresh `jti` without carrying
the watermark or registering a session, so reads on the refreshed bearer hit an
unknown key, get 0, and the gate no-ops. key the watermark per user id, or
carry it in `issue_identity`.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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;
+
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
+ !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE)
+}
+
+/// 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. 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 {
Review Comment:
warning: this park sits inside the per-socket drain loop, which awaits one
frame at a time, so a held metadata read also blocks that connection's queued
`SendMessages`, `PollMessages` and `PING`. the doc above says only this
connection waits - it doesn't say the data plane waits with it.
##########
core/server/src/http/reads.rs:
##########
@@ -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)?;
Review Comment:
warning: the scope was resolved off the pre-wait state machine but the read
re-resolves after it. an entity created during the wait is served with no
permissioner call at all, since a scope miss maps to `Ok(())`. re-resolve the
scope after the wait.
##########
core/simulator/src/lib.rs:
##########
@@ -5097,3 +5105,370 @@ 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.
+ 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;
Review Comment:
warning: `CONVERGE_STEPS` is unreachable. one `sim.step()` is 10ms, which is
one gate poll, so phase 1 burns 50 of the gate's 300 and the read fails at poll
300 - the status assert trips first and this panic never runs. derive it as
`READ_FRONTIER_MAX_POLLS - STALE_WINDOW_STEPS`.
##########
core/server/src/dispatch/submit.rs:
##########
@@ -192,3 +195,27 @@ where
});
rx.recv().await.ok().flatten()
}
+
+/// 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 fn committed_reply_commit(reply: &Message<GenericHeader>) -> Option<u64> {
Review Comment:
nit: `committed_reply_commit` has no unit test, and the binary gate's expiry
arm has never run in one. a table over committed reply, transient rejection,
terminal rejection and eviction would pin it.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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);
Review Comment:
nit: the doc calls this the consensus tick then redefines 10ms by hand;
`consensus::TICK_INTERVAL` is public and already reachable from here. the same
pair is declared again in `http/reads.rs`, and a third 10ms lives in its
recovery barrier.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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;
+
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
+ !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE)
+}
+
+/// 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. 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);
Review Comment:
nit: `read_context` already probed this connection a few lines down, so this
is a second borrow and hash lookup. it has exactly one caller - widen its tuple
and the borrow-scoping comment above goes away too.
##########
core/server/src/http/submit.rs:
##########
@@ -113,6 +113,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);
Review Comment:
warning: on a follower with HTTP forwarding on, control-plane writes are
relayed before the handler runs, so this never fires there and the watermark
stays frozen at the register epoch. the HTTP read-your-writes guarantee doesn't
hold on a forwarding follower.
##########
core/server/src/dispatch/submit.rs:
##########
@@ -192,3 +195,27 @@ where
});
rx.recv().await.ok().flatten()
}
+
+/// 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
Review Comment:
nit: the doc says pre-consensus rejections are excluded, but the filter only
matches codes 57 and 58, and `ReplyHeader.status` - the documented pre-commit
denial channel - is never read. add `status != 0`, stamp `commit: 0` on the
terminal rejection arms, and fix the doc.
##########
core/server/src/http/reads.rs:
##########
@@ -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;
Review Comment:
nit: reads poll uncapped here while writes are capped. compio coalesces
these into one driver timeout so the cost is small, but an event-driven wait
off `fire_commit_notifier` would be the real fix.
##########
core/server/src/http/session.rs:
##########
@@ -269,6 +297,7 @@ mod tests {
data_gate: Mutex::new(FIRST_REQUEST_ID),
registry_token: Cell::new(None),
in_flight_writes: Cell::new(0),
+ metadata_watermark: Cell::new(1),
Review Comment:
nit: this fixture spells the epoch as a literal while `fake_session` below
uses `FIXTURE_EPOCH`. it's field-for-field `fake_session("jwt:test", 7,
u64::MAX)` - delete it and call the helper.
##########
core/server/src/http/reads.rs:
##########
@@ -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(
Review Comment:
nit: no end-to-end test of the HTTP gate at all - seed, record, hold, serve.
the binary plane got two simulator tests; `hold_for_frontier` is only exercised
against a closure.
##########
bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java:
##########
@@ -142,8 +142,14 @@ public void deleteStreamByNumericId() {
@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),
Review Comment:
nit: `!name.equals(lastStreamName)` is vacuously true when `lastStreamName`
is null. not reachable in this feature today, but the rust twin fails loud on a
missing pre-value - add `assertNotNull`.
##########
core/server/src/http/state.rs:
##########
@@ -229,6 +229,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 {
Review Comment:
warning: `forget_session` removes the entry outright on eviction, so this
returns 0 again while the bearer stays valid. a caller that had written then
reads ungated until its next write re-registers - keep a key-scoped floor
outside the session entry.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -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
Review Comment:
nit: `applied_frontier` collides with `InstallOutcome::applied_frontier`,
and the two diverge - install advances the cell to `snapshot_seq` but reports
`snapshot_seq.max(local_applied)`. rename one.
##########
core/server/src/dispatch/submit.rs:
##########
@@ -29,9 +29,12 @@ use crate::dispatch::session_ops::{
submit_register_local_or_forward,
};
use crate::dispatch::upgrade_shard_handle;
+use crate::http::reply::transient_code;
Review Comment:
simplification: this closes a cycle - `http/submit.rs` imports
`committed_reply_commit` from here, which calls back into `http::reply`. move
`transient_code` to `crate::responses`, which both planes already import and
which already open-codes `reply_body`.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -122,6 +123,83 @@ async fn handle_get_me<B, MJ, S, SB>(
.await;
}
+/// 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;
+
+/// 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 const fn read_needs_metadata_frontier(code: u32) -> bool {
+ !matches!(code, DESCRIBE_OPTIONS_CODE | GET_CLUSTER_METADATA_CODE)
+}
+
+/// 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. 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>(
Review Comment:
simplification: this loop is a second copy of `hold_for_frontier` in
`http/reads.rs`, differing only in sleep source and error type. make that one
generic over the sleep and call it from both - about 17 lines and one drift
vector.
##########
core/server/src/boot/mod.rs:
##########
@@ -580,7 +587,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.
Review Comment:
simplification: this seed is copy-pasted verbatim in
`simulator/src/replica.rs`. fold it into one
`seed_applied_frontier_from_consensus()` on the `VsrConsensus`-specialized
impl, since `C` carries no bound on the generic block.
##########
core/simulator/src/client.rs:
##########
@@ -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>();
Review Comment:
simplification: this block and `poll_messages` above are 21 identical lines
apart from code, group and panic text. one `non_replicated_request(code, group,
body)` helper - it must not delegate to `header()`, which advances the request
counter.
##########
core/server/src/dispatch/submit.rs:
##########
@@ -192,3 +195,27 @@ where
});
rx.recv().await.ok().flatten()
}
+
+/// 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 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>())?;
Review Comment:
nit: two silent drops here - the slice `?` swallows a short frame and
`.ok()` swallows a bad cast, neither logs. `warn!` on both.
##########
core/server/src/http/reads.rs:
##########
@@ -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!(
Review Comment:
nit: `/cluster/metadata` is a local route with its own handler and never
calls `read_local`, so this pins a code the HTTP plane can't produce. drop it,
or say it's asserted for the shared predicate.
--
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]
