Validate child instrument IDs in OrderBookDeltas

crates/data/src/engine/mod.rs

@@ -2014,8 +2014,9 @@ impl DataEngine {
         let (parent_start, parent_end) = parent_request_window(parent.as_ref());
         let rebuilt = rebuild_pipeline_response(parent_id, parent.as_ref(), legs);
 
-        // If the rebuild failed (mixed-variant or unsupported-variant legs), drop the
-        // associated `RequestJoin` so its staging maps do not leak. Without this the
+        // If the rebuild failed (mixed-variant, unsupported-variant, or mixed-instrument
+        // `BookDeltas` legs), drop the associated `RequestJoin` so its staging maps do not
+        // leak. Without this the
         // original join request stays in `pending_join_requests` and its
         // `parent_join_request_id` mapping stays live, neither of which will ever
         // resolve through normal flow.
@@ -5470,8 +5471,10 @@ fn log_if_empty_response<T, I: Display>(data: &[T], id: &I, correlation_id: &UUI
 /// Concatenates same-variant leg payloads into a single rebuilt response keyed by `parent_id`.
 ///
 /// Returns `None` when legs are mixed-variant or empty; pipelines only group legs of the same
-/// variant. The rebuilt response inherits `start` and `end` from the parent request when the
-/// parent is a `RequestJoin`; otherwise leg bounds are preserved on the first leg.
+/// variant. `BookDeltas` legs additionally return `None` when their wrapper instruments differ,
+/// since a book-delta batch is keyed by one instrument and cannot carry another's children.
+/// The rebuilt response inherits `start` and `end` from the parent request when the parent is a
+/// `RequestJoin`; otherwise leg bounds are preserved on the first leg.
 fn rebuild_pipeline_response(
     parent_id: UUID4,
     parent: Option<&RequestCommand>,
@@ -5603,6 +5606,24 @@ fn rebuild_pipeline_response(
                     log::error!("Mixed-variant legs in pipeline {parent_id}");
                     return None;
                 };
+
+                // A book-delta batch is keyed by one instrument, so legs for different
+                // instruments cannot be concatenated into a single response. Matched by
+                // value as well as identity, mirroring `OrderBookDeltas::new_checked`,
+                // since legs crossing the FFI boundary do not share an intern pool.
+                let same_instrument = other.instrument_id == acc.instrument_id
+                    || (other.instrument_id.symbol.as_str() == acc.instrument_id.symbol.as_str()
+                        && other.instrument_id.venue.as_str() == acc.instrument_id.venue.as_str());
+
+                if !same_instrument {
+                    log::error!(
+                        "Mixed-instrument BookDeltas legs in pipeline {parent_id}: {} and {}",
+                        acc.instrument_id,
+                        other.instrument_id,
+                    );
+                    return None;
+                }
+
                 acc.data.extend(other.data);
             }
             acc.data.sort_by_key(|d| d.ts_init);

crates/data/tests/engine.rs

@@ -18513,6 +18513,168 @@ fn test_request_join_single_leg_fires_immediately(
     }
 }
 
+fn leg_book_deltas_response(
+    request_id: UUID4,
+    instrument_id: InstrumentId,
+    client_id: ClientId,
+    deltas: Vec<OrderBookDelta>,
+) -> DataResponse {
+    DataResponse::BookDeltas(BookDeltasResponse::new(
+        request_id,
+        client_id,
+        instrument_id,
+        deltas,
+        None,
+        None,
+        UnixNanos::default(),
+        None,
+    ))
+}
+
+#[rstest]
+fn test_request_join_rebuilds_same_instrument_book_deltas_legs(
+    audusd_sim: CurrencyPair,
+    stub_msgbus: Rc<RefCell<MessageBus>>,
+    client_id: ClientId,
+) {
+    let _ = stub_msgbus;
+    let instrument_id = audusd_sim.id;
+    let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
+    let cache: Rc<RefCell<Cache>> = Rc::new(RefCell::new(Cache::default()));
+    // Past the leg ts_init values, so the join's bound-date clamping does not
+    // collapse the parent window to 0.
+    advance_test_clock_to(&clock, 10_000_000_000);
+    let mut data_engine = DataEngine::new(clock, cache, None);
+
+    let leg_a = UUID4::new();
+    let leg_b = UUID4::new();
+    let join_id = UUID4::new();
+
+    data_engine
+        .execute_request(RequestCommand::Join(RequestJoin::new(
+            vec![leg_a, leg_b],
+            None,
+            None,
+            join_id,
+            UnixNanos::default(),
+            None,
+            None,
+        )))
+        .unwrap();
+
+    let (parent_handler, parent_saver) = get_any_saving_handler::<BookDeltasResponse>(Some(
+        Ustr::from("same-instrument-deltas-parent"),
+    ));
+    msgbus::register_response_handler(&join_id, parent_handler);
+
+    data_engine.response(leg_book_deltas_response(
+        leg_a,
+        instrument_id,
+        client_id,
+        vec![delta_with_flag(
+            instrument_id,
+            1_000,
+            RecordFlag::F_LAST as u8,
+        )],
+    ));
+    data_engine.response(leg_book_deltas_response(
+        leg_b,
+        instrument_id,
+        client_id,
+        vec![delta_with_flag(
+            instrument_id,
+            2_000,
+            RecordFlag::F_LAST as u8,
+        )],
+    ));
+
+    let responses = parent_saver.get_messages();
+    assert_eq!(responses.len(), 1);
+    assert_eq!(responses[0].instrument_id, instrument_id);
+    assert_eq!(
+        responses[0]
+            .data
+            .iter()
+            .map(|delta| delta.ts_init.as_u64())
+            .collect::<Vec<_>>(),
+        vec![1_000, 2_000],
+    );
+    assert_eq!(data_engine.pending_join_request_count(), 0);
+}
+
+#[rstest]
+fn test_request_join_mixed_instrument_book_deltas_cleans_up_join_staging(
+    audusd_sim: CurrencyPair,
+    gbpusd_sim: CurrencyPair,
+    stub_msgbus: Rc<RefCell<MessageBus>>,
+    client_id: ClientId,
+) {
+    let _ = stub_msgbus;
+    let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
+    let cache: Rc<RefCell<Cache>> = Rc::new(RefCell::new(Cache::default()));
+    // Past the leg ts_init values, so the deltas survive the parent-window trim and
+    // reach the response handler when the rebuild is not refused.
+    advance_test_clock_to(&clock, 10_000_000_000);
+    let mut data_engine = DataEngine::new(clock, cache, None);
+
+    let leg_a = UUID4::new();
+    let leg_b = UUID4::new();
+    let join_id = UUID4::new();
+
+    data_engine
+        .execute_request(RequestCommand::Join(RequestJoin::new(
+            vec![leg_a, leg_b],
+            None,
+            None,
+            join_id,
+            UnixNanos::default(),
+            None,
+            None,
+        )))
+        .unwrap();
+
+    let (parent_handler, parent_saver) = get_any_saving_handler::<BookDeltasResponse>(Some(
+        Ustr::from("mixed-instrument-deltas-parent"),
+    ));
+    msgbus::register_response_handler(&join_id, parent_handler);
+
+    data_engine.response(leg_book_deltas_response(
+        leg_a,
+        audusd_sim.id,
+        client_id,
+        vec![delta_with_flag(
+            audusd_sim.id,
+            1_000,
+            RecordFlag::F_LAST as u8,
+        )],
+    ));
+    data_engine.response(leg_book_deltas_response(
+        leg_b,
+        gbpusd_sim.id,
+        client_id,
+        vec![delta_with_flag(
+            gbpusd_sim.id,
+            2_000,
+            RecordFlag::F_LAST as u8,
+        )],
+    ));
+
+    assert!(
+        parent_saver.get_messages().is_empty(),
+        "mixed-instrument rebuild must not emit a parent response",
+    );
+    assert_eq!(
+        data_engine.request_pipeline_count(),
+        0,
+        "pipeline state must be cleared after a failed rebuild",
+    );
+    assert_eq!(
+        data_engine.pending_join_request_count(),
+        0,
+        "pending join must be cleared after a failed rebuild to prevent leaks",
+    );
+}
+
 #[rstest]
 fn test_request_join_mixed_variants_cleans_up_join_staging(
     audusd_sim: CurrencyPair,

crates/model/src/data/deltas.rs

@@ -61,7 +61,8 @@ impl OrderBookDeltas {
     ///
     /// # Panics
     ///
-    /// Panics if `deltas` is empty.
+    /// Panics if `deltas` is empty or contains an instrument ID that does not match
+    /// `instrument_id`.
     #[must_use]
     pub fn new(instrument_id: InstrumentId, deltas: Vec<OrderBookDelta>) -> Self {
         Self::new_checked(instrument_id, deltas).expect(FAILED)
@@ -71,7 +72,8 @@ impl OrderBookDeltas {
     ///
     /// # Errors
     ///
-    /// Returns an error if `deltas` is empty.
+    /// Returns an error if `deltas` is empty or contains an instrument ID that does not match
+    /// `instrument_id`.
     ///
     /// # Notes
     ///
@@ -85,6 +87,24 @@ impl OrderBookDeltas {
         deltas: Vec<OrderBookDelta>,
     ) -> anyhow::Result<Self> {
         check_predicate_true(!deltas.is_empty(), "`deltas` cannot be empty")?;
+
+        let mismatch = deltas.iter().enumerate().find(|(_, delta)| {
+            instrument_id != delta.instrument_id
+                && (instrument_id.symbol.as_str() != delta.instrument_id.symbol.as_str()
+                    || instrument_id.venue.as_str() != delta.instrument_id.venue.as_str())
+        });
+
+        if let Some((index, delta)) = mismatch {
+            check_predicate_true(
+                false,
+                &format!(
+                    "`deltas` instrument IDs must match `instrument_id` {instrument_id}, but \
+                     delta at index {index} of {} has {}",
+                    deltas.len(),
+                    delta.instrument_id,
+                ),
+            )?;
+        }
         let last = deltas.last().expect("deltas not empty");
         let flags = last.flags;
         let sequence = last.sequence;
@@ -296,6 +316,40 @@ mod tests {
         assert_eq!(deltas.deltas.len(), 1);
     }
 
+    #[rstest]
+    fn test_order_book_deltas_new_checked_accepts_homogeneous_deltas() {
+        let instrument_id = InstrumentId::from("EURUSD.SIM");
+        let delta1 = create_test_delta();
+        let mut delta2 = create_test_delta();
+        delta2.sequence = 124;
+
+        let result = OrderBookDeltas::new_checked(instrument_id, vec![delta1, delta2]);
+
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap().deltas.len(), 2);
+    }
+
+    #[rstest]
+    #[case::first(0)]
+    #[case::later(1)]
+    fn test_order_book_deltas_new_checked_rejects_mismatched_instrument(
+        #[case] mismatch_index: usize,
+    ) {
+        let instrument_id = InstrumentId::from("EURUSD.SIM");
+        let mut deltas = vec![create_test_delta(), create_test_delta()];
+        deltas[mismatch_index].instrument_id = InstrumentId::from("GBPUSD.SIM");
+
+        let result = OrderBookDeltas::new_checked(instrument_id, deltas);
+
+        assert_eq!(
+            result.unwrap_err().to_string(),
+            format!(
+                "`deltas` instrument IDs must match `instrument_id` EURUSD.SIM, but delta at \
+                 index {mismatch_index} of 2 has GBPUSD.SIM"
+            )
+        );
+    }
+
     #[rstest]
     fn test_order_book_deltas_new_checked_empty_deltas() {
         let instrument_id = InstrumentId::from("EURUSD.SIM");
@@ -481,8 +535,8 @@ mod tests {
 
     #[rstest]
     fn test_order_book_deltas_single_delta() {
-        let instrument_id = InstrumentId::from("BTCUSD.CRYPTO");
         let delta = create_test_delta();
+        let instrument_id = delta.instrument_id;
 
         let deltas = OrderBookDeltas::new(instrument_id, vec![delta]);