fix(ops): #60 cap + observe queryOpSet fan-out size

src/concurrent_engine.rs

@@ -39,6 +39,16 @@ pub struct MetricsBridge {
     pub lazy_load_duration: prometheus::HistogramVec,
     pub compaction_total: prometheus::IntCounterVec,
     pub compaction_duration: prometheus::HistogramVec,
+    /// queryOpSet fan-out size histogram (issue #60). Observed pre-cap so we can
+    /// see what we'd reject if cap were lower.
+    pub query_op_set_fanout_size: prometheus::HistogramVec,
+    /// queryOpSet rejection counter (issue #60). Label `reason="fanout_too_wide"`
+    /// when fan-out exceeds `BITDEX_QUERY_OP_SET_MAX_FANOUT`.
+    pub query_op_set_rejected_total: prometheus::IntCounterVec,
+    /// Cumulative slot mutations from queryOpSet fan-outs. Sums to total work
+    /// the WAL reader has done across all queryOpSets; not the same as the
+    /// `applied` count returned per call.
+    pub query_op_set_applied_slots_total: prometheus::IntCounterVec,
     pub index_name: String,
 }
 /// Commands sent to the flush thread for state transitions that must
@@ -3387,6 +3397,14 @@ impl ConcurrentEngine {
     pub fn set_metrics_bridge(&self, bridge: MetricsBridge) {
         self.metrics_bridge.store(Arc::new(Some(Arc::new(bridge))));
     }
+    /// Read-only handle to the metrics bridge for code paths outside the engine
+    /// (e.g. `ops_processor::apply_query_op_set`). Returns `None` when the server
+    /// layer hasn't wired the bridge yet (boot-time and dump-only test contexts).
+    #[cfg(feature = "server")]
+    pub fn metrics_bridge_handle(&self) -> Option<Arc<MetricsBridge>> {
+        let guard = self.metrics_bridge.load();
+        (**guard).as_ref().map(Arc::clone)
+    }
     /// Get a reference to the bitmap memory cache (for metrics scraping).
     pub fn bitmap_memory_cache(&self) -> &crate::bitmap_memory_cache::BitmapMemoryCache {
         &self.bitmap_memory_cache

src/metrics.rs

@@ -175,6 +175,21 @@ pub struct Metrics {
     /// End-to-end latency of POST /ops WAL append (lock acquisition + write + fsync).
     pub wal_append_duration_seconds: Histogram,
 
+    // -- queryOpSet fan-out cost (issue #60) --
+    /// Histogram of result.ids size per apply_query_op_set call. Buckets cover the
+    /// full observed range from postId-eq narrow matches (1) to wide nsfwLevel-style
+    /// matches (10M+). Used to size the BITDEX_QUERY_OP_SET_MAX_FANOUT cap from real
+    /// prod data instead of guessing.
+    pub query_op_set_fanout_size: HistogramVec,
+    /// Counter incremented when a queryOpSet is rejected for exceeding the fan-out
+    /// cap. Label `reason="fanout_too_wide"`. Alert when rate > 0 — every rejection
+    /// is a missed mutation (data drift) until #62 (paginate-instead-of-skip) lands.
+    pub query_op_set_rejected_total: IntCounterVec,
+    /// Counter of total slot mutations applied via queryOpSet fan-out. Lets us
+    /// distinguish narrow (postId-eq, ~1 slot) from wide (nsfwLevel-eq, millions)
+    /// at the work-unit level rather than the API-call level.
+    pub query_op_set_applied_slots_total: IntCounterVec,
+
     // -- Boot phase breakdown --
     pub boot_phase_seconds: IntGaugeVec,
 }
@@ -868,6 +883,40 @@ impl Metrics {
         )
         .unwrap();
 
+        // queryOpSet fan-out cost (issue #60).
+        // Histogram buckets span the full observed range:
+        // narrow (postId-eq, ~1 slot) → moderate (modelVersionIds, 1K-100K) → wide
+        // (nsfwLevel-eq, 10M+). Powers of 10 with a 22.9M anchor for the local
+        // nsfwLevel=1 size we measured during the OOM repro work.
+        let query_op_set_fanout_buckets = vec![
+            1.0, 10.0, 100.0, 1_000.0, 10_000.0, 100_000.0, 1_000_000.0, 10_000_000.0, 100_000_000.0,
+        ];
+        let query_op_set_fanout_size = HistogramVec::new(
+            HistogramOpts::new(
+                "bitdex_query_op_set_fanout_size",
+                "Number of slots matched by a queryOpSet's filter, observed before per-slot apply",
+            )
+            .buckets(query_op_set_fanout_buckets),
+            &["index"],
+        )
+        .unwrap();
+        let query_op_set_rejected_total = IntCounterVec::new(
+            Opts::new(
+                "bitdex_query_op_set_rejected_total",
+                "queryOpSet ops rejected before per-slot apply (e.g. fan-out exceeded cap)",
+            ),
+            &["index", "reason"],
+        )
+        .unwrap();
+        let query_op_set_applied_slots_total = IntCounterVec::new(
+            Opts::new(
+                "bitdex_query_op_set_applied_slots_total",
+                "Total slot mutations applied via queryOpSet fan-out (sum of result.ids sizes)",
+            ),
+            &["index"],
+        )
+        .unwrap();
+
         // Boot phase breakdown
         let boot_phase_seconds = IntGaugeVec::new(
             Opts::new("bitdex_boot_phase_seconds", "Duration of each boot phase in seconds"),
@@ -1015,6 +1064,9 @@ impl Metrics {
         registry.register(Box::new(wal_ops_written_total.clone())).unwrap();
         registry.register(Box::new(wal_last_applied_timestamp_seconds.clone())).unwrap();
         registry.register(Box::new(wal_append_duration_seconds.clone())).unwrap();
+        registry.register(Box::new(query_op_set_fanout_size.clone())).unwrap();
+        registry.register(Box::new(query_op_set_rejected_total.clone())).unwrap();
+        registry.register(Box::new(query_op_set_applied_slots_total.clone())).unwrap();
         registry.register(Box::new(boot_phase_seconds.clone())).unwrap();
 
         Self {
@@ -1131,6 +1183,9 @@ impl Metrics {
             wal_ops_written_total,
             wal_last_applied_timestamp_seconds,
             wal_append_duration_seconds,
+            query_op_set_fanout_size,
+            query_op_set_rejected_total,
+            query_op_set_applied_slots_total,
             boot_phase_seconds,
         }
     }

src/ops_processor.rs

@@ -23,6 +23,22 @@ use crate::shard_store_doc::PackedValue;
 use crate::shard_store_doc::DocStoreV3;
 use crate::filter::{FilterFieldType, NULL_BITMAP_KEY};
 use crate::ingester::BitmapSink;
+
+/// queryOpSet fan-out cap (issue #60).
+///
+/// Default `usize::MAX` (effectively no cap) — ships first so the new
+/// `bitdex_query_op_set_fanout_size` histogram gathers prod data. Once we have
+/// a fan-out distribution from real traffic, set a finite cap via the
+/// `BITDEX_QUERY_OP_SET_MAX_FANOUT` env var. Reads on every apply (microsecond
+/// cost dwarfed by `execute_query`'s ms-scale work) so operators can tune
+/// without restart by editing the manifest and rolling pods.
+const DEFAULT_MAX_FANOUT: usize = usize::MAX;
+fn max_fanout() -> usize {
+    std::env::var("BITDEX_QUERY_OP_SET_MAX_FANOUT")
+        .ok()
+        .and_then(|v| v.parse::<usize>().ok())
+        .unwrap_or(DEFAULT_MAX_FANOUT)
+}
 use crate::mutation::{value_to_bitmap_key, value_to_sort_u32, FieldRegistry};
 use crate::pg_sync::op_dedup::dedup_ops;
 use crate::pg_sync::ops::{EntityOps, Op};
@@ -1094,6 +1110,39 @@ fn apply_query_op_set<S: BitmapSink>(
         .execute_query(&query)
         .map_err(|e| format!("queryOpSet query failed: {e}"))?;
     let slot_ids = &result.ids;
+
+    // Issue #60: observe fan-out size BEFORE the cap check so the histogram captures
+    // the would-be-rejected upper tail too. The bridge handle is `None` in tests
+    // and dump-only contexts — observation is best-effort.
+    #[cfg(feature = "server")]
+    let metrics = engine.metrics_bridge_handle();
+    #[cfg(feature = "server")]
+    if let Some(ref bridge) = metrics {
+        bridge
+            .query_op_set_fanout_size
+            .with_label_values(&[&bridge.index_name])
+            .observe(slot_ids.len() as f64);
+    }
+
+    let cap = max_fanout();
+    if slot_ids.len() > cap {
+        tracing::warn!(
+            target: "ops_processor",
+            "queryOpSet '{}' matches {} slots, exceeds cap {} — op skipped (data drift)",
+            query_str,
+            slot_ids.len(),
+            cap,
+        );
+        #[cfg(feature = "server")]
+        if let Some(ref bridge) = metrics {
+            bridge
+                .query_op_set_rejected_total
+                .with_label_values(&[&bridge.index_name, "fanout_too_wide"])
+                .inc();
+        }
+        return Ok(0);
+    }
+
     if slot_ids.is_empty() {
         return Ok(0);
     }
@@ -1128,6 +1177,15 @@ fn apply_query_op_set<S: BitmapSink>(
         }
         applied += 1;
     }
+
+    #[cfg(feature = "server")]
+    if let Some(ref bridge) = metrics {
+        bridge
+            .query_op_set_applied_slots_total
+            .with_label_values(&[&bridge.index_name])
+            .inc_by(applied as u64);
+    }
+
     Ok(applied)
 }
 /// Parse a simple filter string like "modelVersionIds eq 456" or "postId eq 789"
@@ -2365,4 +2423,40 @@ mod tests {
             "null add on nullable field should not insert any bitmap bit"
         );
     }
+
+    // ── #60: queryOpSet fan-out cap ─────────────────────────────────────────
+    //
+    // Env-mutating tests must be `#[serial]` because `BITDEX_QUERY_OP_SET_MAX_FANOUT`
+    // is process-global. Each test sets its own value, asserts, restores prior state.
+
+    #[test]
+    #[serial_test::serial]
+    fn test_max_fanout_default_is_usize_max() {
+        std::env::remove_var("BITDEX_QUERY_OP_SET_MAX_FANOUT");
+        assert_eq!(
+            max_fanout(),
+            usize::MAX,
+            "unset env should yield DEFAULT_MAX_FANOUT (usize::MAX)"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn test_max_fanout_env_override_parses() {
+        std::env::set_var("BITDEX_QUERY_OP_SET_MAX_FANOUT", "100000");
+        assert_eq!(max_fanout(), 100_000);
+        std::env::remove_var("BITDEX_QUERY_OP_SET_MAX_FANOUT");
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn test_max_fanout_invalid_env_falls_back_to_default() {
+        std::env::set_var("BITDEX_QUERY_OP_SET_MAX_FANOUT", "not-a-number");
+        assert_eq!(
+            max_fanout(),
+            usize::MAX,
+            "non-numeric env value should fall back to DEFAULT_MAX_FANOUT"
+        );
+        std::env::remove_var("BITDEX_QUERY_OP_SET_MAX_FANOUT");
+    }
 }

src/relay/auth.rs

@@ -88,7 +88,7 @@ mod tests {
         addr.parse().unwrap()
     }
 
-    fn hdr(key: &str, value: &str) -> HeaderMap {
+    fn hdr(key: &'static str, value: &str) -> HeaderMap {
         let mut h = HeaderMap::new();
         h.insert(key, HeaderValue::from_str(value).unwrap());
         h

src/server.rs

@@ -1696,6 +1696,9 @@ fn restore_index(state: &SharedState) -> Result<(), String> {
             lazy_load_duration: state.metrics.lazy_load_duration_seconds.clone(),
             compaction_total: state.metrics.compaction_total.clone(),
             compaction_duration: state.metrics.compaction_duration_seconds.clone(),
+            query_op_set_fanout_size: state.metrics.query_op_set_fanout_size.clone(),
+            query_op_set_rejected_total: state.metrics.query_op_set_rejected_total.clone(),
+            query_op_set_applied_slots_total: state.metrics.query_op_set_applied_slots_total.clone(),
             index_name: def.name.clone(),
         });
         let phase4_elapsed = phase_start.elapsed();
@@ -2012,6 +2015,9 @@ async fn handle_create_index(
         lazy_load_duration: state.metrics.lazy_load_duration_seconds.clone(),
         compaction_total: state.metrics.compaction_total.clone(),
         compaction_duration: state.metrics.compaction_duration_seconds.clone(),
+        query_op_set_fanout_size: state.metrics.query_op_set_fanout_size.clone(),
+        query_op_set_rejected_total: state.metrics.query_op_set_rejected_total.clone(),
+        query_op_set_applied_slots_total: state.metrics.query_op_set_applied_slots_total.clone(),
         index_name: definition.name.clone(),
     });