rustfs/crates/replication/src/stats.rs

// 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.

use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering, Ordering as AtomicOrdering};
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::Mutex;

use crate::resync::ResyncStatusType;

const ROLLING_WINDOW: Duration = Duration::from_secs(60);
const FAILURE_LAST_HOUR_WINDOW: Duration = Duration::from_secs(60 * 60);
const I64_MAX_AS_U128: u128 = 9_223_372_036_854_775_807;

#[derive(Debug)]
pub struct ExponentialMovingAverage {
    pub alpha: f64,
    pub value: AtomicU64,
    pub last_update: Arc<Mutex<SystemTime>>,
}

impl ExponentialMovingAverage {
    pub fn new() -> Self {
        let now = SystemTime::now();
        Self {
            alpha: 0.1,
            value: AtomicU64::new(0_f64.to_bits()),
            last_update: Arc::new(Mutex::new(now)),
        }
    }

    pub fn add_value(&self, value: f64, timestamp: SystemTime) {
        let current_value = f64::from_bits(self.value.load(AtomicOrdering::Relaxed));
        let new_value = if current_value == 0.0 {
            value
        } else {
            self.alpha * value + (1.0 - self.alpha) * current_value
        };
        self.value.store(new_value.to_bits(), AtomicOrdering::Relaxed);

        if let Ok(mut last_update) = self.last_update.try_lock() {
            *last_update = timestamp;
        }
    }

    pub fn get_current_average(&self) -> f64 {
        f64::from_bits(self.value.load(AtomicOrdering::Relaxed))
    }

    pub fn update_exponential_moving_average(&self, now: SystemTime) {
        if let Ok(mut last_update_guard) = self.last_update.try_lock() {
            let last_update = *last_update_guard;
            if let Ok(duration) = now.duration_since(last_update)
                && duration.as_secs() > 0
            {
                let decay = (-duration.as_secs_f64() / 60.0).exp();
                let current_value = f64::from_bits(self.value.load(AtomicOrdering::Relaxed));
                self.value.store((current_value * decay).to_bits(), AtomicOrdering::Relaxed);
                *last_update_guard = now;
            }
        }
    }

    pub fn merge(&self, other: &ExponentialMovingAverage) -> Self {
        let now = SystemTime::now();
        let self_value = f64::from_bits(self.value.load(AtomicOrdering::Relaxed));
        let other_value = f64::from_bits(other.value.load(AtomicOrdering::Relaxed));
        let merged_value = (self_value + other_value) / 2.0;

        let self_time = self.last_update.try_lock().map(|t| *t).unwrap_or(now);
        let other_time = other.last_update.try_lock().map(|t| *t).unwrap_or(now);
        let merged_time = self_time.max(other_time);

        Self {
            alpha: self.alpha,
            value: AtomicU64::new(merged_value.to_bits()),
            last_update: Arc::new(Mutex::new(merged_time)),
        }
    }
}

impl Clone for ExponentialMovingAverage {
    fn clone(&self) -> Self {
        let now = SystemTime::now();
        let value = self.value.load(AtomicOrdering::Relaxed);
        let last_update = self.last_update.try_lock().map(|t| *t).unwrap_or(now);

        Self {
            alpha: self.alpha,
            value: AtomicU64::new(value),
            last_update: Arc::new(Mutex::new(last_update)),
        }
    }
}

impl Default for ExponentialMovingAverage {
    fn default() -> Self {
        Self::new()
    }
}

impl Serialize for ExponentialMovingAverage {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        use serde::ser::SerializeStruct;
        let mut state = serializer.serialize_struct("ExponentialMovingAverage", 3)?;
        state.serialize_field("alpha", &self.alpha)?;
        state.serialize_field("value", &f64::from_bits(self.value.load(AtomicOrdering::Relaxed)))?;
        let last_update = self.last_update.try_lock().map(|t| *t).unwrap_or(SystemTime::UNIX_EPOCH);
        state.serialize_field("last_update", &last_update)?;
        state.end()
    }
}

