Rebase missing-order reconciliation gate onto monotonic receipt time

`handle_missing_order` had a recency gate comparing the order's venue
`ts_last` against `self.clock` now - a cross-axis compare that #4376
hardened against underflow but left on the trading clock. Under a
custom live/sandbox clock factory the trading clock is not wall-paced
(it can run accelerated or sit on a foreign epoch), so that window did
not measure the real settling time it was meant to, and a corrupt
far-future `ts_last` could stall the order's reconciliation.

Drop the venue-`ts_last` gate. The missing-order settling window is now
solely the monotonic `order_local_activity` recency gate (the
`RecencyMap` from the recency-map consolidation), which measures real
receipt-time elapsed at any clock speed. This also removes the
warn-and-defer arm #4376 added for a far-future `ts_last`: with the
cross-axis gate gone there is no longer a failure mode to warn about -
the order simply reconciles once the real grace expires.

Audit of the remaining live timers for the same class: the cache-purge
intervals are deliberately left on the ExecutionEngine clock-timer path
so they stay controlled by the injected Clock for custom-clock callers;
a comment and a conversion test now pin that. Data-engine, order
emulator, and core timing stay domain/deterministic, and
`snapshot_positions_interval_secs` is left as-is (no live monotonic
replacement).

The missing-order test becomes a differential paused-time case: with a
far-future venue `ts_last` present throughout, recent local activity
defers, and after the monotonic grace expires reconciliation proceeds.

Coded by an LLM.

Author: folknor

crates/live/src/execution/manager.rs

@@ -1670,32 +1670,9 @@ impl ExecutionManager {
             return events;
         };
 
-        let ts_now = self.clock.borrow().timestamp_ns();
-        let ts_last = order.ts_last();
-
-        // Domain-time recency gate: ts_last and ts_now are both domain
-        // timestamps, so this stays on self.clock (it is not a real-time
-        // settling window).
-        match ts_now.duration_since(&ts_last) {
-            // Within the recency threshold: order is genuinely too recent, defer.
-            Some(elapsed_ns) if elapsed_ns < self.config.open_check_threshold_ns => {
-                return events;
-            }
-            // Old enough: fall through to the reconciliation checks below.
-            Some(_) => {}
-            // ts_last is ahead of ts_now - impossible under a sane clock.
-            // A corrupted far-future ts_last (for example a double-scaled
-            // timestamp) would otherwise stall this order's reconciliation
-            // forever with no signal, so warn before deferring.
-            None => {
-                log::warn!(
-                    "Order {client_order_id} has venue ts_last {ts_last} ahead of local ts_now {ts_now}; deferring reconciliation"
-                );
-                return events;
-            }
-        }
-
-        // Check local activity threshold
+        // Recent local activity is the real-time settling window for missing
+        // orders. Venue/domain timestamps can be ahead of the trading clock and
+        // must not stall reconciliation.
         if self.order_local_activity.within(
             &client_order_id,
             Duration::from_nanos(self.config.open_check_threshold_ns),

crates/live/src/node/config.rs

@@ -501,6 +501,10 @@ impl From<LiveExecEngineConfig> for ExecutionEngineConfig {
             allow_overfills: config.allow_overfills,
             filter_unclaimed_external_orders: config.filter_unclaimed_external_orders,
             external_clients: config.external_clients,
+            // Keep purge intervals on the ExecutionEngine clock-timer path.
+            // LiveNode also dispatches purge checks from its maintenance loop,
+            // but engine timers must remain controlled by the injected Clock
+            // for callers using a custom live/sandbox clock factory.
             purge_closed_orders_interval_mins: config.purge_closed_orders_interval_mins,
             purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
             purge_closed_positions_interval_mins: config.purge_closed_positions_interval_mins,
@@ -1285,6 +1289,12 @@ mod tests {
             load_cache: false,
             snapshot_positions_interval_secs: Some(30.0),
             filter_unclaimed_external_orders: true,
+            purge_closed_orders_interval_mins: Some(5),
+            purge_closed_orders_buffer_mins: Some(1),
+            purge_closed_positions_interval_mins: Some(10),
+            purge_closed_positions_buffer_mins: Some(2),
+            purge_account_events_interval_mins: Some(15),
+            purge_account_events_lookback_mins: Some(3),
             ..Default::default()
         };
 
@@ -1293,6 +1303,12 @@ mod tests {
         assert!(!converted.load_cache);
         assert_eq!(converted.snapshot_positions_interval_secs, Some(30.0));
         assert!(converted.filter_unclaimed_external_orders);
+        assert_eq!(converted.purge_closed_orders_interval_mins, Some(5));
+        assert_eq!(converted.purge_closed_orders_buffer_mins, Some(1));
+        assert_eq!(converted.purge_closed_positions_interval_mins, Some(10));
+        assert_eq!(converted.purge_closed_positions_buffer_mins, Some(2));
+        assert_eq!(converted.purge_account_events_interval_mins, Some(15));
+        assert_eq!(converted.purge_account_events_lookback_mins, Some(3));
     }
 
     #[rstest]

crates/live/tests/manager.rs

@@ -7617,31 +7617,39 @@ async fn test_check_open_orders_submitted_missing_at_venue_generates_rejected()
     }
 }
 
