fix(s3-output)!: stream uploads via TM PartStream, bound peak RAM

The S3 sink used to accumulate every per-prefix batch in a single
`CodecEncoder<Vec<u8>>` for the lifetime of the run (default mode) or
until `batch_max_mb` crossed. On long runs this OOM'd: with N prefixes
and M MB matched per prefix, peak RAM = N × M before the first byte
was uploaded.

Replace the buffer with a streaming pipeline:

  filter worker → CodecEncoder<ChannelWriter> → mpsc(Bytes, cap=2)
                → EncoderPartStream → tm::io::InputStream::from_part_stream
                → TM multipart upload

`ChannelWriter` is a sync `io::Write` that ships each codec block as a
`Bytes` chunk through a tokio mpsc. `EncoderPartStream` impls TM's
`PartStream`, aggregating up to `multipart_part_mb` before yielding one
`PartData` per poll; the trailing partial is marked `is_last(true)` on
channel EOF. Peak resident per active prefix drops to
~`channel_cap × block_size + part_size` (sub-10 MB at default settings)
regardless of total batch size.

PrefixBatch now holds `upload: Option<ActiveUpload>` — opened lazily on
the first matched line (no empty MPUs), closed on `bytes_sent >=
batch_max_mb` or end-of-run. Each open spawns a driver task that joins
the TM handle and folds per-batch counts into global counters (or
`lines_dropped` + `fatal` on failure). `finish()` closes any trailing
uploads on a blocking thread (codec frame trailer writes hit
`ChannelWriter::write` → `blocking_send`, which panics from a runtime
worker) and awaits all driver handles.

Trade-offs documented in the module header:

- Always-MPU. `from_part_stream` is MPU-only; sub-multipart_threshold
  batches now use 3 S3 API calls instead of 1. Functional only.
- No full-batch retry. Once compressed bytes leave the encoder they're
  gone — TM's per-part retries are the only retry. The sink already
  treated failed uploads as dropped batches, so this is a
  non-regression.

Removed dead pool plumbing as part of the rewrite:

  - S3OutputConfig.upload_tasks (and the YAML knob)
  - OutputCli.s3_upload_tasks
  - --s3-output-upload-tasks CLI flag
  - UploadJob struct, upload_tx flume queue, uploader_task

HTTP sink's `upload_tasks` is unaffected — that path still uses a
worker pool.

Instrumentation: `Inner.peak_inflight_bytes` (sink-global high-water
of bytes resident in mpsc + reader pending across all active uploads)
is now surfaced in `OutputStats.extras` so the e2e suite can assert
it. `report_completion` emits `extras` as a JSON string instead of
Debug repr so consumers (and tests) can parse it.

Test coverage (test-first):

- s3_output_does_not_buffer_whole_run_in_memory — bulky fixture
  (200K matches × 2 prefixes ≈ 56 MB plaintext), default mode, asserts
  peak_inflight_bytes ≤ 12 MB. Pre-fix: 61 MB resident → fails.
  Post-fix: passes well under cap.
- Existing s3_output_end_to_end_{no_batching, batched, gzip, plaintext,
  multipart, json_array} all still pass against the streaming path.
- New unit tests on ChannelWriter (inflight/peak/bytes_sent accounting,
  BrokenPipe on reader drop).

BREAKING: --s3-output-upload-tasks CLI flag and S3OutputConfig.upload_tasks
removed (the per-prefix streaming model has no worker pool to size).

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

Author: eliebleton-manomano

src/config/output.rs

@@ -165,8 +165,6 @@ pub struct S3OutputConfig {
     /// integer for an explicit cap.
     #[serde(default)]
     pub multipart_concurrency: Option<usize>,
-    #[serde(default)]
-    pub upload_tasks: Option<usize>,
     /// Object body framing. Default `json_lines` writes NDJSON. Set
     /// `kind: json_array` to upload a single JSON array per batch. The
     /// object extension is still driven by the codec — override
@@ -482,7 +480,6 @@ mystery: 42
             multipart_threshold_mb: 5,
             multipart_part_mb: 5,
             multipart_concurrency: None,
-            upload_tasks: None,
             format: OutputFormat::default(),
         });
         validate_output(&cfg).unwrap();
@@ -526,7 +523,6 @@ mystery: 42
             multipart_threshold_mb: 5,
             multipart_part_mb: 5,
             multipart_concurrency: None,
