feat(observability): track foreground response-body completion windows (#8894)

* feat(observability): track foreground response-body completion windows

* docs: correct foreground ownership spelling

crates/concurrency/src/foreground.rs

@@ -0,0 +1,568 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+//! Bounded foreground completion windows for background resource feedback.
+//! Recording uses fixed atomic storage, without an allocation, mutex or task
+//! per request. A single reader rotates banks and waits for the retired bank's
+//! writers before reading it. Delayed writers delay publication, not the
+//! recorded window boundary. Percentiles are intervals, never exact values.
+
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
+
+const SUB_BUCKETS: usize = 64;
+const FINITE_BUCKETS: usize = 27 * SUB_BUCKETS;
+const BUCKETS: usize = FINITE_BUCKETS + 1;
+const MAX_MICROS: u64 = u32::MAX as u64;
+
+/// Bounded foreground categories; no bucket, object or request identifiers.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ForegroundClass {
+    /// Object body reads.
+    Read,
+    /// Object writes, including multipart and copy operations.
+    Write,
+    /// Metadata and listing operations.
+    Metadata,
+    /// Other operations, including rejection before typed dispatch.
+    Other,
+}
+
+impl ForegroundClass {
+    /// Stable order shared by window summaries.
+    pub const ALL: [Self; 4] = [Self::Read, Self::Write, Self::Metadata, Self::Other];
+
+    /// Fixed metric label.
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Read => "read",
+            Self::Write => "write",
+            Self::Metadata => "metadata",
+            Self::Other => "other",
+        }
+    }
+
+    const fn index(self) -> usize {
+        match self {
+            Self::Read => 0,
+            Self::Write => 1,
+            Self::Metadata => 2,
+            Self::Other => 3,
+        }
+    }
+}
+
+/// Outcome at the observed response-body boundary, not a client ACK.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum CompletionOutcome {
+    /// Successful response body completed.
+    Success,
+    /// Error response, service error, or body-stream failure.
+    Failed,
+    /// The request/body was dropped before completion.
+    Cancelled,
+}
+
+/// Inclusive bounds containing the observed percentile.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct LatencyBounds {
+    /// Inclusive lower bound in nanoseconds.
+    pub lower_nanos: u64,
+    /// Inclusive upper bound; absent for the overflow bucket.
+    pub upper_nanos: Option<u64>,
+}
+
+/// One class in a complete observation window.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ClassCompletions {
+    /// All finishes, including errors and cancellation.
+    pub finished: u64,
+    /// Error responses, service errors and body-stream failures.
+    pub failed: u64,
+    /// Censored observations; these must not be treated as successful latency.
+    pub cancelled: u64,
+    /// P99 of non-cancelled finishes, when at least 100 samples exist.
+    /// This count permits computing the rank; it is not statistical confidence.
+    pub p99: Option<LatencyBounds>,
+}
+
+/// Clock readings bracketing an atomic event-window cut. A scheduler pause
+/// between these readings is visible, rather than disguised as precise timing.
+#[derive(Clone, Copy, Debug)]
+pub struct WindowBoundary {
+    /// Read immediately before the event cut.
+    pub earliest: Instant,
+    /// Read immediately after the event cut.
+    pub latest: Instant,
+}
+
+/// Complete windows retain their actual cut brackets, even if publication waits.
+#[derive(Clone, Copy, Debug)]
+pub struct ForegroundWindow {
+    /// Start of this event window.
+    pub start: WindowBoundary,
+    /// Cut of the retired bank, possibly older than publication time.
+    pub end: WindowBoundary,
+    /// Request starts in this window, including pre-dispatch rejection.
+    pub started: Option<u64>,
+    /// Requests still alive at the cut; absent after accounting overflow.
+    pub in_flight: Option<u64>,
+    /// Per-class completions in ForegroundClass::ALL order.
+    pub classes: [Option<ClassCompletions>; 4],
+    /// False after a counter overflow or an inconsistent event count.
+    pub valid: bool,
+}
+
+struct ClassBank {
+    latencies: [AtomicU64; BUCKETS],
+    failed: AtomicU64,
+    cancelled: AtomicU64,
+}
+
+impl ClassBank {
+    fn new() -> Self {
+        Self {
+            latencies: std::array::from_fn(|_| AtomicU64::new(0)),
+            failed: AtomicU64::new(0),
+            cancelled: AtomicU64::new(0),
+        }
+    }
+}
+
+struct Bank {
+    writers: AtomicUsize,
+    started: AtomicU64,
+    overflowed: AtomicBool,
+    classes: [ClassBank; 4],
+}
+
+impl Bank {
+    fn new() -> Self {
+        Self {
+            writers: AtomicUsize::new(0),
+            started: AtomicU64::new(0),
+            overflowed: AtomicBool::new(false),
+            classes: std::array::from_fn(|_| ClassBank::new()),
+        }
+    }
+
+    fn increment(&self, counter: &AtomicU64) {
+        if counter.fetch_add(1, Ordering::Relaxed) == u64::MAX {
+            self.overflowed.store(true, Ordering::Relaxed);
+        }
+    }
+}
+
+struct Writer<'a>(&'a Bank);
+
+impl Drop for Writer<'_> {
+    fn drop(&mut self) {
+        self.0.writers.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
+/// One process-wide recorder. Its fixed storage is allocated with the owner.
+/// A request holds only a reference, timestamps and bounded enum state.
+pub struct ForegroundRecorder {
+    created_at: Instant,
+    epoch: AtomicU64,
+    reader_taken: AtomicBool,
+    banks: [Bank; 2],
+}
+
+impl Default for ForegroundRecorder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ForegroundRecorder {
+    /// Construct empty accounting for one lifetime of the owning process.
+    pub fn new() -> Self {
+        Self {
+            created_at: Instant::now(),
+            epoch: AtomicU64::new(0),
+            reader_taken: AtomicBool::new(false),
+            banks: std::array::from_fn(|_| Bank::new()),
+        }
+    }
+
+    fn writer(&self) -> Writer<'_> {
+        loop {
+            let epoch = self.epoch.load(Ordering::SeqCst);
+            let bank = &self.banks[usize::from(epoch & 1 != 0)];
+            bank.writers.fetch_add(1, Ordering::SeqCst);
+            let writer = Writer(bank);
+            // Sequential ordering joins registration, the reader's cut, and
+            // its zero-writer check. A late old-epoch registrant cannot write
+            // into a bank that the reader has begun draining or reusing.
+            if self.epoch.load(Ordering::SeqCst) == epoch {
+                return writer;
+            }
+        }
+    }
+
+    /// Begin at the external service-call boundary. Drop records cancellation
+    /// unless finish has already recorded a response-body outcome.
+    pub fn start(&self) -> ForegroundObservation<'_> {
+        let started_at = Instant::now();
+        let writer = self.writer();
+        writer.0.increment(&writer.0.started);
+        ForegroundObservation {
+            recorder: self,
+            started_at: Some(started_at),
+            class: ForegroundClass::Other,
+        }
+    }
+
+    fn finish(&self, class: ForegroundClass, elapsed: Duration, outcome: CompletionOutcome) {
+        let writer = self.writer();
+        let bank = writer.0;
+        let class = &bank.classes[class.index()];
+        if outcome == CompletionOutcome::Cancelled {
+            bank.increment(&class.cancelled);
+            return;
+        }
+        let bucket = latency_bucket(elapsed);
+        bank.increment(&class.latencies[bucket]);
+        if outcome == CompletionOutcome::Failed {
+            bank.increment(&class.failed);
+        }
+    }
+
+    /// Claim the sole reader. It cannot be replaced after being dropped:
+    /// resetting its carried in-flight count would falsify later windows.
+    pub fn take_reader(&self) -> Option<ForegroundReader<'_>> {
+        self.reader_taken
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .ok()?;
+        Some(ForegroundReader {
+            recorder: self,
+            start: WindowBoundary {
+                earliest: self.created_at,
+                latest: self.created_at,
+            },
+            retired: None,
+            in_flight: Some(0),
+        })
+    }
+}
+
+/// Non-cloneable request ownership, moved from the service future to its body.
+pub struct ForegroundObservation<'a> {
+    recorder: &'a ForegroundRecorder,
+    started_at: Option<Instant>,
+    class: ForegroundClass,
+}
+
+impl ForegroundObservation<'_> {
+    /// Set the first dispatched operation's bounded category.
+    pub fn set_class(&mut self, class: ForegroundClass) {
+        self.class = class;
+    }
+
+    /// Finish once. Later calls or Drop do not produce another observation.
+    pub fn finish(&mut self, outcome: CompletionOutcome) {
+        if let Some(started_at) = self.started_at.take() {
+            self.recorder.finish(self.class, started_at.elapsed(), outcome);
+        }
+    }
+}
+
+impl Drop for ForegroundObservation<'_> {
+    fn drop(&mut self) {
+        self.finish(CompletionOutcome::Cancelled);
+    }
+}
+
+/// Single-reader state; never acquired by a request path.
+pub struct ForegroundReader<'a> {
+    recorder: &'a ForegroundRecorder,
+    start: WindowBoundary,
+    retired: Option<(usize, WindowBoundary)>,
+    in_flight: Option<u64>,
+}
+
+impl ForegroundReader<'_> {
+    /// Cut a new window, or finish publishing an earlier cut. A live retired
+    /// writer returns None without blocking, discarding data or moving its cut.
+    pub fn sample(&mut self) -> Option<ForegroundWindow> {
+        if self.retired.is_none() {
+            let epoch = self.recorder.epoch.load(Ordering::SeqCst);
+            let next = epoch.checked_add(1)?;
+            let earliest = Instant::now();
+            self.recorder.epoch.store(next, Ordering::SeqCst);
+            let cut = WindowBoundary {
+                earliest,
+                latest: Instant::now(),
+            };
+            self.retired = Some((usize::from(epoch & 1 != 0), cut));
+        }
+        let (index, end) = self.retired?;
+        let bank = &self.recorder.banks[index];
+        if bank.writers.load(Ordering::SeqCst) != 0 {
+            return None;
+        }
+
+        let started = bank.started.swap(0, Ordering::Relaxed);
+        let overflowed = bank.overflowed.swap(false, Ordering::Relaxed);
+        let classes = std::array::from_fn(|index| {
+            let class = &bank.classes[index];
+            let failed = class.failed.swap(0, Ordering::Relaxed);
+            let cancelled = class.cancelled.swap(0, Ordering::Relaxed);
+            let buckets = std::array::from_fn::<_, BUCKETS, _>(|i| class.latencies[i].swap(0, Ordering::Relaxed));
+            let non_cancelled = buckets.iter().try_fold(0_u64, |sum, count| sum.checked_add(*count));
+            let finished = non_cancelled.and_then(|total| total.checked_add(cancelled));
+            let finished = finished.filter(|_| !overflowed && non_cancelled.is_some_and(|total| failed <= total))?;
+            Some(ClassCompletions {
+                finished,
+                failed,
+                cancelled,
+                p99: non_cancelled
+                    .filter(|total| *total >= 100)
+                    .and_then(|total| percentile_99(&buckets, total)),
+            })
+        });
+        let started = (!overflowed).then_some(started);
+        let finished = classes
+            .iter()
+            .try_fold(0_u64, |sum, class| sum.checked_add(class.as_ref()?.finished));
+        self.in_flight = self
+            .in_flight
+            .zip(finished)
+            .and_then(|(active, finished)| active.checked_add(started?)?.checked_sub(finished));
+        let window = ForegroundWindow {
+            start: self.start,
+            end,
+            started,
+            in_flight: self.in_flight,
+            classes,
+            valid: self.in_flight.is_some(),
+        };
+        self.start = end;
+        self.retired = None;
+        Some(window)
+    }
+}
+
+fn latency_bucket(elapsed: Duration) -> usize {
+    let micros = elapsed.as_nanos().div_ceil(1_000);
+    let Ok(micros) = u64::try_from(micros) else {
+        return FINITE_BUCKETS;
+    };
+    if micros > MAX_MICROS {
+        return FINITE_BUCKETS;
+    }
+    if micros < SUB_BUCKETS as u64 {
+        return usize::try_from(micros).unwrap_or(FINITE_BUCKETS);
+    }
+    let shift = 63 - micros.leading_zeros() - 6;
+    usize::try_from((u64::from(shift) + 1) * SUB_BUCKETS as u64 + (micros >> shift) - SUB_BUCKETS as u64)
+        .unwrap_or(FINITE_BUCKETS)
+}
+
+fn bucket_bounds(index: usize) -> LatencyBounds {
+    if index == FINITE_BUCKETS {
+        return LatencyBounds {
+            lower_nanos: MAX_MICROS * 1_000 + 1,
+            upper_nanos: None,
+        };
+    }
+    let (lower, upper) = if index < SUB_BUCKETS {
+        (index as u64, index as u64)
+    } else {
+        let shift = index / SUB_BUCKETS - 1;
+        let lower = (SUB_BUCKETS + index % SUB_BUCKETS) as u64 * (1_u64 << shift);
+        (lower, lower + (1_u64 << shift) - 1)
+    };
+    LatencyBounds {
+        lower_nanos: lower.saturating_sub(1) * 1_000 + u64::from(lower != 0),
+        upper_nanos: Some(upper * 1_000),
+    }
+}
+
+fn percentile_99(buckets: &[u64; BUCKETS], total: u64) -> Option<LatencyBounds> {
+    let rank = (u128::from(total) * 99).div_ceil(100);
+    let mut cumulative = 0_u128;
+    for (index, count) in buckets.iter().enumerate() {
+        cumulative += u128::from(*count);
+        if cumulative >= rank {
+            return Some(bucket_bounds(index));
+        }
+    }
+    None
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+
+    #[test]
+    fn percentile_intervals_contain_each_bucket_edge_and_neighbour() {
+        for index in 0..BUCKETS {
+            let bounds = bucket_bounds(index);
+            assert_eq!(latency_bucket(Duration::from_nanos(bounds.lower_nanos)), index);
+            if let Some(upper) = bounds.upper_nanos {
+                assert!(upper >= bounds.lower_nanos);
+                assert_eq!(latency_bucket(Duration::from_nanos(upper)), index);
+                assert_eq!(latency_bucket(Duration::from_nanos(upper + 1)), index + 1);
+            }
+        }
+        assert_eq!(latency_bucket(Duration::MAX), FINITE_BUCKETS);
+    }
+
+    #[test]
+    fn body_ownership_carries_in_flight_counts_across_windows() {
+        let recorder = ForegroundRecorder::new();
+        let mut reader = recorder.take_reader().expect("first reader");
+        let mut request = recorder.start();
+        request.set_class(ForegroundClass::Read);
+        let first = reader.sample().expect("cut with a live request");
+        assert!(first.valid);
+        assert_eq!(first.started, Some(1));
+        assert_eq!(first.in_flight, Some(1));
+        assert_eq!(first.classes[0].expect("valid read count").finished, 0);
+
+        request.finish(CompletionOutcome::Success);
+        request.finish(CompletionOutcome::Failed);
+        drop(request);
+        let second = reader.sample().expect("body finished");
+        assert!(second.valid);
+        assert_eq!(second.started, Some(0));
+        assert_eq!(second.in_flight, Some(0));
+        let read = second.classes[0].expect("valid read summary");
+        assert_eq!((read.finished, read.failed, read.cancelled, read.p99), (1, 0, 0, None));
+        assert_eq!(first.end.earliest, second.start.earliest);
+        assert_eq!(first.end.latest, second.start.latest);
+    }
+
+    #[test]
+    fn cancellations_are_counted_without_fabricating_success_latencies() {
+        let recorder = ForegroundRecorder::new();
+        let mut reader = recorder.take_reader().expect("first reader");
+        for _ in 0..150 {
+            let mut request = recorder.start();
+            request.set_class(ForegroundClass::Write);
+        }
+        let window = reader.sample().expect("cancelled requests");
+        let write = window.classes[1].expect("known write count");
+        assert!(window.valid);
+        assert_eq!(window.in_flight, Some(0));
+        assert_eq!((write.finished, write.cancelled, write.failed, write.p99), (150, 150, 0, None));
+    }
+
+    #[test]
+    fn delayed_retired_writer_preserves_the_original_cut_and_next_bank() {
+        let recorder = ForegroundRecorder::new();
+        let mut reader = recorder.take_reader().expect("first reader");
+        let writer = recorder.writer();
+        assert!(reader.sample().is_none(), "never read a live writer's bank");
+        let original_cut = reader.retired.expect("pending cut").1;
+        let request = recorder.start(); // a start after the cut belongs to the new bank
+        writer.0.increment(&writer.0.started);
+        drop(writer);
+        let previous = reader.sample().expect("retired writer drained");
+        assert_eq!(previous.end.earliest, original_cut.earliest);
+        assert_eq!(previous.end.latest, original_cut.latest);
+        assert_eq!(previous.started, Some(1));
+        assert_eq!(previous.in_flight, Some(1));
+        drop(request);
+        let current = reader.sample().expect("next bank");
+        assert_eq!(current.started, Some(1));
+        assert_eq!(current.in_flight, Some(1));
+        assert_eq!(current.classes[3].expect("unclassified cancellation").cancelled, 1);
+    }
+
+    #[test]
+    fn completed_percentile_uses_rank_and_never_turns_maximum_into_p99() {
+        let recorder = ForegroundRecorder::new();
+        let mut reader = recorder.take_reader().expect("first reader");
+        for i in 0..100 {
+            let writer = recorder.writer();
+            writer.0.increment(&writer.0.started);
+            drop(writer);
+            recorder.finish(
+                ForegroundClass::Read,
+                if i == 99 {
+                    Duration::from_secs(60)
+                } else {
+                    Duration::from_millis(2)
+                },
+                if i == 99 {
+                    CompletionOutcome::Failed
+                } else {
+                    CompletionOutcome::Success
+                },
+            );
+        }
+        let window = reader.sample().expect("complete window");
+        let read = window.classes[0].expect("read summary");
+        assert_eq!((read.finished, read.failed, read.cancelled), (100, 1, 0));
+        let p99 = read.p99.expect("100 non-cancelled samples");
+        assert!(p99.lower_nanos <= 2_000_000);
+        assert!(p99.upper_nanos.is_some_and(|upper| (2_000_000..3_000_000).contains(&upper)));
+    }
+
+    #[test]
+    fn overflow_stays_unknown_and_reader_cannot_reset_event_accounting() {
+        let recorder = ForegroundRecorder::new();
+        {
+            let mut reader = recorder.take_reader().expect("first reader");
+            recorder.banks[0].started.store(u64::MAX, Ordering::Relaxed);
+            drop(recorder.start());
+            let invalid = reader.sample().expect("overflow window");
+            assert!(!invalid.valid);
+            assert_eq!(invalid.started, None);
+            assert_eq!(invalid.in_flight, None);
+            assert_eq!(invalid.classes, [None; 4]);
+            let later = reader.sample().expect("later empty window");
+            assert!(!later.valid, "a lost carried count cannot recover by inventing zero active work");
+        }
+        assert!(recorder.take_reader().is_none(), "reader replacement would lose live request ownership");
+    }
+
+    #[test]
+    fn concurrent_rotation_preserves_every_start_finish_and_cancellation() {
+        let recorder = Arc::new(ForegroundRecorder::new());
+        let mut reader = recorder.take_reader().expect("first reader");
+        let workers: Vec<_> = (0..4)
+            .map(|_| {
+                let recorder = Arc::clone(&recorder);
+                std::thread::spawn(move || {
+                    for index in 0..10_000 {
+                        let mut request = recorder.start();
+                        request.set_class(ForegroundClass::Read);
+                        if index % 10 != 0 {
+                            request.finish(CompletionOutcome::Success);
+                        }
+                    }
+                })
+            })
+            .collect();
+        let (mut started, mut finished, mut cancelled) = (0, 0, 0);
+        let mut add_window = |window: ForegroundWindow| {
+            assert!(window.valid, "concurrent event cuts must stay balanced");
+            started += window.started.expect("valid start count");
+            let reads = window.classes[0].expect("valid read count");
+            finished += reads.finished;
+            cancelled += reads.cancelled;
+        };
+        while workers.iter().any(|worker| !worker.is_finished()) {
+            if let Some(window) = reader.sample() {
+                add_window(window);
+            }
+            std::thread::yield_now();
+        }
+        for worker in workers {
+            worker.join().expect("writer must finish");
+        }
+        add_window(reader.sample().expect("drain after writers join"));
+        // A previously retired window may have been published by the last
+        // sample; cut once more to include the current, still open bank.
+        let final_window = reader.sample().expect("final open bank");
+        assert_eq!(final_window.in_flight, Some(0));
+        add_window(final_window);
+        assert_eq!((started, finished, cancelled), (40_000, 40_000, 4_000));
+    }
+}