impl<'de> Deserialize<'de> for ExponentialMovingAverage {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(Deserialize)]
        struct ExponentialMovingAverageData {
            alpha: f64,
            value: f64,
            last_update: SystemTime,
        }

        let data = ExponentialMovingAverageData::deserialize(deserializer)?;
        Ok(Self {
            alpha: data.alpha,
            value: AtomicU64::new(data.value.to_bits()),
            last_update: Arc::new(Mutex::new(data.last_update)),
        })
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct XferStats {
    pub avg: f64,
    pub curr: f64,
    pub peak: f64,
    pub measure: ExponentialMovingAverage,
}

impl XferStats {
    pub fn new() -> Self {
        Self {
            avg: 0.0,
            curr: 0.0,
            peak: 0.0,
            measure: ExponentialMovingAverage::new(),
        }
    }

    pub fn add_size(&mut self, size: i64, duration: Duration) {
        if duration.as_nanos() > 0 {
            let rate = (size as f64) / duration.as_secs_f64();
            self.curr = rate;
            if rate > self.peak {
                self.peak = rate;
            }
            self.measure.add_value(rate, SystemTime::now());
            self.avg = self.measure.get_current_average();
        }
    }

    pub fn clone_stats(&self) -> Self {
        Self {
            avg: self.avg,
            curr: self.curr,
            peak: self.peak,
            measure: self.measure.clone(),
        }
    }

    pub fn merge(&self, other: &XferStats) -> Self {
        Self {
            avg: (self.avg + other.avg) / 2.0,
            curr: self.curr + other.curr,
            peak: self.peak.max(other.peak),
            measure: self.measure.merge(&other.measure),
        }
    }

    pub fn update_exponential_moving_average(&mut self, now: SystemTime) {
        self.measure.update_exponential_moving_average(now);
        self.avg = self.measure.get_current_average();
    }
}

impl Default for XferStats {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Debug, Default, Serialize, Deserialize)]
pub struct InQueueStats {
    pub bytes: i64,
    pub count: i64,
    #[serde(skip)]
    pub now_bytes: AtomicI64,
    #[serde(skip)]
    pub now_count: AtomicI64,
}

#[derive(Debug, Clone)]
struct QueueSample {
    observed_at: Instant,
    bytes: i64,
    count: i64,
}

impl Clone for InQueueStats {
    fn clone(&self) -> Self {
        Self {
            bytes: self.bytes,
            count: self.count,
            now_bytes: AtomicI64::new(self.now_bytes.load(Ordering::Relaxed)),
            now_count: AtomicI64::new(self.now_count.load(Ordering::Relaxed)),
        }
    }
}

impl InQueueStats {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn add_current(&self, bytes: i64, count: i64) {
        self.now_bytes.fetch_add(bytes.max(0), Ordering::Relaxed);
        self.now_count.fetch_add(count.max(0), Ordering::Relaxed);
    }

    pub fn subtract_current(&self, bytes: i64, count: i64) {
        saturating_atomic_sub(&self.now_bytes, bytes.max(0));
        saturating_atomic_sub(&self.now_count, count.max(0));
    }

    pub fn get_current_bytes(&self) -> i64 {
        self.now_bytes.load(Ordering::Relaxed)
    }

    pub fn get_current_count(&self) -> i64 {
        self.now_count.load(Ordering::Relaxed)
    }
}

fn saturating_atomic_sub(value: &AtomicI64, delta: i64) {
    if delta == 0 {
        return;
    }

    let _ = value.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(delta).max(0)));
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct InQueueMetric {
    pub curr: InQueueStats,
    pub avg: InQueueStats,
    pub max: InQueueStats,
    pub last_minute: InQueueStats,
    #[serde(skip)]
    samples: VecDeque<QueueSample>,
}

