This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/iggy-server-hardening in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c93fa9c648211d396a4a0acec85955ef2404caf2 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Jul 24 13:48:00 2026 +0200 fix(server-ng): fail reads loud when restart recovery outruns its deadline The post-restart read barrier waited a fixed 15s for the recovered WAL suffix to re-apply, then warned and served the local state machine anyway - silently returning rolled-back state, including login validating against rolled-back credentials. The 15s was sized for the old fixed 5s heartbeat; heartbeat_timeout is now operator-tunable with no ceiling, so a slower heartbeat makes worst-case recovery (election plus ViewChangeStatus backstop plus suffix recommit, roughly heartbeat + 7s) outrun a fixed deadline. Scale the deadline to max(15s, 3 x heartbeat_timeout), computed at the barrier arm site and stored on the consensus object beside the barrier, and turn expiry into a structured failure: the read path returns a retryable 503 with Retry-After, matching the not-primary response shape, instead of serving data the client may have seen rolled back. Login handlers map the same expiry to their existing transient 503. Saturating multiply because the heartbeat has no config ceiling. --- core/consensus/src/impls.rs | 17 ++++++ core/server-ng/src/bootstrap.rs | 70 +++++++++++++++++++++- core/server-ng/src/http/error.rs | 27 +++++++++ core/server-ng/src/http/handlers.rs | 15 ++++- core/server-ng/src/http/reads.rs | 114 ++++++++++++++++++++++++++++-------- 5 files changed, 215 insertions(+), 28 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index d81c308f7..a7188e94d 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -37,6 +37,7 @@ use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE}; use std::cell::{Cell, RefCell}; use std::collections::VecDeque; use std::rc::Rc; +use std::time::Duration; /// Injected time source for primary-stamped prepare timestamps. /// @@ -790,6 +791,10 @@ where /// Commit point the recovered WAL suffix must re-reach before admitting /// client requests as primary (`0` = no recovered suffix pending). recovery_barrier: Cell<u64>, + /// Wall-clock budget the recovered suffix has to re-commit before a waiter + /// on [`Self::recovery_barrier`] gives up. Armed together with the barrier; + /// `ZERO` when no suffix is pending (barrier `0`), which no waiter reads. + recovery_deadline: Cell<Duration>, /// True while this replica declines the primaryship its (stale) recovered /// view assigns it (see `init_as_backup`). `is_primary()` is pure view /// math, so without this flag a restarted view-N primary would still pass @@ -915,6 +920,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { view: Cell::new(0), log_view: Cell::new(0), recovery_barrier: Cell::new(0), + recovery_deadline: Cell::new(Duration::ZERO), ceded_primaryship: Cell::new(false), status: Cell::new(Status::Recovering), sequencer: LocalSequencer::new(0), @@ -1135,6 +1141,17 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.recovery_barrier.set(required_commit); } + /// Deadline paired with [`Self::recovery_barrier`]; only meaningful while the + /// barrier is armed (non-zero). + #[must_use] + pub const fn recovery_deadline(&self) -> Duration { + self.recovery_deadline.get() + } + + pub fn set_recovery_deadline(&self, deadline: Duration) { + self.recovery_deadline.set(deadline); + } + pub fn set_view(&mut self, view: u32) { self.view.set(view); } diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index a626b6a3c..58021ac40 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -960,6 +960,7 @@ async fn shard_main( cluster_heartbeat_ticks(config), commit_broadcast_ticks(config), prepare_retransmit_ticks(config), + recovery_barrier_deadline(config.cluster.heartbeat_timeout.get_duration()), ); (Some(consensus), Some(journal), snapshot) } else { @@ -1707,6 +1708,32 @@ pub(crate) fn cluster_heartbeat_ticks(config: &ServerNgConfig) -> u64 { duration_to_ticks(config.cluster.heartbeat_timeout.get_duration()) } +/// Floor for the post-restart read-recovery deadline (see +/// [`recovery_barrier_deadline`]). At and below the 5s default heartbeat the +/// worst-case recovery is dominated by the heartbeat-independent term - the +/// `ViewChangeStatus` backstop plus election ceremony and suffix recommit, +/// empirically ~7s - so the scaled value must never fall under this or a +/// fast-heartbeat cluster would 503 legitimate reads mid-recovery. +const RECOVERY_BARRIER_DEADLINE_FLOOR: Duration = Duration::from_secs(15); + +/// Heartbeat multiplier for the recovery deadline: a slower heartbeat stretches +/// election and suffix recommit proportionally. 3x reproduces the empirically +/// chosen 15s margin at the 5s default (3 x 5s = 15s) and holds that safety +/// factor as the heartbeat grows. +const RECOVERY_BARRIER_HEARTBEAT_MULTIPLIER: u32 = 3; + +/// How long the post-restart read path waits for the recovered WAL suffix to +/// re-commit before failing loud (retryable 503): the larger of the +/// heartbeat-independent floor and a heartbeat-scaled window. Derived from +/// `[cluster] heartbeat_timeout` rather than its own knob so the two cannot +/// drift; see `await_recovery_barrier` for the read-side wait. +pub(crate) fn recovery_barrier_deadline(heartbeat: Duration) -> Duration { + // saturating: heartbeat_timeout has no config ceiling, plain `*` panics + heartbeat + .saturating_mul(RECOVERY_BARRIER_HEARTBEAT_MULTIPLIER) + .max(RECOVERY_BARRIER_DEADLINE_FLOOR) +} + /// `[cluster] commit_broadcast_interval` in consensus ticks: how often the /// primary broadcasts its commit point, the cluster's liveness feed. Applied /// to every consensus group, matching `cluster_heartbeat_ticks`. @@ -1734,6 +1761,7 @@ fn restore_metadata_consensus( normal_heartbeat_ticks: u64, commit_message_ticks: u64, prepare_ticks: u64, + recovery_deadline: Duration, ) -> VsrConsensus<Rc<IggyMessageBus>> { let mut consensus = VsrConsensus::new( cluster_id, @@ -1813,10 +1841,11 @@ fn restore_metadata_consensus( // primary; via StartView adoption + the local commit walk on a rejoined // backup), serving reads would show pre-restart state that clients already // saw acked -- gate them on the barrier regardless of role. If the suffix - // never committed cluster-wide, the barrier times out on the read path and - // serving resumes (`await_recovery_barrier`). + // never re-commits cluster-wide, the read path fails loud with a retryable + // 503 once the paired deadline expires (`await_recovery_barrier`). if commit_watermark < restored_op { consensus.set_recovery_barrier(restored_op); + consensus.set_recovery_deadline(recovery_deadline); } // Re-pipeline the prepared-but-uncommitted suffix so the primary's @@ -3112,6 +3141,43 @@ mod tests { ); } + #[test] + fn recovery_barrier_deadline_holds_the_floor_for_small_heartbeats() { + // Below the 5s default the heartbeat-independent recovery term (~7s of + // ViewChangeStatus backstop plus ceremony) dominates, so the floor + // governs however small the heartbeat is; 3 x 5s lands exactly on it. + assert_eq!( + recovery_barrier_deadline(Duration::from_secs(1)), + RECOVERY_BARRIER_DEADLINE_FLOOR + ); + assert_eq!( + recovery_barrier_deadline(Duration::from_secs(5)), + RECOVERY_BARRIER_DEADLINE_FLOOR + ); + } + + #[test] + fn recovery_barrier_deadline_scales_past_the_floor_for_large_heartbeats() { + // Once 3 x heartbeat clears the floor the scaled window governs, so a + // slow-heartbeat cluster is not failed 503 before its longer recovery + // can finish. + assert_eq!( + recovery_barrier_deadline(Duration::from_secs(10)), + Duration::from_secs(30) + ); + assert_eq!( + recovery_barrier_deadline(Duration::from_secs(15)), + Duration::from_secs(45) + ); + } + + #[test] + fn recovery_barrier_deadline_saturates_instead_of_panicking() { + // heartbeat_timeout has no config ceiling, so the multiply must + // saturate rather than abort boot on an absurd parseable value. + assert_eq!(recovery_barrier_deadline(Duration::MAX), Duration::MAX); + } + #[test] fn default_commit_broadcast_interval_matches_consensus_constant() { // The config default lives in core/server-ng/config.toml (a string, diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs index 59fa369ec..998e03238 100644 --- a/core/server-ng/src/http/error.rs +++ b/core/server-ng/src/http/error.rs @@ -411,6 +411,12 @@ pub(in crate::http) enum ReadError { /// query, so the caller re-issues the read against the leader (see /// [`primary_redirect_response`]). RedirectToPrimary(String), + /// The post-restart read-recovery barrier expired with the recovered WAL + /// suffix still uncommitted: serving now could show state that rolls back + /// history a client already saw acked. Fail-closed 503 via the shared + /// [`service_unavailable`] body, retryable once the cluster re-commits the + /// suffix. + RecoveryIncomplete, /// 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. @@ -426,6 +432,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::Timeout => gateway_timeout_response( "partition_read_timeout", "the partition owner did not answer the read in time; retry", @@ -560,4 +567,24 @@ mod tests { Some("http://[::1]:8080/streams?consistency=linearizable".to_owned()) ); } + + #[test] + fn recovery_incomplete_renders_retryable_503_like_not_primary() { + // Barrier expiry must render as the shared retryable 503: the same + // status and Retry-After hint as the not-primary 503, so an SDK treats + // it as a connection-level retry rather than a terminal error. + let recovery = ReadError::RecoveryIncomplete.into_response(); + assert_eq!(recovery.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + recovery.headers().get(RETRY_AFTER), + Some(&HeaderValue::from(RETRY_AFTER_SECONDS)) + ); + + let not_primary = ReadError::NotPrimary.into_response(); + assert_eq!(recovery.status(), not_primary.status()); + assert_eq!( + recovery.headers().get(RETRY_AFTER), + not_primary.headers().get(RETRY_AFTER) + ); + } } diff --git a/core/server-ng/src/http/handlers.rs b/core/server-ng/src/http/handlers.rs index 6ca70039d..4adf953e4 100644 --- a/core/server-ng/src/http/handlers.rs +++ b/core/server-ng/src/http/handlers.rs @@ -176,8 +176,13 @@ pub(in crate::http) async fn login_user( ) -> Result<Json<IdentityInfo>, CustomError> { // Credential verification is a consensus-free STM read; hold it while a // recovered WAL suffix (which may carry the user's create/password ops) - // re-commits, like every other local read. - SendWrapper::new(crate::http::reads::await_recovery_barrier(&state.shard)).await; + // re-commits, like every other local read. On barrier expiry, fail with a + // retryable 503 rather than validating against rolled-back credentials: + // this route's error currency is `IggyError -> CustomError`, and + // `TransientNotCommitted` is the variant `CustomError` renders 503. + SendWrapper::new(crate::http::reads::await_recovery_barrier(&state.shard)) + .await + .map_err(|_| IggyError::TransientNotCommitted)?; let user_id = verify_login_credentials( &state.shard, &command.username, @@ -191,7 +196,11 @@ pub(in crate::http) async fn login_with_personal_access_token( State(state): State<HttpState>, Json(command): Json<LoginWithPersonalAccessToken>, ) -> Result<Json<IdentityInfo>, CustomError> { - SendWrapper::new(crate::http::reads::await_recovery_barrier(&state.shard)).await; + // Same recovery-barrier wait and retryable-503-on-expiry mapping as + // `login_user`. + SendWrapper::new(crate::http::reads::await_recovery_barrier(&state.shard)) + .await + .map_err(|_| IggyError::TransientNotCommitted)?; let user_id = verify_pat_credentials(&state.shard, command.token.expose_secret()) .map_err(|error| login_error_to_iggy(&error))?; issue_identity(&state, user_id) diff --git a/core/server-ng/src/http/reads.rs b/core/server-ng/src/http/reads.rs index afc2397f6..a9b433a02 100644 --- a/core/server-ng/src/http/reads.rs +++ b/core/server-ng/src/http/reads.rs @@ -90,7 +90,7 @@ pub(in crate::http) async fn read_local( body: &[u8], rule: impl FnOnce(&Permissioner, u32) -> Result<(), IggyError>, ) -> Result<Bytes, ReadError> { - await_recovery_barrier(&state.shard).await; + await_recovery_barrier(&state.shard).await?; authorize_read(state, identity, consistency, rule)?; match build_non_replicated_response( &state.shard, @@ -106,25 +106,54 @@ pub(in crate::http) async fn read_local( } } +/// 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. +#[derive(Debug, PartialEq, Eq)] +enum BarrierWait { + /// Barrier met, or none armed: serve the read. + Ready, + /// Barrier unmet and the deadline has passed: fail loud. + Expired, + /// Barrier unmet, deadline still ahead: keep polling. + Pending, +} + +/// Decide the barrier outcome from the armed barrier, the locally applied commit +/// point, and whether the deadline has passed. A met barrier wins over an +/// expired deadline, so recovery that completes as the deadline lands still +/// serves rather than 503-ing. +const fn barrier_state(barrier: u64, commit_min: u64, expired: bool) -> BarrierWait { + if barrier == 0 || commit_min >= barrier { + BarrierWait::Ready + } else if expired { + BarrierWait::Expired + } else { + BarrierWait::Pending + } +} + /// Hold a local read while the recovered WAL suffix re-commits. /// /// Recovery re-pipelines prepared-but-uncommitted ops that clients saw /// committed before the restart; JWT-authenticated HTTP reads skip consensus /// entirely, so without this wait they can observe state that rolls back /// committed history in the first few hundred milliseconds after a restart. -/// No-op (`recovery_barrier() == 0`) outside that window. Bounded: the -/// suffix needs a backup quorum to re-commit, so serve anyway after the -/// deadline rather than wedging reads on a partitioned cluster. -pub(in crate::http) async fn await_recovery_barrier(shard: &Rc<ServerNgShard>) { - // Must outlast a full post-restart convergence: the peers' election - // (several heartbeat timeouts in the worst case) plus the new primary - // re-committing the journaled suffix. A shorter deadline expires mid - // view-change and serves pre-restart state that clients saw acked. - const DEADLINE: std::time::Duration = std::time::Duration::from_secs(15); +/// `Ok(())` immediately when no suffix is pending (`recovery_barrier() == 0`). +/// +/// Bounded by the barrier's paired deadline (scaled from the configured +/// heartbeat; see `recovery_barrier_deadline`). If the suffix has not +/// re-committed by then the read fails loud with a retryable 503 +/// ([`ReadError::RecoveryIncomplete`]) instead of silently serving pre-restart +/// state a client already saw acked; the caller retries against a converged +/// cluster. +pub(in crate::http) async fn await_recovery_barrier( + shard: &Rc<ServerNgShard>, +) -> Result<(), ReadError> { const POLL: std::time::Duration = std::time::Duration::from_millis(10); let Some(consensus) = shard.plane.metadata().consensus.as_ref() else { - return; + return Ok(()); }; let barrier = consensus.recovery_barrier(); // Gate on commit_MIN (locally applied), not commit_max (known committed): @@ -132,20 +161,24 @@ pub(in crate::http) async fn await_recovery_barrier(shard: &Rc<ServerNgShard>) { // journal applying ops, and this task interleaves with that walk at its // await points -- a commit_max gate would serve state from before the // suffix applied (e.g. a pre-restart password change not yet visible). - if barrier == 0 || consensus.commit_min() >= barrier { - return; + if barrier_state(barrier, consensus.commit_min(), false) == BarrierWait::Ready { + return Ok(()); } - let deadline = std::time::Instant::now() + DEADLINE; - while consensus.commit_min() < barrier { - if std::time::Instant::now() >= deadline { - tracing::warn!( - barrier, - commit_min = consensus.commit_min(), - "recovered suffix still unapplied past deadline; serving read anyway" - ); - return; + let deadline = std::time::Instant::now() + consensus.recovery_deadline(); + loop { + let expired = std::time::Instant::now() >= deadline; + match barrier_state(barrier, consensus.commit_min(), expired) { + BarrierWait::Ready => return Ok(()), + BarrierWait::Expired => { + tracing::warn!( + barrier, + commit_min = consensus.commit_min(), + "recovered suffix still unapplied past deadline; failing read with retryable 503" + ); + return Err(ReadError::RecoveryIncomplete); + } + BarrierWait::Pending => compio::time::sleep(POLL).await, } - compio::time::sleep(POLL).await; } } @@ -228,3 +261,38 @@ pub(in crate::http) fn authorize_data_plane( .users() .authorize(|permissioner| rule(permissioner, user_id, stream_id, topic_id)) } + +#[cfg(test)] +mod tests { + use super::{BarrierWait, barrier_state}; + + #[test] + fn barrier_state_ready_when_no_barrier_armed() { + assert_eq!(barrier_state(0, 0, false), BarrierWait::Ready); + assert_eq!(barrier_state(0, 0, true), BarrierWait::Ready); + } + + #[test] + fn barrier_state_ready_when_commit_reached_barrier() { + assert_eq!(barrier_state(5, 5, false), BarrierWait::Ready); + assert_eq!(barrier_state(5, 6, false), BarrierWait::Ready); + } + + #[test] + fn barrier_state_pending_while_unmet_before_deadline() { + assert_eq!(barrier_state(5, 3, false), BarrierWait::Pending); + } + + #[test] + fn barrier_state_expires_when_unmet_past_deadline() { + // Red before the fail-loud change: an expired barrier used to serve the + // read (a bare `()`), now an unmet barrier past its deadline is a + // distinct terminal outcome the wait maps to a retryable 503. + assert_eq!(barrier_state(5, 3, true), BarrierWait::Expired); + } + + #[test] + fn barrier_state_met_wins_over_expired_deadline() { + assert_eq!(barrier_state(5, 5, true), BarrierWait::Ready); + } +}
