From 711b6d434d8b37d866af60bcffe593b67efbcfcd Mon Sep 17 00:00:00 2001
From: jgrusewski <jgrusewski@localhost>
Date: Sat, 16 May 2026 15:00:34 +0200
Subject: [PATCH] obs(alpha_pipeline): per-chunk progress logs for cluster
 diagnostics
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

After the cluster's silent-failure incident (75 min between 'alpha
pipeline inputs' and exit, no intervening log line), instrument the
parallel path so any future hang or panic localises to the specific
chunk that died. Each chunk emits start + done lines with chunk_idx,
emit range, warmup_start, row count, and elapsed seconds; the wrapper
also logs dispatch (chunk count + threads + chunk_size) and overall
completion.

Local 1Q smoke (16-thread box):
- dispatching 16 chunks (chunk_size=35485, warmup_bars=2000)
- chunk 0 (cold-start, no warmup shortcut): 19.2s
- chunks 1-15:                              13.9-16.5s
- all-chunks-complete log fires before fxcache write

Cost: ~17 info lines per precompute run — negligible. Worth the
observability when the pipeline takes minutes and any future kill
needs root-cause.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 crates/ml-features/src/alpha_pipeline.rs | 44 ++++++++++++++++++++----
 1 file changed, 38 insertions(+), 6 deletions(-)

diff --git a/crates/ml-features/src/alpha_pipeline.rs b/crates/ml-features/src/alpha_pipeline.rs
index ebfc4404c..7b9f09733 100644
--- a/crates/ml-features/src/alpha_pipeline.rs
+++ b/crates/ml-features/src/alpha_pipeline.rs
@@ -143,20 +143,30 @@ pub fn extract_alpha_features(
     // ── Chunk the emit range ──
     let n_chunks = rayon::current_num_threads().max(1).min(n_output);
     let chunk_size = n_output.div_ceil(n_chunks);
-    let chunks: Vec<(usize, usize)> = (0..n_chunks)
+    let chunks: Vec<(usize, usize, usize)> = (0..n_chunks)
         .map(|k| {
             let emit_start = k * chunk_size;
             let emit_end = ((k + 1) * chunk_size).min(n_output);
-            (emit_start, emit_end)
+            (k, emit_start, emit_end)
         })
-        .filter(|(s, e)| s < e)
+        .filter(|(_, s, e)| s < e)
         .collect();
 
+    tracing::info!(
+        "alpha pipeline: dispatching {} chunks (rayon threads={}, chunk_size={}, warmup_bars={})",
+        chunks.len(),
+        rayon::current_num_threads(),
+        chunk_size,
+        WARMUP_BARS
+    );
+    let t_pipeline_start = std::time::Instant::now();
+
     // Parallel per-chunk computation; flatten into the final contiguous Vec.
-    chunks
+    let features: Vec<Vec<f32>> = chunks
         .par_iter()
-        .flat_map_iter(|&(emit_start, emit_end)| {
+        .flat_map_iter(|&(chunk_idx, emit_start, emit_end)| {
             process_chunk(
+                chunk_idx,
                 bars,
                 snapshots,
                 trades,
@@ -167,12 +177,20 @@ pub fn extract_alpha_features(
             )
             .into_iter()
         })
-        .collect()
+        .collect();
+
+    tracing::info!(
+        "alpha pipeline: all chunks complete, {} rows emitted in {:.1}s",
+        features.len(),
+        t_pipeline_start.elapsed().as_secs_f64()
+    );
+    features
 }
 
 /// Process one chunk [emit_start..emit_end) with fresh aggregator state,
 /// pre-rolling [`WARMUP_BARS`] leading bars to saturate stateful blocks.
 fn process_chunk(
+    chunk_idx: usize,
     bars: &[OHLCVBar],
     snapshots: &[Mbp10Snapshot],
     trades: &[Mbp10Trade],
@@ -182,6 +200,14 @@ fn process_chunk(
     emit_end: usize,
 ) -> Vec<Vec<f32>> {
     let warmup_start = emit_start.saturating_sub(WARMUP_BARS);
+    let t_chunk_start = std::time::Instant::now();
+    tracing::info!(
+        "alpha chunk[{}] start: emit=[{}..{}) warmup_start={}",
+        chunk_idx,
+        emit_start,
+        emit_end,
+        warmup_start
+    );
 
     // ── Stateful aggregators (persistent across bars) ──
     // Block A-E factory aggregators
@@ -494,6 +520,12 @@ fn process_chunk(
         }
     }
 
+    tracing::info!(
+        "alpha chunk[{}] done: rows={} elapsed={:.1}s",
+        chunk_idx,
+        features.len(),
+        t_chunk_start.elapsed().as_secs_f64()
+    );
     features
 }