Measure position-reconciliation grace on the monotonic clock (#4366)

The continuous position-consistency check has a grace window: when the node
recently observed local activity on a position, check_position_discrepancy
suppresses a cache-vs-venue discrepancy for position_check_threshold_ns,
giving the local execution pipeline a moment to settle before synthesizing
an EXTERNAL reconciliation position.

That window is inherently a real-time duration — you want N real seconds of
cover. But it was measured on self.clock: last_activity was stamped from
self.clock (and before that carried from the venue event's ts_event), and
the threshold was compared against a self.clock delta. self.clock is not
guaranteed to advance at wall rate — a live node can be given a non-wall
clock through a clock factory (e.g. a simulated or replay venue driving the
node clock). When that clock runs fast or is anchored to a different epoch,
the delta is not real elapsed time and the threshold no longer means real
cover: an accelerated clock shrinks the window by its speed and the grace
can collapse before the first reconciliation poll.

Measure the grace on the monotonic dst::time clock instead: stamp
last_activity from dst::time::Instant and compare via elapsed() against the
threshold as a Duration. The window is now real elapsed time at any clock
speed or epoch. This also aligns the grace with the reconciliation loop
itself, which already schedules its polls on dst::time (LiveNode::run) — the
grace was the only part measuring durations on self.clock. The continuous
position check is a live-only feature and never runs under backtest, so a
monotonic grace clock has no effect on backtest determinism.

record_position_activity no longer takes a ts_event parameter and no longer
reads self.clock; position_local_activity now holds monotonic instants.

Add a regression test that records activity, races self.clock ~954 days
ahead in a single step, and asserts the grace still holds; it fails against
a self.clock-based grace and passes with the monotonic grace.

Coded by an LLM.

Author: folknor

crates/live/src/execution/manager.rs

@@ -18,14 +18,15 @@
 //! This module provides the execution manager for reconciling execution state between
 //! the local cache and connected venues, as well as purging old state during live trading.
 
-use std::{cell::RefCell, fmt::Debug, rc::Rc, str::FromStr, sync::LazyLock};
+use std::{cell::RefCell, fmt::Debug, rc::Rc, str::FromStr, sync::LazyLock, time::Duration};
 
 use indexmap::{IndexMap, IndexSet};
 use nautilus_common::{
     cache::Cache,
     clients::ExecutionClient,
     clock::Clock,
     enums::{LogColor, LogLevel},
+    live::dst,
     log_info,
     messages::{
         ExecutionReport,
@@ -271,7 +272,8 @@ pub struct ExecutionManager {
     recon_check_retries: IndexMap<ClientOrderId, u32>,
     ts_last_query: IndexMap<ClientOrderId, UnixNanos>,
     order_local_activity_ns: IndexMap<ClientOrderId, UnixNanos>,
-    position_local_activity_ns: IndexMap<InstrumentAccountKey, UnixNanos>,
+    // Monotonic (`dst::time`) instants, not `self.clock`; see `record_position_activity`.
+    position_local_activity: IndexMap<InstrumentAccountKey, dst::time::Instant>,
     position_recon_retries: IndexMap<InstrumentAccountKey, u32>,
     recent_fills_cache: IndexMap<TradeId, UnixNanos>,
 }
@@ -305,7 +307,7 @@ impl ExecutionManager {
             recon_check_retries: IndexMap::new(),
             ts_last_query: IndexMap::new(),
             order_local_activity_ns: IndexMap::new(),
-            position_local_activity_ns: IndexMap::new(),
+            position_local_activity: IndexMap::new(),
             position_recon_retries: IndexMap::new(),
             recent_fills_cache: IndexMap::new(),
         }
@@ -1439,14 +1441,21 @@ impl ExecutionManager {
     }
 
     /// Records position activity for reconciliation tracking, scoped per (instrument, account).
-    pub fn record_position_activity(
-        &mut self,
-        instrument_id: InstrumentId,
-        account_id: AccountId,
-        ts_event: UnixNanos,
-    ) {
-        self.position_local_activity_ns
-            .insert((instrument_id, account_id), ts_event);
+    ///
+    /// The activity is stamped from the monotonic `dst::time` clock (real elapsed
+    /// time), **not** from `self.clock` and **not** from the venue event's
+    /// `ts_event`. The position-discrepancy grace is a real-time settling window:
+    /// give the local pipeline a moment to catch up before flagging a
+    /// cache-vs-venue gap. That is inherently wall/monotonic time; you want N
+    /// real seconds of cover regardless of the trading clock's epoch or speed.
+    /// `self.clock` can be driven off wall time (e.g. an accelerated simulated
+    /// venue), which would shrink the window by the clock's speed; the venue
+    /// `ts_event` lives on yet another axis. Measuring against the same monotonic
+    /// clock the reconciliation loop already schedules on keeps the grace honest.
+    /// See `check_position_discrepancy`.
+    pub fn record_position_activity(&mut self, instrument_id: InstrumentId, account_id: AccountId) {
+        self.position_local_activity
+            .insert((instrument_id, account_id), dst::time::Instant::now());
     }
 
     /// Returns the current position-reconciliation retry count for the given
@@ -1498,11 +1507,7 @@ impl ExecutionManager {
                 if let Some(coid) = client_order_id {
                     self.record_local_activity(coid);
                 }
-                self.record_position_activity(
-                    fill_report.instrument_id,
-                    fill_report.account_id,
-                    fill_report.ts_event,
-                );
+                self.record_position_activity(fill_report.instrument_id, fill_report.account_id);
             }
             ExecutionReport::OrderWithFills(order_report, fills) => {
                 if let Some(client_order_id) = &order_report.client_order_id
@@ -1522,15 +1527,13 @@ impl ExecutionManager {
                     self.record_position_activity(
                         fill_report.instrument_id,
                         fill_report.account_id,
-                        fill_report.ts_event,
                     );
                 }
             }
             ExecutionReport::Position(position_report) => {
                 self.record_position_activity(
                     position_report.instrument_id,
                     position_report.account_id,
-                    position_report.ts_last,
                 );
             }
             ExecutionReport::MassStatus(_) => {
@@ -1737,8 +1740,10 @@ impl ExecutionManager {
 
         let ts_now = self.clock.borrow().timestamp_ns();
 
-        if let Some(&last_activity) = self.position_local_activity_ns.get(&key)
-            && (ts_now - last_activity) < self.config.position_check_threshold_ns
+        // Grace window measured on the monotonic `dst::time` clock; see `record_position_activity`.
+        if let Some(&last_activity) = self.position_local_activity.get(&key)
+            && last_activity.elapsed()
+                < Duration::from_nanos(self.config.position_check_threshold_ns)
         {
             log::debug!(
                 "Skipping position reconciliation for {instrument_id}: recent activity within threshold"

crates/live/src/node/mod.rs

@@ -1166,7 +1166,6 @@ impl LiveNode {
                                     self.exec_manager.record_position_activity(
                                         fill.instrument_id,
                                         fill.account_id,
-                                        fill.ts_event,
                                     );
                                     self.exec_manager.mark_fill_processed(fill.trade_id);
                                 }
@@ -1406,11 +1405,8 @@ impl LiveNode {
             self.exec_manager
                 .record_local_activity(event.client_order_id());
             if let OrderEventAny::Filled(fill) = event {
-                self.exec_manager.record_position_activity(
-                    fill.instrument_id,
-                    fill.account_id,
-                    fill.ts_event,
-                );
+                self.exec_manager
+                    .record_position_activity(fill.instrument_id, fill.account_id);
                 self.exec_manager.mark_fill_processed(fill.trade_id);
             }
             self.kernel.exec_engine.borrow_mut().process(event);

crates/live/tests/manager.rs

@@ -8110,9 +8110,10 @@ async fn test_position_check_activity_throttle_independent_per_account() {
     ctx.add_position(&pos_b);
 
     // Simulate recent activity on account A only: B must remain unthrottled.
-    let ts_now = ctx.clock.borrow().get_time_ns();
+    // Activity is stamped from the monotonic `dst::time` clock inside
+    // `record_position_activity`, so no explicit timestamp is passed.
     ctx.manager
-        .record_position_activity(instrument_id, account_a, ts_now);
+        .record_position_activity(instrument_id, account_a);
 
     // No venue report for B: treated as flat, so a discrepancy.
     let mock_client = MockPositionExecutionClient::new(vec![], vec![]);
@@ -8134,6 +8135,86 @@ async fn test_position_check_activity_throttle_independent_per_account() {
     );
 }
 
+#[tokio::test]
+async fn test_position_check_grace_survives_accelerated_trading_clock() {
+    // The position-reconciliation grace is measured on the monotonic clock, not
+    // `self.clock` (see `record_position_activity`). Here we record local activity,
+    // then jump `self.clock` ~954 days forward in a single step, standing in for a
+    // trading clock that has raced ahead while only milliseconds of real time have
+    // elapsed, and assert the grace still suppresses the discrepancy.
+    let config = ExecutionManagerConfig {
+        position_check_retries: 3,
+        position_check_threshold_ns: 60_000_000_000, // 60s of real cover
+        ..Default::default()
+    };
+    let mut ctx = TestContext::with_config(config);
+    let instrument = test_instrument();
+    let instrument_id = instrument.id();
+    let account = AccountId::from("BINANCE-A");
+
+    ctx.add_instrument(instrument.clone());
+    ctx.add_margin_account(account);
+
+    // Observe a fill: records local activity on the monotonic clock. The venue
+    // event timestamps do not feed the grace and are left arbitrary.
+    let ts_event = UnixNanos::from(1_000_000_000);
+    let fill_report = FillReport::new(
+        account,
+        instrument_id,
+        VenueOrderId::from("V-1"),
+        TradeId::from("T-1"),
+        OrderSide::Buy,
+        Quantity::from("0.06"),
+        Price::from("3000.00"),
+        Money::new(0.0, Currency::USDT()),
+        LiquiditySide::Taker,
+        Some(ClientOrderId::from("O-1")),
+        None,     // venue_position_id
+        ts_event, // ts_event
+        ts_event, // ts_init
+        None,     // report_id
+    );
+    ctx.manager
+        .observe_execution_report(&ExecutionReport::Fill(Box::new(fill_report)));
+
+    // Race the trading clock ~954 days ahead. A `self.clock`-based grace would now
+    // read "activity was 954 days ago" and fire; the monotonic grace must not.
+    let accelerated_jump_ns: u64 = 954 * 86_400 * 1_000_000_000; // 954 days in ns
+    ctx.advance_time(accelerated_jump_ns);
+
+    // Cache is flat but the venue reports a 0.06 long: a genuine discrepancy. The
+    // only thing between it and a synthesized EXTERNAL position is the grace.
+    let venue_report = PositionStatusReport::new(
+        account,
+        instrument_id,
+        PositionSideSpecified::Long,
+        Quantity::from("0.06"),
+        UnixNanos::from(accelerated_jump_ns),
+        UnixNanos::from(accelerated_jump_ns),
+        None, // report_id
+        None, // venue_position_id
+        Some(dec!(3000.00)),
+    );
+    let mock_client = MockPositionExecutionClient::new(vec![], vec![venue_report]);
+    let clients: Vec<&dyn ExecutionClient> = vec![&mock_client];
+
+    let events = ctx.manager.check_positions_consistency(&clients).await;
+
+    assert!(
+        events.is_empty(),
+        "grace must survive an accelerated trading clock (it is measured on the \
+         monotonic clock); got {} reconciliation event(s), the grace regressed \
+         to self.clock",
+        events.len(),
+    );
+    assert_eq!(
+        ctx.manager
+            .position_recon_retry_count(&(instrument_id, account)),
+        0,
+        "grace path must return before the retry counter is touched",
+    );
+}
+
 #[tokio::test]
 async fn test_check_positions_consistency_processes_only_discrepant_account() {
     // With cache and venue agreeing on account A and disagreeing on account B,