crates/concurrency/src/lib.rs

@@ -39,6 +39,7 @@ pub use rustfs_io_core::OperationProgress;
 pub mod adaptive;
 mod backpressure;
 mod deadlock;
+pub mod foreground;
 mod queue;
 pub mod workers;
 pub mod workload;

crates/io-metrics/src/s3_http_metrics.rs

@@ -252,6 +252,12 @@ pub struct S3HttpRequestGuard {
 }
 
 impl S3HttpRequestGuard {
+    /// First typed external operation seen so far, including after a pending
+    /// service poll. This does not finish or transfer HTTP outcome accounting.
+    pub fn operation(&self) -> Option<S3Operation> {
+        S3Operation::ALL.get(self.operation).copied()
+    }
+
     pub fn is_active() -> bool {
         CURRENT_OPERATION.try_with(|_| ()).is_ok()
     }

docs/operations/scanner-benchmark-runbook.md

@@ -397,6 +397,14 @@ GETs are bounded (`--workers`, default 8, maximum 64); each body is read in 1 Mi
 chunks. This checks S3-visible content and versions, not every on-disk erasure
 fragment, and does not replace the seeded repair or performance gates.
 
+#### In-Process Foreground Windows
+
+When scanner or heal is enabled, a one-second process-local sampler tracks external S3 requests from the outer service call through response-body completion, stream failure or cancellation. It runs independently of the metrics export switch and stops with background services. Requests and their existing body buffers are not copied or moved to another task. The start boundary is `Service::call`; connection establishment and Tower readiness waits before that call are outside this window. A successful body observation means the server produced its final frame, not that the client received or acknowledged it. HEAD and bodyless HTTP statuses finish at response headers. Admin, health, console, STS and internal RPC traffic are excluded by the existing external S3 route classifier.
+
+Windows preserve request starts, in-flight requests and completed, failed and cancelled counts in four fixed classes: `read`, `write`, `metadata` and `other`. Cancellation is a censored observation and contributes no completed-latency sample. P99 is an inclusive nanosecond interval over non-cancelled completions, including errors, and is unavailable below 100 samples; 100 samples permit rank calculation but do not establish statistical confidence. The overflow bucket has no upper bound. The sampler brackets each atomic window cut with two monotonic clock readings. A writer delayed across a cut delays publication without changing those readings. Consumers must check validity, age and cut uncertainty; a missing or low-sample window does not prove that foreground work is idle.
+
+With metrics export enabled, `rustfs_background_foreground_window_*` gauges expose validity, cut uncertainty, duration, starts, in-flight work and per-class counts and percentile bounds. Check the class and percentile availability gauges before reading retained numeric values. Exported floating-point gauges are diagnostic; the in-process counters retain integer precision. These local windows neither align different nodes nor provide the exact client-side percentile required by the SSD ABBA gate. They supply observations, without enabling an adaptive admission policy or completing performance acceptance.
+
 #### Histogram Window Evidence
 
 Use `scripts/scanner_otlp_window.py` to subtract two cumulative explicit-histogram or monotonic integer-counter snapshots before deriving window evidence. It reads OTLP JSON, or binary OTLP with `--format protobuf` when `opentelemetry-proto` is installed in the active Python environment. Pin and record that dependency in the adapter environment. Keep each original export and its source-node association; output includes the SHA-256 of both inputs.

rustfs/src/foreground_observation.rs

@@ -0,0 +1,236 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+//! In-process foreground windows for background protection. The HTTP service
+//! owns each observation through response-body completion, failure or drop.
+//! This is server-side production of the body, not proof of client receipt.
+
+use rustfs_concurrency::foreground::{ForegroundObservation, ForegroundRecorder, ForegroundWindow};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{LazyLock, RwLock};
+use std::time::Duration;
+use tokio_util::sync::CancellationToken;
+
+static RECORDER: LazyLock<ForegroundRecorder> = LazyLock::new(ForegroundRecorder::new);
+static ENABLED: AtomicBool = AtomicBool::new(false);
+static LATEST: RwLock<Option<ForegroundWindow>> = RwLock::new(None);
+
+pub(crate) fn start_request() -> Option<ForegroundObservation<'static>> {
+    ENABLED.load(Ordering::Acquire).then(|| RECORDER.start())
+}
+
+/// Latest complete event window. Consumers must check validity, cut-bracket
+/// uncertainty and age; an old or low-sample window is not evidence of idle.
+pub fn latest_window() -> Option<ForegroundWindow> {
+    *LATEST.read().unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+struct RecorderLifetime;
+
+impl Drop for RecorderLifetime {
+    fn drop(&mut self) {
+        ENABLED.store(false, Ordering::Release);
+        let last = LATEST.write().unwrap_or_else(std::sync::PoisonError::into_inner).take();
+        if let Some(window) = last {
+            // Clear earlier exports even if the export gate has since changed.
+            // The original cut and numeric values are not rewritten as zero.
+            invalidate_window(window);
+        }
+    }
+}
+
+pub(crate) fn start_sampler(cancel: CancellationToken) -> Option<tokio::task::JoinHandle<()>> {
+    if cancel.is_cancelled() {
+        return None;
+    }
+    let mut reader = RECORDER.take_reader()?;
+    let lifetime = RecorderLifetime;
+    ENABLED.store(true, Ordering::Release);
+    Some(tokio::spawn(async move {
+        // Also clear availability if the worker is dropped before its first poll.
+        let _lifetime = lifetime;
+        let mut tick = tokio::time::interval(Duration::from_secs(1));
+        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        loop {
+            tokio::select! {
+                biased;
+                _ = cancel.cancelled() => break,
+                _ = tick.tick() => {
+                    if let Some(window) = reader.sample() {
+                        *LATEST.write().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(window);
+                        if rustfs_io_metrics::metrics_enabled() {
+                            record_window(window);
+                        }
+                    }
+                }
+            }
+        }
+    }))
+}
+
+fn invalidate_window(mut window: ForegroundWindow) {
+    window.valid = false;
+    window.started = None;
+    window.in_flight = None;
+    window.classes = [None; 4];
+    record_window(window);
+}
+
+fn record_window(window: ForegroundWindow) {
+    // Exported f64 gauges are diagnostic. Control consumers use the exact
+    // integer fields in latest_window, never these potentially rounded values.
+    metrics::gauge!("rustfs_background_foreground_window_valid").set(f64::from(window.valid));
+    metrics::gauge!("rustfs_background_foreground_window_cut_uncertainty_seconds")
+        .set(window.end.latest.duration_since(window.end.earliest).as_secs_f64());
+    metrics::gauge!("rustfs_background_foreground_window_duration_seconds")
+        .set(window.end.latest.duration_since(window.start.earliest).as_secs_f64());
+    if let Some(active) = window.in_flight {
+        metrics::gauge!("rustfs_background_foreground_window_in_flight").set(active as f64);
+    }
+    if let Some(started) = window.started {
+        metrics::gauge!("rustfs_background_foreground_window_starts").set(started as f64);
+    }
+    for (class, counts) in rustfs_concurrency::foreground::ForegroundClass::ALL
+        .into_iter()
+        .zip(window.classes)
+    {
+        let label = class.as_str();
+        metrics::gauge!("rustfs_background_foreground_window_class_available", "class" => label).set(f64::from(counts.is_some()));
+        let p99 = counts.and_then(|counts| counts.p99);
+        metrics::gauge!("rustfs_background_foreground_window_p99_available", "class" => label).set(f64::from(p99.is_some()));
+        metrics::gauge!("rustfs_background_foreground_window_p99_upper_available", "class" => label)
+            .set(f64::from(p99.and_then(|bounds| bounds.upper_nanos).is_some()));
+        let Some(counts) = counts else {
+            continue;
+        };
+        metrics::gauge!("rustfs_background_foreground_window_completions", "class" => label).set(counts.finished as f64);
+        metrics::gauge!("rustfs_background_foreground_window_failures", "class" => label).set(counts.failed as f64);
+        metrics::gauge!("rustfs_background_foreground_window_cancellations", "class" => label).set(counts.cancelled as f64);
+        if let Some(bounds) = counts.p99 {
+            metrics::gauge!("rustfs_background_foreground_window_p99_lower_nanoseconds", "class" => label)
+                .set(bounds.lower_nanos as f64);
+            if let Some(upper) = bounds.upper_nanos {
+                metrics::gauge!("rustfs_background_foreground_window_p99_upper_nanoseconds", "class" => label).set(upper as f64);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+    use rustfs_concurrency::foreground::{ClassCompletions, LatencyBounds, WindowBoundary};
+    use std::time::Instant;
+
+    #[test]
+    fn unavailable_windows_clear_availability_without_fabricating_values() {
+        for stage in 0..4 {
+            let recorder = DebuggingRecorder::new();
+            let now = Instant::now();
+            let mut window = ForegroundWindow {
+                start: WindowBoundary {
+                    earliest: now,
+                    latest: now,
+                },
+                end: WindowBoundary {
+                    earliest: now,
+                    latest: now,
+                },
+                started: Some(100),
+                in_flight: Some(0),
+                classes: [Some(ClassCompletions {
+                    finished: 100,
+                    failed: 0,
+                    cancelled: 0,
+                    p99: Some(LatencyBounds {
+                        lower_nanos: 1_001,
+                        upper_nanos: Some(2_000),
+                    }),
+                }); 4],
+                valid: true,
+            };
+            metrics::with_local_recorder(&recorder, || {
+                record_window(window);
+                if stage == 1 {
+                    window.classes[0].as_mut().expect("known read class").p99 = None;
+                    record_window(window);
+                }
+                if stage == 2 {
+                    window.classes[0] = None;
+                    window.in_flight = None;
+                    window.started = None;
+                    window.valid = false;
+                    record_window(window);
+                }
+                if stage == 3 {
+                    invalidate_window(window);
+                }
+            });
+            let values = recorder.snapshotter().snapshot().into_vec();
+            let gauge = |suffix: &str| {
+                let name = format!("rustfs_background_foreground_window_{suffix}");
+                values.iter().find_map(|(key, _, _, value)| {
+                    (key.key().name() == name
+                        && key
+                            .key()
+                            .labels()
+                            .any(|label| label.key() == "class" && label.value() == "read"))
+                    .then(|| match value {
+                        DebugValue::Gauge(value) => value.into_inner(),
+                        _ => panic!("expected gauge {name}"),
+                    })
+                })
+            };
+            assert_eq!(gauge("class_available"), Some(if stage < 2 { 1.0 } else { 0.0 }));
+            assert_eq!(gauge("p99_available"), Some(if stage == 0 { 1.0 } else { 0.0 }));
+            assert_eq!(gauge("p99_upper_available"), Some(if stage == 0 { 1.0 } else { 0.0 }));
+            assert_eq!(gauge("p99_lower_nanoseconds"), Some(1_001.0));
+            assert_eq!(gauge("p99_upper_nanoseconds"), Some(2_000.0));
+            assert!(values.iter().any(|(key, _, _, value)| {
+                key.key().name() == "rustfs_background_foreground_window_valid"
+                    && matches!(value, DebugValue::Gauge(value) if value.into_inner() == if stage < 2 { 1.0 } else { 0.0 })
+            }));
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn sampler_cancellation_disables_recording_without_resetting_accounting() {
+        let recorder = DebuggingRecorder::new();
+        let _metrics = metrics::set_default_local_recorder(&recorder);
+        let cancelled = CancellationToken::new();
+        cancelled.cancel();
+        assert!(start_sampler(cancelled).is_none());
+        assert!(start_request().is_none());
+        let cancel = CancellationToken::new();
+        let worker = start_sampler(cancel.clone()).expect("first live sampler");
+        let request = start_request().expect("recording enabled before first sampler poll");
+        drop(request);
+        tokio::time::sleep(Duration::from_millis(1)).await;
+        let window = latest_window().expect("sampler publishes actual events before shutdown");
+        assert!(window.valid);
+        assert_eq!((window.started, window.in_flight), (Some(1), Some(0)));
+        assert_eq!(window.classes[3].expect("undispatched request").cancelled, 1);
+        // Model a prior export without changing the process-global metrics gate.
+        record_window(window);
+        cancel.cancel();
+        tokio::time::timeout(Duration::from_secs(1), worker)
+            .await
+            .expect("bounded shutdown")
+            .expect("sampler task");
+        assert!(start_request().is_none());
+        assert!(latest_window().is_none());
+        let values = recorder.snapshotter().snapshot().into_vec();
+        assert!(
+            values.iter().any(|(key, _, _, value)| {
+                key.key().name() == "rustfs_background_foreground_window_valid"
+                    && matches!(value, DebugValue::Gauge(value) if value.into_inner() == 0.0)
+            }),
+            "shutdown must invalidate earlier exported windows"
+        );
+        assert!(
+            start_sampler(CancellationToken::new()).is_none(),
+            "never reset carried in-flight accounting"
+        );
+    }
+}

rustfs/src/lib.rs

@@ -86,6 +86,7 @@ pub mod delete_tail_activity;
 pub mod diagnose;
 pub mod embedded;
 pub mod error;
+pub mod foreground_observation;
 pub mod init;
 pub mod inspect;
 pub(crate) mod kms_deletion_gate;

rustfs/src/server/http.rs

@@ -114,7 +114,9 @@ type Http3CoreService = tower::util::BoxCloneService<
 #[cfg(feature = "http3")]
 type Http3ExternalService = tower::util::BoxCloneService<
     HttpRequest<http3::RequestBody>,
-    Response<tower_http::body::UnsyncBoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>>,
+    Response<
+        super::layer::ForegroundResponseBody<tower_http::body::UnsyncBoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>>,
+    >,
     Box<dyn std::error::Error + Send + Sync>,
 >;
 use tracing::{Level, Span, debug, error, info, instrument, trace, warn};
@@ -2123,14 +2125,16 @@ impl<A, B> PathDispatchService<A, B> {
     }
 }
 
-impl<A, B> Service<HttpRequest<Incoming>> for PathDispatchService<A, B>
+impl<A, B, ExternalBody, InternodeBody> Service<HttpRequest<Incoming>> for PathDispatchService<A, B>
 where
-    A: Service<HttpRequest<Incoming>> + Clone + Send + 'static,
+    A: Service<HttpRequest<Incoming>, Response = Response<ExternalBody>> + Clone + Send + 'static,
     A::Future: Send + 'static,
-    B: Service<HttpRequest<Incoming>, Response = A::Response, Error = A::Error> + Clone + Send + 'static,
+    B: Service<HttpRequest<Incoming>, Response = Response<InternodeBody>, Error = A::Error> + Clone + Send + 'static,
     B::Future: Send + 'static,
+    ExternalBody: 'static,
+    InternodeBody: 'static,
 {
-    type Response = A::Response;
+    type Response = Response<http_body_util::Either<ExternalBody, InternodeBody>>;
     type Error = A::Error;
     type Future = Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
 
@@ -2143,9 +2147,17 @@ where
 
     fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
         if Self::is_internode_path(req.uri().path()) {
-            Box::pin(self.internode.call(req))
+            let future = self.internode.call(req);
+            Box::pin(async move {
+                let (parts, body) = future.await?.into_parts();
+                Ok(Response::from_parts(parts, http_body_util::Either::Right(body)))
+            })
         } else {
-            Box::pin(self.external.call(req))
+            let future = self.external.call(req);
+            Box::pin(async move {
+                let (parts, body) = future.await?.into_parts();
+                Ok(Response::from_parts(parts, http_body_util::Either::Left(body)))
+            })
         }
     }
 }
@@ -2854,7 +2866,11 @@ mod tests {
     >;
     type RateLimitTestStack = tower::util::BoxCloneService<
         HttpRequest<s3s::Body>,
-        Response<tower_http::body::UnsyncBoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>>,
+        Response<
+            super::super::layer::ForegroundResponseBody<
+                tower_http::body::UnsyncBoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>,
+            >,
+        >,
         Box<dyn std::error::Error + Send + Sync>,
     >;
 
@@ -5028,6 +5044,66 @@ mod tests {
         let _ = service;
     }
 
+    #[tokio::test]
+    async fn path_dispatch_preserves_distinct_response_bodies() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("route listener");
+        let address = listener.local_addr().expect("route address");
+        let server = tokio::spawn(async move {
+            for _ in 0..2 {
+                let (stream, _) = listener.accept().await.expect("route connection");
+                let external = tower::service_fn(|_: HttpRequest<Incoming>| async {
+                    Ok::<_, Infallible>(
+                        Response::builder()
+                            .header("x-lane", "external")
+                            .body(Full::new(Bytes::from_static(b"external body")))
+                            .expect("external response"),
+                    )
+                });
+                let internode = tower::service_fn(|_: HttpRequest<Incoming>| async {
+                    let frames =
+                        futures::stream::iter([Ok::<_, Infallible>(http_body::Frame::data(Bytes::from_static(b"rpc body")))]);
+                    Ok::<_, Infallible>(
+                        Response::builder()
+                            .status(201)
+                            .header("x-lane", "internode")
+                            .body(http_body_util::StreamBody::new(frames))
+                            .expect("internode response"),
+                    )
+                });
+                hyper::server::conn::http1::Builder::new()
+                    .serve_connection(
+                        TokioIo::new(stream),
+                        TowerToHyperService::new(PathDispatchService::new(external, internode)),
+                    )
+                    .await
+                    .expect("route connection completion");
+            }
+        });
+        let client = reqwest::Client::builder()
+            .no_proxy()
+            .timeout(Duration::from_secs(5))
+            .build()
+            .expect("route client");
+        for (path, status, lane, body) in [
+            ("/bucket/object", 200, "external", "external body"),
+            ("/rustfs/rpc/example", 201, "internode", "rpc body"),
+        ] {
+            let response = client
+                .get(format!("http://{address}{path}"))
+                .header(http::header::CONNECTION, "close")
+                .send()
+                .await
+                .expect("route response");
+            assert_eq!(response.status().as_u16(), status);
+            assert_eq!(response.headers()["x-lane"], lane);
+            assert_eq!(response.bytes().await.expect("route body"), body.as_bytes());
+        }
+        tokio::time::timeout(Duration::from_secs(5), server)
+            .await
+            .expect("bounded route shutdown")
+            .expect("route server");
+    }
+
     // backlog#806-35: in-flight gauge must be adjusted exactly once per request.
 
     #[derive(Clone, Copy)]

rustfs/src/server/layer.rs

@@ -43,6 +43,7 @@ use rustfs_common::GlobalReadiness;
 use rustfs_common::trace_bus::{
     TelemetryTraceEvent, TelemetryTraceOperation, TelemetryTraceStatus, telemetry_trace_emit, telemetry_trace_subscriber_count,
 };
+use rustfs_concurrency::foreground::{CompletionOutcome, ForegroundClass, ForegroundObservation};
 use rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard;
 use rustfs_obs::HTTP_SERVER_LOG_TARGET;
 #[cfg(feature = "swift")]
@@ -54,6 +55,7 @@ use s3s::S3Error;
 use s3s::S3ErrorCode;
 use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
+use std::cell::Cell;
 use std::future::Future;
 use std::net::{IpAddr, SocketAddr};
 use std::pin::Pin;
@@ -213,12 +215,16 @@ fn uses_server_owned_s3_request_id<B>(req: &HttpRequest<B>, console_redirect_ena
 #[derive(Clone, Default)]
 pub struct ExternalRequestContextLayer {
     console_redirect_enabled: bool,
+    #[cfg(test)]
+    feedback_recorder: Option<&'static rustfs_concurrency::foreground::ForegroundRecorder>,
 }
 
 impl ExternalRequestContextLayer {
     pub(crate) fn new(console_redirect_enabled: bool) -> Self {
         Self {
             console_redirect_enabled,
+            #[cfg(test)]
+            feedback_recorder: None,
         }
     }
 }
@@ -230,6 +236,8 @@ impl<S> Layer<S> for ExternalRequestContextLayer {
         ExternalRequestContextService {
             inner,
             console_redirect_enabled: self.console_redirect_enabled,
+            #[cfg(test)]
+            feedback_recorder: self.feedback_recorder,
         }
     }
 }
@@ -238,13 +246,15 @@ impl<S> Layer<S> for ExternalRequestContextLayer {
 pub struct ExternalRequestContextService<S> {
     inner: S,
     console_redirect_enabled: bool,
+    #[cfg(test)]
+    feedback_recorder: Option<&'static rustfs_concurrency::foreground::ForegroundRecorder>,
 }
 
 impl<S, B, ResBody> Service<HttpRequest<B>> for ExternalRequestContextService<S>
 where
     S: Service<HttpRequest<B>, Response = Response<ResBody>>,
 {
-    type Response = Response<ResBody>;
+    type Response = Response<ForegroundResponseBody<ResBody>>;
     type Error = S::Error;
     type Future = ExternalRequestContextFuture<S::Future>;
 
@@ -279,16 +289,31 @@ where
 
         // This outer boundary includes readiness, rate-limit and auth
         // rejections. Metric attribution never depends on an enabled span.
+        let mut feedback = is_s3
+            .then(|| {
+                #[cfg(test)]
+                if let Some(recorder) = self.feedback_recorder {
+                    return Some(recorder.start());
+                }
+                crate::foreground_observation::start_request()
+            })
+            .flatten();
+        let head_request = req.method() == Method::HEAD;
         let mut metrics = is_s3.then(|| s3_http_request_guard(req.method().as_str()));
         let inner = match metrics.as_mut() {
             Some(metrics) => metrics.in_scope(|| self.inner.call(req)),
             None => self.inner.call(req),
         };
+        if let Some(feedback) = feedback.as_mut() {
+            feedback.set_class(foreground_class(metrics.as_ref().and_then(S3HttpRequestGuard::operation)));
+        }
         ExternalRequestContextFuture {
             inner,
             request_id,
             is_s3,
             metrics,
+            feedback,
+            head_request,
         }
     }
 }
@@ -336,6 +361,8 @@ pin_project! {
         request_id: Option<HeaderValue>,
         is_s3: bool,
         metrics: Option<S3HttpRequestGuard>,
+        feedback: Option<ForegroundObservation<'static>>,
+        head_request: bool,
     }
 }
 
@@ -343,7 +370,7 @@ impl<F, ResBody, E> Future for ExternalRequestContextFuture<F>
 where
     F: Future<Output = Result<Response<ResBody>, E>>,
 {
-    type Output = Result<Response<ResBody>, E>;
+    type Output = Result<Response<ForegroundResponseBody<ResBody>>, E>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
@@ -351,12 +378,18 @@ where
             Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)),
             None => this.inner.poll(cx),
         };
+        if let Some(feedback) = this.feedback.as_mut() {
+            feedback.set_class(foreground_class(this.metrics.as_ref().and_then(S3HttpRequestGuard::operation)));
+        }
         let mut response = match result {
             Poll::Ready(Ok(response)) => response,
             Poll::Ready(Err(error)) => {
                 if let Some(metrics) = this.metrics.as_mut() {
                     metrics.service_error();
                 }
+                if let Some(mut feedback) = this.feedback.take() {
+                    feedback.finish(CompletionOutcome::Failed);
+                }
                 return Poll::Ready(Err(error));
             }
             Poll::Pending => return Poll::Pending,
@@ -374,7 +407,126 @@ where
             }
         }
 