impl InQueueMetric {
    fn observe(&mut self, observed_at: Instant) {
        let bytes = self.curr.now_bytes.load(Ordering::Relaxed);
        let count = self.curr.now_count.load(Ordering::Relaxed);

        self.curr.bytes = bytes;
        self.curr.count = count;
        self.samples.push_back(QueueSample {
            observed_at,
            bytes,
            count,
        });

        while self
            .samples
            .front()
            .is_some_and(|sample| observed_at.duration_since(sample.observed_at) > ROLLING_WINDOW)
        {
            self.samples.pop_front();
        }

        if self.samples.is_empty() {
            self.avg = InQueueStats::default();
            self.max = InQueueStats::default();
            self.last_minute = InQueueStats::default();
            return;
        }

        let sample_count = self.samples.len() as i64;
        let total_bytes = self.samples.iter().map(|sample| sample.bytes).sum::<i64>();
        let total_count = self.samples.iter().map(|sample| sample.count).sum::<i64>();
        let max_bytes = self.samples.iter().map(|sample| sample.bytes).max().unwrap_or(0);
        let max_count = self.samples.iter().map(|sample| sample.count).max().unwrap_or(0);

        self.avg.bytes = total_bytes / sample_count;
        self.avg.count = total_count / sample_count;
        self.max.bytes = max_bytes;
        self.max.count = max_count;
        self.last_minute.bytes = self.avg.bytes;
        self.last_minute.count = self.avg.count;
    }

    pub fn snapshot(&self) -> Self {
        let mut snapshot = self.clone();
        snapshot.samples.clear();
        snapshot.curr.bytes = snapshot.curr.now_bytes.load(Ordering::Relaxed);
        snapshot.curr.count = snapshot.curr.now_count.load(Ordering::Relaxed);
        snapshot
    }

    pub fn merge(&self, other: &InQueueMetric) -> Self {
        Self {
            curr: InQueueStats {
                bytes: self.curr.bytes.saturating_add(other.curr.bytes),
                count: self.curr.count.saturating_add(other.curr.count),
                now_bytes: AtomicI64::new(
                    self.curr
                        .now_bytes
                        .load(Ordering::Relaxed)
                        .saturating_add(other.curr.now_bytes.load(Ordering::Relaxed)),
                ),
                now_count: AtomicI64::new(
                    self.curr
                        .now_count
                        .load(Ordering::Relaxed)
                        .saturating_add(other.curr.now_count.load(Ordering::Relaxed)),
                ),
            },
            avg: InQueueStats {
                bytes: self.avg.bytes.saturating_add(other.avg.bytes) / 2,
                count: self.avg.count.saturating_add(other.avg.count) / 2,
                ..Default::default()
            },
            max: InQueueStats {
                bytes: self.max.bytes.max(other.max.bytes),
                count: self.max.count.max(other.max.count),
                ..Default::default()
            },
            last_minute: InQueueStats {
                bytes: self.last_minute.bytes.saturating_add(other.last_minute.bytes),
                count: self.last_minute.count.saturating_add(other.last_minute.count),
                ..Default::default()
            },
            samples: VecDeque::new(),
        }
    }
}

#[derive(Debug, Default)]
pub struct QueueCache {
    pub bucket_stats: HashMap<String, InQueueMetric>,
    pub sr_queue_stats: InQueueMetric,
}

impl QueueCache {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn inc(&mut self, bucket: &str, size: i64) {
        let stats = self.bucket_stats.entry(bucket.to_string()).or_default();
        stats.curr.add_current(size, 1);
        self.sr_queue_stats.curr.add_current(size, 1);
    }

    pub fn dec(&mut self, bucket: &str, size: i64) {
        let stats = self.bucket_stats.entry(bucket.to_string()).or_default();
        stats.curr.subtract_current(size, 1);
        self.sr_queue_stats.curr.subtract_current(size, 1);
    }

    pub fn update(&mut self) {
        let observed_at = Instant::now();
        self.sr_queue_stats.observe(observed_at);
        for stats in self.bucket_stats.values_mut() {
            stats.observe(observed_at);
        }
    }

    pub fn get_bucket_stats(&self, bucket: &str) -> InQueueMetric {
        self.bucket_stats.get(bucket).map(InQueueMetric::snapshot).unwrap_or_default()
    }