-#[rstest]
-#[tokio::test]
-async fn test_check_open_orders_defers_when_venue_ts_last_is_ahead() {
-    // A corrupted far-future ts_last must defer reconciliation without
-    // rejecting or panicking.
+#[cfg_attr(
+    not(all(feature = "simulation", madsim)),
+    tokio::test(start_paused = true)
+)]
+#[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
+async fn test_check_open_orders_missing_gate_uses_local_activity_not_venue_ts_last() {
+    // A corrupted far-future ts_last must not stall missing-order reconciliation
+    // after the local activity grace expires.
     let config = ExecutionManagerConfig {
-        open_check_threshold_ns: 0,
+        open_check_threshold_ns: 200_000_000,
         open_check_missing_retries: 1,
         open_check_open_only: false,
         ..Default::default()
     };
     let mut ctx = TestContext::with_config(config);
     ctx.add_instrument(test_instrument());
 
-    let mut order = create_limit_order(
+    let order = create_limit_order(
         "O-AHEAD",
         test_instrument_id(),
         OrderSide::Buy,
         "10.0",
         "100.0",
     );
     let submitted = TestOrderEventStubs::submitted(&order, test_account_id());
-    order.apply(submitted).unwrap();
+    ctx.add_order(order);
+    let order = ctx.cache.borrow_mut().update_order(&submitted).unwrap();
 
-    let future_ts = UnixNanos::from(10_000_000_000);
+    let future_ts = ctx
+        .clock
+        .borrow()
+        .timestamp_ns()
+        .saturating_add_ns(10_000_000_000_u64);
     let accepted = OrderEventAny::Accepted(
         OrderAcceptedSpec::builder()
             .trader_id(order.trader_id())
@@ -7654,8 +7662,9 @@ async fn test_check_open_orders_defers_when_venue_ts_last_is_ahead() {
             .ts_init(future_ts)
             .build(),
     );
-    order.apply(accepted).unwrap();
-    ctx.add_order(order);
+    let client_order_id = order.client_order_id();
+    ctx.cache.borrow_mut().update_order(&accepted).unwrap();
+    ctx.manager.record_local_activity(client_order_id);
 
     let mock_client = MockExecutionClient::new(vec![]);
     let clients: Vec<&dyn ExecutionClient> = vec![&mock_client];
@@ -7664,7 +7673,17 @@ async fn test_check_open_orders_defers_when_venue_ts_last_is_ahead() {
 
     assert!(
         events.is_empty(),
-        "a far-future venue ts_last must defer reconciliation, not reject or panic",
+        "recent local activity should defer reconciliation",
+    );
+
+    advance_clock(dst::time::Duration::from_millis(250)).await;
+
+    let events = ctx.manager.check_open_orders(&clients).await;
+
+    assert_eq!(events.len(), 1);
+    assert!(
+        matches!(&events[0], OrderEventAny::Rejected(rejected) if rejected.client_order_id == client_order_id),
+        "far-future venue ts_last must not keep deferring reconciliation",
     );
 }