-        Poll::Ready(Ok(response))
+        let status = response.status();
+        let suppressed = *this.head_request
+            || status.is_informational()
+            || status == StatusCode::NO_CONTENT
+            || status == StatusCode::NOT_MODIFIED;
+        Poll::Ready(Ok(response.map(|body| {
+            ForegroundResponseBody::new(
+                body,
+                this.feedback.take(),
+                !status.is_client_error() && !status.is_server_error(),
+                suppressed,
+            )
+        })))
+    }
+}
+
+fn foreground_class(operation: Option<rustfs_s3_ops::S3Operation>) -> ForegroundClass {
+    use rustfs_s3_ops::S3Operation;
+    match operation {
+        Some(S3Operation::GetObject | S3Operation::SelectObjectContent) => ForegroundClass::Read,
+        Some(
+            S3Operation::PutObject
+            | S3Operation::CopyObject
+            | S3Operation::UploadPart
+            | S3Operation::UploadPartCopy
+            | S3Operation::CompleteMultipartUpload
+            | S3Operation::AbortMultipartUpload
+            | S3Operation::DeleteObject
+            | S3Operation::DeleteObjects,
+        ) => ForegroundClass::Write,
+        Some(_) => ForegroundClass::Metadata,
+        None => ForegroundClass::Other,
+    }
+}
+
+pin_project! {
+    /// Inline ownership of a foreground observation through the response body.
+    /// EOF means the server produced the body, not that the client acknowledged
+    /// its bytes. Other drops are censored and never successful latency samples.
+    pub struct ForegroundResponseBody<B> {
+        #[pin]
+        inner: B,
+        // Body::is_end_stream only has a shared reference. Empty responses can
+        // finish there without being polled; Cell adds no mutex or allocation.
+        observation: Cell<Option<ForegroundObservation<'static>>>,
+        succeeded: bool,
+    }
+}
+
+fn finish_foreground_observation(observation: &Cell<Option<ForegroundObservation<'_>>>, outcome: CompletionOutcome) {
+    if let Some(mut observation) = observation.take() {
+        observation.finish(outcome);
+    }
+}
+
+impl<B> ForegroundResponseBody<B> {
+    fn new(inner: B, observation: Option<ForegroundObservation<'static>>, succeeded: bool, suppressed: bool) -> Self {
+        let body = Self {
+            inner,
+            observation: Cell::new(observation),
+            succeeded,
+        };
+        if suppressed {
+            body.complete();
+        }
+        body
+    }
+
+    fn complete(&self) {
+        finish_foreground_observation(
+            &self.observation,
+            if self.succeeded {
+                CompletionOutcome::Success
+            } else {
+                CompletionOutcome::Failed
+            },
+        );
+    }
+}
+
+impl<B: Body> Body for ForegroundResponseBody<B> {
+    type Data = B::Data;
+    type Error = B::Error;
+
+    fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
+        let mut this = self.project();
+        let result = this.inner.as_mut().poll_frame(cx);
+        if this.observation.get_mut().is_none() {
+            return result;
+        }
+        let outcome = match &result {
+            Poll::Ready(Some(Err(_))) => Some(CompletionOutcome::Failed),
+            Poll::Ready(None) => Some(if *this.succeeded {
+                CompletionOutcome::Success
+            } else {
+                CompletionOutcome::Failed
+            }),
+            Poll::Ready(Some(Ok(_))) if this.inner.as_ref().get_ref().is_end_stream() => Some(if *this.succeeded {
+                CompletionOutcome::Success
+            } else {
+                CompletionOutcome::Failed
+            }),
+            _ => None,
+        };
+        if let Some(outcome) = outcome {
+            finish_foreground_observation(this.observation, outcome);
+        }
+        result
+    }
+
+    fn is_end_stream(&self) -> bool {
+        let finished = self.inner.is_end_stream();
+        if finished {
+            self.complete();
+        }
+        finished
+    }
+
+    fn size_hint(&self) -> http_body::SizeHint {
+        self.inner.size_hint()
     }
 }
 
