diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 10100a435..dc275bc81 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -166,7 +166,7 @@ uuid = { workspace = true, features = ["v4", "fast-rng", "serde", "macro-diagnos reed-solomon-erasure = { workspace = true, features = ["simd-accel"] } reed-solomon-simd = { workspace = true } lazy_static.workspace = true -moka = { workspace = true, features = ["future"] } +moka = { workspace = true, features = ["future", "sync"] } rustfs-lock.workspace = true rustfs-io-metrics.workspace = true regex = { workspace = true } @@ -185,7 +185,7 @@ hyper-rustls = { workspace = true, default-features = false, features = ["native hostname.workspace = true rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] } rustls-pki-types.workspace = true -tokio = { workspace = true, features = ["io-util", "sync", "signal", "fs", "rt-multi-thread"] } +tokio = { workspace = true, features = ["io-util", "sync", "signal", "fs", "rt-multi-thread", "time"] } tonic = { workspace = true, features = ["gzip", "deflate"] } xxhash-rust = { workspace = true, features = ["xxh64", "xxh3"] } tower = { workspace = true, features = ["timeout"] } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 9fdb7fca9..6fed68316 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -146,6 +146,14 @@ pub mod bucket { } pub mod on_demand_migration { + pub use crate::bucket::on_demand_migration::{ + ApplyOutcome, BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, + Breaker, BreakerState, BreakerTransition, BreakerVerdict, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, GaugeGuard, + LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup, + OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason, + PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS, + SourceLatencySnapshot, source_client_spec, + }; pub use crate::bucket::on_demand_migration::{ ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, diff --git a/crates/ecstore/src/bucket/on_demand_migration/breaker.rs b/crates/ecstore/src/bucket/on_demand_migration/breaker.rs new file mode 100644 index 000000000..c41305f1d --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/breaker.rs @@ -0,0 +1,362 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! Per-bucket three-state circuit breaker protecting an on-demand migration +//! source (rustfs/backlog#2152). +//! +//! `Closed` lets every request through and counts consecutive failures +//! inside a sliding window; reaching the threshold opens the breaker. `Open` +//! rejects everything until the open duration elapses, then moves to +//! `HalfOpen`, which admits a single probe: success closes the breaker, +//! failure re-opens it. Timing uses `tokio::time::Instant` so tests can drive +//! it with `tokio::time::pause`. +//! +//! Only transport-level failures count (`Throttled`, `Timeout`, `Connect`, +//! `ServerError`). `NotFound` is a healthy answer and resets the failure +//! streak; `AccessDenied`, `Unsupported` and `Other` are configuration or +//! object problems that neither open nor close the breaker. + +use super::source_client::SourceError; +use parking_lot::Mutex; +use serde::{Deserialize, Serialize}; +use std::time::Duration; +use tokio::time::Instant; + +/// Consecutive counted failures that open the breaker. +pub const BREAKER_FAILURE_THRESHOLD: u32 = 5; +/// Failures further apart than this do not accumulate. +pub const BREAKER_FAILURE_WINDOW: Duration = Duration::from_secs(30); +/// How long an open breaker rejects before admitting a probe. +pub const BREAKER_OPEN_DURATION: Duration = Duration::from_secs(30); +/// Probes admitted while half-open. +pub const BREAKER_HALF_OPEN_MAX_PROBES: u32 = 1; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum BreakerState { + Closed, + Open, + HalfOpen, +} + +impl BreakerState { + pub fn as_str(self) -> &'static str { + match self { + BreakerState::Closed => "closed", + BreakerState::Open => "open", + BreakerState::HalfOpen => "half_open", + } + } +} + +/// A state change the caller may want to log. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct BreakerTransition { + pub from: BreakerState, + pub to: BreakerState, +} + +/// How a source result is scored by the breaker. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BreakerVerdict { + /// Resets the failure streak; closes a half-open breaker. + Success, + /// Counts toward the threshold; re-opens a half-open breaker. + Failure, + /// Leaves the breaker untouched. + Neutral, +} + +impl BreakerVerdict { + /// `None` is a successful source call. + pub fn for_result(error: Option<&SourceError>) -> Self { + match error { + None | Some(SourceError::NotFound) => BreakerVerdict::Success, + Some(SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_)) => { + BreakerVerdict::Failure + } + Some(SourceError::AccessDenied | SourceError::Unsupported(_) | SourceError::Other(_)) => BreakerVerdict::Neutral, + } + } +} + +#[derive(Debug)] +struct Inner { + state: BreakerState, + consecutive_failures: u32, + last_failure_at: Option, + opened_at: Option, + half_open_probes: u32, +} + +#[derive(Debug)] +pub struct Breaker { + inner: Mutex, +} + +impl Default for Breaker { + fn default() -> Self { + Self::new() + } +} + +impl Breaker { + pub fn new() -> Self { + Self { + inner: Mutex::new(Inner { + state: BreakerState::Closed, + consecutive_failures: 0, + last_failure_at: None, + opened_at: None, + half_open_probes: 0, + }), + } + } + + /// Current state after applying the open-duration timeout. + pub fn state(&self) -> BreakerState { + let mut inner = self.inner.lock(); + Self::advance(&mut inner, Instant::now()); + inner.state + } + + /// Whether a request may reach the source right now. Consumes the + /// half-open probe budget when it grants one. + pub fn allow_request(&self) -> bool { + let mut inner = self.inner.lock(); + Self::advance(&mut inner, Instant::now()); + match inner.state { + BreakerState::Closed => true, + BreakerState::Open => false, + BreakerState::HalfOpen => { + if inner.half_open_probes < BREAKER_HALF_OPEN_MAX_PROBES { + inner.half_open_probes += 1; + true + } else { + false + } + } + } + } + + /// Scores a source result; returns the transition it caused, if any. + pub fn record(&self, verdict: BreakerVerdict) -> Option { + match verdict { + BreakerVerdict::Success => self.record_success(), + BreakerVerdict::Failure => self.record_failure(), + BreakerVerdict::Neutral => None, + } + } + + pub fn record_success(&self) -> Option { + let mut inner = self.inner.lock(); + let now = Instant::now(); + Self::advance(&mut inner, now); + inner.consecutive_failures = 0; + inner.last_failure_at = None; + match inner.state { + BreakerState::Closed => None, + // A success while open can only come from a request admitted + // before the breaker opened; it says nothing about recovery. + BreakerState::Open => None, + BreakerState::HalfOpen => Some(Self::transition(&mut inner, BreakerState::Closed, now)), + } + } + + pub fn record_failure(&self) -> Option { + let mut inner = self.inner.lock(); + let now = Instant::now(); + Self::advance(&mut inner, now); + match inner.state { + BreakerState::Closed => { + let within_window = inner + .last_failure_at + .is_some_and(|last| now.saturating_duration_since(last) <= BREAKER_FAILURE_WINDOW); + inner.consecutive_failures = if within_window { inner.consecutive_failures + 1 } else { 1 }; + inner.last_failure_at = Some(now); + if inner.consecutive_failures >= BREAKER_FAILURE_THRESHOLD { + Some(Self::transition(&mut inner, BreakerState::Open, now)) + } else { + None + } + } + BreakerState::Open => None, + BreakerState::HalfOpen => Some(Self::transition(&mut inner, BreakerState::Open, now)), + } + } + + fn advance(inner: &mut Inner, now: Instant) { + if inner.state == BreakerState::Open + && inner + .opened_at + .is_some_and(|opened| now.saturating_duration_since(opened) >= BREAKER_OPEN_DURATION) + { + Self::transition(inner, BreakerState::HalfOpen, now); + } + } + + fn transition(inner: &mut Inner, to: BreakerState, now: Instant) -> BreakerTransition { + let from = inner.state; + inner.state = to; + match to { + BreakerState::Open => { + inner.opened_at = Some(now); + inner.half_open_probes = 0; + } + BreakerState::HalfOpen => { + inner.half_open_probes = 0; + } + BreakerState::Closed => { + inner.opened_at = None; + inner.half_open_probes = 0; + inner.consecutive_failures = 0; + inner.last_failure_at = None; + } + } + BreakerTransition { from, to } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn server_error() -> SourceError { + SourceError::ServerError(503) + } + + #[tokio::test(start_paused = true)] + async fn five_failures_open_then_half_open_after_timeout() { + let breaker = Breaker::new(); + for i in 0..BREAKER_FAILURE_THRESHOLD - 1 { + assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&server_error()))), None, "failure {i}"); + assert_eq!(breaker.state(), BreakerState::Closed); + } + assert_eq!( + breaker.record(BreakerVerdict::for_result(Some(&server_error()))), + Some(BreakerTransition { + from: BreakerState::Closed, + to: BreakerState::Open + }) + ); + assert_eq!(breaker.state(), BreakerState::Open); + assert!(!breaker.allow_request()); + + tokio::time::advance(BREAKER_OPEN_DURATION - Duration::from_secs(1)).await; + assert!(!breaker.allow_request()); + assert_eq!(breaker.state(), BreakerState::Open); + + tokio::time::advance(Duration::from_secs(1)).await; + assert_eq!(breaker.state(), BreakerState::HalfOpen); + assert!(breaker.allow_request(), "one probe is admitted"); + assert!(!breaker.allow_request(), "second probe is rejected"); + } + + #[tokio::test(start_paused = true)] + async fn half_open_probe_success_closes_and_failure_reopens() { + let breaker = Breaker::new(); + for _ in 0..BREAKER_FAILURE_THRESHOLD { + breaker.record_failure(); + } + tokio::time::advance(BREAKER_OPEN_DURATION).await; + assert!(breaker.allow_request()); + assert_eq!( + breaker.record_failure(), + Some(BreakerTransition { + from: BreakerState::HalfOpen, + to: BreakerState::Open + }) + ); + assert!(!breaker.allow_request()); + + tokio::time::advance(BREAKER_OPEN_DURATION).await; + assert!(breaker.allow_request()); + assert_eq!( + breaker.record_success(), + Some(BreakerTransition { + from: BreakerState::HalfOpen, + to: BreakerState::Closed + }) + ); + assert_eq!(breaker.state(), BreakerState::Closed); + assert!(breaker.allow_request()); + // The streak restarts from zero after closing. + for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 { + assert_eq!(breaker.record_failure(), None); + } + assert_eq!(breaker.state(), BreakerState::Closed); + } + + #[tokio::test(start_paused = true)] + async fn failures_outside_window_do_not_accumulate() { + let breaker = Breaker::new(); + for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 { + breaker.record_failure(); + } + tokio::time::advance(BREAKER_FAILURE_WINDOW + Duration::from_secs(1)).await; + assert_eq!(breaker.record_failure(), None, "stale streak restarts at one"); + assert_eq!(breaker.state(), BreakerState::Closed); + } + + #[test] + fn not_found_and_access_denied_do_not_count() { + let breaker = Breaker::new(); + for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 { + breaker.record(BreakerVerdict::for_result(Some(&server_error()))); + } + assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::AccessDenied))), None); + assert_eq!(breaker.state(), BreakerState::Closed); + // AccessDenied is neutral: the streak is still one short of opening. + assert_eq!( + breaker.record(BreakerVerdict::for_result(Some(&SourceError::Unsupported("sse-c".into())))), + None + ); + assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::Other("x".into())))), None); + // NotFound is a healthy answer and resets the streak entirely. + assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::NotFound))), None); + for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 { + assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::Timeout))), None); + } + assert_eq!(breaker.state(), BreakerState::Closed); + } + + #[test] + fn verdicts_cover_every_source_error_class() { + assert_eq!(BreakerVerdict::for_result(None), BreakerVerdict::Success); + assert_eq!(BreakerVerdict::for_result(Some(&SourceError::NotFound)), BreakerVerdict::Success); + for failure in [ + SourceError::Throttled, + SourceError::Timeout, + SourceError::Connect("refused".into()), + SourceError::ServerError(500), + ] { + assert_eq!(BreakerVerdict::for_result(Some(&failure)), BreakerVerdict::Failure, "{failure:?}"); + } + for neutral in [ + SourceError::AccessDenied, + SourceError::Unsupported("sse-c".into()), + SourceError::Other("x".into()), + ] { + assert_eq!(BreakerVerdict::for_result(Some(&neutral)), BreakerVerdict::Neutral, "{neutral:?}"); + } + } + + #[test] + fn state_labels_are_stable() { + assert_eq!(BreakerState::Closed.as_str(), "closed"); + assert_eq!(BreakerState::Open.as_str(), "open"); + assert_eq!(BreakerState::HalfOpen.as_str(), "half_open"); + assert_eq!(serde_json::to_string(&BreakerState::HalfOpen).unwrap(), "\"half_open\""); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/mod.rs b/crates/ecstore/src/bucket/on_demand_migration/mod.rs index 11491e1c4..b2b4c1a08 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/mod.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/mod.rs @@ -15,14 +15,33 @@ //! On-Demand Migration (ODM): a bucket can name an external S3-compatible //! source bucket; GET misses are served from that source and backfilled //! locally. This module owns the bucket-level configuration model -//! (`on-demand-migration.json` in the bucket metadata file); the runtime is -//! layered on top of it by later tasks (rustfs/backlog#2147). +//! (`on-demand-migration.json` in the bucket metadata file), the source +//! client, and the per-node runtime (`sys`) that turns configs into live +//! clients guarded by a breaker, a negative cache, singleflight and a pull +//! concurrency limit (rustfs/backlog#2147). +pub mod breaker; pub mod config; +pub mod negative_cache; pub mod source_client; +pub mod stats; +pub mod sys; +pub use breaker::{ + BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker, + BreakerState, BreakerTransition, BreakerVerdict, +}; pub use config::{ ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, }; +pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache}; +pub use stats::{ + GaugeGuard, LastSourceError, LatencyBucketSnapshot, OdmOp, OdmOutcome, OdmStats, OdmStatsSnapshot, PullFailureReason, + PullPath, SOURCE_LATENCY_BUCKET_BOUNDS_MS, SourceLatencySnapshot, +}; +pub use sys::{ + ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError, + OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec, +}; diff --git a/crates/ecstore/src/bucket/on_demand_migration/negative_cache.rs b/crates/ecstore/src/bucket/on_demand_migration/negative_cache.rs new file mode 100644 index 000000000..13f0de71b --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/negative_cache.rs @@ -0,0 +1,130 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! Per-bucket cache of keys the source answered 404 for +//! (rustfs/backlog#2152). A hit short-circuits the source lookup for +//! `policy.negative_cache_ttl_secs`; a TTL of zero disables the cache. +//! +//! Entries are never invalidated on a local PUT: once the object exists +//! locally the handler never consults ODM for it, so a stale negative entry +//! is harmless. + +use std::time::Duration; + +/// Upper bound on remembered keys per bucket; LRU eviction beyond it. +pub const NEGATIVE_CACHE_MAX_ENTRIES: u64 = 100_000; + +#[derive(Debug)] +pub struct NegativeCache { + cache: Option>, + ttl: Duration, +} + +impl NegativeCache { + /// `ttl == 0` builds a disabled cache that never records anything. + pub fn new(ttl: Duration) -> Self { + Self::with_capacity(ttl, NEGATIVE_CACHE_MAX_ENTRIES) + } + + pub fn with_capacity(ttl: Duration, max_entries: u64) -> Self { + let cache = (!ttl.is_zero()).then(|| { + moka::sync::Cache::builder() + .max_capacity(max_entries) + .time_to_live(ttl) + .build() + }); + Self { cache, ttl } + } + + pub fn is_enabled(&self) -> bool { + self.cache.is_some() + } + + pub fn ttl(&self) -> Duration { + self.ttl + } + + /// Whether `key` is currently remembered as absent on the source. + pub fn contains(&self, key: &str) -> bool { + self.cache.as_ref().is_some_and(|cache| cache.get(key).is_some()) + } + + /// Remembers `key` as absent; no-op when disabled. + pub fn insert(&self, key: &str) { + if let Some(cache) = &self.cache { + cache.insert(key.to_string(), ()); + } + } + + /// Forgets `key` (e.g. after an admin-triggered backfill found it). + pub fn remove(&self, key: &str) { + if let Some(cache) = &self.cache { + cache.invalidate(key); + } + } + + /// Approximate live entry count, for status snapshots only. + pub fn len(&self) -> u64 { + self.cache.as_ref().map_or(0, |cache| { + cache.run_pending_tasks(); + cache.entry_count() + }) + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn entry_expires_after_ttl() { + let cache = NegativeCache::new(Duration::from_millis(80)); + assert!(cache.is_enabled()); + cache.insert("a/x"); + assert!(cache.contains("a/x")); + assert!(!cache.contains("a/y")); + std::thread::sleep(Duration::from_millis(160)); + assert!(!cache.contains("a/x"), "entry must expire after the TTL"); + } + + #[test] + fn zero_ttl_disables_the_cache() { + let cache = NegativeCache::new(Duration::ZERO); + assert!(!cache.is_enabled()); + cache.insert("a/x"); + assert!(!cache.contains("a/x")); + assert!(cache.is_empty()); + } + + #[test] + fn remove_forgets_a_key() { + let cache = NegativeCache::new(Duration::from_secs(30)); + cache.insert("a/x"); + cache.remove("a/x"); + assert!(!cache.contains("a/x")); + } + + #[test] + fn capacity_bounds_entries() { + let cache = NegativeCache::with_capacity(Duration::from_secs(30), 4); + for i in 0..64 { + cache.insert(&format!("k{i}")); + } + assert!(cache.len() <= 4, "len {} exceeds capacity", cache.len()); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/stats.rs b/crates/ecstore/src/bucket/on_demand_migration/stats.rs new file mode 100644 index 000000000..d76d1c123 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/stats.rs @@ -0,0 +1,527 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! Per-bucket on-demand migration counters (rustfs/backlog#2152). +//! +//! `OdmStats` is lock-free and survives config rebuilds; `snapshot()` turns +//! it into the serializable `OdmStatsSnapshot` that the metrics collector +//! and the admin status route (ODM-10/14/15) consume. Field names and label +//! values are a wire contract: the golden JSON test below pins them. + +use super::breaker::BreakerState; +use super::source_client::SourceError; +use parking_lot::Mutex; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; +use time::OffsetDateTime; + +/// Request operations that can enter ODM. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum OdmOp { + Get, + Head, +} + +impl OdmOp { + pub const ALL: [OdmOp; 2] = [OdmOp::Get, OdmOp::Head]; + + pub fn as_str(self) -> &'static str { + match self { + OdmOp::Get => "get", + OdmOp::Head => "head", + } + } +} + +/// How a request that entered ODM ended. `local_hit` is deliberately absent: +/// requests served locally never reach the runtime. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum OdmOutcome { + SourceHit, + SourceMiss, + SourceError, + BreakerOpen, + NegativeCached, + Filtered, + Unsupported, +} + +impl OdmOutcome { + pub const ALL: [OdmOutcome; 7] = [ + OdmOutcome::SourceHit, + OdmOutcome::SourceMiss, + OdmOutcome::SourceError, + OdmOutcome::BreakerOpen, + OdmOutcome::NegativeCached, + OdmOutcome::Filtered, + OdmOutcome::Unsupported, + ]; + + pub fn as_str(self) -> &'static str { + match self { + OdmOutcome::SourceHit => "source_hit", + OdmOutcome::SourceMiss => "source_miss", + OdmOutcome::SourceError => "source_error", + OdmOutcome::BreakerOpen => "breaker_open", + OdmOutcome::NegativeCached => "negative_cached", + OdmOutcome::Filtered => "filtered", + OdmOutcome::Unsupported => "unsupported", + } + } +} + +/// Which pipeline stored a pulled object locally. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PullPath { + /// Streamed to the client and written locally in one pass. + Inline, + /// Pulled by a background task after a partial/large read. + Background, + /// Pulled by the backfill job. + Backfill, +} + +impl PullPath { + pub const ALL: [PullPath; 3] = [PullPath::Inline, PullPath::Background, PullPath::Backfill]; + + pub fn as_str(self) -> &'static str { + match self { + PullPath::Inline => "inline", + PullPath::Background => "background", + PullPath::Backfill => "backfill", + } + } +} + +/// Why a pull did not produce a local object. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PullFailureReason { + SourceNotFound, + SourceAccessDenied, + SourceThrottled, + SourceTimeout, + SourceConnect, + SourceServerError, + SourceUnsupported, + SourceOther, + /// Source bytes did not match the ETag advertised by HEAD/GET. + EtagMismatch, + /// The local write (internal PUT) failed. + LocalWrite, + /// The bucket state was removed or the process is shutting down. + Canceled, + /// The background pull queue was full. + QueueFull, +} + +impl PullFailureReason { + pub const ALL: [PullFailureReason; 12] = [ + PullFailureReason::SourceNotFound, + PullFailureReason::SourceAccessDenied, + PullFailureReason::SourceThrottled, + PullFailureReason::SourceTimeout, + PullFailureReason::SourceConnect, + PullFailureReason::SourceServerError, + PullFailureReason::SourceUnsupported, + PullFailureReason::SourceOther, + PullFailureReason::EtagMismatch, + PullFailureReason::LocalWrite, + PullFailureReason::Canceled, + PullFailureReason::QueueFull, + ]; + + pub fn as_str(self) -> &'static str { + match self { + PullFailureReason::SourceNotFound => "source_not_found", + PullFailureReason::SourceAccessDenied => "source_access_denied", + PullFailureReason::SourceThrottled => "source_throttled", + PullFailureReason::SourceTimeout => "source_timeout", + PullFailureReason::SourceConnect => "source_connect", + PullFailureReason::SourceServerError => "source_server_error", + PullFailureReason::SourceUnsupported => "source_unsupported", + PullFailureReason::SourceOther => "source_other", + PullFailureReason::EtagMismatch => "etag_mismatch", + PullFailureReason::LocalWrite => "local_write", + PullFailureReason::Canceled => "canceled", + PullFailureReason::QueueFull => "queue_full", + } + } +} + +impl From<&SourceError> for PullFailureReason { + fn from(err: &SourceError) -> Self { + match err { + SourceError::NotFound => PullFailureReason::SourceNotFound, + SourceError::AccessDenied => PullFailureReason::SourceAccessDenied, + SourceError::Throttled => PullFailureReason::SourceThrottled, + SourceError::Timeout => PullFailureReason::SourceTimeout, + SourceError::Connect(_) => PullFailureReason::SourceConnect, + SourceError::ServerError(_) => PullFailureReason::SourceServerError, + SourceError::Unsupported(_) => PullFailureReason::SourceUnsupported, + SourceError::Other(_) => PullFailureReason::SourceOther, + } + } +} + +/// Upper bounds (milliseconds) of the source latency histogram buckets; the +/// implicit last bucket is unbounded. Roughly logarithmic from 5 ms to 60 s. +pub const SOURCE_LATENCY_BUCKET_BOUNDS_MS: [u64; 14] = [ + 5, 10, 20, 50, 100, 200, 500, 1_000, 2_000, 5_000, 10_000, 20_000, 30_000, 60_000, +]; + +#[derive(Debug, Default)] +struct LatencyHistogram { + /// One counter per bound plus one for the overflow bucket. + buckets: [AtomicU64; SOURCE_LATENCY_BUCKET_BOUNDS_MS.len() + 1], + count: AtomicU64, + sum_ms: AtomicU64, +} + +impl LatencyHistogram { + fn observe(&self, latency: Duration) { + let ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX); + let index = SOURCE_LATENCY_BUCKET_BOUNDS_MS + .iter() + .position(|bound| ms <= *bound) + .unwrap_or(SOURCE_LATENCY_BUCKET_BOUNDS_MS.len()); + self.buckets[index].fetch_add(1, Ordering::Relaxed); + self.count.fetch_add(1, Ordering::Relaxed); + self.sum_ms.fetch_add(ms, Ordering::Relaxed); + } + + fn snapshot(&self) -> SourceLatencySnapshot { + let mut cumulative = 0; + let buckets = SOURCE_LATENCY_BUCKET_BOUNDS_MS + .iter() + .zip(self.buckets.iter()) + .map(|(bound, counter)| { + cumulative += counter.load(Ordering::Relaxed); + LatencyBucketSnapshot { + le_ms: *bound, + count: cumulative, + } + }) + .collect(); + SourceLatencySnapshot { + buckets, + count: self.count.load(Ordering::Relaxed), + sum_ms: self.sum_ms.load(Ordering::Relaxed), + } + } +} + +/// The most recent source failure, kept for operators: class only, never the +/// key or the message (which may echo attacker-controlled input). +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LastSourceError { + pub class: String, + #[serde(with = "time::serde::rfc3339")] + pub at: OffsetDateTime, +} + +#[derive(Debug, Default)] +pub struct OdmStats { + requests_total: [[AtomicU64; OdmOutcome::ALL.len()]; OdmOp::ALL.len()], + pulled_bytes_total: AtomicU64, + pulled_objects_total: [AtomicU64; PullPath::ALL.len()], + pull_failures_total: [AtomicU64; PullFailureReason::ALL.len()], + inflight_pulls: AtomicU64, + queue_depth: AtomicU64, + source_latency: LatencyHistogram, + last_source_error: Mutex>, +} + +impl OdmStats { + pub fn new() -> Self { + Self::default() + } + + pub fn record_request(&self, op: OdmOp, outcome: OdmOutcome) { + self.requests_total[op as usize][outcome as usize].fetch_add(1, Ordering::Relaxed); + } + + pub fn record_pulled_bytes(&self, bytes: u64) { + self.pulled_bytes_total.fetch_add(bytes, Ordering::Relaxed); + } + + pub fn record_pulled_object(&self, path: PullPath) { + self.pulled_objects_total[path as usize].fetch_add(1, Ordering::Relaxed); + } + + pub fn record_pull_failure(&self, reason: PullFailureReason) { + self.pull_failures_total[reason as usize].fetch_add(1, Ordering::Relaxed); + } + + pub fn record_source_latency(&self, latency: Duration) { + self.source_latency.observe(latency); + } + + pub fn record_source_error(&self, err: &SourceError) { + self.record_source_error_at(err, OffsetDateTime::now_utc()); + } + + pub fn record_source_error_at(&self, err: &SourceError, at: OffsetDateTime) { + *self.last_source_error.lock() = Some(LastSourceError { + class: err.class_label().to_string(), + at, + }); + } + + pub fn last_source_error(&self) -> Option { + self.last_source_error.lock().clone() + } + + pub fn inflight_pulls(&self) -> u64 { + self.inflight_pulls.load(Ordering::Relaxed) + } + + pub fn queue_depth(&self) -> u64 { + self.queue_depth.load(Ordering::Relaxed) + } + + /// RAII increment of `inflight_pulls`. + pub fn inflight_guard(self: &Arc) -> GaugeGuard { + GaugeGuard::new(Arc::clone(self), OdmGauge::InflightPulls) + } + + /// RAII increment of `queue_depth`. + pub fn queue_guard(self: &Arc) -> GaugeGuard { + GaugeGuard::new(Arc::clone(self), OdmGauge::QueueDepth) + } + + fn gauge(&self, gauge: OdmGauge) -> &AtomicU64 { + match gauge { + OdmGauge::InflightPulls => &self.inflight_pulls, + OdmGauge::QueueDepth => &self.queue_depth, + } + } + + /// Read-only, side-effect-free copy of every counter. The breaker lives + /// next to the stats in the bucket state; its state is passed in so the + /// snapshot stays a single document. + pub fn snapshot(&self, breaker_state: BreakerState) -> OdmStatsSnapshot { + let mut requests_total = BTreeMap::new(); + for op in OdmOp::ALL { + let mut by_outcome = BTreeMap::new(); + for outcome in OdmOutcome::ALL { + by_outcome.insert( + outcome.as_str().to_string(), + self.requests_total[op as usize][outcome as usize].load(Ordering::Relaxed), + ); + } + requests_total.insert(op.as_str().to_string(), by_outcome); + } + let pulled_objects_total = PullPath::ALL + .iter() + .map(|path| { + ( + path.as_str().to_string(), + self.pulled_objects_total[*path as usize].load(Ordering::Relaxed), + ) + }) + .collect(); + let pull_failures_total = PullFailureReason::ALL + .iter() + .map(|reason| { + ( + reason.as_str().to_string(), + self.pull_failures_total[*reason as usize].load(Ordering::Relaxed), + ) + }) + .collect(); + OdmStatsSnapshot { + requests_total, + pulled_bytes_total: self.pulled_bytes_total.load(Ordering::Relaxed), + pulled_objects_total, + pull_failures_total, + inflight_pulls: self.inflight_pulls(), + queue_depth: self.queue_depth(), + source_latency: self.source_latency.snapshot(), + last_source_error: self.last_source_error(), + breaker_state, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum OdmGauge { + InflightPulls, + QueueDepth, +} + +/// Increments a gauge on creation and decrements it on drop. Owns its +/// `OdmStats` so it can live inside the pull slot handed to callers. +#[derive(Debug)] +pub struct GaugeGuard { + stats: Arc, + gauge: OdmGauge, +} + +impl GaugeGuard { + fn new(stats: Arc, gauge: OdmGauge) -> Self { + stats.gauge(gauge).fetch_add(1, Ordering::Relaxed); + Self { stats, gauge } + } +} + +impl Drop for GaugeGuard { + fn drop(&mut self) { + self.stats.gauge(self.gauge).fetch_sub(1, Ordering::Relaxed); + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LatencyBucketSnapshot { + /// Upper bound of the bucket in milliseconds. + pub le_ms: u64, + /// Cumulative observations at or below `le_ms`. + pub count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceLatencySnapshot { + pub buckets: Vec, + /// Total observations, including those above the last bound. + pub count: u64, + pub sum_ms: u64, +} + +/// Serializable copy of [`OdmStats`]. Every key is snake_case and every +/// label set is fixed, so consumers can rely on the document shape. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct OdmStatsSnapshot { + /// `op -> outcome -> count`. + pub requests_total: BTreeMap>, + pub pulled_bytes_total: u64, + /// `path -> count`. + pub pulled_objects_total: BTreeMap, + /// `reason -> count`. + pub pull_failures_total: BTreeMap, + pub inflight_pulls: u64, + pub queue_depth: u64, + pub source_latency: SourceLatencySnapshot, + pub last_source_error: Option, + pub breaker_state: BreakerState, +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use time::macros::datetime; + + #[test] + fn snapshot_matches_golden_json() { + let stats = Arc::new(OdmStats::new()); + stats.record_request(OdmOp::Get, OdmOutcome::SourceHit); + stats.record_request(OdmOp::Get, OdmOutcome::SourceHit); + stats.record_request(OdmOp::Head, OdmOutcome::NegativeCached); + stats.record_pulled_bytes(4096); + stats.record_pulled_object(PullPath::Inline); + stats.record_pull_failure(PullFailureReason::from(&SourceError::Timeout)); + stats.record_source_latency(Duration::from_millis(3)); + stats.record_source_latency(Duration::from_millis(750)); + stats.record_source_latency(Duration::from_secs(90)); + stats.record_source_error_at(&SourceError::ServerError(502), datetime!(2026-09-02 10:00:00 UTC)); + let _inflight = stats.inflight_guard(); + let _queued = stats.queue_guard(); + + let snapshot = stats.snapshot(BreakerState::HalfOpen); + let actual = serde_json::to_value(&snapshot).unwrap(); + let expected = json!({ + "requests_total": { + "get": { + "breaker_open": 0, "filtered": 0, "negative_cached": 0, "source_error": 0, + "source_hit": 2, "source_miss": 0, "unsupported": 0 + }, + "head": { + "breaker_open": 0, "filtered": 0, "negative_cached": 1, "source_error": 0, + "source_hit": 0, "source_miss": 0, "unsupported": 0 + } + }, + "pulled_bytes_total": 4096, + "pulled_objects_total": { "backfill": 0, "background": 0, "inline": 1 }, + "pull_failures_total": { + "canceled": 0, "etag_mismatch": 0, "local_write": 0, "queue_full": 0, + "source_access_denied": 0, "source_connect": 0, "source_not_found": 0, "source_other": 0, + "source_server_error": 0, "source_throttled": 0, "source_timeout": 1, "source_unsupported": 0 + }, + "inflight_pulls": 1, + "queue_depth": 1, + "source_latency": { + "buckets": [ + { "le_ms": 5, "count": 1 }, { "le_ms": 10, "count": 1 }, { "le_ms": 20, "count": 1 }, + { "le_ms": 50, "count": 1 }, { "le_ms": 100, "count": 1 }, { "le_ms": 200, "count": 1 }, + { "le_ms": 500, "count": 1 }, { "le_ms": 1000, "count": 2 }, { "le_ms": 2000, "count": 2 }, + { "le_ms": 5000, "count": 2 }, { "le_ms": 10000, "count": 2 }, { "le_ms": 20000, "count": 2 }, + { "le_ms": 30000, "count": 2 }, { "le_ms": 60000, "count": 2 } + ], + "count": 3, + "sum_ms": 90753 + }, + "last_source_error": { "class": "server_error", "at": "2026-09-02T10:00:00Z" }, + "breaker_state": "half_open" + }); + assert_eq!(actual, expected); + + let round_trip: OdmStatsSnapshot = serde_json::from_value(actual).unwrap(); + assert_eq!(round_trip, snapshot); + } + + #[test] + fn gauges_return_to_zero_when_guards_drop() { + let stats = Arc::new(OdmStats::new()); + { + let _a = stats.inflight_guard(); + let _b = stats.inflight_guard(); + let _c = stats.queue_guard(); + assert_eq!(stats.inflight_pulls(), 2); + assert_eq!(stats.queue_depth(), 1); + } + assert_eq!(stats.inflight_pulls(), 0); + assert_eq!(stats.queue_depth(), 0); + } + + #[test] + fn pull_failure_reason_covers_every_source_error_class() { + let cases = [ + (SourceError::NotFound, PullFailureReason::SourceNotFound), + (SourceError::AccessDenied, PullFailureReason::SourceAccessDenied), + (SourceError::Throttled, PullFailureReason::SourceThrottled), + (SourceError::Timeout, PullFailureReason::SourceTimeout), + (SourceError::Connect("x".into()), PullFailureReason::SourceConnect), + (SourceError::ServerError(500), PullFailureReason::SourceServerError), + (SourceError::Unsupported("x".into()), PullFailureReason::SourceUnsupported), + (SourceError::Other("x".into()), PullFailureReason::SourceOther), + ]; + for (err, reason) in cases { + assert_eq!(PullFailureReason::from(&err), reason, "{err:?}"); + assert_eq!(serde_json::to_string(&reason).unwrap(), format!("\"{}\"", reason.as_str())); + } + } + + #[test] + fn label_lists_are_exhaustive_and_unique() { + let outcomes: std::collections::BTreeSet<_> = OdmOutcome::ALL.iter().map(|o| o.as_str()).collect(); + assert_eq!(outcomes.len(), OdmOutcome::ALL.len()); + let reasons: std::collections::BTreeSet<_> = PullFailureReason::ALL.iter().map(|r| r.as_str()).collect(); + assert_eq!(reasons.len(), PullFailureReason::ALL.len()); + let paths: std::collections::BTreeSet<_> = PullPath::ALL.iter().map(|p| p.as_str()).collect(); + assert_eq!(paths.len(), PullPath::ALL.len()); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/sys.rs b/crates/ecstore/src/bucket/on_demand_migration/sys.rs new file mode 100644 index 000000000..366b1d5a9 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/sys.rs @@ -0,0 +1,1235 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! Per-node on-demand migration runtime (rustfs/backlog#2152). +//! +//! `OnDemandMigrationSys` turns each bucket's persisted +//! [`OnDemandMigrationConfig`] into a live [`BucketOdmState`]: a +//! [`SourceClient`], a circuit breaker, a negative cache, a per-key +//! singleflight table, a pull concurrency limit and counters. Its lifecycle +//! follows the bucket metadata cache through the publish hook registered in +//! [`ON_DEMAND_MIGRATION_CONFIG_HOOK`]; the hook fires on every cache install +//! path (initial load, admin update, peer reload, refresh loop, lazy load). +//! +//! Change detection compares the config by value (`PartialEq`) rather than +//! by `updated_at`: the hook does not carry the timestamp, fetching it would +//! re-enter the metadata system from inside its own publish path, and a +//! byte-identical config never needs a new client anyway. +//! +//! Client construction is async (TLS material may be read from disk), so +//! the hook does not build inline: `publish` removes state synchronously and +//! spawns `apply` for installs, and a per-call generation number makes sure +//! a slower, older install can never overwrite a newer one. +//! +//! The module switch (`RUSTFS_ON_DEMAND_MIGRATION_ENABLED`, default off) is +//! injected by the `rustfs` binary through [`OnDemandMigrationSys::set_module_enabled`] +//! before bucket metadata loads; this crate never reads the environment. + +use super::breaker::{Breaker, BreakerState, BreakerTransition, BreakerVerdict}; +use super::config::{ + ON_DEMAND_MIGRATION_CONFIG_HOOK, OnDemandMigrationConfig, PathStyle as ConfigPathStyle, Provider, SourceConfig, +}; +use super::negative_cache::NegativeCache; +use super::source_client::{SourceClient, SourceClientSpec, SourceError, SourceProvider, SourceTimeouts}; +use super::stats::{GaugeGuard, OdmStats, OdmStatsSnapshot, PullFailureReason}; +use crate::bucket::remote_s3_client::{PathStyle as ClientPathStyle, RemoteCredentials, RemoteS3ClientError}; +use parking_lot::{Mutex, RwLock}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::fmt; +use std::num::NonZeroU64; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::Duration; +use time::OffsetDateTime; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, watch}; +use tokio::time::Instant; +use tokio_util::sync::CancellationToken; +use tracing::{info, warn}; +use url::Url; + +const EVENT_ODM_BUCKET_STATE_APPLIED: &str = "odm_bucket_state_applied"; +const EVENT_ODM_BREAKER_TRANSITION: &str = "odm_breaker_transition"; +const EVENT_ODM_SOURCE_ERROR: &str = "odm_source_error"; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_ON_DEMAND_MIGRATION: &str = "on_demand_migration"; + +/// Minimum spacing between `EVENT_ODM_SOURCE_ERROR` records per bucket. +const SOURCE_ERROR_LOG_INTERVAL: Duration = Duration::from_secs(10); + +pub static GLOBAL_ON_DEMAND_MIGRATION_SYS: OnceLock = OnceLock::new(); + +/// Why a configured bucket has no usable source client. Surfaced through +/// `resolve` as [`OdmLookup::Unavailable`] and through status snapshots. +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +pub enum OdmStateError { + /// `source.credentials` is `null`; the shared client builder has no + /// anonymous mode yet (rustfs/backlog#2149 follow-up). + #[error("anonymous source access is not supported yet; configure source credentials")] + AnonymousUnsupported, + #[error("source client could not be built: {0}")] + ClientBuild(String), +} + +/// One-shot verdict for a `(bucket, key)` lookup. `None` from +/// [`OnDemandMigrationSys::resolve`] means "do not intervene"; every `Some` +/// carries the bucket state so the handler can record its outcome. +#[derive(Clone, Debug)] +pub enum OdmLookup { + /// The source answered 404 for this key recently. + NegativeCached { state: Arc }, + /// The breaker rejects source traffic right now. + BreakerOpen { state: Arc }, + /// The bucket is configured but its client could not be built. + Unavailable { + state: Arc, + error: OdmStateError, + }, + /// Go to the source. + Ready { state: Arc }, +} + +impl OdmLookup { + pub fn state(&self) -> &Arc { + match self { + OdmLookup::NegativeCached { state } + | OdmLookup::BreakerOpen { state } + | OdmLookup::Unavailable { state, .. } + | OdmLookup::Ready { state } => state, + } + } +} + +/// What `apply` did with a bucket. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ApplyOutcome { + /// No state before, none after. + NotDesired, + /// The state was removed and its cancellation token fired. + Removed, + /// Same config as the installed state; nothing rebuilt. + Unchanged, + /// First state for this bucket. + Installed, + /// A newer config replaced the previous state (counters carried over). + Rebuilt, + /// A later `apply`/`publish` for the same bucket won the race. + Superseded, +} + +/// Result of one pull as seen by the singleflight leader and its followers. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PullOutcome { + pub etag: Option, + pub size: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +#[error("{}: {message}", reason.as_str())] +pub struct PullError { + pub reason: PullFailureReason, + pub message: String, +} + +impl PullError { + pub fn new(reason: PullFailureReason, message: impl Into) -> Self { + Self { + reason, + message: message.into(), + } + } + + pub fn canceled(message: impl Into) -> Self { + Self::new(PullFailureReason::Canceled, message) + } +} + +impl From<&SourceError> for PullError { + fn from(err: &SourceError) -> Self { + Self::new(PullFailureReason::from(err), err.to_string()) + } +} + +pub type PullResult = Result; + +/// Outcome of [`BucketOdmState::acquire_pull_slot`]. +#[derive(Debug)] +pub enum PullSlot { + /// This caller performs the pull and must call [`PullLeader::complete`]. + Leader(PullLeader), + /// Another caller is pulling the same key; await [`PullFollower::wait`]. + Follower(PullFollower), +} + +/// Held by the single puller of a key. Dropping it without `complete` +/// fails every follower with [`PullFailureReason::Canceled`]. +pub struct PullLeader { + state: Arc, + key: String, + tx: watch::Sender>, + _permit: OwnedSemaphorePermit, + _inflight: GaugeGuard, + completed: bool, +} + +impl fmt::Debug for PullLeader { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("PullLeader") + .field("bucket", &self.state.bucket) + .field("key", &self.key) + .field("completed", &self.completed) + .finish_non_exhaustive() + } +} + +impl PullLeader { + pub fn key(&self) -> &str { + &self.key + } + + pub fn state(&self) -> &Arc { + &self.state + } + + /// Publishes the result to every follower and releases the key. + pub fn complete(mut self, result: PullResult) { + self.completed = true; + self.tx.send_replace(Some(result)); + } +} + +impl Drop for PullLeader { + fn drop(&mut self) { + if !self.completed { + self.tx + .send_replace(Some(Err(PullError::canceled("pull leader exited without a result")))); + } + self.state.inflight.lock().remove(&self.key); + } +} + +#[derive(Debug)] +pub struct PullFollower { + rx: watch::Receiver>, +} + +impl PullFollower { + /// Resolves once the leader completes (or disappears). + pub async fn wait(mut self) -> PullResult { + match self.rx.wait_for(|result| result.is_some()).await { + Ok(result) => result + .clone() + .unwrap_or_else(|| Err(PullError::canceled("pull leader vanished"))), + Err(_) => Err(PullError::canceled("pull leader vanished")), + } + } +} + +/// Removes the singleflight entry if the leader gives up before it exists. +struct InflightEntryGuard<'a> { + state: &'a BucketOdmState, + key: &'a str, + tx: &'a watch::Sender>, + armed: bool, +} + +impl Drop for InflightEntryGuard<'_> { + fn drop(&mut self) { + if self.armed { + self.tx + .send_replace(Some(Err(PullError::canceled("pull leader canceled before starting")))); + self.state.inflight.lock().remove(self.key); + } + } +} + +/// Live runtime for one bucket. Built by `apply`, replaced wholesale on a +/// config change (counters excepted), removed when the config goes away. +pub struct BucketOdmState { + bucket: String, + config: OnDemandMigrationConfig, + applied_at: OffsetDateTime, + endpoint_host: String, + client: Result, OdmStateError>, + breaker: Breaker, + negative_cache: NegativeCache, + inflight: Mutex>>>, + pull_semaphore: Arc, + stats: Arc, + cancel: CancellationToken, + last_source_error_logged_at: Mutex>, +} + +impl fmt::Debug for BucketOdmState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BucketOdmState") + .field("bucket", &self.bucket) + .field("provider", &self.config.source.provider) + .field("endpoint_host", &self.endpoint_host) + .field("applied_at", &self.applied_at) + .field("client", &self.client.as_ref().map(|_| "ready")) + .field("breaker", &self.breaker.state()) + .field("cancelled", &self.cancel.is_cancelled()) + .finish_non_exhaustive() + } +} + +impl BucketOdmState { + async fn build(bucket: &str, config: &OnDemandMigrationConfig, stats: Arc) -> Arc { + let spec = source_client_spec(config); + let client = if config.source.credentials.is_none() { + Err(OdmStateError::AnonymousUnsupported) + } else { + SourceClient::new(&spec).await.map(Arc::new).map_err(|err| match err { + RemoteS3ClientError::MissingCredentials => OdmStateError::AnonymousUnsupported, + other => OdmStateError::ClientBuild(other.to_string()), + }) + }; + let policy = &config.policy; + Arc::new(Self { + bucket: bucket.to_string(), + endpoint_host: endpoint_host(&config.source), + config: config.clone(), + applied_at: OffsetDateTime::now_utc(), + client, + breaker: Breaker::new(), + negative_cache: NegativeCache::new(Duration::from_secs(policy.negative_cache_ttl_secs)), + inflight: Mutex::new(HashMap::new()), + pull_semaphore: Arc::new(Semaphore::new(policy.max_concurrent_pulls.max(1) as usize)), + stats, + cancel: CancellationToken::new(), + last_source_error_logged_at: Mutex::new(None), + }) + } + + pub fn bucket(&self) -> &str { + &self.bucket + } + + pub fn config(&self) -> &OnDemandMigrationConfig { + &self.config + } + + pub fn applied_at(&self) -> OffsetDateTime { + self.applied_at + } + + /// Host of the source endpoint, safe to log. + pub fn endpoint_host(&self) -> &str { + &self.endpoint_host + } + + pub fn client(&self) -> Result<&Arc, &OdmStateError> { + self.client.as_ref() + } + + pub fn breaker(&self) -> &Breaker { + &self.breaker + } + + pub fn negative_cache(&self) -> &NegativeCache { + &self.negative_cache + } + + pub fn stats(&self) -> &Arc { + &self.stats + } + + /// Fires when this state is replaced or removed; background pulls + /// started for it must exit. + pub fn cancel_token(&self) -> CancellationToken { + self.cancel.clone() + } + + pub fn is_cancelled(&self) -> bool { + self.cancel.is_cancelled() + } + + /// Whether `filter.prefix` admits this local key. + pub fn matches_prefix(&self, key: &str) -> bool { + self.config + .filter + .prefix + .as_deref() + .is_none_or(|prefix| key.starts_with(prefix)) + } + + /// The bucket-level part of [`OnDemandMigrationSys::resolve`]. + pub fn resolve_key(self: &Arc, key: &str) -> Option { + if !self.matches_prefix(key) { + return None; + } + if let Err(error) = &self.client { + return Some(OdmLookup::Unavailable { + state: Arc::clone(self), + error: error.clone(), + }); + } + if self.negative_cache.contains(key) { + return Some(OdmLookup::NegativeCached { state: Arc::clone(self) }); + } + if !self.breaker.allow_request() { + return Some(OdmLookup::BreakerOpen { state: Arc::clone(self) }); + } + Some(OdmLookup::Ready { state: Arc::clone(self) }) + } + + /// Records the outcome of one source call: latency, breaker scoring, + /// `last_source_error`, a rate-limited log line, and the negative cache + /// on `NotFound`. + pub fn observe_source(&self, latency: Duration, key: &str, error: Option<&SourceError>) { + self.stats.record_source_latency(latency); + if let Some(transition) = self.breaker.record(BreakerVerdict::for_result(error)) { + self.log_breaker_transition(transition); + } + let Some(err) = error else { + return; + }; + if matches!(err, SourceError::NotFound) { + self.negative_cache.insert(key); + return; + } + self.stats.record_source_error(err); + if self.should_log_source_error() { + warn!( + event = EVENT_ODM_SOURCE_ERROR, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + bucket = %self.bucket, + provider = %self.config.source.provider, + endpoint_host = %self.endpoint_host, + error_class = err.class_label(), + "On-demand migration source request failed" + ); + } + } + + fn should_log_source_error(&self) -> bool { + let now = Instant::now(); + let mut last = self.last_source_error_logged_at.lock(); + if last.is_some_and(|at| now.saturating_duration_since(at) < SOURCE_ERROR_LOG_INTERVAL) { + return false; + } + *last = Some(now); + true + } + + fn log_breaker_transition(&self, transition: BreakerTransition) { + match transition.to { + BreakerState::Open => warn!( + event = EVENT_ODM_BREAKER_TRANSITION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + state = transition.to.as_str(), + previous_state = transition.from.as_str(), + bucket = %self.bucket, + provider = %self.config.source.provider, + endpoint_host = %self.endpoint_host, + "On-demand migration breaker opened; source traffic suspended" + ), + BreakerState::Closed | BreakerState::HalfOpen => info!( + event = EVENT_ODM_BREAKER_TRANSITION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + state = transition.to.as_str(), + previous_state = transition.from.as_str(), + bucket = %self.bucket, + provider = %self.config.source.provider, + endpoint_host = %self.endpoint_host, + "On-demand migration breaker state changed" + ), + } + } + + /// Singleflight plus concurrency limit for pulling `key`. The first + /// caller per key becomes the leader and waits for one of + /// `max_concurrent_pulls` permits (`queue_depth` counts that wait); + /// later callers for the same key become followers and never touch the + /// semaphore. Fails with `Canceled` when the state is torn down. + pub async fn acquire_pull_slot(self: &Arc, key: &str) -> Result { + if self.cancel.is_cancelled() { + return Err(PullError::canceled("bucket on-demand migration state was removed")); + } + let tx = { + let mut inflight = self.inflight.lock(); + if let Some(rx) = inflight.get(key) { + return Ok(PullSlot::Follower(PullFollower { rx: rx.clone() })); + } + let (tx, rx) = watch::channel(None); + inflight.insert(key.to_string(), rx); + tx + }; + + let mut entry_guard = InflightEntryGuard { + state: self, + key, + tx: &tx, + armed: true, + }; + let permit = { + let _queued = self.stats.queue_guard(); + tokio::select! { + permit = Arc::clone(&self.pull_semaphore).acquire_owned() => permit, + _ = self.cancel.cancelled() => { + return Err(PullError::canceled("bucket on-demand migration state was removed")); + } + } + }; + let permit = permit.map_err(|_| PullError::canceled("pull semaphore closed"))?; + entry_guard.armed = false; + drop(entry_guard); + + Ok(PullSlot::Leader(PullLeader { + state: Arc::clone(self), + key: key.to_string(), + tx, + _permit: permit, + _inflight: self.stats.inflight_guard(), + completed: false, + })) + } + + /// Keys currently being pulled (leader registered). + pub fn inflight_keys(&self) -> usize { + self.inflight.lock().len() + } + + pub fn snapshot(&self) -> OdmBucketSnapshot { + OdmBucketSnapshot { + bucket: self.bucket.clone(), + provider: self.config.source.provider.as_str().to_string(), + endpoint_host: self.endpoint_host.clone(), + applied_at: self.applied_at, + client_error: self.client.as_ref().err().map(|err| err.to_string()), + negative_cache_entries: self.negative_cache.len(), + inflight_keys: self.inflight_keys() as u64, + max_concurrent_pulls: self.config.policy.max_concurrent_pulls, + stats: self.stats.snapshot(self.breaker.state()), + } + } +} + +/// Read-only status of one bucket's runtime, for admin/status consumers. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct OdmBucketSnapshot { + pub bucket: String, + pub provider: String, + pub endpoint_host: String, + #[serde(with = "time::serde::rfc3339")] + pub applied_at: OffsetDateTime, + /// `Some` when the source client could not be built. + pub client_error: Option, + pub negative_cache_entries: u64, + pub inflight_keys: u64, + pub max_concurrent_pulls: u32, + pub stats: OdmStatsSnapshot, +} + +/// Maps the persisted config onto the client spec. `path_style` +/// `virtual` becomes the client's `VirtualHost`; `auto` is left for the +/// client to resolve per provider. `first_byte_ms` is the SDK read timeout; +/// `idle_ms` applies to body streaming and is enforced by the pull pipeline. +pub fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClientSpec { + let source = &config.source; + let policy = &config.policy; + SourceClientSpec { + endpoint: source.effective_endpoint(), + region: source.effective_region().to_string(), + bucket: source.bucket.clone(), + source_prefix: config.filter.source_prefix.clone(), + provider: source_provider(source.provider), + path_style: match source.path_style { + ConfigPathStyle::Auto => ClientPathStyle::Auto, + ConfigPathStyle::Path => ClientPathStyle::Path, + ConfigPathStyle::Virtual => ClientPathStyle::VirtualHost, + }, + credentials: source.credentials.as_ref().map(|credentials| RemoteCredentials { + access_key: credentials.access_key.clone(), + secret_key: credentials.secret_key.clone(), + session_token: credentials.session_token.clone(), + expiration: None, + account_id: String::new(), + }), + skip_tls_verify: source.tls.skip_verify, + ca_cert_pem: source.tls.ca_cert_pem.clone(), + timeouts: SourceTimeouts { + connect: Duration::from_millis(policy.source_timeout.connect_ms), + read: Duration::from_millis(policy.source_timeout.first_byte_ms), + }, + bandwidth_limit: policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new), + } +} + +fn source_provider(provider: Provider) -> SourceProvider { + match provider { + Provider::S3 => SourceProvider::S3, + Provider::Aws => SourceProvider::Aws, + Provider::Minio => SourceProvider::Minio, + Provider::Rustfs => SourceProvider::Rustfs, + Provider::R2 => SourceProvider::R2, + Provider::Gcs => SourceProvider::Gcs, + } +} + +fn endpoint_host(source: &SourceConfig) -> String { + Url::parse(&source.effective_endpoint()) + .ok() + .and_then(|url| url.host_str().map(str::to_ascii_lowercase)) + .unwrap_or_default() +} + +#[derive(Default)] +struct BucketSlot { + /// Generation of the last `apply`/`publish` that touched this bucket. + generation: u64, + state: Option>, +} + +/// Process-wide ODM runtime; see the module docs. +pub struct OnDemandMigrationSys { + module_enabled: AtomicBool, + buckets: RwLock>, + generation: AtomicU64, +} + +impl fmt::Debug for OnDemandMigrationSys { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OnDemandMigrationSys") + .field("module_enabled", &self.is_module_enabled()) + .field("buckets", &self.bucket_names()) + .finish() + } +} + +impl Default for OnDemandMigrationSys { + fn default() -> Self { + Self::new() + } +} + +impl OnDemandMigrationSys { + /// A detached instance; production code uses [`Self::get`]. + pub fn new() -> Self { + Self { + module_enabled: AtomicBool::new(false), + buckets: RwLock::new(HashMap::new()), + generation: AtomicU64::new(0), + } + } + + pub fn get() -> &'static Self { + GLOBAL_ON_DEMAND_MIGRATION_SYS.get_or_init(Self::new) + } + + /// Publishes the module switch resolved by the binary. Default `false`. + pub fn set_module_enabled(&self, enabled: bool) { + self.module_enabled.store(enabled, Ordering::Relaxed); + } + + pub fn is_module_enabled(&self) -> bool { + self.module_enabled.load(Ordering::Relaxed) + } + + /// Registers `publish` as the bucket-metadata publish hook. Returns + /// `false` when a hook was already registered. + pub fn register_config_hook(&'static self) -> bool { + ON_DEMAND_MIGRATION_CONFIG_HOOK + .set(Box::new(move |bucket, config| self.publish(bucket, config))) + .is_ok() + } + + /// Hook entry point: removals apply immediately, installs are spawned + /// (client construction is async). Requires a Tokio runtime for the + /// install path; without one the config is logged and skipped. + pub fn publish(&'static self, bucket: &str, config: Option<&OnDemandMigrationConfig>) { + let generation = self.next_generation(); + let Some(config) = self.desired(config) else { + self.remove_with_generation(bucket, generation); + return; + }; + if self.is_unchanged(bucket, config, generation) { + return; + } + let Ok(handle) = tokio::runtime::Handle::try_current() else { + warn!( + event = EVENT_ODM_BUCKET_STATE_APPLIED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + result = "skipped", + bucket = %bucket, + provider = %config.source.provider, + endpoint_host = %endpoint_host(&config.source), + "On-demand migration config published outside a Tokio runtime; state not built" + ); + return; + }; + let bucket = bucket.to_string(); + let config = config.clone(); + handle.spawn(async move { + self.apply_with_generation(&bucket, Some(&config), generation).await; + }); + } + + /// Installs, rebuilds, or removes the bucket state for `config`. + /// Idempotent: the same config on an installed bucket is a no-op. + pub async fn apply(&self, bucket: &str, config: Option<&OnDemandMigrationConfig>) -> ApplyOutcome { + let generation = self.next_generation(); + self.apply_with_generation(bucket, config, generation).await + } + + async fn apply_with_generation( + &self, + bucket: &str, + config: Option<&OnDemandMigrationConfig>, + generation: u64, + ) -> ApplyOutcome { + let Some(config) = self.desired(config) else { + return self.remove_with_generation(bucket, generation); + }; + if self.is_unchanged(bucket, config, generation) { + return ApplyOutcome::Unchanged; + } + let stats = self.state(bucket).map(|state| Arc::clone(&state.stats)).unwrap_or_default(); + let state = BucketOdmState::build(bucket, config, stats).await; + + let (outcome, previous) = { + let mut buckets = self.buckets.write(); + let slot = buckets.entry(bucket.to_string()).or_default(); + if slot.generation > generation { + (ApplyOutcome::Superseded, None) + } else { + slot.generation = generation; + let previous = slot.state.replace(Arc::clone(&state)); + let outcome = if previous.is_some() { + ApplyOutcome::Rebuilt + } else { + ApplyOutcome::Installed + }; + (outcome, previous) + } + }; + match outcome { + ApplyOutcome::Superseded => { + state.cancel.cancel(); + } + _ => { + if let Some(previous) = previous { + previous.cancel.cancel(); + } + info!( + event = EVENT_ODM_BUCKET_STATE_APPLIED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + result = if outcome == ApplyOutcome::Rebuilt { "rebuilt" } else { "installed" }, + bucket = %bucket, + provider = %config.source.provider, + endpoint_host = %state.endpoint_host, + client_ready = state.client.is_ok(), + "On-demand migration bucket state applied" + ); + } + } + outcome + } + + /// Removes a bucket's state (idempotent), cancelling its token. + pub fn remove(&self, bucket: &str) -> ApplyOutcome { + let generation = self.next_generation(); + self.remove_with_generation(bucket, generation) + } + + /// One-shot lookup: module switch, bucket state, prefix filter, + /// client availability, negative cache, breaker, in that order. + pub fn resolve(&self, bucket: &str, key: &str) -> Option { + if !self.is_module_enabled() { + return None; + } + self.state(bucket)?.resolve_key(key) + } + + pub fn state(&self, bucket: &str) -> Option> { + self.buckets.read().get(bucket).and_then(|slot| slot.state.clone()) + } + + pub fn bucket_names(&self) -> Vec { + let mut names: Vec = self + .buckets + .read() + .iter() + .filter(|(_, slot)| slot.state.is_some()) + .map(|(name, _)| name.clone()) + .collect(); + names.sort(); + names + } + + pub fn bucket_snapshot(&self, bucket: &str) -> Option { + self.state(bucket).map(|state| state.snapshot()) + } + + /// Snapshot of every configured bucket, sorted by name. + pub fn snapshot(&self) -> Vec { + let states: Vec> = self.buckets.read().values().filter_map(|slot| slot.state.clone()).collect(); + let mut snapshots: Vec = states.iter().map(|state| state.snapshot()).collect(); + snapshots.sort_by(|a, b| a.bucket.cmp(&b.bucket)); + snapshots + } + + fn next_generation(&self) -> u64 { + self.generation.fetch_add(1, Ordering::Relaxed) + 1 + } + + fn desired<'c>(&self, config: Option<&'c OnDemandMigrationConfig>) -> Option<&'c OnDemandMigrationConfig> { + config.filter(|config| config.enabled && self.is_module_enabled()) + } + + /// Claims `generation` for the bucket when the installed state already + /// matches `config` and has a usable client. + fn is_unchanged(&self, bucket: &str, config: &OnDemandMigrationConfig, generation: u64) -> bool { + let mut buckets = self.buckets.write(); + let Some(slot) = buckets.get_mut(bucket) else { + return false; + }; + let unchanged = slot + .state + .as_ref() + .is_some_and(|state| state.client.is_ok() && state.config == *config); + if unchanged && slot.generation < generation { + slot.generation = generation; + } + unchanged + } + + fn remove_with_generation(&self, bucket: &str, generation: u64) -> ApplyOutcome { + let removed = { + let mut buckets = self.buckets.write(); + let Some(slot) = buckets.get_mut(bucket) else { + return ApplyOutcome::NotDesired; + }; + if slot.generation > generation { + return ApplyOutcome::Superseded; + } + slot.generation = generation; + slot.state.take() + }; + let Some(state) = removed else { + return ApplyOutcome::NotDesired; + }; + state.cancel.cancel(); + info!( + event = EVENT_ODM_BUCKET_STATE_APPLIED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + result = "removed", + bucket = %bucket, + provider = %state.config.source.provider, + endpoint_host = %state.endpoint_host, + "On-demand migration bucket state removed" + ); + ApplyOutcome::Removed + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bucket::on_demand_migration::breaker::BREAKER_FAILURE_THRESHOLD; + use crate::bucket::on_demand_migration::config::{FilterConfig, PolicyConfig, SourceCredentials, SourceTimeout, TlsConfig}; + use std::sync::atomic::AtomicUsize; + use tokio::sync::Barrier; + + fn config(prefix: Option<&str>) -> OnDemandMigrationConfig { + OnDemandMigrationConfig { + version: 1, + enabled: true, + source: SourceConfig { + provider: Provider::Minio, + endpoint: Some("https://Source.Example.com:9000".to_string()), + region: "auto".to_string(), + bucket: "legacy".to_string(), + path_style: ConfigPathStyle::Auto, + credentials: Some(SourceCredentials { + access_key: "AK".to_string(), + secret_key: "SK".to_string(), + session_token: None, + }), + tls: TlsConfig::default(), + }, + filter: FilterConfig { + prefix: prefix.map(str::to_string), + source_prefix: None, + }, + policy: PolicyConfig::default(), + } + } + + fn enabled_sys() -> OnDemandMigrationSys { + let sys = OnDemandMigrationSys::new(); + sys.set_module_enabled(true); + sys + } + + fn ready_state(lookup: Option) -> Arc { + match lookup { + Some(OdmLookup::Ready { state }) => state, + other => panic!("expected Ready, got {other:?}"), + } + } + + #[tokio::test] + async fn apply_is_idempotent_rebuilds_on_change_and_removes() { + let sys = enabled_sys(); + let cfg = config(None); + assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed); + let first = ready_state(sys.resolve("b", "any")); + assert!(first.client().is_ok()); + + assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Unchanged); + let again = ready_state(sys.resolve("b", "any")); + assert!(Arc::ptr_eq(&first, &again), "identical config must not rebuild"); + assert!(!first.is_cancelled()); + + let mut changed = cfg.clone(); + changed.filter.prefix = Some("docs/".to_string()); + assert_eq!(sys.apply("b", Some(&changed)).await, ApplyOutcome::Rebuilt); + let rebuilt = ready_state(sys.resolve("b", "docs/x")); + assert!(!Arc::ptr_eq(&first, &rebuilt)); + assert!(first.is_cancelled(), "old state token fires on rebuild"); + assert!(!rebuilt.is_cancelled()); + assert!(Arc::ptr_eq(first.stats(), rebuilt.stats()), "counters survive a rebuild"); + + assert_eq!(sys.apply("b", None).await, ApplyOutcome::Removed); + assert!(rebuilt.is_cancelled()); + assert!(sys.resolve("b", "docs/x").is_none()); + assert!(sys.state("b").is_none()); + assert_eq!(sys.apply("b", None).await, ApplyOutcome::NotDesired); + } + + #[tokio::test] + async fn disabled_config_or_module_switch_removes_state() { + let sys = enabled_sys(); + let cfg = config(None); + sys.apply("b", Some(&cfg)).await; + let state = sys.state("b").expect("installed"); + + let mut disabled = cfg.clone(); + disabled.enabled = false; + assert_eq!(sys.apply("b", Some(&disabled)).await, ApplyOutcome::Removed); + assert!(state.is_cancelled()); + + sys.apply("b", Some(&cfg)).await; + let state = sys.state("b").expect("installed again"); + sys.set_module_enabled(false); + assert!(sys.resolve("b", "k").is_none(), "switch off: resolve never intervenes"); + assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Removed); + assert!(state.is_cancelled()); + + let off = OnDemandMigrationSys::new(); + assert!(!off.is_module_enabled(), "default switch is off"); + assert_eq!(off.apply("c", Some(&cfg)).await, ApplyOutcome::NotDesired); + } + + #[tokio::test] + async fn resolve_order_prefix_negative_cache_breaker() { + let sys = enabled_sys(); + sys.apply("b", Some(&config(Some("a/")))).await; + + assert!(sys.resolve("b", "b/x").is_none(), "prefix mismatch"); + assert!(sys.resolve("other", "a/x").is_none(), "unknown bucket"); + let state = ready_state(sys.resolve("b", "a/x")); + + state.observe_source(Duration::from_millis(1), "a/gone", Some(&SourceError::NotFound)); + assert!(matches!(sys.resolve("b", "a/gone"), Some(OdmLookup::NegativeCached { .. }))); + assert!(matches!(sys.resolve("b", "a/x"), Some(OdmLookup::Ready { .. }))); + + for _ in 0..BREAKER_FAILURE_THRESHOLD { + state.observe_source(Duration::from_millis(5), "a/x", Some(&SourceError::ServerError(503))); + } + assert_eq!(state.breaker().state(), BreakerState::Open); + assert!(matches!(sys.resolve("b", "a/x"), Some(OdmLookup::BreakerOpen { .. }))); + // Negative cache still wins over the breaker for its own keys. + assert!(matches!(sys.resolve("b", "a/gone"), Some(OdmLookup::NegativeCached { .. }))); + assert_eq!(state.stats().last_source_error().map(|e| e.class), Some("server_error".to_string())); + } + + #[tokio::test] + async fn anonymous_source_is_a_typed_state_error() { + let sys = enabled_sys(); + let mut cfg = config(None); + cfg.source.credentials = None; + assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed); + let state = sys.state("b").unwrap(); + assert_eq!(state.client().err(), Some(&OdmStateError::AnonymousUnsupported)); + match sys.resolve("b", "k") { + Some(OdmLookup::Unavailable { error, .. }) => assert_eq!(error, OdmStateError::AnonymousUnsupported), + other => panic!("expected Unavailable, got {other:?}"), + } + let snapshot = sys.bucket_snapshot("b").unwrap(); + assert!(snapshot.client_error.as_deref().unwrap().contains("anonymous")); + // A failed client is rebuilt on the next apply of the same config. + assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Rebuilt); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn singleflight_admits_one_leader_per_key() { + let sys = enabled_sys(); + sys.apply("b", Some(&config(None))).await; + let state = sys.state("b").unwrap(); + let barrier = Arc::new(Barrier::new(100)); + let leaders = Arc::new(AtomicUsize::new(0)); + let mut tasks = Vec::new(); + for _ in 0..100 { + let state = Arc::clone(&state); + let barrier = Arc::clone(&barrier); + let leaders = Arc::clone(&leaders); + tasks.push(tokio::spawn(async move { + let slot = state.acquire_pull_slot("same-key").await.unwrap(); + barrier.wait().await; + match slot { + PullSlot::Leader(leader) => { + leaders.fetch_add(1, Ordering::SeqCst); + assert_eq!(state.stats().inflight_pulls(), 1); + leader.complete(Ok(PullOutcome { + etag: Some("abc".into()), + size: 7, + })); + Ok(PullOutcome { + etag: Some("abc".into()), + size: 7, + }) + } + PullSlot::Follower(follower) => follower.wait().await, + } + })); + } + for task in tasks { + let result = task.await.unwrap(); + assert_eq!(result.unwrap().size, 7); + } + assert_eq!(leaders.load(Ordering::SeqCst), 1); + assert_eq!(state.inflight_keys(), 0); + assert_eq!(state.stats().inflight_pulls(), 0); + assert_eq!(state.stats().queue_depth(), 0); + } + + #[tokio::test] + async fn followers_fail_when_leader_drops_without_result() { + let sys = enabled_sys(); + sys.apply("b", Some(&config(None))).await; + let state = sys.state("b").unwrap(); + let leader = match state.acquire_pull_slot("k").await.unwrap() { + PullSlot::Leader(leader) => leader, + PullSlot::Follower(_) => panic!("first caller must lead"), + }; + let follower = match state.acquire_pull_slot("k").await.unwrap() { + PullSlot::Follower(follower) => follower, + PullSlot::Leader(_) => panic!("second caller must follow"), + }; + drop(leader); + let err = follower.wait().await.unwrap_err(); + assert_eq!(err.reason, PullFailureReason::Canceled); + // The key is free again. + assert!(matches!(state.acquire_pull_slot("k").await.unwrap(), PullSlot::Leader(_))); + } + + #[tokio::test] + async fn pull_semaphore_waits_and_reports_queue_depth() { + let sys = enabled_sys(); + let mut cfg = config(None); + cfg.policy.max_concurrent_pulls = 2; + sys.apply("b", Some(&cfg)).await; + let state = sys.state("b").unwrap(); + + let first = state.acquire_pull_slot("k1").await.unwrap(); + let second = state.acquire_pull_slot("k2").await.unwrap(); + assert_eq!(state.stats().inflight_pulls(), 2); + + let waiter = { + let state = Arc::clone(&state); + tokio::spawn(async move { state.acquire_pull_slot("k3").await }) + }; + tokio::time::timeout(Duration::from_millis(200), async { + while state.stats().queue_depth() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("third caller must register as queued"); + assert_eq!(state.stats().queue_depth(), 1); + assert!(!waiter.is_finished(), "third caller waits instead of being rejected"); + + drop(first); + let third = tokio::time::timeout(Duration::from_secs(2), waiter) + .await + .expect("permit handed over") + .unwrap() + .unwrap(); + assert!(matches!(third, PullSlot::Leader(_))); + assert_eq!(state.stats().queue_depth(), 0); + assert_eq!(state.stats().inflight_pulls(), 2); + drop(second); + drop(third); + assert_eq!(state.stats().inflight_pulls(), 0); + } + + #[tokio::test] + async fn removal_cancels_queued_pull_and_rejects_new_ones() { + let sys = enabled_sys(); + let mut cfg = config(None); + cfg.policy.max_concurrent_pulls = 1; + sys.apply("b", Some(&cfg)).await; + let state = sys.state("b").unwrap(); + let _held = state.acquire_pull_slot("k1").await.unwrap(); + let waiter = { + let state = Arc::clone(&state); + tokio::spawn(async move { state.acquire_pull_slot("k2").await }) + }; + tokio::time::timeout(Duration::from_millis(200), async { + while state.stats().queue_depth() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!(sys.remove("b"), ApplyOutcome::Removed); + let err = waiter.await.unwrap().unwrap_err(); + assert_eq!(err.reason, PullFailureReason::Canceled); + assert_eq!(state.stats().queue_depth(), 0); + assert_eq!(state.inflight_keys(), 1, "held leader still registered"); + assert_eq!(state.acquire_pull_slot("k3").await.unwrap_err().reason, PullFailureReason::Canceled); + } + + #[test] + fn source_client_spec_maps_every_field() { + let mut cfg = config(Some("local/")); + cfg.source.path_style = ConfigPathStyle::Virtual; + cfg.source.credentials.as_mut().unwrap().session_token = Some("tok".to_string()); + cfg.source.tls = TlsConfig { + skip_verify: true, + ca_cert_pem: Some("-----BEGIN CERTIFICATE-----".to_string()), + }; + cfg.filter.source_prefix = Some("old/".to_string()); + cfg.policy.source_timeout = SourceTimeout { + connect_ms: 1500, + first_byte_ms: 2500, + idle_ms: 3500, + }; + cfg.policy.bandwidth_limit_bytes_per_sec = Some(1 << 20); + + let spec = source_client_spec(&cfg); + assert_eq!(spec.endpoint, "https://Source.Example.com:9000"); + assert_eq!(spec.region, "us-east-1", "auto maps to the signing fallback"); + assert_eq!(spec.bucket, "legacy"); + assert_eq!(spec.source_prefix.as_deref(), Some("old/")); + assert_eq!(spec.provider, SourceProvider::Minio); + assert_eq!(spec.path_style, ClientPathStyle::VirtualHost); + let credentials = spec.credentials.as_ref().unwrap(); + assert_eq!(credentials.access_key, "AK"); + assert_eq!(credentials.secret_key, "SK"); + assert_eq!(credentials.session_token.as_deref(), Some("tok")); + assert_eq!(credentials.expiration, None); + assert!(spec.skip_tls_verify); + assert!(spec.ca_cert_pem.is_some()); + assert_eq!(spec.timeouts.connect, Duration::from_millis(1500)); + assert_eq!(spec.timeouts.read, Duration::from_millis(2500)); + assert_eq!(spec.bandwidth_limit, NonZeroU64::new(1 << 20)); + + let mut aws = config(None); + aws.source.provider = Provider::Aws; + aws.source.endpoint = None; + aws.source.region = "eu-west-1".to_string(); + aws.source.credentials = None; + let spec = source_client_spec(&aws); + assert_eq!(spec.endpoint, "https://s3.eu-west-1.amazonaws.com"); + assert_eq!(spec.provider, SourceProvider::Aws); + assert_eq!(spec.path_style, ClientPathStyle::Auto); + assert!(spec.credentials.is_none()); + assert_eq!(endpoint_host(&aws.source), "s3.eu-west-1.amazonaws.com"); + assert_eq!(endpoint_host(&cfg.source), "source.example.com"); + + for (provider, expected) in [ + (Provider::S3, SourceProvider::S3), + (Provider::Rustfs, SourceProvider::Rustfs), + (Provider::R2, SourceProvider::R2), + (Provider::Gcs, SourceProvider::Gcs), + ] { + assert_eq!(source_provider(provider), expected); + } + } + + #[tokio::test] + async fn publish_spawns_install_and_removes_synchronously() { + let sys: &'static OnDemandMigrationSys = Box::leak(Box::new(enabled_sys())); + let cfg = config(None); + sys.publish("p", Some(&cfg)); + tokio::time::timeout(Duration::from_secs(5), async { + while sys.state("p").is_none() { + tokio::task::yield_now().await; + } + }) + .await + .expect("publish must install the state in the background"); + let state = sys.state("p").unwrap(); + + sys.publish("p", Some(&cfg)); + tokio::task::yield_now().await; + assert!(Arc::ptr_eq(&state, &sys.state("p").unwrap()), "unchanged publish is a no-op"); + + sys.publish("p", None); + assert!(sys.state("p").is_none(), "removal is synchronous"); + assert!(state.is_cancelled()); + } + + #[tokio::test] + async fn stale_install_cannot_overwrite_a_later_removal() { + let sys = enabled_sys(); + let cfg = config(None); + let older = sys.next_generation(); + let newer = sys.next_generation(); + assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::NotDesired); + // The removal above did not create a slot; simulate an install that + // started before it and finishes after. + sys.apply_with_generation("b", Some(&cfg), older).await; + assert!(sys.state("b").is_some(), "no slot yet, so the older install lands"); + + let installed = sys.state("b").unwrap(); + let older = sys.next_generation(); + let newer = sys.next_generation(); + assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::Removed); + assert!(installed.is_cancelled()); + assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded); + assert!(sys.state("b").is_none(), "the stale install is discarded"); + } + + #[tokio::test] + async fn snapshot_lists_buckets_sorted_and_serializes() { + let sys = enabled_sys(); + sys.apply("zeta", Some(&config(None))).await; + sys.apply("alpha", Some(&config(Some("a/")))).await; + assert_eq!(sys.bucket_names(), vec!["alpha".to_string(), "zeta".to_string()]); + let snapshots = sys.snapshot(); + assert_eq!(snapshots.len(), 2); + assert_eq!(snapshots[0].bucket, "alpha"); + assert_eq!(snapshots[0].provider, "minio"); + assert_eq!(snapshots[0].endpoint_host, "source.example.com"); + assert_eq!(snapshots[0].client_error, None); + assert_eq!(snapshots[0].stats.breaker_state, BreakerState::Closed); + let json = serde_json::to_string(&snapshots[0]).unwrap(); + assert!(!json.contains("SK"), "snapshot must not carry credentials"); + let round_trip: OdmBucketSnapshot = serde_json::from_str(&json).unwrap(); + assert_eq!(round_trip, snapshots[0]); + let debug = format!("{:?}", sys.state("alpha").unwrap()); + assert!(!debug.contains("SK"), "Debug must not carry credentials: {debug}"); + } +} diff --git a/rustfs/src/module_switches.rs b/rustfs/src/module_switches.rs index ced58b0ec..a9ba1e710 100644 --- a/rustfs/src/module_switches.rs +++ b/rustfs/src/module_switches.rs @@ -35,9 +35,14 @@ pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED"; pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL"; pub(crate) const ENV_BITROT_SELFTEST_ENABLE: &str = "RUSTFS_BITROT_SELFTEST_ENABLE"; pub(crate) const ENV_BITROT_SELFTEST_STRICT: &str = "RUSTFS_BITROT_SELFTEST_STRICT"; +/// On-demand migration module switch (rustfs/backlog#2152). Off until GA +/// (rustfs/backlog#2163) so every intermediate PR ships dark. +pub(crate) const ENV_ON_DEMAND_MIGRATION_ENABLED: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED"; +pub(crate) const DEFAULT_ON_DEMAND_MIGRATION_ENABLED: bool = false; static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE); static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE); +static ON_DEMAND_MIGRATION_MODULE_ENABLED: AtomicBool = AtomicBool::new(DEFAULT_ON_DEMAND_MIGRATION_ENABLED); /// Whether the data scanner is enabled, defaulting to on. pub(crate) fn scanner_enabled_from_env() -> bool { @@ -80,3 +85,51 @@ pub fn is_notify_module_enabled() -> bool { pub(crate) fn set_notify_module_enabled(enabled: bool) { NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed); } + +/// Whether the on-demand migration module is enabled, defaulting to off. +/// Read once at startup by `startup_bucket_metadata` and published below. +pub(crate) fn on_demand_migration_enabled_from_env() -> bool { + rustfs_utils::get_env_bool(ENV_ON_DEMAND_MIGRATION_ENABLED, DEFAULT_ON_DEMAND_MIGRATION_ENABLED) +} + +/// Last published on-demand migration module state. +pub fn is_on_demand_migration_module_enabled() -> bool { + ON_DEMAND_MIGRATION_MODULE_ENABLED.load(Ordering::Relaxed) +} + +/// Publish the on-demand migration module state resolved at startup. The +/// ecstore runtime receives the same value through +/// `OnDemandMigrationSys::set_module_enabled`, since ecstore cannot read +/// this crate. +pub(crate) fn set_on_demand_migration_module_enabled(enabled: bool) { + ON_DEMAND_MIGRATION_MODULE_ENABLED.store(enabled, Ordering::Relaxed); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn on_demand_migration_switch_defaults_off_and_follows_env() { + temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, None::<&str>, || { + assert!(!on_demand_migration_enabled_from_env()); + }); + temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("true"), || { + assert!(on_demand_migration_enabled_from_env()); + }); + temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("not-a-bool"), || { + assert!(!on_demand_migration_enabled_from_env(), "unparsable values keep the default"); + }); + } + + #[test] + fn on_demand_migration_switch_publishes_to_the_cell() { + // The cell is process-global; restore it so sibling tests observe the default. + let before = is_on_demand_migration_module_enabled(); + set_on_demand_migration_module_enabled(true); + assert!(is_on_demand_migration_module_enabled()); + set_on_demand_migration_module_enabled(false); + assert!(!is_on_demand_migration_module_enabled()); + set_on_demand_migration_module_enabled(before); + } +} diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index dec3f50c1..fc336e990 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -12,10 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::module_switches::{on_demand_migration_enabled_from_env, set_on_demand_migration_module_enabled}; use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions}; use crate::storage_api::startup::bucket_metadata::{ - ECStore, Error as StorageError, Result as StorageResult, get_global_replication_pool, init_bucket_metadata_sys, - reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config, + ECStore, Error as StorageError, OnDemandMigrationSys, Result as StorageResult, get_global_replication_pool, + init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config, }; use std::{ io::{Error as IoError, Result as IoResult}, @@ -24,11 +25,13 @@ use std::{ }; use tokio_util::sync::CancellationToken; +const EVENT_ON_DEMAND_MIGRATION_RUNTIME_INITIALIZED: &str = "on_demand_migration_runtime_initialized"; const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_CANCELED: &str = "replication_resync_startup_background_canceled"; const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_COMPLETED: &str = "replication_resync_startup_background_completed"; const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_FAILED: &str = "replication_resync_startup_background_failed"; const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_STARTED: &str = "replication_resync_startup_background_started"; const LOG_COMPONENT_STARTUP_BUCKET_METADATA: &str = "startup_bucket_metadata"; +const LOG_SUBSYSTEM_ON_DEMAND_MIGRATION: &str = "on_demand_migration"; const LOG_SUBSYSTEM_REPLICATION: &str = "replication"; const METRIC_REPLICATION_RESYNC_STARTUP_BACKGROUND_DURATION_SECONDS: &str = "rustfs_replication_resync_startup_background_duration_seconds"; @@ -58,6 +61,7 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc, c let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); try_migrate_bucket_metadata(store.clone()).await; + init_on_demand_migration_runtime(); init_bucket_metadata_sys(store.clone(), buckets.clone()).await; try_migrate_iam_config(store).await; spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx.clone(), false); @@ -79,12 +83,33 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance try_migrate_bucket_metadata(store.clone()).await; try_migrate_iam_config(store.clone()).await; + init_on_demand_migration_runtime(); init_bucket_metadata_sys(store, buckets.clone()).await; spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx, true); Ok(buckets) } +/// Publishes the on-demand migration module switch and registers the +/// runtime's config hook before bucket metadata is loaded, so every cache +/// install path (initial load included) reaches `OnDemandMigrationSys` +/// (rustfs/backlog#2152). Idempotent across embedded and server startups. +fn init_on_demand_migration_runtime() { + let enabled = on_demand_migration_enabled_from_env(); + set_on_demand_migration_module_enabled(enabled); + let sys = OnDemandMigrationSys::get(); + sys.set_module_enabled(enabled); + let hook_registered = sys.register_config_hook(); + tracing::info!( + event = EVENT_ON_DEMAND_MIGRATION_RUNTIME_INITIALIZED, + component = LOG_COMPONENT_STARTUP_BUCKET_METADATA, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + state = if enabled { "enabled" } else { "disabled" }, + hook_registered, + "On-demand migration runtime initialized" + ); +} + fn spawn_bucket_resync_startup_reconcile(buckets: Vec, ctx: CancellationToken, init_resync_after_reconcile: bool) { tokio::spawn(async move { describe_bucket_resync_startup_background_metrics(); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 258a1737e..066780d3b 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -407,8 +407,8 @@ pub(crate) mod ecstore_bucket { #[cfg(test)] pub(crate) use rustfs_ecstore::api::bucket::lifecycle::tier_delete_journal::test_util::install_all_v6_fleet_capability_proof; pub(crate) use rustfs_ecstore::api::bucket::{ - bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, policy_sys, - replication, tagging, target, utils, + bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, on_demand_migration, + policy_sys, replication, tagging, target, utils, }; pub(crate) use rustfs_ecstore::api::bucket::{quota, versioning, versioning_sys}; } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 3f3b20e0c..2f7a442a9 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -290,6 +290,7 @@ pub(crate) mod startup { } } + pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::OnDemandMigrationSys; pub(crate) use crate::storage::storage_api::{ ECStore, Error, Result, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config,