simulator_client/
transaction_step.rs

1use std::time::Duration;
2
3use simulator_api::{
4    BacktestRequest, BacktestResponse, ContinueParams, ContinueToParams, DiscoveryBatchEvent,
5    PausedEvent,
6};
7
8use crate::{BacktestClientError, BacktestClientResult, BacktestSession};
9
10/// A batch discovery paired with the pause location immediately before it.
11#[derive(Debug, Clone)]
12pub struct DiscoveryPause {
13    /// The discovery event that triggered the pause.
14    pub discovery: DiscoveryBatchEvent,
15    /// Where the session paused; state through `batch_index - 1` is visible via RPC.
16    pub paused: PausedEvent,
17}
18
19/// Result of a single [`BacktestSession::advance_to_discovery`] call.
20#[derive(Debug)]
21pub enum DiscoveryStepResult {
22    /// Session paused before the discovered batch. Inspect state via RPC or
23    /// simulate custom transactions, then call `advance_to_discovery` again.
24    Paused(DiscoveryPause),
25    /// No more discoveries in the session range.
26    Completed,
27}
28
29impl BacktestSession {
30    /// Wait for the next batch matching a registered discovery filter and pause
31    /// immediately before it executes.
32    ///
33    /// `DiscoveryBatch` events are emitted by the server as a background scan
34    /// independent of execution — no `Continue` is needed to trigger them.
35    /// This method consumes the next queued `DiscoveryBatch`, sends `ContinueTo`
36    /// to execute up to (but not including) that batch, then waits for `Paused`.
37    ///
38    /// While paused, the session's RPC endpoint reflects state through
39    /// `batch_index - 1` of the discovered slot — no transaction in the
40    /// discovered batch has executed yet.
41    pub async fn advance_to_discovery(
42        &mut self,
43        offset: Option<u32>,
44        timeout: Option<Duration>,
45    ) -> BacktestClientResult<DiscoveryStepResult> {
46        // Phase 1: fetch next DiscoveryBatch or fail if none arrives within the timeout.
47        let discovery: DiscoveryBatchEvent = loop {
48            let Some(response) = self.next_response(timeout).await? else {
49                return Err(BacktestClientError::Closed {
50                    reason: "websocket closed while waiting for DiscoveryBatch".to_string(),
51                });
52            };
53            match response {
54                BacktestResponse::DiscoveryBatch(event) => break event,
55                BacktestResponse::Completed { .. } => return Ok(DiscoveryStepResult::Completed),
56                BacktestResponse::Error(err) => return Err(BacktestClientError::Remote(err)),
57                // Server is paused waiting for a Continue. Send one so it can advance
58                // to the next discovery or reach end-of-range and emit Completed.
59                BacktestResponse::ReadyForContinue => {
60                    self.send(
61                        &BacktestRequest::Continue(ContinueParams {
62                            advance_count: u64::MAX,
63                            ..Default::default()
64                        }),
65                        timeout,
66                    )
67                    .await?;
68                }
69                other => {
70                    tracing::warn!("ignoring {other:?} while waiting for DiscoveryBatch");
71                }
72            }
73        };
74
75        // Phase 2: execute up to (not including) the discovered batch if offset is 0.
76        // Execute past the discovered batch if offset is > 0.
77        let offset = offset.unwrap_or(0);
78        self.send(
79            &BacktestRequest::ContinueTo(ContinueToParams {
80                slot: discovery.slot,
81                batch_index: Some(discovery.batch_index + offset),
82            }),
83            timeout,
84        )
85        .await?;
86
87        // Phase 3: consume until Paused.
88        // Extra DiscoveryBatch events are held in a local buffer (not self.backlog)
89        // so next_response keeps reading from the websocket. After Paused arrives
90        // they are flushed to self.backlog for the next advance_to_discovery call.
91        let mut pending_discoveries: Vec<BacktestResponse> = Vec::new();
92        let paused: PausedEvent = loop {
93            let Some(response) = self.next_response(timeout).await? else {
94                return Err(BacktestClientError::Closed {
95                    reason: "websocket closed while waiting for Paused after ContinueTo"
96                        .to_string(),
97                });
98            };
99            match response {
100                BacktestResponse::Paused(event) => {
101                    self.ready_for_continue = true;
102                    for d in pending_discoveries {
103                        self.push_backlog(d);
104                    }
105                    break event;
106                }
107                next @ BacktestResponse::DiscoveryBatch(_) => {
108                    pending_discoveries.push(next);
109                }
110                BacktestResponse::SlotNotification(_)
111                | BacktestResponse::Status { .. }
112                | BacktestResponse::Success
113                | BacktestResponse::ReadyForContinue => {}
114                BacktestResponse::Completed { .. } => {
115                    return Ok(DiscoveryStepResult::Completed);
116                }
117                BacktestResponse::Error(err) => return Err(BacktestClientError::Remote(err)),
118                other => {
119                    return Err(BacktestClientError::UnexpectedResponse {
120                        context: "waiting for Paused after ContinueTo",
121                        response: Box::new(other),
122                    });
123                }
124            }
125        };
126
127        Ok(DiscoveryStepResult::Paused(DiscoveryPause {
128            discovery,
129            paused,
130        }))
131    }
132}
simulator_client/transaction_step.rs

simulator_client/
transaction_step.rs