@@ -2456,6 +2608,332 @@ fn rewrite_double_slash_root(uri: &Uri) -> Option<Uri> {
 
 #[cfg(test)]
 mod tests {
+    #[tokio::test]
+    async fn foreground_feedback_unobserved_body_is_a_direct_passthrough() {
+        struct EndProbe(Full<Bytes>);
+        impl Body for EndProbe {
+            type Data = Bytes;
+            type Error = Infallible;
+
+            fn poll_frame(
+                mut self: Pin<&mut Self>,
+                cx: &mut Context<'_>,
+            ) -> Poll<Option<Result<http_body::Frame<Bytes>, Infallible>>> {
+                Pin::new(&mut self.0).poll_frame(cx)
+            }
+
+            fn is_end_stream(&self) -> bool {
+                panic!("unobserved frame polling must not add EOF queries")
+            }
+
+            fn size_hint(&self) -> http_body::SizeHint {
+                self.0.size_hint()
+            }
+        }
+
+        let mut body =
+            ForegroundResponseBody::new(EndProbe(Full::new(Bytes::from_static(b"unobserved body"))), None, false, false);
+        assert_eq!(body.size_hint().exact(), Some(15));
+        assert_eq!(
+            body.frame().await.expect("frame").expect("data").into_data().expect("bytes"),
+            Bytes::from_static(b"unobserved body")
+        );
+        assert!(body.frame().await.is_none());
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_empty_head_and_bodyless_responses_finish_once() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let mut service = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        }
+        .layer(tower::service_fn(|req: Request<()>| async move {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::HeadObject);
+            let status = match req.uri().path() {
+                "/bucket/no-content" => StatusCode::NO_CONTENT,
+                "/bucket/not-modified" => StatusCode::NOT_MODIFIED,
+                _ => StatusCode::OK,
+            };
+            Ok::<_, Infallible>(
+                Response::builder()
+                    .status(status)
+                    .body(Empty::<Bytes>::new())
+                    .expect("empty response"),
+            )
+        }));
+        for (method, path) in [
+            (Method::GET, "/bucket/empty"),
+            (Method::HEAD, "/bucket/head"),
+            (Method::GET, "/bucket/no-content"),
+            (Method::GET, "/bucket/not-modified"),
+        ] {
+            let response = service
+                .call(Request::builder().method(method).uri(path).body(()).expect("request"))
+                .await
+                .expect("response");
+            assert!(response.body().is_end_stream());
+            assert!(response.body().is_end_stream(), "repeated inspection is idempotent");
+            drop(response);
+        }
+        let window = reader.sample().expect("completed empty bodies");
+        let metadata = window.classes[2].expect("metadata count");
+        assert_eq!(window.started, Some(4));
+        assert_eq!(
+            (window.in_flight, metadata.finished, metadata.failed, metadata.cancelled),
+            (Some(0), 4, 0, 0)
+        );
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_preserves_service_and_body_errors() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let layer = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        };
+        let mut service = layer.layer(tower::service_fn(|_: Request<()>| async {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::PutObject);
+            Err::<Response<Empty<Bytes>>, _>(io::Error::new(io::ErrorKind::PermissionDenied, "service sentinel"))
+        }));
+        let error = match service
+            .call(Request::builder().uri("/bucket/object").body(()).expect("request"))
+            .await
+        {
+            Err(error) => error,
+            Ok(_) => panic!("service error must be preserved"),
+        };
+        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
+        assert_eq!(error.to_string(), "service sentinel");
+
+        let mut streaming = layer.layer(tower::service_fn(|_: Request<()>| async {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::GetObject);
+            let frames = futures::stream::iter([
+                Ok(http_body::Frame::data(Bytes::from_static(b"partial"))),
+                Err(io::Error::new(io::ErrorKind::UnexpectedEof, "body sentinel")),
+            ]);
+            Ok::<_, Infallible>(Response::new(http_body_util::StreamBody::new(frames)))
+        }));
+        let mut response = streaming
+            .call(Request::builder().uri("/bucket/object").body(()).expect("request"))
+            .await
+            .expect("headers");
+        assert_eq!(
+            response
+                .body_mut()
+                .frame()
+                .await
+                .expect("partial frame")
+                .expect("data frame")
+                .into_data()
+                .expect("bytes"),
+            Bytes::from_static(b"partial")
+        );
+        let error = response
+            .body_mut()
+            .frame()
+            .await
+            .expect("error frame")
+            .expect_err("must preserve body error");
+        assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof);
+        assert_eq!(error.to_string(), "body sentinel");
+        drop(response);
+        let window = reader.sample().expect("failed requests");
+        assert_eq!((window.started, window.in_flight), (Some(2), Some(0)));
+        for class in [0, 1] {
+            let counts = window.classes[class].expect("known class");
+            assert_eq!((counts.finished, counts.failed, counts.cancelled), (1, 1, 0));
+        }
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_http_failure_and_trailers_are_preserved() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let mut service = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        }
+        .layer(tower::service_fn(|_: Request<()>| async {
+            let mut trailers = HeaderMap::new();
+            trailers.insert("x-body-sentinel", HeaderValue::from_static("unchanged"));
+            let frames = futures::stream::iter([
+                Ok::<_, Infallible>(http_body::Frame::data(Bytes::from_static(b"failure"))),
+                Ok(http_body::Frame::trailers(trailers)),
+            ]);
+            Ok::<_, Infallible>(
+                Response::builder()
+                    .status(503)
+                    .body(http_body_util::StreamBody::new(frames))
+                    .expect("response"),
+            )
+        }));
+        let mut response = service
+            .call(Request::builder().uri("/bucket/object").body(()).expect("request"))
+            .await
+            .expect("headers");
+        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(
+            response
+                .body_mut()
+                .frame()
+                .await
+                .expect("data")
+                .expect("frame")
+                .into_data()
+                .expect("bytes"),
+            Bytes::from_static(b"failure")
+        );
+        assert_eq!(reader.sample().expect("before trailers").in_flight, Some(1));
+        let trailers = response
+            .body_mut()
+            .frame()
+            .await
+            .expect("trailers")
+            .expect("frame")
+            .into_trailers()
+            .expect("trailers");
+        assert_eq!(trailers["x-body-sentinel"], "unchanged");
+        assert!(response.body_mut().frame().await.is_none());
+        drop(response);
+        let window = reader.sample().expect("complete failure body");
+        let other = window.classes[3].expect("pre-dispatch failure");
+        assert_eq!((window.in_flight, other.finished, other.failed, other.cancelled), (Some(0), 1, 1, 0));
+    }
+
+    #[test]
+    fn foreground_feedback_classifies_synchronous_dispatch_before_first_poll() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let mut service = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        }
+        .layer(tower::service_fn(|_: Request<()>| {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::PutObject);
+            std::future::pending::<Result<Response<Empty<Bytes>>, Infallible>>()
+        }));
+        let request = service.call(Request::builder().uri("/bucket/object").body(()).expect("request"));
+        drop(request);
+        let window = reader.sample().expect("cancelled unpolled request");
+        let write = window.classes[1].expect("dispatched write");
+        assert_eq!((window.started, window.in_flight, write.cancelled), (Some(1), Some(0), 1));
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_excludes_control_plane_and_internal_requests() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let mut service = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        }
+        .layer(tower::service_fn(|_: Request<()>| async {
+            Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
+        }));
+        for path in ["/health/ready", "/rustfs/admin/v3/info", "/minio/health/live"] {
+            let response = service
+                .call(Request::builder().uri(path).body(()).expect("request"))
+                .await
+                .expect("response");
+            assert!(response.body().is_end_stream());
+        }
+        let window = reader.sample().expect("no S3 events");
+        assert_eq!((window.started, window.in_flight), (Some(0), Some(0)));
+        assert!(window.classes.iter().all(|class| class.expect("known class").finished == 0));
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_keeps_body_ownership_after_response_headers() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let layer = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        };
+        let inner = tower::service_fn(|_: Request<()>| async {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::GetObject);
+            Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"body is still owned"))))
+        });
+        let mut response = layer
+            .layer(inner)
+            .call(Request::builder().uri("/bucket/object").body(()).expect("request"))
+            .await
+            .expect("response headers");
+        let headers = reader.sample().expect("header boundary");
+        assert_eq!(headers.started, Some(1));
+        assert_eq!(headers.in_flight, Some(1), "headers do not finish the body");
+        assert_eq!(headers.classes[0].expect("read count").finished, 0);
+        let frame = response.body_mut().frame().await.expect("data frame").expect("body success");
+        assert_eq!(frame.into_data().expect("data"), Bytes::from_static(b"body is still owned"));
+        drop(response);
+        let body = reader.sample().expect("body boundary");
+        let read = body.classes[0].expect("read count");
+        assert_eq!((body.in_flight, read.finished, read.failed, read.cancelled), (Some(0), 1, 0, 0));
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_records_dropped_body_as_censored() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let mut service = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        }
+        .layer(tower::service_fn(|_: Request<()>| async {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::GetObject);
+            Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"undrained"))))
+        }));
+        let response = service
+            .call(Request::builder().uri("/bucket/object").body(()).expect("request"))
+            .await
+            .expect("response");
+        drop(response);
+        let window = reader.sample().expect("cancelled body");
+        assert_eq!(window.started, Some(1));
+        let read = window.classes[0].expect("read count");
+        assert_eq!((window.in_flight, read.finished, read.cancelled, read.p99), (Some(0), 1, 1, None));
+    }
+
+    #[tokio::test]
+    async fn foreground_feedback_counts_cancellation_before_headers() {
+        use rustfs_concurrency::foreground::ForegroundRecorder;
+        let recorder = Box::leak(Box::new(ForegroundRecorder::new()));
+        let mut reader = recorder.take_reader().expect("isolated reader");
+        let mut service = ExternalRequestContextLayer {
+            feedback_recorder: Some(recorder),
+            ..Default::default()
+        }
+        .layer(tower::service_fn(|_: Request<()>| async {
+            rustfs_io_metrics::record_s3_op(rustfs_s3_ops::S3Operation::PutObject);
+            std::future::pending::<Result<Response<Full<Bytes>>, Infallible>>().await
+        }));
+        let mut pending = Box::pin(
+            service.call(
+                Request::builder()
+                    .method("PUT")
+                    .uri("/bucket/object")
+                    .body(())
+                    .expect("request"),
+            ),
+        );
+        assert!(matches!(futures::poll!(pending.as_mut()), Poll::Pending));
+        drop(pending);
+        let window = reader.sample().expect("cancelled service future");
+        assert_eq!(window.started, Some(1));
+        let write = window.classes[1].expect("write count");
+        assert_eq!((window.in_flight, write.finished, write.cancelled), (Some(0), 1, 1));
+    }
+
     #[tokio::test]
     async fn console_prefix_process_case_browser_redirect() {
         if std::env::var_os("RUSTFS_TEST_CONSOLE_PREFIX_PROCESS").is_none() {

rustfs/src/startup_background.rs

@@ -57,7 +57,7 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
     // (rustfs/backlog#1873).
     run_startup_bitrot_self_test(bitrot_selftest_enabled_from_env(), bitrot_selftest_strict_from_env()).await?;
 
-    let _ = create_ahm_services_cancel_token();
+    let background_cancel = create_ahm_services_cancel_token();
 
     let plan = resolve_background_service_plan(scanner_enabled_from_env(), heal_enabled_from_env());
 
@@ -95,6 +95,10 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
 
     init_on_demand_migration_backfill_runtime(store).await;
 
+    if plan.heal_manager_enabled {
+        let _ = crate::foreground_observation::start_sampler(background_cancel);
+    }
+
     Ok(plan.scanner_enabled)
 }