feat(observability): diagnose node-local S3 write failures (#7407)

crates/utils/src/logging.rs

@@ -13,6 +13,47 @@
 // limitations under the License.
 
 use std::fmt;
+use std::sync::OnceLock;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Bounds a repetitive diagnostic without changing its underlying counters.
+/// Each emitted event includes the number suppressed since the previous one.
+pub struct LogThrottle {
+    interval_ms: u64,
+    last_ms: AtomicU64,
+    suppressed: AtomicU64,
+}
+
+impl LogThrottle {
+    pub const fn new(interval_ms: u64) -> Self {
+        Self {
+            interval_ms,
+            last_ms: AtomicU64::new(u64::MAX),
+            suppressed: AtomicU64::new(0),
+        }
+    }
+
+    pub fn claim(&self) -> Option<u64> {
+        static ANCHOR: OnceLock<std::time::Instant> = OnceLock::new();
+        let now = ANCHOR.get_or_init(std::time::Instant::now).elapsed().as_millis();
+        self.claim_at(u64::try_from(now).unwrap_or(u64::MAX - 1))
+    }
+
+    fn claim_at(&self, now: u64) -> Option<u64> {
+        let last = self.last_ms.load(Ordering::Relaxed);
+        if (last == u64::MAX || now.saturating_sub(last) >= self.interval_ms)
+            && self
+                .last_ms
+                .compare_exchange(last, now, Ordering::Relaxed, Ordering::Relaxed)
+                .is_ok()
+        {
+            Some(self.suppressed.swap(0, Ordering::Relaxed))
+        } else {
+            self.suppressed.fetch_add(1, Ordering::Relaxed);
+            None
+        }
+    }
+}
 
 #[derive(Clone, Copy)]
 pub struct MaskedAccessKey<'a>(pub &'a str);
@@ -51,7 +92,32 @@ impl fmt::Debug for MaskedAccessKey<'_> {
 
 #[cfg(test)]
 mod tests {
-    use super::MaskedAccessKey;
+    use super::{LogThrottle, MaskedAccessKey};
+
+    #[test]
+    fn log_throttle_emits_once_per_interval_and_reports_suppression() {
+        let throttle = LogThrottle::new(5_000);
+        assert_eq!(throttle.claim_at(0), Some(0));
+        assert_eq!(throttle.claim_at(1), None);
+        assert_eq!(throttle.claim_at(4_999), None);
+        assert_eq!(throttle.claim_at(5_000), Some(2));
+        assert_eq!(throttle.claim_at(5_001), None);
+    }
+
+    #[test]
+    fn log_throttle_allows_only_one_concurrent_claim() {
+        let throttle = LogThrottle::new(5_000);
+        let reported = std::thread::scope(|scope| {
+            let threads: Vec<_> = (0..16).map(|_| scope.spawn(|| throttle.claim_at(0))).collect();
+            let emitted: Vec<_> = threads
+                .into_iter()
+                .filter_map(|thread| thread.join().expect("claim worker"))
+                .collect();
+            assert_eq!(emitted.len(), 1);
+            emitted[0]
+        });
+        assert_eq!(reported + throttle.claim_at(5_000).expect("next window"), 15);
+    }
 
     #[test]
     fn masks_short_values() {