-            upload_tasks: None,
             format: OutputFormat::default(),
         });
         let err = validate_output(&cfg).unwrap_err();
@@ -545,7 +541,6 @@ mystery: 42
             multipart_threshold_mb: 5,
             multipart_part_mb: 5,
             multipart_concurrency: None,
-            upload_tasks: None,
             format: OutputFormat::default(),
         });
         let err = validate_output(&cfg).unwrap_err();
@@ -632,7 +627,6 @@ dir: /tmp/out
             multipart_threshold_mb: threshold_mb,
             multipart_part_mb: part_mb,
             multipart_concurrency: concurrency,
-            upload_tasks: None,
             format: OutputFormat::default(),
         })
     }

src/config/resolve.rs

@@ -78,7 +78,6 @@ pub struct OutputCli {
     pub s3_multipart_threshold_mb: Option<u64>,
     pub s3_multipart_part_mb: Option<u64>,
     pub s3_multipart_concurrency: Option<usize>,
-    pub s3_upload_tasks: Option<usize>,
 
     // shared
     pub compression_format: Option<CodecFormat>,
@@ -150,7 +149,6 @@ impl OutputCli {
             s3_multipart_concurrency,
             "--s3-output-multipart-concurrency"
         );
-        check!(s3_upload_tasks, "--s3-output-upload-tasks");
         check!(compression_format, "--compression-format");
         check!(compression_level, "--compression-level");
         check!(file_path_template, "--output-path-template");
@@ -271,7 +269,6 @@ fn build_from_cli(cli: &OutputCli) -> Result<OutputConfig> {
                 multipart_threshold_mb: cli.s3_multipart_threshold_mb.unwrap_or(5),
                 multipart_part_mb: cli.s3_multipart_part_mb.unwrap_or(5),
                 multipart_concurrency: cli.s3_multipart_concurrency,
-                upload_tasks: cli.s3_upload_tasks,
                 format: cli_format(cli),
             })
         }
@@ -337,7 +334,6 @@ fn reject_inactive_flags(kind: OutputKind, cli: &OutputCli) -> Result<()> {
         s3_multipart_concurrency,
         "--s3-output-multipart-concurrency"
     );
-    reject_unless!(is_s3, s3_upload_tasks, "--s3-output-upload-tasks");
     reject_unless!(is_file, file_path_template, "--output-path-template");
 
     if !bad.is_empty() {

src/main.rs

@@ -264,10 +264,6 @@ struct Cli {
     /// pass a positive integer for an explicit cap.
     #[arg(long)]
     s3_output_multipart_concurrency: Option<usize>,
-
-    /// Number of concurrent uploader tasks for the s3 output.
-    #[arg(long)]
-    s3_output_upload_tasks: Option<usize>,
 }
 
 /// CLI mirror of [`CodecFormat`] — kept separate so clap's `ValueEnum`
@@ -321,7 +317,6 @@ impl Cli {
             s3_multipart_threshold_mb: self.s3_output_multipart_threshold_mb,
             s3_multipart_part_mb: self.s3_output_multipart_part_mb,
             s3_multipart_concurrency: self.s3_output_multipart_concurrency,
-            s3_upload_tasks: self.s3_output_upload_tasks,
             compression_format: self.compression_format.map(Into::into),
             compression_level: self.compression_level,
             file_path_template: self.output_path_template.clone(),
@@ -814,7 +809,11 @@ fn report_completion(
         compression_ratio = compression_ratio,
         plaintext_mbps = if elapsed_s > 0.0 { plaintext_mb / elapsed_s } else { 0.0 },
         pattern = cli.line_pattern_regex.as_deref().unwrap_or("(all lines)"),
-        extras = ?stats.extras,
+        // Render `extras` as a JSON object string so downstream log
+        // consumers (and the e2e tests) can parse it back; the previous
+        // `?` Debug format produced an ad-hoc Rust-style repr that
+        // wasn't machine-readable.
+        extras = %serde_json::to_string(&stats.extras).unwrap_or_else(|_| "{}".to_string()),
         "Search completed"
     );
 }

src/pipeline/mod.rs

@@ -7,6 +7,7 @@ pub mod observer;
 pub mod orchestrator;
 pub mod output;
 pub mod path_template;
+pub mod s3_streaming;
 pub mod s3_writer;
 pub mod streaming_writer;
 pub mod void_writer;