feat(progress): disambiguate filter bottleneck via sink-busy gauge + S3 inflight

The progress reporter's `bottleneck` field had been lying for a while:
`filter` triggered whenever the decompressed-line channel was full,
but downstream of that channel sits a lot more than the regex matcher
— `sink.ingest` does the codec compression, framing, per-prefix mutex
handoff, and (for S3) an mpsc send that blocks when TM can't drain.
A no-regex S3 run getting reported as `bottleneck=filter` was the
prompt: the label couldn't tell the user whether their CPU was pegged
on zstd or whether S3 was the floor.

Fix it with two new signals, both cheap:

1. `workers_in_ingest` — sink-agnostic AtomicUsize gauge. Filter
   workers bump it via an RAII guard around every `sink.ingest`
   call; the progress reporter samples it instantaneously at each
   tick. ≥ half the filter workers inside ingest means the sink is
   the lid; < half means workers are spending their time on
   regex / channel receive / upstream stages.

2. S3-specific drill-down — when the gauge says "sink is the lid",
   look at the S3 sink's `inflight_bytes` (resident bytes across
   the per-upload mpsc channels and reader pending buffers) and
   `active_uploads` count. High inflight per upload means TM is
   slow to drain parts → network-bound. Low inflight means the
   codec is the producer-side cost → CPU-bound.

Wired through a new `SinkObservability` struct on the `OutputSink`
trait. Default is empty; S3 implements; file / http / void return
empty (they don't have meaningful internal buffers to expose at
this level — for file, `iostat` is the next step).

New label set, all underscore-separated, no parens:

  download           dc channel mostly empty
  filter             dc full, sink not busy
  sink_s3_codec      dc full, sink busy, S3 mpsc empty
  sink_s3_network    dc full, sink busy, S3 mpsc backed up
  sink_file          dc full, sink busy, file sink
  sink_http          dc full, sink busy, HTTP sink (unusual)
  sink_void          dc full, sink busy, void sink (~never)
  sink_busy          fallback for unknown sink kinds
  compress (HTTP)    http line channel saturated  [unchanged]
  upload   (HTTP)    http batch channel saturated [unchanged]

Existing HTTP path inherits the gauge for the same disambiguation:
its old `filter` label is now split into `filter` vs `sink_http`
the same way.

The `Search progress` log line now also carries `workers_in_ingest`
and (when applicable) `sink_inflight_bytes` + `sink_active_uploads`
as structured fields so post-run log analysis has the raw signals
alongside the classifier's verdict.

Documentation: new "Reading the `bottleneck` label" section in
README.md enumerates every label, the signals it's computed from,
and operator guidance for what to investigate next per label.

Tests: 10 new classifier tests + 1 RAII gauge test cover every
label transition, the half-of-workers threshold for sink-busy,
the conservative fall-back to `sink_s3_codec` on missing inflight
signal, and the older HTTP priority order under the new signature.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: eliebleton-manomano

README.md