    pub fn get_site_stats(&self) -> InQueueMetric {
        self.sr_queue_stats.snapshot()
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ProxyMetric {
    pub get_total: i64,
    pub get_failed: i64,
    pub get_tag_total: i64,
    pub get_tag_failed: i64,
    pub put_total: i64,
    pub put_failed: i64,
    pub put_tag_total: i64,
    pub put_tag_failed: i64,
    pub delete_tag_total: i64,
    pub delete_tag_failed: i64,
    pub head_total: i64,
    pub head_failed: i64,
}

impl ProxyMetric {
    pub fn add(&mut self, other: &ProxyMetric) {
        self.get_total += other.get_total;
        self.get_failed += other.get_failed;
        self.get_tag_total += other.get_tag_total;
        self.get_tag_failed += other.get_tag_failed;
        self.put_total += other.put_total;
        self.put_failed += other.put_failed;
        self.put_tag_total += other.put_tag_total;
        self.put_tag_failed += other.put_tag_failed;
        self.delete_tag_total += other.delete_tag_total;
        self.delete_tag_failed += other.delete_tag_failed;
        self.head_total += other.head_total;
        self.head_failed += other.head_failed;
    }
}

#[derive(Debug, Clone, Default)]
pub struct ProxyStatsCache {
    bucket_stats: HashMap<String, ProxyMetric>,
}

impl ProxyStatsCache {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn inc(&mut self, bucket: &str, api: &str, is_err: bool) {
        let metric = self.bucket_stats.entry(bucket.to_string()).or_default();

        match api {
            "GetObject" => {
                metric.get_total += 1;
                if is_err {
                    metric.get_failed += 1;
                }
            }
            "GetObjectTagging" => {
                metric.get_tag_total += 1;
                if is_err {
                    metric.get_tag_failed += 1;
                }
            }
            "PutObject" => {
                metric.put_total += 1;
                if is_err {
                    metric.put_failed += 1;
                }
            }
            "PutObjectTagging" => {
                metric.put_tag_total += 1;
                if is_err {
                    metric.put_tag_failed += 1;
                }
            }
            "HeadObject" => {
                metric.head_total += 1;
                if is_err {
                    metric.head_failed += 1;
                }
            }
            "DeleteObjectTagging" => {
                metric.delete_tag_total += 1;
                if is_err {
                    metric.delete_tag_failed += 1;
                }
            }
            _ => {}
        }
    }

    pub fn get_bucket_stats(&self, bucket: &str) -> ProxyMetric {
        self.bucket_stats.get(bucket).cloned().unwrap_or_default()
    }

    pub fn bucket_names(&self) -> impl Iterator<Item = &str> {
        self.bucket_stats.keys().map(String::as_str)
    }

    pub fn get_site_stats(&self) -> ProxyMetric {
        let mut total = ProxyMetric::default();
        for metric in self.bucket_stats.values() {
            total.add(metric);
        }
        total
    }
}

#[derive(Debug, Clone)]
struct FailureSample {
    observed_at: Instant,
    size: i64,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct FailStats {
    pub count: i64,
    pub size: i64,
    /// Rolling-window snapshots refreshed at collection time
    /// ([`Self::refresh_windows`]). The raw samples (`recent`) are process
    /// local (serde-skipped), so these fields are what survives the peer-RPC
    /// wire and [`Self::merge`]-based cluster aggregation.
    #[serde(default)]
    pub last_minute: FailedMetric,
    #[serde(default)]
    pub last_hour: FailedMetric,
    #[serde(skip)]
    recent: VecDeque<FailureSample>,
}

impl FailStats {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn add_size<E>(&mut self, size: i64, _err: Option<&E>) {
        let observed_at = Instant::now();
        self.count = self.count.saturating_add(1);
        self.size = self.size.saturating_add(size);
        self.recent.push_back(FailureSample { observed_at, size });
        self.prune(observed_at);
    }

    /// Recompute the serializable rolling-window snapshots from the local
    /// samples. Called at the collection point (per-node stats snapshot),
    /// never on the failure hot path — the two deque scans are O(window) and
    /// `add_size` runs under the bucket-stats write lock. Only meaningful on
    /// the live per-node struct: a deserialized or merged struct has no
    /// samples, and refreshing it would wipe the aggregated windows.
    pub fn refresh_windows(&mut self) {
        self.last_minute = self.recent_since(Duration::from_secs(60));
        self.last_hour = self.recent_since(Duration::from_secs(3600));
    }

    fn prune(&mut self, observed_at: Instant) {
        while self
            .recent
            .front()
            .is_some_and(|sample| observed_at.duration_since(sample.observed_at) > FAILURE_LAST_HOUR_WINDOW)
        {
            self.recent.pop_front();
        }
    }

    pub fn recent_since(&self, window: Duration) -> FailedMetric {
        let now = Instant::now();
        let mut count = 0i64;
        let mut size = 0i64;
        for sample in self.recent.iter().rev() {
            if now.duration_since(sample.observed_at) > window {
                break;
            }
            count += 1;
            size += sample.size;
        }
        FailedMetric { count, size }
    }

    /// Both rolling windows from one walk of the samples. `short` must be the
    /// narrower window; the walk stops at `long`. Callers that need both (the
    /// per-node site snapshot) would otherwise scan the deque twice while
    /// holding the bucket-stats read lock, and the deque is only bounded by
    /// the one-hour window - an unreachable target under load fills it.
    pub fn recent_windows(&self, short: Duration, long: Duration) -> (FailedMetric, FailedMetric) {
        let now = Instant::now();
        let mut short_metric = FailedMetric::default();
        let mut long_metric = FailedMetric::default();
        for sample in self.recent.iter().rev() {
            let age = now.duration_since(sample.observed_at);
            if age > long {
                break;
            }
            if age <= short {
                short_metric.count += 1;
                short_metric.size += sample.size;
            }
            long_metric.count += 1;
            long_metric.size += sample.size;
        }
        (short_metric, long_metric)
    }

    pub fn merge(&self, other: &FailStats) -> Self {
        Self {
            count: self.count.saturating_add(other.count),
            size: self.size.saturating_add(other.size),
            // The window snapshots sum across nodes; the raw samples do not
            // travel and stay empty on aggregated structs.
            last_minute: FailedMetric {
                count: self.last_minute.count.saturating_add(other.last_minute.count),
                size: self.last_minute.size.saturating_add(other.last_minute.size),
            },
            last_hour: FailedMetric {
                count: self.last_hour.count.saturating_add(other.last_hour.count),
                size: self.last_hour.size.saturating_add(other.last_hour.size),
            },
            recent: VecDeque::new(),
        }
    }

    pub fn to_metric(&self) -> FailedMetric {
        FailedMetric {
            count: self.count,
            size: self.size,
        }
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct FailedMetric {
    pub count: i64,
    pub size: i64,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LatencyStats {
    pub avg: f64,
    pub curr: f64,
    pub max: f64,
}

impl LatencyStats {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn update(&mut self, _size: i64, duration: Duration) {
        let latency = duration.as_millis() as f64;
        self.curr = latency;
        if latency > self.max {
            self.max = latency;
        }
        self.avg = (self.avg + latency) / 2.0;
    }

    pub fn merge(&self, other: &LatencyStats) -> Self {
        Self {
            avg: (self.avg + other.avg) / 2.0,
            curr: self.curr.max(other.curr),
            max: self.max.max(other.max),
        }
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BucketReplicationStat {
    pub replicated_size: i64,
    pub replicated_count: i64,
    pub failed: FailedMetric,
    pub fail_stats: FailStats,
    pub latency: LatencyStats,
    pub xfer_rate_lrg: XferStats,
    pub xfer_rate_sml: XferStats,
    pub bandwidth_limit_bytes_per_sec: i64,
    pub current_bandwidth_bytes_per_sec: f64,
    #[serde(default)]
    pub latency_scope: ReplicationMetricScope,
    #[serde(default)]
    pub bandwidth_scope: ReplicationMetricScope,
}

impl BucketReplicationStat {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn update_xfer_rate(&mut self, size: i64, duration: Duration) {
        // Same boundary as the worker-pool split and minio-go's
        // Large/Small transfer-summary labels: >= 128 MiB is "large".
        if size >= crate::runtime::MIN_LARGE_OBJ_SIZE {
            self.xfer_rate_lrg.add_size(size, duration);
        } else {
            self.xfer_rate_sml.add_size(size, duration);
        }
    }

    pub fn set_node_local_bandwidth(&mut self, limit_bytes_per_sec: i64, current_bytes_per_sec: f64) {
        self.bandwidth_limit_bytes_per_sec = limit_bytes_per_sec;
        self.current_bandwidth_bytes_per_sec = current_bytes_per_sec;
        self.bandwidth_scope = ReplicationMetricScope::NodeLocal;
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReplicationMetricScope {
    #[default]
    Unavailable,
    NodeLocal,
    ClusterAggregated,
    PartialCluster,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QueueStats {
    pub nodes: Vec<QueueNode>,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QueueNode {
    pub q_stats: InQueueMetric,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BucketReplicationStats {
    pub stats: HashMap<String, BucketReplicationStat>,
    pub replica_size: i64,
    pub replica_count: i64,
    pub replicated_size: i64,
    pub replicated_count: i64,
    pub q_stat: InQueueMetric,
    #[serde(default)]
    pub resync_started_count: i64,
    #[serde(default)]
    pub resync_completed_count: i64,
    #[serde(default)]
    pub resync_failed_count: i64,
    #[serde(default)]
    pub resync_canceled_count: i64,
    #[serde(default)]
    pub resync_duration_ms: i64,
    #[serde(default)]
    pub provider_available: bool,
    #[serde(default)]
    pub cluster_complete: bool,
    #[serde(default)]
    pub observed_node_count: u32,
    #[serde(default)]
    pub expected_node_count: u32,
    #[serde(default)]
    pub queue_scope: ReplicationMetricScope,
}

impl BucketReplicationStats {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn is_empty(&self) -> bool {
        self.stats.is_empty()
            && self.replica_size == 0
            && self.replicated_size == 0
            && self.resync_started_count == 0
            && self.resync_completed_count == 0
            && self.resync_failed_count == 0
            && self.resync_canceled_count == 0
            && self.resync_duration_ms == 0
    }

    pub fn has_replication_usage(&self) -> bool {
        self.replica_size > 0
            || self.replicated_size > 0
            || self.resync_started_count > 0
            || self.resync_completed_count > 0
            || self.resync_failed_count > 0
            || self.resync_canceled_count > 0
            || self.resync_duration_ms > 0
            || !self.stats.is_empty()
    }

    pub fn clone_stats(&self) -> Self {
        let mut snapshot = self.clone();
        for stat in snapshot.stats.values_mut() {
            stat.failed = stat.fail_stats.to_metric();
        }
        snapshot
    }

    pub fn mark_node_local_provider_available(&mut self) {
        self.provider_available = true;
        self.cluster_complete = true;
        self.observed_node_count = 1;
        self.expected_node_count = 1;
    }

    pub fn record_resync_status(&mut self, status: ResyncStatusType, duration: Option<Duration>) {
        match status {
            ResyncStatusType::ResyncStarted => self.resync_started_count += 1,
            ResyncStatusType::ResyncCompleted => self.resync_completed_count += 1,
            ResyncStatusType::ResyncFailed => self.resync_failed_count += 1,
            ResyncStatusType::ResyncCanceled => self.resync_canceled_count += 1,
            ResyncStatusType::NoResync | ResyncStatusType::ResyncPending => return,
        }

        if let Some(duration) = duration {
            let duration_ms = duration.as_millis().min(I64_MAX_AS_U128);
            if let Ok(duration_ms) = i64::try_from(duration_ms) {
                self.resync_duration_ms = self.resync_duration_ms.saturating_add(duration_ms);
            }
        }
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BucketStats {
    pub uptime: i64,
    pub replication_stats: BucketReplicationStats,
    pub queue_stats: QueueStats,
    pub proxy_stats: ProxyMetric,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct SRMetricsSummary {
    pub uptime: i64,
    pub queued: InQueueMetric,
    pub active_workers: ActiveWorkerStat,
    pub metrics: HashMap<String, i64>,
    pub proxied: ProxyMetric,
    pub replica_size: i64,
    pub replica_count: i64,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ActiveWorkerStat {
    pub curr: i32,
    pub max: i32,
    pub avg: f64,
    #[serde(skip)]
    samples: VecDeque<WorkerSample>,
}

#[derive(Debug, Clone)]
struct WorkerSample {
    observed_at: Instant,
    workers: i32,
}

impl ActiveWorkerStat {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn get(&self) -> Self {
        self.clone()
    }

    pub fn update(&mut self, curr: i32) {
        let observed_at = Instant::now();
        self.curr = curr;
        self.samples.push_back(WorkerSample {
            observed_at,
            workers: curr,
        });

        while self
            .samples
            .front()
            .is_some_and(|sample| observed_at.duration_since(sample.observed_at) > ROLLING_WINDOW)
        {
            self.samples.pop_front();
        }

        if self.samples.is_empty() {
            self.max = curr;
            self.avg = curr as f64;
            return;
        }

        self.max = self.samples.iter().map(|sample| sample.workers).max().unwrap_or(curr);
        let total = self.samples.iter().map(|sample| sample.workers as i64).sum::<i64>();
        self.avg = total as f64 / self.samples.len() as f64;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn in_queue_metric_observe_updates_rolling_stats() {
        let mut metric = InQueueMetric::default();
        metric.curr.add_current(128, 4);
        metric.observe(Instant::now());

        metric.curr.add_current(128, 2);
        metric.observe(Instant::now());

        assert_eq!(metric.curr.bytes, 256);
        assert_eq!(metric.curr.count, 6);
        assert_eq!(metric.max.bytes, 256);
        assert_eq!(metric.max.count, 6);
        assert_eq!(metric.last_minute.bytes, 192);
        assert_eq!(metric.last_minute.count, 5);
    }

    #[test]
    fn queue_cache_decrement_saturates_at_zero() {
        let mut cache = QueueCache::new();
        cache.inc("bucket", 128);
        cache.dec("bucket", 512);
        cache.dec("bucket", 1);

        let bucket = cache.get_bucket_stats("bucket");
        let site = cache.get_site_stats();
        assert_eq!(bucket.curr.count, 0);
        assert_eq!(bucket.curr.bytes, 0);
        assert_eq!(site.curr.count, 0);
        assert_eq!(site.curr.bytes, 0);
    }

    #[test]
    fn fail_stats_recent_since_tracks_windows() {
        let mut stats = FailStats::default();
        stats.add_size(64, None::<&()>);
        stats.add_size(32, None::<&()>);

        let last_minute = stats.recent_since(Duration::from_secs(60));
        let last_hour = stats.recent_since(Duration::from_secs(60 * 60));
        assert_eq!(last_minute.count, 2);
        assert_eq!(last_minute.size, 96);
        assert_eq!(last_hour.count, 2);
        assert_eq!(last_hour.size, 96);
    }

    #[test]
    fn fail_stats_recent_windows_matches_two_separate_scans() {
        let mut stats = FailStats::default();
        stats.add_size(64, None::<&()>);
        stats.add_size(32, None::<&()>);

        let (minute, hour) = stats.recent_windows(Duration::from_secs(60), Duration::from_secs(60 * 60));
        let expected_minute = stats.recent_since(Duration::from_secs(60));
        let expected_hour = stats.recent_since(Duration::from_secs(60 * 60));

        assert_eq!((minute.count, minute.size), (expected_minute.count, expected_minute.size));
        assert_eq!((hour.count, hour.size), (expected_hour.count, expected_hour.size));
        assert_eq!(minute.count, 2);
        assert_eq!(hour.size, 96);

        let empty = FailStats::default();
        let (minute, hour) = empty.recent_windows(Duration::from_secs(60), Duration::from_secs(60 * 60));
        assert_eq!((minute.count, minute.size, hour.count, hour.size), (0, 0, 0, 0));
    }

    #[test]
    fn fail_stats_saturate_instead_of_wrapping() {
        let mut stats = FailStats {
            count: i64::MAX,
            size: i64::MAX,
            ..Default::default()
        };

        stats.add_size(1, None::<&()>);
        let merged = stats.merge(&FailStats {
            count: 1,
            size: 1,
            ..Default::default()
        });

        assert_eq!(stats.count, i64::MAX);
        assert_eq!(stats.size, i64::MAX);
        assert_eq!(merged.count, i64::MAX);
        assert_eq!(merged.size, i64::MAX);
    }

    #[test]
    fn active_worker_stat_update_tracks_rolling_avg_and_max() {
        let mut stats = ActiveWorkerStat::default();
        stats.update(2);
        stats.update(6);
        stats.update(4);

        assert_eq!(stats.curr, 4);
        assert_eq!(stats.max, 6);
        assert_eq!(stats.avg, 4.0);
    }
}