nautilus_dydx/execution/
block_time.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Block time monitoring for dYdX short-term order expiration estimation.
17//!
18//! This module provides [`BlockTimeMonitor`], a component that tracks rolling average
19//! block times from WebSocket data to enable accurate estimation of short-term order
20//! expiration in wall-clock time.
21//!
22//! # Overview
23//!
24//! dYdX short-term orders expire by block height (typically 20 blocks). Without knowing
25//! the actual block time, it's impossible to estimate when an order will expire in
26//! wall-clock time. This monitor captures block timestamps from WebSocket updates and
27
28use std::{
29    collections::VecDeque,
30    sync::atomic::{AtomicU64, Ordering},
31};
32
33use jiff::Timestamp;
34use parking_lot::RwLock;
35
36/// Default rolling window size for block time averaging.
37///
38/// 100 blocks at ~500ms/block = ~50 seconds of data.
39pub const DEFAULT_BLOCK_TIME_WINDOW_SIZE: usize = 100;
40
41/// Default block time in milliseconds (dYdX mainnet ~500ms).
42///
43/// Used as fallback when insufficient samples are available.
44pub const DEFAULT_BLOCK_TIME_MS: u64 = 500;
45
46/// Minimum number of samples required before trusting the rolling average.
47///
48/// Below this threshold, [`BlockTimeMonitor::estimated_seconds_per_block`] returns `None`
49/// and [`BlockTimeMonitor::seconds_per_block_or_default`] uses the default value.
50pub const MIN_SAMPLES_FOR_ESTIMATE: usize = 5;
51
52/// Minimum valid block time in milliseconds.
53///
54/// Any calculated block time below this threshold is considered invalid
55/// (likely due to clock skew, data corruption, or integer division truncation).
56/// When detected, the monitor falls back to the default block time.
57pub const MIN_VALID_BLOCK_TIME_MS: f64 = 50.0;
58
59/// Internal rolling window buffer for block samples.
60///
61/// Uses a `VecDeque` for O(1) push/pop operations with bounded memory.
62/// Includes deduplication to skip repeated block heights during rapid replays.
63#[derive(Debug)]
64struct BlockTimeWindow {
65    /// Circular buffer of (height, timestamp) samples.
66    samples: VecDeque<(u64, Timestamp)>,
67    /// Maximum capacity of the window.
68    capacity: usize,
69    /// Last recorded block height for deduplication.
70    last_height: Option<u64>,
71}
72
73impl BlockTimeWindow {
74    /// Creates a new window with specified capacity.
75    fn new(capacity: usize) -> Self {
76        Self {
77            samples: VecDeque::with_capacity(capacity),
78            capacity,
79            last_height: None,
80        }
81    }
82
83    /// Records a new block sample.
84    ///
85    /// Skips duplicate block heights to prevent redundant entries during rapid
86    /// block replays where the same height may be reported multiple times.
87    fn record(&mut self, height: u64, time: Timestamp) {
88        // Skip duplicate heights (rapid replays often repeat same block)
89        if self.last_height == Some(height) {
90            return;
91        }
92        self.last_height = Some(height);
93
94        // Maintain bounded size: remove oldest when at capacity
95        if self.samples.len() >= self.capacity {
96            self.samples.pop_front();
97        }
98        self.samples.push_back((height, time));
99    }
100
101    /// Returns the number of samples in the window.
102    fn sample_count(&self) -> usize {
103        self.samples.len()
104    }
105
106    /// Computes the average seconds per block from the rolling window.
107    ///
108    /// Returns `None` if fewer than [`MIN_SAMPLES_FOR_ESTIMATE`] samples are available.
109    fn average_seconds_per_block(&self) -> Option<f64> {
110        let sample_count = self.sample_count();
111        if sample_count < MIN_SAMPLES_FOR_ESTIMATE {
112            return None;
113        }
114
115        // Sort samples by height to compute deltas between consecutive blocks
116        let mut sorted: Vec<_> = self.samples.iter().copied().collect();
117        sorted.sort_by_key(|(height, _)| *height);
118
119        let mut total_delta_ms: i64 = 0;
120        let mut delta_count: usize = 0;
121
122        for window in sorted.windows(2) {
123            let (h1, t1) = &window[0];
124            let (h2, t2) = &window[1];
125
126            // Skip duplicate heights (shouldn't happen with deduplication, but be safe)
127            let height_diff = h2.saturating_sub(*h1);
128            if height_diff == 0 {
129                continue;
130            }
131
132            let Ok(time_diff_ms) = i64::try_from(t1.duration_until(*t2).as_millis()) else {
133                continue;
134            };
135
136            if time_diff_ms <= 0 {
137                continue; // Invalid time difference (clock skew or reorg)
138            }
139
140            // Normalize time difference by height difference for multi-block gaps
141            let ms_per_block = time_diff_ms / height_diff as i64;
142            total_delta_ms += ms_per_block;
143            delta_count += 1;
144        }
145
146        if delta_count == 0 {
147            return None;
148        }
149
150        let avg_ms = total_delta_ms as f64 / delta_count as f64;
151
152        // Validate: block time must be at least MIN_VALID_BLOCK_TIME_MS
153        // to avoid division issues with unrealistically small values
154        if avg_ms < MIN_VALID_BLOCK_TIME_MS {
155            return None;
156        }
157
158        Some(avg_ms / 1000.0)
159    }
160}
161
162/// Monitors block times and provides estimation utilities for order expiration.
163///
164/// Thread-safe component that tracks rolling average block times from WebSocket data.
165/// Uses atomic operations for the hot path (height reads) and a read-write lock for
166/// less frequent operations (window updates, time estimation).
167#[derive(Debug)]
168pub struct BlockTimeMonitor {
169    /// Current block height (atomic for fast reads on hot path).
170    current_height: AtomicU64,
171    /// Current block timestamp.
172    current_time: RwLock<Option<Timestamp>>,
173    /// Rolling window for block time averaging.
174    window: RwLock<BlockTimeWindow>,
175}
176
177impl Default for BlockTimeMonitor {
178    fn default() -> Self {
179        Self::new()
180    }
181}
182
183impl BlockTimeMonitor {
184    /// Creates a new [`BlockTimeMonitor`] with default window size.
185    #[must_use]
186    pub fn new() -> Self {
187        Self::with_window_size(DEFAULT_BLOCK_TIME_WINDOW_SIZE)
188    }
189
190    /// Creates a new [`BlockTimeMonitor`] with custom window size.
191    #[must_use]
192    pub fn with_window_size(window_size: usize) -> Self {
193        Self {
194            current_height: AtomicU64::new(0),
195            current_time: RwLock::new(None),
196            window: RwLock::new(BlockTimeWindow::new(window_size)),
197        }
198    }
199
200    /// Records a new block from WebSocket data.
201    ///
202    /// Should be called whenever a block height update is received.
203    /// Updates the current height atomically and adds the sample to the rolling window.
204    pub fn record_block(&self, height: u64, time: Timestamp) {
205        // Update current height atomically (hot path)
206        self.current_height.store(height, Ordering::Release);
207
208        // Update current time
209        *self.current_time.write() = Some(time);
210
211        // Add to rolling window
212        self.window.write().record(height, time);
213    }
214
215    /// Returns the current block height.
216    ///
217    /// This is a fast, lock-free read suitable for hot paths.
218    #[must_use]
219    pub fn current_block_height(&self) -> u64 {
220        self.current_height.load(Ordering::Acquire)
221    }
222
223    /// Returns the timestamp of the most recent block.
224    #[must_use]
225    pub fn current_block_time(&self) -> Option<Timestamp> {
226        *self.current_time.read()
227    }
228
229    /// Returns the estimated seconds per block based on rolling average.
230    ///
231    /// Returns `None` if fewer than [`MIN_SAMPLES_FOR_ESTIMATE`] samples are available.
232    #[must_use]
233    pub fn estimated_seconds_per_block(&self) -> Option<f64> {
234        self.window.read().average_seconds_per_block()
235    }
236
237    /// Returns estimated seconds per block, falling back to default if unavailable.
238    ///
239    /// Uses [`DEFAULT_BLOCK_TIME_MS`] (500ms) when insufficient samples.
240    #[must_use]
241    pub fn seconds_per_block_or_default(&self) -> f64 {
242        self.estimated_seconds_per_block()
243            .unwrap_or(DEFAULT_BLOCK_TIME_MS as f64 / 1000.0)
244    }
245
246    /// Estimates how many blocks will occur in the given duration.
247    ///
248    /// Uses the rolling average if available, otherwise falls back to default block time.
249    /// Result is capped at `u32::MAX` to prevent overflow from edge cases.
250    #[must_use]
251    pub fn estimate_blocks_for_duration(&self, duration_secs: f64) -> u32 {
252        let secs_per_block = self.seconds_per_block_or_default();
253        let blocks = (duration_secs / secs_per_block).ceil();
254        // Cap at u32::MAX to prevent overflow from infinity or very large values
255        blocks.min(f64::from(u32::MAX)) as u32
256    }
257
258    /// Estimates the wall-clock time when a specific block height will be reached.
259    ///
260    /// Returns `None` if:
261    /// - Insufficient samples for reliable estimation
262    /// - No current block time available
263    /// - Target block is in the past
264    #[must_use]
265    pub fn estimate_expiry_time(&self, expiry_block: u64) -> Option<Timestamp> {
266        let current_height = self.current_block_height();
267        let current_time = self.current_block_time()?;
268        let secs_per_block = self.estimated_seconds_per_block()?;
269
270        if expiry_block <= current_height {
271            // Block already passed
272            return None;
273        }
274
275        let blocks_remaining = expiry_block - current_height;
276        let seconds_remaining = blocks_remaining as f64 * secs_per_block;
277
278        Some(
279            current_time
280                + jiff::SignedDuration::from_millis((seconds_remaining * 1000.0).round() as i64),
281        )
282    }
283
284    /// Estimates remaining lifetime in seconds for an order expiring at the given block.
285    ///
286    /// Returns `None` if insufficient data or block already passed.
287    #[must_use]
288    pub fn estimate_remaining_lifetime_secs(&self, expiry_block: u64) -> Option<f64> {
289        let current_height = self.current_block_height();
290
291        if expiry_block <= current_height {
292            return Some(0.0);
293        }
294
295        let blocks_remaining = expiry_block - current_height;
296        let secs_per_block = self.estimated_seconds_per_block()?;
297
298        Some(blocks_remaining as f64 * secs_per_block)
299    }
300
301    /// Returns `true` if the monitor has enough samples for reliable estimation.
302    #[must_use]
303    pub fn is_ready(&self) -> bool {
304        self.window.read().sample_count() >= MIN_SAMPLES_FOR_ESTIMATE
305    }
306
307    /// Returns the number of samples collected in the rolling window.
308    #[must_use]
309    pub fn sample_count(&self) -> usize {
310        self.window.read().sample_count()
311    }
312}
313
314#[cfg(test)]
315mod tests {
316    use jiff::SignedDuration;
317    use rstest::rstest;
318
319    use super::*;
320
321    #[rstest]
322    fn test_new_monitor_not_ready() {
323        let monitor = BlockTimeMonitor::new();
324        assert!(!monitor.is_ready());
325        assert_eq!(monitor.current_block_height(), 0);
326        assert!(monitor.estimated_seconds_per_block().is_none());
327    }
328
329    #[rstest]
330    fn test_record_updates_height() {
331        let monitor = BlockTimeMonitor::new();
332        let now = Timestamp::now();
333
334        monitor.record_block(100, now);
335        assert_eq!(monitor.current_block_height(), 100);
336
337        monitor.record_block(101, now + SignedDuration::from_millis(500));
338        assert_eq!(monitor.current_block_height(), 101);
339    }
340
341    #[rstest]
342    fn test_seconds_per_block_or_default_before_ready() {
343        let monitor = BlockTimeMonitor::new();
344        let default = DEFAULT_BLOCK_TIME_MS as f64 / 1000.0;
345        assert!((monitor.seconds_per_block_or_default() - default).abs() < 0.001);
346    }
347
348    #[rstest]
349    fn test_becomes_ready_after_min_samples() {
350        let monitor = BlockTimeMonitor::new();
351        let mut time = Timestamp::now();
352
353        for i in 0..MIN_SAMPLES_FOR_ESTIMATE {
354            monitor.record_block(100 + i as u64, time);
355            time += SignedDuration::from_millis(500);
356        }
357
358        assert!(monitor.is_ready());
359    }
360
361    #[rstest]
362    fn test_average_block_time_calculation() {
363        let monitor = BlockTimeMonitor::new();
364        let mut time = Timestamp::now();
365        let block_time_ms = 500;
366
367        // Record enough samples with consistent 500ms block time
368        for i in 0..10 {
369            monitor.record_block(100 + i as u64, time);
370            time += SignedDuration::from_millis(block_time_ms);
371        }
372
373        let estimated = monitor.estimated_seconds_per_block().unwrap();
374        assert!(
375            (estimated - 0.5).abs() < 0.1,
376            "Expected ~0.5s, was {estimated}"
377        );
378    }
379
380    #[rstest]
381    fn test_estimate_blocks_for_duration() {
382        let monitor = BlockTimeMonitor::new();
383        let mut time = Timestamp::now();
384
385        // Set up with 500ms block time
386        for i in 0..10 {
387            monitor.record_block(100 + i as u64, time);
388            time += SignedDuration::from_millis(500);
389        }
390
391        // 10 seconds should be ~20 blocks at 500ms/block
392        let blocks = monitor.estimate_blocks_for_duration(10.0);
393        assert!((18..=22).contains(&blocks), "Expected ~20, was {blocks}");
394    }
395
396    #[rstest]
397    fn test_estimate_expiry_time() {
398        let monitor = BlockTimeMonitor::new();
399        let start_time = Timestamp::now();
400        let mut time = start_time;
401
402        // Set up with 500ms block time, ending at block 109
403        for i in 0..10 {
404            monitor.record_block(100 + i as u64, time);
405            time += SignedDuration::from_millis(500);
406        }
407
408        // After loop: current block is 109, current_block_time = time - 500ms
409        // Expiry at block 129 = 20 blocks from 109
410        let expiry_time = monitor.estimate_expiry_time(129).unwrap();
411        // Expected: current_block_time + (20 blocks * 500ms)
412        let current_block_time = time - SignedDuration::from_millis(500);
413        let expected = current_block_time + SignedDuration::from_millis(20 * 500);
414
415        let diff_ms = expiry_time.duration_since(expected).as_millis().abs();
416        assert!(diff_ms < 1000, "Expected ~{expected}, was {expiry_time}");
417    }
418
419    #[rstest]
420    fn test_estimate_expiry_time_past_block() {
421        let monitor = BlockTimeMonitor::new();
422        let time = Timestamp::now();
423
424        monitor.record_block(100, time);
425
426        // Block 50 is in the past
427        assert!(monitor.estimate_expiry_time(50).is_none());
428    }
429
430    #[rstest]
431    fn test_estimate_remaining_lifetime() {
432        let monitor = BlockTimeMonitor::new();
433        let mut time = Timestamp::now();
434
435        // Set up with 500ms block time
436        for i in 0..10 {
437            monitor.record_block(100 + i as u64, time);
438            time += SignedDuration::from_millis(500);
439        }
440
441        // Current height is 109, expiry at 129 (20 blocks)
442        let remaining = monitor.estimate_remaining_lifetime_secs(129).unwrap();
443        assert!(
444            (remaining - 10.0).abs() < 1.0,
445            "Expected ~10s, was {remaining}"
446        );
447    }
448
449    #[rstest]
450    fn test_circular_buffer_wraps() {
451        let monitor = BlockTimeMonitor::with_window_size(5);
452        let mut time = Timestamp::now();
453
454        // Record more samples than window size
455        for i in 0..10 {
456            monitor.record_block(100 + i as u64, time);
457            time += SignedDuration::from_millis(500);
458        }
459
460        // Should still have only 5 samples
461        assert_eq!(monitor.sample_count(), 5);
462        assert!(monitor.is_ready());
463    }
464
465    #[rstest]
466    fn test_handles_non_consecutive_blocks() {
467        let monitor = BlockTimeMonitor::new();
468        let mut time = Timestamp::now();
469
470        // Record blocks with a gap (100, 101, 102, 105, 106)
471        monitor.record_block(100, time);
472        time += SignedDuration::from_millis(500);
473        monitor.record_block(101, time);
474        time += SignedDuration::from_millis(500);
475        monitor.record_block(102, time);
476        time += SignedDuration::from_millis(1500); // Skip 3 blocks
477        monitor.record_block(105, time);
478        time += SignedDuration::from_millis(500);
479        monitor.record_block(106, time);
480
481        // Should still calculate a reasonable estimate
482        assert!(monitor.is_ready());
483        let estimated = monitor.estimated_seconds_per_block().unwrap();
484        // Expect ~500ms per block even with the gap
485        assert!(
486            (estimated - 0.5).abs() < 0.2,
487            "Expected ~0.5s, was {estimated}"
488        );
489    }
490
491    #[rstest]
492    fn test_deduplicates_same_block_height() {
493        let monitor = BlockTimeMonitor::with_window_size(10);
494        let time = Timestamp::now();
495
496        // Record same block height multiple times (rapid replay scenario)
497        monitor.record_block(100, time);
498        monitor.record_block(100, time + SignedDuration::from_millis(10));
499        monitor.record_block(100, time + SignedDuration::from_millis(20));
500        monitor.record_block(100, time + SignedDuration::from_millis(30));
501
502        // Should only have 1 sample due to deduplication
503        assert_eq!(monitor.sample_count(), 1);
504    }
505
506    #[rstest]
507    fn test_rapid_replay_bounded_memory() {
508        let monitor = BlockTimeMonitor::with_window_size(5);
509        let mut time = Timestamp::now();
510
511        // Simulate rapid replay: 1000 block updates
512        for i in 0..1000 {
513            monitor.record_block(100 + i as u64, time);
514            time += SignedDuration::from_millis(500);
515        }
516
517        // Buffer should never exceed capacity
518        assert_eq!(monitor.sample_count(), 5);
519        assert!(monitor.is_ready());
520
521        // Estimate should still be valid
522        let estimated = monitor.estimated_seconds_per_block().unwrap();
523        assert!(
524            (estimated - 0.5).abs() < 0.1,
525            "Expected ~0.5s, was {estimated}"
526        );
527    }
528
529    #[rstest]
530    fn test_rapid_replay_with_duplicate_heights() {
531        let monitor = BlockTimeMonitor::with_window_size(10);
532        let mut time = Timestamp::now();
533
534        // Simulate rapid replay with duplicates: each block reported 3 times
535        for block in 100..110 {
536            for _ in 0..3 {
537                monitor.record_block(block, time);
538                time += SignedDuration::from_millis(100);
539            }
540            time += SignedDuration::from_millis(200); // Actual block time ~500ms
541        }
542
543        // Should have exactly 10 samples (one per unique block)
544        assert_eq!(monitor.sample_count(), 10);
545    }
546
547    #[rstest]
548    fn test_rejects_unrealistically_small_block_times() {
549        let monitor = BlockTimeMonitor::with_window_size(10);
550        let time = Timestamp::now();
551
552        // Record blocks with extremely small time differences (1ms per block)
553        // This is unrealistic and should be rejected
554        for i in 0..10 {
555            monitor.record_block(100 + i as u64, time + SignedDuration::from_millis(i));
556        }
557
558        // Should have samples but estimated time should be None (below threshold)
559        assert!(monitor.is_ready());
560        assert!(
561            monitor.estimated_seconds_per_block().is_none(),
562            "Expected None for unrealistically small block times"
563        );
564
565        // Should fall back to default
566        let default = super::DEFAULT_BLOCK_TIME_MS as f64 / 1000.0;
567        assert!((monitor.seconds_per_block_or_default() - default).abs() < 0.001);
568    }
569
570    #[rstest]
571    fn test_estimate_blocks_handles_large_duration() {
572        let monitor = BlockTimeMonitor::new();
573        // Without samples, uses default (0.5s per block)
574
575        // Very large duration should not overflow
576        let blocks = monitor.estimate_blocks_for_duration(f64::MAX);
577        assert_eq!(blocks, u32::MAX);
578    }
579
580    #[rstest]
581    fn test_estimate_blocks_handles_zero_duration() {
582        let monitor = BlockTimeMonitor::new();
583
584        let blocks = monitor.estimate_blocks_for_duration(0.0);
585        assert_eq!(blocks, 0);
586    }
587}
nautilus_dydx/execution/block_time.rs

nautilus_dydx/execution/
block_time.rs