@@ -325,3 +325,33 @@ Usage: bucket-scrapper [OPTIONS] --start <START>
 **CPU**: Use samply with the profiling build profile.
 
 **Memory**: `cargo build --profile profiling --features dhat-heap`, then submit the generated `profiler.json` to [dh_view](https://nnethercote.github.io/dh_view/dh_view.html).
+
+## Reading the `bottleneck` label
+
+Every `Search progress` log record carries a `bottleneck` field that names the stage most likely limiting throughput at the time the report was emitted. The classifier reads several channel-fill percentages plus a sink-busy gauge; the label is a one-word summary of the dominant signal.
+
+### Signals it looks at
+
+- **`dc_pct`** — fill percentage of the decompressed-line channel between the download/decompress stage and the filter workers. High means the downstream stages can't keep up with what's being decompressed.
+- **`workers_in_ingest`** — instantaneous count of filter workers currently inside `sink.ingest`. Compared against the total filter-worker count: ≥ half means the sink is where time is going (codec compression, framing, sink-internal queues), < half means filter-side work (regex / channel receive) is what's eating the workers.
+- **`sink_inflight_bytes`** + **`sink_active_uploads`** — S3 sink only. Bytes resident in the per-upload mpsc channels and reader pending buffers, scaled by how many uploads are currently open. High means TM is slow to drain parts to S3; low while workers are stuck in `sink.ingest` means the codec is the producer-side cost.
+- HTTP-mode only: **`line_pct`** and **`batch_pct`** — fill of the HTTP writer's internal line and batch channels. Provide direct visibility into the compressor and uploader stages.
+
+### Label meanings
+
+- **`download`** — `dc_pct` is low. The download stage isn't filling the line channel fast enough. Causes: S3 per-connection throughput, network bandwidth, low `--max-parallel`, or storage class.
+- **`filter`** — `dc_pct` high, `workers_in_ingest` low. Filter workers are spending their time on regex matching or waiting on channel receives. Causes: an expensive regex, lots of non-matching lines, or simply too few filter workers (`--filter-tasks`).
+- **`sink_s3_codec`** — `dc_pct` high, `workers_in_ingest` high, sink mpsc is roughly empty. Workers are in `sink.ingest` but bytes are leaving fast — the codec (zstd / gzip) is the producer-side cost. Try `--compression-format none`, raise the level only with eyes on this label, or look at per-prefix lock contention if you have many concurrent prefixes hitting the same per-prefix mutex.
+- **`sink_s3_network`** — `dc_pct` high, `workers_in_ingest` high, sink mpsc is backed up. `ChannelWriter::blocking_send` is waiting because TM / S3 isn't accepting parts fast enough. Causes: network bandwidth ceiling, `multipart_concurrency` set too low, S3 throttling.
+- **`sink_file`** — `dc_pct` high, `workers_in_ingest` high, file sink. Codec or the OS write path is the lid. The file sink doesn't have an internal queue to look at — reach for `iostat`/`vmstat`/`dmesg` to see whether it's the filesystem, dm-crypt, an NBD/EBS volume, or just disk pressure.
+- **`sink_http`** — `dc_pct` high, `workers_in_ingest` high, HTTP sink, but the HTTP writer's own channels aren't full. Unusual; usually you'd see `compress` or `upload` for HTTP-bound runs.
+- **`compress`** *(HTTP only)* — the HTTP writer's line channel is full. The compressor task pool isn't keeping up. Raise `--http-compressor-tasks` (or rely on the auto-inferred default).
+- **`upload`** *(HTTP only)* — the HTTP writer's batch channel is full. The uploader pool can't get batches out to the API fast enough. Raise `--http-upload-tasks`, raise `--max-upload-rate`, or check the upstream API.
+- **`sink_void`** — should never realistically appear; void's ingest is a counter bump. If it shows up, something is wrong.
+- **`sink_busy`** — generic fallback if the sink kind doesn't match a known label. Means: filter workers are stuck inside `sink.ingest` but we couldn't be more specific.
+
+### Caveats
+
+- The classifier reports the *dominant* signal at sampling time. A flapping pipeline (e.g. download bursts followed by sink bursts) will rotate labels across consecutive reports — that's a useful signal in itself.
+- The threshold for "channel saturated" is 80% fill; for "sink busy" it's ≥ half of filter workers inside `sink.ingest`. Both are heuristics and may need adjustment as workloads shift.
+- For the S3 sink, the codec-vs-network drill-down doesn't see *inside* the AWS transfer manager — once bytes leave our `ChannelWriter`, we lose visibility. If TM has its own internal queueing under pressure, we'd report `sink_s3_codec` (low local inflight) even though the actual bottleneck is downstream. Use the upload throughput numbers and `multipart_concurrency` setting alongside the label.

src/pipeline/mod.rs

@@ -20,7 +20,7 @@ pub use http_sink::HttpOutputSink;
 pub use http_writer::{HttpResultWriter, HttpWriterConfig, HttpWriterStats};
 pub use observer::{ChannelObserver, DownloadObserver, PipelineObserver};
 pub use orchestrator::{StreamingDownloader, StreamingDownloaderConfig};
-pub use output::{OutputSink, OutputStats};
+pub use output::{OutputSink, OutputStats, SinkObservability};
 pub use s3_writer::S3OutputSink;
 pub use streaming_writer::{FileWriterStats, SharedFileWriter};
 pub use void_writer::VoidOutputSink;

src/pipeline/orchestrator.rs

@@ -217,6 +217,13 @@ impl StreamingDownloader {
         let filter_lines_in = Arc::new(AtomicUsize::new(0));
         let filter_bytes_in = Arc::new(AtomicUsize::new(0));
         let workers_alive = Arc::new(AtomicUsize::new(self.config.filter_tasks));
+        // Sink-agnostic "how many filter workers are inside sink.ingest
+        // right now" gauge. Filter workers bump it via a RAII guard
+        // around each `sink.ingest` call; the progress reporter samples
+        // it instantaneously to distinguish `filter` vs `sink_*` labels.
+        let workers_in_ingest: crate::progress::IngestGauge = Arc::new(AtomicUsize::new(0));
+        let sink_obs = sink.sink_observability();
+        let sink_kind = sink.type_name();
 
         let progress = Arc::new(Mutex::new(PipelineProgress::new(
             objects.len(),
@@ -230,6 +237,10 @@ impl StreamingDownloader {
             filter_lines_in.clone(),
             filter_bytes_in.clone(),
             workers_alive.clone(),
+            self.config.filter_tasks,
+            workers_in_ingest.clone(),
+            sink_obs,
+            sink_kind,
         )));
 
         // Emit initial progress at t=0 so charts always have a starting point
@@ -282,6 +293,7 @@ impl StreamingDownloader {
             let filter_bytes_in = filter_bytes_in.clone();
             let fe = fatal_error.clone();
             let wa = workers_alive.clone();
+            let wii = workers_in_ingest.clone();
 
             worker_set.spawn(async move {
                 let result = Self::filter_worker(
@@ -294,6 +306,7 @@ impl StreamingDownloader {
                     filter_lines_in,
                     filter_bytes_in,
                     fe,
+                    wii,
                 )
                 .await;
                 wa.fetch_sub(1, Ordering::Relaxed);
@@ -854,6 +867,7 @@ impl StreamingDownloader {
         filter_lines_in: Arc<AtomicUsize>,
         filter_bytes_in: Arc<AtomicUsize>,
         fatal_error: Option<Arc<AtomicBool>>,
+        workers_in_ingest: crate::progress::IngestGauge,
     ) -> Result<usize> {
         let result = tokio::task::spawn_blocking(move || -> Result<usize> {
             let mut local = 0usize;
@@ -875,6 +889,12 @@ impl StreamingDownloader {
                 filter_bytes_in.fetch_add(line_len, Ordering::Relaxed);
 
                 if searcher.matches_line(&line.data) {
+                    // RAII gauge so the progress reporter can tell whether
+                    // filter workers are stuck inside the sink (codec /
+                    // mpsc send / I/O) or actually doing filter-side work.
+                    // Drop on the same line frees the slot regardless of
+                    // panics / early returns.
+                    let _ingest_guard = crate::progress::IngestGuard::new(&workers_in_ingest);
                     sink.ingest(&line.source.prefix, &line.data)?;
                     local += 1;
                     match_count.fetch_add(1, Ordering::Relaxed);

src/pipeline/output.rs

@@ -10,7 +10,7 @@ use anyhow::Result;
 use serde_json::{Map, Value};
 use std::future::Future;
 use std::pin::Pin;
-use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize};
 use std::sync::Arc;
 
 /// Boxed future used by [`OutputSink::finish`] to make the trait `dyn`-safe
@@ -89,4 +89,34 @@ pub trait OutputSink: Send + Sync {
 
     /// One-line label describing the sink for end-of-run logging.
     fn type_name(&self) -> &'static str;
+
+    /// Optional handle to per-sink internal-state metrics, sampled by
+    /// the progress reporter. Used to disambiguate the bottleneck label:
+    /// when filter workers are stuck inside `sink.ingest`, this tells us
+    /// *why* — is the sink's own outbound queue full (network-bound) or
+    /// is the queue empty because the producer side is slow (codec-bound)?
+    ///
+    /// Default returns empty; sinks with meaningful internal buffering
+    /// (currently just S3) override.
+    fn sink_observability(&self) -> SinkObservability {
+        SinkObservability::default()
+    }
+}
+
+/// Per-sink internal-state metrics exposed to the progress reporter.
+///
+/// Both fields are optional: a sink that doesn't have a meaningful
+/// internal queue (file, void) returns `None` for both, and the
+/// progress reporter treats the absence of the signal conservatively
+/// — see `classify_bottleneck_non_http` for the exact rules.
+#[derive(Debug, Clone, Default)]
+pub struct SinkObservability {
+    /// Bytes currently resident in sink-internal channels and reader
+    /// pending buffers. Read by the progress reporter on each tick.
+    pub inflight_bytes: Option<Arc<AtomicU64>>,
+    /// Count of upload contexts currently open — used to scale the
+    /// per-upload "backed up" threshold against. For the S3 sink this
+    /// is the number of multipart uploads with a still-living
+    /// `ChannelWriter` or a still-running TM driver task.
+    pub active_uploads: Option<Arc<AtomicUsize>>,
 }

src/pipeline/s3_writer.rs

@@ -110,7 +110,7 @@ use aws_sdk_s3_transfer_manager as tm;
 use bytes::Bytes;
 use serde_json::json;
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use tokio::runtime::Handle;
 use tokio::task::JoinHandle;
@@ -223,6 +223,12 @@ struct Inner {
     /// `OutputStats.extras` so the e2e suite can assert the sink doesn't
     /// buffer the whole run in memory.
     peak_inflight_bytes: Arc<AtomicU64>,
+    /// Count of currently-open TM upload contexts. Incremented when
+    /// `open_upload` spawns a driver task; decremented at the end of
+    /// that driver task (i.e. once TM finalizes the multipart). Sampled
+    /// by the progress reporter via `SinkObservability` to scale the
+    /// per-upload "channel backed up" threshold.
+    active_uploads: Arc<AtomicUsize>,
     fatal: Arc<AtomicBool>,
     /// Once `true`, the sink is finalized and `ingest` returns an error
     /// instead of opening a fresh upload. Prevents `flush_batch`-style
@@ -295,6 +301,7 @@ impl S3OutputSink {
             objects_written: AtomicU64::new(0),
             inflight_bytes: Arc::new(AtomicU64::new(0)),
             peak_inflight_bytes: Arc::new(AtomicU64::new(0)),
+            active_uploads: Arc::new(AtomicUsize::new(0)),
             fatal: Arc::new(AtomicBool::new(false)),
             finished: AtomicBool::new(false),
         });
@@ -389,6 +396,13 @@ impl S3OutputSink {
             .initiate()
             .map_err(|e| anyhow!("S3 sink: failed to initiate upload for `{key}`: {e}"))?;
 
+        // From this moment until the TM driver task exits, this upload
+        // contributes to `inflight_bytes` (channel + reader pending) and
+        // counts as an "active upload" for the progress reporter's
+        // backpressure heuristic. Decrement in the driver task's
+        // closure below.
+        inner.active_uploads.fetch_add(1, Ordering::Relaxed);
+
         let batch_stats = Arc::new(BatchStats::default());
 
         // Spawn a tiny driver task that awaits TM completion, then folds
@@ -401,6 +415,9 @@ impl S3OutputSink {
             let bytes_sent_for_task = bytes_sent.clone();
             async move {
                 let result = upload_handle.join().await;
+                // Whatever TM's outcome, this upload is no longer active —
+                // its mpsc is fully drained and its driver is exiting.
+                inner.active_uploads.fetch_sub(1, Ordering::Relaxed);
                 let lines = stats.lines.load(Ordering::Relaxed);
                 let plaintext = stats.plaintext.load(Ordering::Relaxed);
                 let bytes = bytes_sent_for_task.load(Ordering::Relaxed);
@@ -745,6 +762,13 @@ impl OutputSink for S3OutputSink {
     fn type_name(&self) -> &'static str {
         "s3"
     }
+
+    fn sink_observability(&self) -> crate::pipeline::SinkObservability {
+        crate::pipeline::SinkObservability {
+            inflight_bytes: Some(self.inner.inflight_bytes.clone()),
+            active_uploads: Some(self.inner.active_uploads.clone()),
+        }
+    }
 }
 
 #[cfg(test)]