feat(tier): report cluster tier stats and count tier requests (#7110)

`GET /v3/tier-stats` answered from whichever process received the
request, returning that node's rolling 24-hour transition counters as
if they were cluster totals, and the `TierRequestsSuccess` and
`TierRequestsFailure` metric names had no producer at all.

The body now separates the two quantities a tier carries. Stored
inventory comes from the persisted scanner usage snapshot, which is
already cluster-wide; rolling activity is summed over every member
through a new read-only `TierDailyStats` peer RPC. Rings are merged
rather than added, so an idle node's expired hours age out, and each
node counts only its own committed transitions, so a retry is counted
once. Coverage travels with the numbers: `activity.status` names the
reporting members and the ones that could not be asked, timed out, or
answered with a ring this build refuses to merge, and per-tier
inventory is absent rather than zero when the snapshot has no
accounting. The version 1 body stays reachable at `?format=legacy`.

Tier request counters are recorded at the two seams every remote
request passes through, so a new provider is counted by construction,
with a closed operation/outcome label set that can never grow a tier
name, endpoint or object key.

Closes rustfs/backlog#2207

Co-authored-by: cxymds <cxymds@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -118,7 +118,7 @@ pub mod bucket {
         }
 
         pub mod tier_last_day_stats {
-            pub use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats};
+            pub use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats, TierDailyStatsWire};
         }
 
         pub mod tier_sweeper {
@@ -472,7 +472,7 @@ pub mod notification {
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::rotate_cross_pool_fence_fleet_proof_for_test;
     pub use crate::services::notification_sys::{
-        CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
+        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
         acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
         new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
     };

crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs

@@ -19,6 +19,7 @@
 #![allow(clippy::all)]
 
 use rustfs_data_usage::TierStats;
+use serde::{Deserialize, Serialize};
 use sha2::Sha256;
 use std::collections::HashMap;
 use std::ops::Sub;
@@ -27,7 +28,24 @@ use tracing::{error, warn};
 
 pub type DailyAllTierStats = HashMap<String, LastDayTierStats>;
 
-#[derive(Clone)]
+/// One bin per hour of the rolling day. The bin index is the UTC hour, so the
+/// array is a ring the writer ages forward rather than a queue.
+pub const TIER_DAILY_STATS_BINS: usize = 24;
+
+/// Interchange form of [`LastDayTierStats`] for the internode tier-stats RPC.
+///
+/// The in-memory type keeps its bins private because the ring is only
+/// meaningful together with `updated_at`; this type carries both across the
+/// wire and is validated back into the ring by [`LastDayTierStats::from_wire`].
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TierDailyStatsWire {
+    pub bins: Vec<TierStats>,
+    /// Seconds since the Unix epoch. Bins are hour-resolution, so a coarser
+    /// timestamp than the in-memory `OffsetDateTime` loses nothing.
+    pub updated_at_unix_secs: i64,
+}
+
+#[derive(Clone, Debug)]
 pub struct LastDayTierStats {
     bins: [TierStats; 24],
     updated_at: OffsetDateTime,
@@ -80,11 +98,57 @@ impl LastDayTierStats {
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
-    )]
-    fn merge(&self, m: LastDayTierStats) -> LastDayTierStats {
+    /// The rolling ring as observed, without aging it forward.
+    ///
+    /// Only meaningful together with [`LastDayTierStats::updated_at`]: a bin
+    /// belongs to the hour of its index within the day that ends at
+    /// `updated_at`.
+    pub fn bins(&self) -> &[TierStats; TIER_DAILY_STATS_BINS] {
+        &self.bins
+    }
+
+    pub fn updated_at(&self) -> OffsetDateTime {
+        self.updated_at
+    }
+
+    pub fn to_wire(&self) -> TierDailyStatsWire {
+        TierDailyStatsWire {
+            bins: self.bins.to_vec(),
+            updated_at_unix_secs: self.updated_at.unix_timestamp(),
+        }
+    }
+
+    /// Rebuild the ring from a peer's response.
+    ///
+    /// A ring of the wrong width or an unrepresentable timestamp is corrupt
+    /// peer input, not a zero sample: it returns an error so the caller can
+    /// report the node as non-reporting instead of merging a plausible but
+    /// wrong day into a cluster total.
+    pub fn from_wire(wire: TierDailyStatsWire) -> Result<Self, std::io::Error> {
+        let bins: [TierStats; TIER_DAILY_STATS_BINS] = wire.bins.try_into().map_err(|bins: Vec<TierStats>| {
+            std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!("tier daily stats must carry {TIER_DAILY_STATS_BINS} bins, got {}", bins.len()),
+            )
+        })?;
+        let updated_at = OffsetDateTime::from_unix_timestamp(wire.updated_at_unix_secs).map_err(|err| {
+            std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!("tier daily stats carry an unrepresentable timestamp: {err}"),
+            )
+        })?;
+
+        Ok(Self { bins, updated_at })
+    }
+
+    /// Combine two independently observed rings.
+    ///
+    /// Each node counts only the transitions it completed itself, so summing
+    /// bins across nodes is a cluster total rather than a double count. The
+    /// older ring is aged forward to the newer one's clock first, so a node
+    /// that stopped transitioning hours ago contributes its still-current
+    /// bins and not its expired ones.
+    pub fn merge(&self, m: LastDayTierStats) -> LastDayTierStats {
         let mut cl = self.clone();
         let mut cm = m;
         let mut merged = LastDayTierStats::default();
@@ -108,6 +172,7 @@ impl LastDayTierStats {
 #[cfg(test)]
 mod test {
     use super::*;
+    use time::Duration;
 
     #[test]
     fn total_sums_all_recorded_stats() {
@@ -132,4 +197,83 @@ mod test {
             }
         );
     }
+
+    fn sample(total_size: u64) -> TierStats {
+        TierStats {
+            total_size,
+            num_versions: 1,
+            num_objects: 1,
+        }
+    }
+
+    #[test]
+    fn wire_round_trip_preserves_the_ring_and_its_clock() {
+        let mut stats = LastDayTierStats::default();
+        stats.add_stats(sample(10));
+
+        let restored = LastDayTierStats::from_wire(stats.to_wire()).expect("a ring this node produced must decode");
+
+        assert_eq!(restored.bins(), stats.bins(), "every bin must survive the wire");
+        assert_eq!(
+            restored.updated_at().unix_timestamp(),
+            stats.updated_at().unix_timestamp(),
+            "the ring's clock must survive the wire"
+        );
+    }
+
+    #[test]
+    fn a_ring_of_the_wrong_width_is_rejected() {
+        let mut wire = LastDayTierStats::default().to_wire();
+        wire.bins.pop();
+
+        let err = LastDayTierStats::from_wire(wire).expect_err("a short ring must not decode as a zero day");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
+    #[test]
+    fn an_unrepresentable_clock_is_rejected() {
+        let mut wire = LastDayTierStats::default().to_wire();
+        wire.updated_at_unix_secs = i64::MIN;
+
+        let err = LastDayTierStats::from_wire(wire).expect_err("an unrepresentable clock must not decode");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
+    #[test]
+    fn merge_sums_two_nodes_that_transitioned_in_the_same_hour() {
+        let mut left = LastDayTierStats::default();
+        left.add_stats(sample(10));
+        let mut right = LastDayTierStats::default();
+        right.add_stats(sample(20));
+
+        assert_eq!(
+            left.merge(right).total(),
+            TierStats {
+                total_size: 30,
+                num_versions: 2,
+                num_objects: 2,
+            },
+            "each node counts only its own completions, so a merge is a cluster total"
+        );
+    }
+
+    #[test]
+    fn merge_ages_out_a_peer_ring_older_than_a_day() {
+        let mut stale = LastDayTierStats::default();
+        stale.add_stats(sample(10));
+        stale.updated_at -= Duration::days(2);
+
+        let mut fresh = LastDayTierStats::default();
+        fresh.add_stats(sample(20));
+
+        assert_eq!(
+            fresh.merge(stale).total(),
+            TierStats {
+                total_size: 20,
+                num_versions: 1,
+                num_objects: 1,
+            },
+            "a node that stopped transitioning more than a day ago must not keep contributing"
+        );
+    }
 }

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -23,6 +23,7 @@ use crate::storage_api_contracts::internode::{
     SCANNER_ACTIVITY_V6_PROTOCOL_VERSION,
 };
 use crate::{
+    bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats, TierDailyStatsWire},
     bucket::replication::BucketStats,
     disk::disk_store::{get_drive_active_check_interval, get_drive_active_check_timeout},
     layout::endpoints::EndpointServerPools,
@@ -48,7 +49,7 @@ use rustfs_protos::proto_gen::node_service::{
     LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ReplacementRecoveryStatusRequest,
     ScannerActivityRequest, ScannerActivityResponse, ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest,
     ScannerPublicationLeaseResponse, ServerInfoRequest, SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest,
-    StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest,
+    StartProfilingRequest, StopRebalanceRequest, TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest,
     TierMutationControlResponse, TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest,
     node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient,
 };
@@ -986,6 +987,40 @@ impl PeerRestClient {
         Ok(cpus)
     }
 
+    /// This peer's own rolling-day transition counters, per remote tier.
+    ///
+    /// The response is untrusted peer input: a ring of the wrong width or an
+    /// unrepresentable clock is rejected here rather than merged, so a corrupt
+    /// answer makes the node non-reporting instead of silently shifting a
+    /// cluster total.
+    pub async fn tier_daily_stats(&self) -> Result<DailyAllTierStats> {
+        self.finalize_result(self.tier_daily_stats_inner().await).await
+    }
+
+    async fn tier_daily_stats_inner(&self) -> Result<DailyAllTierStats> {
+        let mut client = self.get_client().await?;
+        let request = Request::new(TierDailyStatsRequest {});
+
+        let response = client.tier_daily_stats(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(peer_failure_without_details("tier_daily_stats", None));
+        }
+
+        let mut buf = Deserializer::new(Cursor::new(response.tier_daily_stats));
+        let wire: HashMap<String, TierDailyStatsWire> = Deserialize::deserialize(&mut buf)?;
+
+        wire.into_iter()
+            .map(|(tier, stats)| {
+                LastDayTierStats::from_wire(stats)
+                    .map(|stats| (tier, stats))
+                    .map_err(Error::from)
+            })
+            .collect()
+    }
+
     pub async fn get_net_info(&self) -> Result<NetInfo> {
         self.finalize_result(self.get_net_info_inner().await).await
     }

crates/ecstore/src/services/notification_sys.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::bucket::lifecycle::tier_last_day_stats::DailyAllTierStats;
 use crate::cluster::rpc::{PeerRestClient, ScannerPeerActivity, ScannerPublicationLease, TierConfigReloadOutcome};
 use crate::diagnostics::admin_server_info::get_commit_id;
 use crate::disk::DiskAPI;
@@ -50,6 +51,7 @@ const LOG_SUBSYSTEM_NOTIFICATION: &str = "notification";
 const EVENT_NOTIFICATION_PEER_PROPAGATION: &str = "notification_peer_propagation";
 const EVENT_NOTIFICATION_CAPABILITY_PROBE: &str = "notification_capability_probe";
 const SCANNER_ACTIVITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
+const TIER_DAILY_STATS_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const TIER_CONFIG_RELOAD_RETRY_BASE: Duration = Duration::from_millis(100);
 const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROBE_INTERVAL: Duration = Duration::from_secs(10);
@@ -973,6 +975,120 @@ impl NotificationSys {
     }
 }
 
+/// Rolling tier activity summed over every cluster member that answered, with
+/// the reporting coverage behind the sum.
+///
+/// The coverage is part of the result rather than a log line: a sum over a
+/// subset of the cluster is not a cluster total, and a caller that renders it
+/// as one is the defect this type exists to prevent.
+pub struct ClusterTierDailyStats {
+    pub stats: DailyAllTierStats,
+    /// Members whose rolling day is included, always at least this node.
+    pub nodes_reporting: usize,
+    /// Members this deployment expects to hear from, including this node.
+    pub nodes_expected: usize,
+    /// Members that could not be asked, could not answer, or answered with a
+    /// ring this build refuses to merge. Sorted, and named by grid host.
+    pub unavailable_nodes: Vec<String>,
+}
+
+impl ClusterTierDailyStats {
+    pub fn is_complete(&self) -> bool {
+        self.unavailable_nodes.is_empty() && self.nodes_reporting == self.nodes_expected
+    }
+}
+
+/// Fold one member's rolling day into the running cluster total.
+///
+/// Merging (rather than adding totals) ages each member's ring to the newer
+/// clock first, so a member that stopped transitioning yesterday contributes
+/// only the hours still inside the rolling day.
+fn merge_tier_daily_stats(into: &mut DailyAllTierStats, from: DailyAllTierStats) {
+    for (tier, stats) in from {
+        match into.remove(&tier) {
+            Some(existing) => {
+                into.insert(tier, existing.merge(stats));
+            }
+            None => {
+                into.insert(tier, stats);
+            }
+        }
+    }
+}
+
+impl NotificationSys {
+    /// Sum this node's rolling tier activity with every reachable peer's.
+    ///
+    /// Each node records only the transitions it completed itself, so the sum
+    /// is a cluster total and a retried transition is counted once, by the
+    /// node that finally committed it. Peers are probed concurrently under a
+    /// per-peer deadline so one black-holed member cannot hold the admin
+    /// request open.
+    pub async fn tier_daily_stats(&self, local: DailyAllTierStats) -> ClusterTierDailyStats {
+        let mut stats = local;
+        let nodes_expected = self.peer_clients.len() + 1;
+        let mut nodes_reporting = 1;
+        let mut unavailable_nodes = Vec::new();
+
+        let mut probes = Vec::with_capacity(self.peer_clients.len());
+        for (idx, client) in self.peer_clients.iter().enumerate() {
+            let host = self.tier_daily_stats_peer_host(idx, client.as_ref());
+            probes.push(async move {
+                let Some(client) = client.as_ref() else {
+                    return (host, Err(Error::other("peer is not reachable")));
+                };
+                // The peer is already named by the caller's `peer` log field
+                // and by `unavailable_nodes`, so the deadline is reported as
+                // the typed variant rather than a formatted fragment.
+                let result = timeout(TIER_DAILY_STATS_PROBE_TIMEOUT, client.tier_daily_stats())
+                    .await
+                    .unwrap_or(Err(Error::Timeout));
+                (host, result)
+            });
+        }
+
+        for (host, result) in join_all(probes).await {
+            match result {
+                Ok(peer_stats) => {
+                    nodes_reporting += 1;
+                    merge_tier_daily_stats(&mut stats, peer_stats);
+                }
+                Err(err) => {
+                    warn!(
+                        event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                        peer = host,
+                        error = %err,
+                        "tier daily stats peer did not report"
+                    );
+                    unavailable_nodes.push(host);
+                }
+            }
+        }
+
+        unavailable_nodes.sort();
+        ClusterTierDailyStats {
+            stats,
+            nodes_reporting,
+            nodes_expected,
+            unavailable_nodes,
+        }
+    }
+
+    /// Name a peer slot even when no client was ever built for it, so an
+    /// unreachable member is reported by host instead of disappearing.
+    fn tier_daily_stats_peer_host(&self, idx: usize, client: Option<&PeerRestClient>) -> String {
+        if let Some(client) = client {
+            return client.grid_host.clone();
+        }
+        self.peer_topology_hosts
+            .get(idx)
+            .cloned()
+            .unwrap_or_else(|| format!("peer[{idx}]"))
+    }
+}
+
 pub struct NotificationPeerErr {
     pub host: String,
     pub err: Option<Error>,
@@ -2935,6 +3051,65 @@ fn aggregate_scanner_dirty_usage_acknowledgement_results(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::lifecycle::tier_last_day_stats::LastDayTierStats;
+    use rustfs_data_usage::TierStats;
+
+    fn ring(total_size: u64) -> LastDayTierStats {
+        let mut stats = LastDayTierStats::default();
+        stats.add_stats(TierStats {
+            total_size,
+            num_versions: 1,
+            num_objects: 1,
+        });
+        stats
+    }
+
+    #[test]
+    fn merging_peer_rings_sums_each_tier_without_double_counting() {
+        let mut cluster = DailyAllTierStats::from([("WARM".to_string(), ring(10))]);
+
+        merge_tier_daily_stats(
+            &mut cluster,
+            DailyAllTierStats::from([("WARM".to_string(), ring(20)), ("COLD".to_string(), ring(5))]),
+        );
+
+        assert_eq!(
+            cluster.get("WARM").expect("the shared tier must survive the merge").total(),
+            TierStats {
+                total_size: 30,
+                num_versions: 2,
+                num_objects: 2,
+            },
+            "both nodes' completions belong in the cluster total"
+        );
+        assert_eq!(
+            cluster.get("COLD").expect("a tier only one node saw must be kept").total(),
+            TierStats {
+                total_size: 5,
+                num_versions: 1,
+                num_objects: 1,
+            }
+        );
+    }
+
+    #[test]
+    fn a_single_member_result_is_complete_and_a_missing_peer_is_not() {
+        let complete = ClusterTierDailyStats {
+            stats: DailyAllTierStats::new(),
+            nodes_reporting: 1,
+            nodes_expected: 1,
+            unavailable_nodes: Vec::new(),
+        };
+        assert!(complete.is_complete(), "a single-member deployment reports its whole cluster");
+
+        let partial = ClusterTierDailyStats {
+            stats: DailyAllTierStats::new(),
+            nodes_reporting: 1,
+            nodes_expected: 2,
+            unavailable_nodes: vec!["node-b:9000".to_string()],
+        };
+        assert!(!partial.is_complete(), "a silent member must make the sum partial");
+    }
 
     #[test]
     fn cross_pool_policy_versions_authorize_only_their_supported_protocols() {

crates/ecstore/src/services/tier/tier_config.rs

@@ -38,7 +38,7 @@ const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[
     ("eu-south-1", "https://s3.it-1.wasabisys.com"),
 ];
 
-#[derive(Serialize, Deserialize, Default, Debug, Clone)]
+#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq)]
 pub enum TierType {
     #[default]
     Unsupported,

crates/ecstore/src/services/tier/warm_backend.rs

@@ -43,6 +43,7 @@ use rustfs_s3_client::{
     api_put_object::{AdvancedPutOptions, PutObjectOptions},
     transition_api::{ReadCloser, ReaderImpl},
 };
+use rustfs_scanner_metrics::metrics::{TierRequestOperation, TierRequestOutcome, global_metrics};
 use rustfs_utils::egress::validate_outbound_url;
 use rustfs_utils::http::headers::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _,
@@ -346,6 +347,110 @@ pub(crate) fn optimal_part_size(object_size: i64, min_part_size: i64) -> Result<
     Ok(part_size)
 }
 
+/// Counts every remote-tier request exactly once, whatever backend performs it.
+///
+/// The counters are the only production update path for the `tier` request
+/// metrics, and they must not grow with tier names, endpoints or object keys,
+/// so the wrapper deliberately records nothing but the fixed operation and
+/// outcome labels. Wrapping here rather than inside each provider keeps a new
+/// backend counted by construction, and keeps one call per request: the
+/// overridden `remove_exact` delegates to the inner backend, so the inner
+/// default's own `remove` cannot count the same request a second time.
+struct MeteredWarmBackend {
+    inner: WarmBackendImpl,
+}
+
+impl MeteredWarmBackend {
+    fn record<T>(operation: TierRequestOperation, result: Result<T, std::io::Error>) -> Result<T, std::io::Error> {
+        let outcome = match &result {
+            Ok(_) => TierRequestOutcome::Success,
+            Err(err) => TierRequestOutcome::from_error(err),
+        };
+        global_metrics().record_tier_request(operation, outcome);
+        result
+    }
+}
+
+#[async_trait::async_trait]
+impl WarmBackend for MeteredWarmBackend {
+    /// Delegated without a counter: only one backend issues a remote request
+    /// here, and every other one takes the trait default, so a `validate`
+    /// counter would mostly record requests that never happened.
+    async fn validate(&self) -> Result<(), std::io::Error> {
+        self.inner.validate().await
+    }
+
+    /// A local check of a value this process already holds; it issues no
+    /// remote request, so it is delegated without a counter.
+    fn validate_remote_version_id(&self, remote_version_id: &str) -> Result<(), std::io::Error> {
+        self.inner.validate_remote_version_id(remote_version_id)
+    }
+
+    async fn put(&self, object: &str, r: ReaderImpl, length: i64) -> Result<String, std::io::Error> {
+        Self::record(TierRequestOperation::Put, self.inner.put(object, r, length).await)
+    }
+
+    async fn put_with_meta(
+        &self,
+        object: &str,
+        r: ReaderImpl,
+        length: i64,
+        meta: HashMap<String, String>,
+    ) -> Result<String, std::io::Error> {
+        Self::record(TierRequestOperation::Put, self.inner.put_with_meta(object, r, length, meta).await)
+    }
+
+    async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
+        Self::record(TierRequestOperation::Get, self.inner.get(object, rv, opts).await)
+    }
+
+    async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error> {
+        Self::record(TierRequestOperation::Remove, self.inner.remove(object, rv).await)
+    }
+
+    async fn remove_exact(&self, object: &str, rv: &str) -> Result<(), std::io::Error> {
+        Self::record(TierRequestOperation::Remove, self.inner.remove_exact(object, rv).await)
+    }
+
+    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
+        let result = self.inner.probe_transition_candidate(object).await;
+        // `Unsupported` is the trait default: the backend issued no request,
+        // so counting it would inflate the probe counter on every backend that
+        // does not implement probing.
+        if matches!(result, Ok(TransitionCandidateProbe::Unsupported)) {
+            return result;
+        }
+        Self::record(TierRequestOperation::Probe, result)
+    }
+
+    async fn in_use(&self) -> Result<bool, std::io::Error> {
+        Self::record(TierRequestOperation::InUse, self.inner.in_use().await)
+    }
+}
+
+/// The reconciler is a second remote probe path, reached from recovery rather
+/// than from the `WarmBackend` handle, so it needs its own counter: a `probe`
+/// counter that saw only one of the two paths would read as a complete count
+/// while missing every recovery probe.
+struct MeteredTransitionCandidateReconciler {
+    inner: Box<dyn TransitionCandidateReconciler + Send + Sync + 'static>,
+}
+
+#[async_trait::async_trait]
+impl TransitionCandidateReconciler for MeteredTransitionCandidateReconciler {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: TransitionCandidateIdentity,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        let result = self.inner.probe_transition_candidate_for(object, identity).await;
+        if matches!(result, Ok(TransitionCandidateProbe::Unsupported)) {
+            return result;
+        }
+        MeteredWarmBackend::record(TierRequestOperation::Probe, result)
+    }
+}
+
 pub async fn check_warm_backend(w: Option<&WarmBackendImpl>) -> Result<(), AdminError> {
     let w = w.ok_or_else(|| ERR_TIER_NOT_FOUND.clone())?;
     w.validate().await.map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?;
@@ -593,6 +698,8 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
         status_code: StatusCode::BAD_REQUEST,
     })?;
 
+    let d: WarmBackendImpl = Box::new(MeteredWarmBackend { inner: d });
+
     if probe {
         d.validate().await.map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?;
     }
@@ -641,7 +748,7 @@ pub(crate) async fn new_transition_candidate_reconciler(
         ),
         _ => return Ok(None),
     };
-    Ok(Some(reconciler))
+    Ok(Some(Box::new(MeteredTransitionCandidateReconciler { inner: reconciler })))
 }
 
 #[cfg(test)]
