evm_fork_cache/reactive/
ingress_timing.rs

1//! Optional observational clocks supplied by an application-owned transport.
2use alloy_primitives::B256;
3use std::{sync::Arc, time::Instant};
4
5/// One complete message observed before JSON-RPC decoding. Identity must name
6/// the concrete receipt, not merely the payload or block. No execution authority.
7#[derive(Clone, Copy, Debug, PartialEq, Eq)]
8pub struct TransportReceipt {
9    identity: B256,
10    received: Instant,
11    decoded: Instant,
12}
13impl TransportReceipt {
14    /// Reject inconsistent clocks instead of fabricating zero elapsed time.
15    pub fn new(identity: B256, received: Instant, decoded: Instant) -> Option<Self> {
16        (received <= decoded).then_some(Self {
17            identity,
18            received,
19            decoded,
20        })
21    }
22    /// Application-defined receipt identity, never a metric label.
23    pub const fn identity(self) -> B256 {
24        self.identity
25    }
26    /// Complete-message receive time, before envelope decoding.
27    pub const fn received(self) -> Instant {
28        self.received
29    }
30    /// JSON-RPC envelope decoded time, before pubsub delivery.
31    pub const fn decoded(self) -> Instant {
32        self.decoded
33    }
34}
35
36/// Lookup for the exact raw payload consumed by a pubsub receiver. Implementations
37/// must be bounded, nonblocking and observational. Missing, evicted or ambiguous
38/// duplicate matches must return None; never pick an arbitrary earlier receipt.
39/// The callback adds no transport read and must not mutate execution state.
40///
41/// A backend-neutral, one-shot in-memory observer. A real transport must maintain
42/// bounded, unambiguous associations; this fixture accepts one known delivery only.
43/// No exporter, metric labels, payload logging, or async runtime is required.
44///
45/// ```
46/// use evm_fork_cache::reactive::{TransportReceipt, TransportReceiptLookup};
47/// use alloy_primitives::B256;
48/// use std::{sync::{Arc, Mutex}, time::Instant};
49///
50/// let received = Instant::now();
51/// let receipt = TransportReceipt::new(B256::repeat_byte(1), received, Instant::now()).unwrap();
52/// let slot = Mutex::new(Some(receipt));
53/// let lookup: TransportReceiptLookup = Arc::new(move |raw| {
54///     if raw != r#"{"fixture":true}"# { return None; }
55///     slot.try_lock().ok()?.take()
56/// });
57/// let observed = lookup(r#"{"fixture":true}"#).unwrap();
58/// tracing::debug!(receipt_id = %observed.identity(),
59///     decode_ns = observed.decoded().duration_since(observed.received()).as_nanos(),
60///     "transport receipt observed");
61/// assert!(lookup(r#"{"fixture":true}"#).is_none());
62/// ```
63pub type TransportReceiptLookup = Arc<dyn Fn(&str) -> Option<TransportReceipt> + Send + Sync>;
64
65/// Timing preserved with a canonical subscriber input through its bounded queue.
66#[derive(Clone, Copy, Debug, PartialEq, Eq)]
67pub struct CanonicalIngressTiming {
68    receipt: TransportReceipt,
69    dequeued: Instant,
70    decoded: Instant,
71}
72impl CanonicalIngressTiming {
73    /// Bind source receipt to the actual dequeue and typed payload decode.
74    pub fn new(receipt: TransportReceipt, dequeued: Instant, decoded: Instant) -> Option<Self> {
75        (receipt.decoded <= dequeued && dequeued <= decoded).then_some(Self {
76            receipt,
77            dequeued,
78            decoded,
79        })
80    }
81    /// Concrete transport receipt.
82    pub const fn receipt(self) -> TransportReceipt {
83        self.receipt
84    }
85    /// Raw subscription notification dequeue time.
86    pub const fn dequeued(self) -> Instant {
87        self.dequeued
88    }
89    /// Typed domain payload decode completion time.
90    pub const fn decoded(self) -> Instant {
91        self.decoded
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use std::time::Duration;
99    #[test]
100    fn queue_age_is_retained_and_invalid_clocks_remain_missing() {
101        let received = Instant::now();
102        let envelope = received + Duration::from_micros(5);
103        let dequeue = envelope + Duration::from_millis(30);
104        let decode = dequeue + Duration::from_micros(10);
105        let receipt = TransportReceipt::new(B256::repeat_byte(1), received, envelope).unwrap();
106        let timing = CanonicalIngressTiming::new(receipt, dequeue, decode).unwrap();
107        assert_eq!(timing.receipt().received(), received);
108        assert_eq!(
109            timing.dequeued().duration_since(receipt.decoded()),
110            Duration::from_millis(30)
111        );
112        assert!(TransportReceipt::new(B256::ZERO, envelope, received).is_none());
113        assert!(CanonicalIngressTiming::new(receipt, received, decode).is_none());
114    }
115}
evm_fork_cache/reactive/ingress_timing.rs

evm_fork_cache/reactive/
ingress_timing.rs