@@ -655,6 +762,156 @@ mod tests {
 
     const PROBE_VERSION: &str = "remote-v2";
 
+    struct CountingBackend {
+        put_result: fn() -> Result<String, std::io::Error>,
+        removes: Arc<AtomicUsize>,
+    }
+
+    #[async_trait::async_trait]
+    impl WarmBackend for CountingBackend {
+        async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
+            (self.put_result)()
+        }
+
+        async fn put_with_meta(
+            &self,
+            object: &str,
+            r: ReaderImpl,
+            length: i64,
+            _meta: HashMap<String, String>,
+        ) -> Result<String, std::io::Error> {
+            self.put(object, r, length).await
+        }
+
+        async fn get(&self, _object: &str, _rv: &str, _opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
+            Err(std::io::Error::other("unused"))
+        }
+
+        async fn remove(&self, _object: &str, _rv: &str) -> Result<(), std::io::Error> {
+            self.removes.fetch_add(1, Ordering::SeqCst);
+            Ok(())
+        }
+
+        async fn in_use(&self) -> Result<bool, std::io::Error> {
+            Ok(false)
+        }
+    }
+
+    fn tier_cell(operation: TierRequestOperation, outcome: TierRequestOutcome) -> u64 {
+        global_metrics()
+            .tier_request_counts()
+            .into_iter()
+            .find(|count| count.operation == operation && count.outcome == outcome)
+            .map(|count| count.count)
+            .expect("every operation/outcome cell must be reported")
+    }
+
+    #[tokio::test]
+    async fn a_metered_put_counts_its_outcome_once() {
+        let before_success = tier_cell(TierRequestOperation::Put, TierRequestOutcome::Success);
+        let before_failure = tier_cell(TierRequestOperation::Put, TierRequestOutcome::BackendError);
+        let backend = MeteredWarmBackend {
+            inner: Box::new(CountingBackend {
+                put_result: || Ok("remote-v1".to_string()),
+                removes: Arc::new(AtomicUsize::new(0)),
+            }),
+        };
+
+        backend
+            .put("object", ReaderImpl::Body(Bytes::from_static(b"payload")), 7)
+            .await
+            .expect("the fake backend accepts the put");
+
+        assert_eq!(
+            tier_cell(TierRequestOperation::Put, TierRequestOutcome::Success),
+            before_success + 1,
+            "an acknowledged put must be counted once"
+        );
+        assert_eq!(
+            tier_cell(TierRequestOperation::Put, TierRequestOutcome::BackendError),
+            before_failure,
+            "a success must not also increment a failure cell"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_metered_put_failure_is_classified_by_error_kind() {
+        let before_timeout = tier_cell(TierRequestOperation::Put, TierRequestOutcome::Timeout);
+        let backend = MeteredWarmBackend {
+            inner: Box::new(CountingBackend {
+                put_result: || Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "endpoint stalled")),
+                removes: Arc::new(AtomicUsize::new(0)),
+            }),
+        };
+
+        backend
+            .put("object", ReaderImpl::Body(Bytes::from_static(b"payload")), 7)
+            .await
+            .expect_err("the fake backend rejects the put");
+
+        assert_eq!(
+            tier_cell(TierRequestOperation::Put, TierRequestOutcome::Timeout),
+            before_timeout + 1,
+            "a timed-out request must land in the timeout cell"
+        );
+    }
+
+    fn tier_probe_total() -> u64 {
+        global_metrics()
+            .tier_request_counts()
+            .into_iter()
+            .filter(|count| count.operation == TierRequestOperation::Probe)
+            .map(|count| count.count)
+            .sum()
+    }
+
+    #[tokio::test]
+    async fn an_unsupported_probe_is_not_counted_as_a_request() {
+        let before = tier_probe_total();
+        let backend = MeteredWarmBackend {
+            inner: Box::new(CountingBackend {
+                put_result: || Ok(String::new()),
+                removes: Arc::new(AtomicUsize::new(0)),
+            }),
+        };
+
+        let probe = backend
+            .probe_transition_candidate("object")
+            .await
+            .expect("the trait default answers without a remote request");
+
+        assert_eq!(probe, TransitionCandidateProbe::Unsupported);
+        assert_eq!(
+            tier_probe_total(),
+            before,
+            "a backend that issues no probe request must not appear in the probe counters"
+        );
+    }
+
+    #[tokio::test]
+    async fn an_exact_remove_is_counted_once_not_twice() {
+        let before = tier_cell(TierRequestOperation::Remove, TierRequestOutcome::Success);
+        let removes = Arc::new(AtomicUsize::new(0));
+        let backend = MeteredWarmBackend {
+            inner: Box::new(CountingBackend {
+                put_result: || Ok(String::new()),
+                removes: Arc::clone(&removes),
+            }),
+        };
+
+        backend
+            .remove_exact("object", "remote-v1")
+            .await
+            .expect("the fake backend accepts the remove");
+
+        assert_eq!(removes.load(Ordering::SeqCst), 1, "the inner backend performs one request");
+        assert_eq!(
+            tier_cell(TierRequestOperation::Remove, TierRequestOutcome::Success),
+            before + 1,
+            "the default remove_exact must not count its own inner remove a second time"
+        );
+    }
+
     struct RejectingValidationBackend {
         validations: Arc<AtomicUsize>,
         puts: Arc<AtomicUsize>,