Skip to main content

simulator_api/
lib.rs

1use std::{
2    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
3    fmt,
4    str::FromStr,
5};
6
7use base64::{
8    DecodeError as Base64DecodeError, Engine as _, engine::general_purpose::STANDARD as BASE64,
9};
10use serde::{Deserialize, Serialize};
11use solana_account::AccountSharedData;
12use solana_address::{Address, error::ParseAddressError};
13
14mod session_id;
15mod slot_range;
16
17pub use session_id::{ControlSessionId, SessionId};
18pub use slot_range::{InvertedSlotRange, SlotRange};
19
20pub mod subscribe_config;
21pub mod usage;
22pub mod ws_compression;
23
24/// Backtest RPC methods exposed to the client.
25#[derive(Debug, Serialize, Deserialize)]
26#[serde(tag = "method", content = "params", rename_all = "camelCase")]
27pub enum BacktestRequest {
28    CreateBacktestSession(CreateBacktestSessionRequest),
29    Continue(ContinueParams),
30    ContinueTo(ContinueToParams),
31    ContinueSessionV1(ContinueSessionRequestV1),
32    ContinueToSessionV1(ContinueToSessionRequestV1),
33    CloseBacktestSession,
34    CloseSessionV1(CloseSessionRequestV1),
35    AttachBacktestSession {
36        session_id: SessionId,
37        /// Last sequence number the client received. Responses after this sequence
38        /// will be replayed from the session's buffer. None = replay entire buffer.
39        last_sequence: Option<u64>,
40    },
41    /// Sent after reattaching and rebuilding any dependent subscriptions.
42    /// Allows the manager to resume a session that was paused for handoff.
43    ResumeAttachedSession,
44    AttachParallelControlSessionV2 {
45        control_session_id: ControlSessionId,
46        /// Last per-session sequence number received by the client. Responses after
47        /// these sequence numbers will be replayed from the manager's per-session
48        /// replay store. Missing sessions replay their entire retained history.
49        #[serde(default)]
50        last_sequences: BTreeMap<SessionId, u64>,
51    },
52}
53
54/// Versioned payload for `CreateBacktestSession`.
55///
56/// - `V0` keeps backwards-compatible shape by using `CreateSessionParams` directly.
57/// - `V1` keeps the same shape and adds `parallel`.
58#[derive(Debug, Clone, Serialize, Deserialize)]
59#[serde(untagged)]
60pub enum CreateBacktestSessionRequest {
61    V1(CreateBacktestSessionRequestV1),
62    V0(CreateSessionParams),
63}
64
65impl CreateBacktestSessionRequest {
66    pub fn into_request_options(self) -> CreateBacktestSessionRequestOptions {
67        match self {
68            Self::V0(request) => CreateBacktestSessionRequestOptions {
69                request,
70                parallel: false,
71            },
72            Self::V1(CreateBacktestSessionRequestV1 { request, parallel }) => {
73                CreateBacktestSessionRequestOptions { request, parallel }
74            }
75        }
76    }
77
78    pub fn into_request_and_parallel(self) -> (CreateSessionParams, bool) {
79        let options = self.into_request_options();
80        (options.request, options.parallel)
81    }
82}
83
84impl From<CreateSessionParams> for CreateBacktestSessionRequest {
85    fn from(value: CreateSessionParams) -> Self {
86        Self::V0(value)
87    }
88}
89
90impl From<CreateBacktestSessionRequestV1> for CreateBacktestSessionRequest {
91    fn from(value: CreateBacktestSessionRequestV1) -> Self {
92        Self::V1(value)
93    }
94}
95
96#[derive(Debug, Clone, Serialize, Deserialize)]
97#[serde(rename_all = "camelCase")]
98pub struct CreateBacktestSessionRequestV1 {
99    #[serde(flatten)]
100    pub request: CreateSessionParams,
101    pub parallel: bool,
102}
103
104#[derive(Debug, Clone)]
105pub struct CreateBacktestSessionRequestOptions {
106    pub request: CreateSessionParams,
107    pub parallel: bool,
108}
109
110#[derive(Debug, Serialize, Deserialize)]
111#[serde(rename_all = "camelCase")]
112pub struct ContinueSessionRequestV1 {
113    pub session_id: SessionId,
114    pub request: ContinueParams,
115}
116
117#[derive(Debug, Clone, Serialize, Deserialize)]
118#[serde(rename_all = "camelCase")]
119pub struct ContinueToSessionRequestV1 {
120    pub session_id: SessionId,
121    pub request: ContinueToParams,
122}
123
124#[derive(Debug, Serialize, Deserialize)]
125#[serde(rename_all = "camelCase")]
126pub struct CloseSessionRequestV1 {
127    pub session_id: SessionId,
128}
129
130/// A filter registered at session creation describing which upcoming batches
131/// the session should announce ahead of execution via
132/// [`BacktestResponse::DiscoveryBatch`] (and its session-event twins). Each
133/// filter describes an event of interest (e.g. a specific program executing);
134/// the session "discovers" the batch in which that event will occur and
135/// emits a `DiscoveryBatch` so the client can pause immediately before it.
136#[serde_with::serde_as]
137#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
138#[serde(tag = "kind", content = "value", rename_all = "camelCase")]
139pub enum DiscoveryFilter {
140    /// Discover batches containing a transaction that invokes this program.
141    ProgramExecuted(#[serde_as(as = "serde_with::DisplayFromStr")] Address),
142}
143
144/// Per-transaction facts a [`DiscoveryFilter`] can inspect when deciding
145/// whether to match. Callers build this once per transaction and feed it to
146/// every registered filter; new variants add fields here rather than growing
147/// the [`DiscoveryFilter::matches`] signature.
148pub struct TxMatchContext<'a> {
149    /// Programs invoked by the transaction (top-level + CPI observed in logs).
150    pub invoked_programs: &'a HashSet<Address>,
151}
152
153impl DiscoveryFilter {
154    /// Return `true` when this filter is satisfied by the transaction
155    /// described by `ctx`.
156    pub fn matches(&self, ctx: &TxMatchContext<'_>) -> bool {
157        match self {
158            Self::ProgramExecuted(target) => ctx.invoked_programs.contains(target),
159        }
160    }
161}
162
163/// The two mints a swap trades, without a direction: a maker quotes a pair both ways, so
164/// `(a, b)` and `(b, a)` are the same pair. Normalized on construction and on deserialization,
165/// which is what makes the set lookup direction-agnostic.
166#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
167#[serde(try_from = "[String; 2]", into = "[String; 2]")]
168pub struct MintPair([Address; 2]);
169
170impl MintPair {
171    pub fn new(one: Address, other: Address) -> Self {
172        Self(if one <= other {
173            [one, other]
174        } else {
175            [other, one]
176        })
177    }
178
179    pub fn mints(&self) -> (Address, Address) {
180        (self.0[0], self.0[1])
181    }
182}
183
184#[derive(Debug, thiserror::Error)]
185#[error("invalid mint {mint} in pair")]
186pub struct ParseMintPairError {
187    mint: String,
188    source: ParseAddressError,
189}
190
191impl TryFrom<[String; 2]> for MintPair {
192    type Error = ParseMintPairError;
193
194    fn try_from([one, other]: [String; 2]) -> Result<Self, Self::Error> {
195        fn parse(mint: String) -> Result<Address, ParseMintPairError> {
196            Address::from_str(&mint).map_err(|source| ParseMintPairError { mint, source })
197        }
198        Ok(Self::new(parse(one)?, parse(other)?))
199    }
200}
201
202impl From<MintPair> for [String; 2] {
203    fn from(pair: MintPair) -> Self {
204        let (one, other) = pair.mints();
205        [one.to_string(), other.to_string()]
206    }
207}
208
209/// Restricts which detected swaps are re-quoted. A transaction is rerouted when at least one
210/// of its swaps trades a listed pair; the rest replay unchanged, counted as `swaps_filtered`.
211#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
212#[serde(rename_all = "camelCase")]
213pub struct RerouteFilter {
214    /// A swap matches when the mints it trades are one of these pairs, either direction.
215    #[serde(default)]
216    pub pairs: BTreeSet<MintPair>,
217}
218
219impl RerouteFilter {
220    /// An empty filter admits nothing, which creation-time validation rejects: omit the filter
221    /// to admit everything.
222    pub fn is_empty(&self) -> bool {
223        self.pairs.is_empty()
224    }
225
226    pub fn matches_mints(&self, input: &Address, output: &Address) -> bool {
227        self.pairs.contains(&MintPair::new(*input, *output))
228    }
229
230    /// Whether any listed pair trades `mint`. An arbitrage cycle trades one mint with itself, so
231    /// it is admitted by the book naming that mint rather than by a pair it can never match.
232    pub fn mentions_mint(&self, mint: &Address) -> bool {
233        self.pairs.iter().any(|pair| {
234            let (one, other) = pair.mints();
235            one == *mint || other == *mint
236        })
237    }
238}
239
240/// How an action's transactions run. Only meaningful for an action that carries
241/// transactions: one that carries none is a schedule entry for `account_overrides`.
242#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
243#[serde(rename_all = "camelCase")]
244pub enum ActionKind {
245    /// Runs against session state without committing. Each transaction's writes are
246    /// visible to the next one in the same action and to nothing else, so the replay is
247    /// unaffected.
248    #[default]
249    Simulate,
250    /// Executes and commits to the session ledger. The writes are visible to every later
251    /// slot, including the historical transactions that follow — a deliberate change to
252    /// the replayed world, not a simulation of one. Rejected under `replay_account_state`,
253    /// where the recording restamps account state and the commit would be lost.
254    Send,
255}
256
257/// Where a [`ScheduledAction`] fires.
258#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
259#[serde(tag = "at", rename_all = "camelCase")]
260pub enum ActionAnchor {
261    /// At the start of each slot, before its transactions execute and before
262    /// reroute quoting. Fires the entry for the current slot, if any.
263    BeforeSlot {
264        #[serde(default, with = "compressed")]
265        slots: Vec<u64>,
266    },
267    /// At the end of each slot, after its transactions execute but before it
268    /// finalizes. Fires the entry for the current slot, if any.
269    AfterSlot {
270        #[serde(default, with = "compressed")]
271        slots: Vec<u64>,
272    },
273    /// Before each batch matching `filter` executes. The whole sequence fires at
274    /// each matching batch.
275    BeforeMatch { filter: DiscoveryFilter },
276    /// After each batch matching `filter` commits. The whole sequence fires at
277    /// each matching batch.
278    AfterMatch { filter: DiscoveryFilter },
279}
280
281/// Upper bound on a decompressed transactions blob, guarding the server against
282/// a decompression bomb in client input.
283const MAX_ACTION_TRANSACTIONS_BYTES: u64 = 64 * 1024 * 1024;
284
285/// serde adapter that stores any value as a base64-encoded, zstd-compressed JSON
286/// blob on the wire, so transaction bytes never travel raw. Used via
287/// `#[serde(with = "compressed")]` on [`ScheduledAction`]'s `transactions` field
288/// and [`ActionAnchor`]'s `slots` field.
289mod compressed {
290    use std::io::Read;
291
292    use base64::Engine as _;
293    use serde::{Deserialize, Deserializer, Serialize, Serializer, de::DeserializeOwned};
294
295    use super::{BASE64, MAX_ACTION_TRANSACTIONS_BYTES};
296
297    /// zstd compression level; `0` selects zstd's default (level 3).
298    const COMPRESSION_LEVEL: i32 = 0;
299
300    pub fn serialize<T, S>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
301    where
302        T: Serialize,
303        S: Serializer,
304    {
305        let json = serde_json::to_vec(value).map_err(serde::ser::Error::custom)?;
306        let compressed = zstd::stream::encode_all(json.as_slice(), COMPRESSION_LEVEL)
307            .map_err(serde::ser::Error::custom)?;
308        serializer.serialize_str(&BASE64.encode(compressed))
309    }
310
311    pub fn deserialize<'de, T, D>(deserializer: D) -> Result<T, D::Error>
312    where
313        T: DeserializeOwned,
314        D: Deserializer<'de>,
315    {
316        let blob = String::deserialize(deserializer)?;
317        let compressed = BASE64
318            .decode(blob.as_bytes())
319            .map_err(serde::de::Error::custom)?;
320        let mut json = Vec::new();
321        zstd::stream::read::Decoder::new(compressed.as_slice())
322            .and_then(|d| Read::take(d, MAX_ACTION_TRANSACTIONS_BYTES).read_to_end(&mut json))
323            .map_err(serde::de::Error::custom)?;
324        serde_json::from_slice(&json).map_err(serde::de::Error::custom)
325    }
326}
327
328/// A sequence of transactions the server runs automatically during a backtest,
329/// with results streamed over the `actionSubscribe` subscription. An action carrying
330/// no transactions is declarative instead: it posts `account_overrides` to the
331/// reroute layer and streams nothing.
332#[serde_with::serde_as]
333#[derive(Debug, Clone, Serialize, Deserialize)]
334#[serde(rename_all = "camelCase")]
335pub struct ScheduledAction {
336    pub anchor: ActionAnchor,
337    /// Ignored when `transactions` is empty — that action is a schedule entry, not a run.
338    #[serde(default)]
339    pub kind: ActionKind,
340    /// Base64-encoded transactions, carried over the wire as a base64-encoded,
341    /// zstd-compressed blob. For a slot anchor (`beforeSlot`/`afterSlot`),
342    /// entry `i` fires at `slots[i]` — the two lists must be the same length,
343    /// and repeating a slot runs its entries there in order as one sequence.
344    /// For match anchors the whole list fires, in order, at each matching
345    /// batch. For `simulate`, each transaction's writes are visible to the next
346    /// without committing to the session ledger. Empty makes the action a
347    /// schedule entry for `account_overrides` instead of a run.
348    #[serde(default, with = "compressed")]
349    pub transactions: Vec<String>,
350    /// Account state overrides, per `simulateTransaction` `modifyAccountStates`
351    /// semantics. Alongside transactions they seed the first one's simulation and
352    /// are invalid for `kind: send`; alone they are the whole payload, applied to
353    /// the reroute layer at each `beforeSlot` slot and carried forward per account
354    /// until a later action names the same account.
355    #[serde(default)]
356    pub account_overrides: AccountModifications,
357    /// Accounts whose post-execution state to return, following
358    /// `simulateTransaction` semantics; reflects cumulative state after the
359    /// final transaction.
360    #[serde_as(as = "Vec<serde_with::DisplayFromStr>")]
361    #[serde(default)]
362    pub return_accounts: Vec<Address>,
363    /// Optional client tag echoed back on each `actionSubscribe` result to correlate it to this action.
364    #[serde(default)]
365    pub label: Option<String>,
366}
367
368impl ScheduledAction {
369    /// Whether this action posts `account_overrides` to the reroute layer instead of
370    /// running transactions. Carrying none is what makes it declarative.
371    pub fn is_override(&self) -> bool {
372        self.transactions.is_empty()
373    }
374}
375
376/// Slot-keyed account overrides, carried forward per account: an entry stays
377/// active from its slot until a later entry names the same account.
378#[derive(Debug, Clone, Default)]
379pub struct OverrideSchedule {
380    entries: BTreeMap<u64, HashMap<Address, AccountSharedData>>,
381}
382
383impl From<BTreeMap<u64, HashMap<Address, AccountSharedData>>> for OverrideSchedule {
384    fn from(entries: BTreeMap<u64, HashMap<Address, AccountSharedData>>) -> Self {
385        Self { entries }
386    }
387}
388
389impl OverrideSchedule {
390    pub fn is_empty(&self) -> bool {
391        self.entries.is_empty()
392    }
393
394    /// Slots at which some entry activates, ascending.
395    pub fn slots(&self) -> impl Iterator<Item = u64> + '_ {
396        self.entries.keys().copied()
397    }
398
399    /// Everything active at `slot`, resolved per account.
400    pub fn active_at(&self, slot: u64) -> HashMap<Address, AccountSharedData> {
401        self.entries
402            .range(..=slot)
403            .flat_map(|(_, accounts)| {
404                accounts
405                    .iter()
406                    .map(|(key, account)| (*key, account.clone()))
407            })
408            .collect()
409    }
410
411    pub fn active_for(&self, key: &Address, slot: u64) -> Option<&AccountSharedData> {
412        self.entries
413            .range(..=slot)
414            .rev()
415            .find_map(|(_, accounts)| accounts.get(key))
416    }
417
418    /// Final values activated in `(after, up_to]`, deduplicated per account.
419    pub fn newly_active(
420        &self,
421        after: Option<u64>,
422        up_to: u64,
423    ) -> Vec<(Address, AccountSharedData)> {
424        let lower = after.map_or(std::ops::Bound::Unbounded, std::ops::Bound::Excluded);
425        self.entries
426            .range((lower, std::ops::Bound::Included(up_to)))
427            .flat_map(|(_, accounts)| {
428                accounts
429                    .iter()
430                    .map(|(key, account)| (*key, account.clone()))
431            })
432            .collect::<HashMap<_, _>>()
433            .into_iter()
434            .collect()
435    }
436}
437
438/// Validate and compile the transaction-less actions into the slot-keyed schedule the
439/// reroute layer consumes. Same-slot actions merge account-wise, later wins.
440fn decode_override_actions(actions: &[ScheduledAction]) -> Result<OverrideSchedule, String> {
441    let mut schedule: BTreeMap<u64, HashMap<Address, AccountSharedData>> = BTreeMap::new();
442    for action in actions.iter().filter(|action| action.is_override()) {
443        if action.account_overrides.0.is_empty() {
444            return Err("an action must carry transactions or accountOverrides".to_string());
445        }
446        // Without this an action that meant to send would post overrides instead, silently.
447        if action.kind == ActionKind::Send {
448            return Err("kind: send requires transactions".to_string());
449        }
450        let ActionAnchor::BeforeSlot { slots } = &action.anchor else {
451            return Err(
452                "an action with no transactions posts accountOverrides to the reroute layer \
453                 and requires a beforeSlot anchor"
454                    .to_string(),
455            );
456        };
457        let accounts = decode_override_accounts(&action.account_overrides)?;
458        for slot in slots {
459            schedule
460                .entry(*slot)
461                .or_default()
462                .extend(accounts.iter().map(|(key, data)| (*key, data.clone())));
463        }
464    }
465    Ok(OverrideSchedule::from(schedule))
466}
467
468/// Decode account modifications into engine override form, rejecting
469/// executable/program accounts (cached bytecode makes program-code overrides
470/// stale).
471#[allow(
472    deprecated,
473    reason = "`PROGRAM_OWNERS` remains the canonical loader-owner list"
474)]
475pub fn decode_account_modifications(
476    modifications: &AccountModifications,
477) -> Result<HashMap<Address, AccountSharedData>, String> {
478    modifications
479        .0
480        .iter()
481        .map(|(address, data)| {
482            let account = data
483                .to_account()
484                .map_err(|error| format!("invalid account data for {address}: {error}"))?;
485            if account.executable || solana_account::PROGRAM_OWNERS.contains(&account.owner) {
486                return Err(format!(
487                    "overriding executable/program accounts is not supported (account {address})"
488                ));
489            }
490            Ok((*address, AccountSharedData::from(account)))
491        })
492        .collect()
493}
494
495/// [`decode_account_modifications`] plus rejection of token accounts, which the
496/// reroute-only read path cannot honor consistently.
497fn decode_override_accounts(
498    modifications: &AccountModifications,
499) -> Result<HashMap<Address, AccountSharedData>, String> {
500    let accounts = decode_account_modifications(modifications)?;
501    if let Some(address) = modifications
502        .0
503        .iter()
504        .find_map(|(address, data)| TOKEN_PROGRAM_IDS.contains(&data.owner).then_some(address))
505    {
506        return Err(format!(
507            "override for {address}: token accounts cannot be overridden \
508             (prepare-stage reads would desync from simulation)"
509        ));
510    }
511    Ok(accounts)
512}
513
514/// Server-side ceiling on [`CreateSessionParams::disconnect_timeout_secs`]:
515/// 900 seconds (15 minutes). A larger request is silently clamped to this, so
516/// clients must keep their reconnect budget under it to survive a detach.
517pub const MAX_DISCONNECT_TIMEOUT_SECS: u16 = 900;
518
519/// A router whose historical swaps a session can re-quote. The wire spelling; the session maps
520/// each onto its own venue type.
521#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
522#[serde(rename_all = "lowercase", try_from = "String")]
523pub enum SwapVenue {
524    Jupiter,
525    Okx,
526    Titan,
527    Dflow,
528}
529
530impl SwapVenue {
531    /// Matches the serialized spelling and the venue labels in the per-requote rows.
532    pub const fn as_str(self) -> &'static str {
533        match self {
534            Self::Jupiter => "jupiter",
535            Self::Okx => "okx",
536            Self::Titan => "titan",
537            Self::Dflow => "dflow",
538        }
539    }
540}
541
542impl std::fmt::Display for SwapVenue {
543    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
544        f.write_str(self.as_str())
545    }
546}
547
548#[derive(Debug, thiserror::Error)]
549pub enum ParseVenueError {
550    #[error("unknown swap venue {0:?}; expected one of jupiter, okx, titan, dflow")]
551    Unknown(String),
552    /// A run that admits no venue drives the whole reroute apparatus and re-quotes nothing.
553    #[error("name at least one venue (jupiter, okx, titan, dflow)")]
554    NoVenues,
555}
556
557impl TryFrom<String> for SwapVenue {
558    type Error = ParseVenueError;
559
560    fn try_from(value: String) -> Result<Self, Self::Error> {
561        value.parse()
562    }
563}
564
565impl std::str::FromStr for SwapVenue {
566    type Err = ParseVenueError;
567
568    fn from_str(s: &str) -> Result<Self, Self::Err> {
569        match s.trim().to_ascii_lowercase().as_str() {
570            "jupiter" => Ok(Self::Jupiter),
571            "okx" => Ok(Self::Okx),
572            "titan" => Ok(Self::Titan),
573            "dflow" => Ok(Self::Dflow),
574            _ => Err(ParseVenueError::Unknown(s.to_string())),
575        }
576    }
577}
578
579/// The venues a session re-quotes. Jupiter alone by default: the router builds Jupiter
580/// replacements, so other venues' order flow is a cross-router counterfactual a run asks for
581/// deliberately. Parsed from one comma-separated string so a flag or an env var carries the set.
582#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
583#[serde(transparent)]
584pub struct RerouteVenues(BTreeSet<SwapVenue>);
585
586impl Default for RerouteVenues {
587    fn default() -> Self {
588        Self([SwapVenue::Jupiter].into_iter().collect())
589    }
590}
591
592impl RerouteVenues {
593    pub fn new(venues: impl IntoIterator<Item = SwapVenue>) -> Self {
594        Self(venues.into_iter().collect())
595    }
596
597    pub fn contains(&self, venue: SwapVenue) -> bool {
598        self.0.contains(&venue)
599    }
600
601    pub fn is_empty(&self) -> bool {
602        self.0.is_empty()
603    }
604}
605
606impl std::fmt::Display for RerouteVenues {
607    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
608        let mut venues = self.0.iter();
609        if let Some(first) = venues.next() {
610            write!(f, "{first}")?;
611        }
612        venues.try_for_each(|venue| write!(f, ",{venue}"))
613    }
614}
615
616impl std::str::FromStr for RerouteVenues {
617    type Err = ParseVenueError;
618
619    fn from_str(s: &str) -> Result<Self, Self::Err> {
620        let venues: BTreeSet<SwapVenue> = s
621            .split(',')
622            .filter(|venue| !venue.trim().is_empty())
623            .map(str::parse)
624            .collect::<Result<_, _>>()?;
625        if venues.is_empty() {
626            return Err(ParseVenueError::NoVenues);
627        }
628        Ok(Self(venues))
629    }
630}
631
632/// Re-quoting historical swaps through the metis router: each admitted swap is re-quoted and the
633/// routed transaction is *simulated* just before the original executes — never committed, so the
634/// replay is unaffected. Each simulation reads the state of the batch its original belonged to;
635/// reroutes never observe one another.
636///
637/// Presence is the switch, which makes "a knob but no rerouting" unrepresentable rather than a
638/// validation error.
639#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
640#[serde(rename_all = "camelCase")]
641pub struct RerouteParams {
642    #[serde(default)]
643    pub venues: RerouteVenues,
644    /// Restrict re-quoting to the mint pairs of interest — a maker's own book. Absent: every
645    /// detected swap on an admitted venue is re-quoted.
646    #[serde(default, skip_serializing_if = "Option::is_none")]
647    pub filter: Option<RerouteFilter>,
648    /// Also re-quote swaps whose L1 transaction failed, when the allowlist admits the failure as
649    /// one re-quoting can answer (a slippage rejection, or a failure after the swap already ran).
650    ///
651    /// Off by default: L1 realised nothing for these, so every comparison against the original is
652    /// undefined and they are tracked as their own population — `swaps_detected` and
653    /// `swaps_rerouted` exclude them, and every row carries `original_failed`.
654    #[serde(default)]
655    pub detect_failed_l1_swaps: bool,
656    /// Also re-quote arbitrage cycles — swaps whose input and output mint are the same. Off by
657    /// default: the router prices a cycle and then refuses to build the route it found.
658    #[serde(default)]
659    pub circular_arbs: bool,
660}
661
662/// Parameters required to start a new backtest session.
663#[serde_with::serde_as]
664#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
665#[serde(rename_all = "camelCase")]
666pub struct CreateSessionParams {
667    /// First slot (inclusive) to replay.
668    pub start_slot: u64,
669    /// Last slot (inclusive) to replay.
670    pub end_slot: u64,
671    #[serde_as(as = "BTreeSet<serde_with::DisplayFromStr>")]
672    #[serde(default)]
673    #[builder(default)]
674    /// Skip transactions signed by these addresses.
675    pub signer_filter: BTreeSet<Address>,
676    /// When true, include a session summary with transaction statistics in client-facing
677    /// `Completed` responses. Summary generation remains enabled internally for metrics.
678    #[serde(default)]
679    #[builder(default)]
680    pub send_summary: bool,
681    /// Re-quote historical swaps through the metis router. Absent: plain replay.
682    #[serde(default, skip_serializing_if = "Option::is_none")]
683    pub reroute: Option<RerouteParams>,
684    /// Maximum seconds to wait for ECS capacity-related startup retries before
685    /// failing session creation. If not set (or 0), capacity errors fail immediately.
686    #[serde(default)]
687    pub capacity_wait_timeout_secs: Option<u16>,
688    /// Maximum seconds to keep the session alive after the control websocket disconnects.
689    /// If not set (or 0), the session tears down immediately on disconnect.
690    /// Clamped to [`MAX_DISCONNECT_TIMEOUT_SECS`] (900 seconds / 15 minutes).
691    #[serde(default)]
692    pub disconnect_timeout_secs: Option<u16>,
693    /// Extra compute units to add to each transaction's `SetComputeUnitLimit` budget.
694    /// Useful when replaying with an account override whose program uses more CU than
695    /// the original, causing otherwise-healthy transactions to run out of budget.
696    /// Only applied when a `SetComputeUnitLimit` instruction is already present.
697    #[serde(default)]
698    pub extra_compute_units: Option<u32>,
699    /// Events of interest the session should watch for. When an upcoming
700    /// batch matches any filter, the server emits
701    /// [`BacktestResponse::DiscoveryBatch`] (and its session-event twins)
702    /// ahead of execution so the client can follow up with
703    /// [`BacktestRequest::ContinueTo`] to pause before the batch. Empty
704    /// means no batch discoveries are performed (existing behaviour).
705    #[serde(default, skip_serializing_if = "Vec::is_empty")]
706    #[builder(default)]
707    pub discoveries: Vec<DiscoveryFilter>,
708    /// Transactions the server runs automatically during the backtest, with
709    /// results streamed over `actionSubscribe`.
710    ///
711    /// Compressed as a whole rather than per action: a per-slot schedule repeats
712    /// near-identical account bytes thousands of times, and that redundancy is
713    /// only visible across actions. Compressing each action alone leaves a
714    /// 10k-slot schedule near the websocket message limit; compressing the
715    /// vector takes the same schedule from ~28 MiB to ~1 MiB.
716    #[serde(default, with = "compressed", skip_serializing_if = "Vec::is_empty")]
717    #[builder(default)]
718    pub actions: Vec<ScheduledAction>,
719    /// Replay accounts states for this range instead of executing transactions
720    /// — a state-only acceleration.
721    #[serde(default)]
722    #[builder(default)]
723    pub replay_account_state: bool,
724}
725
726/// What account state replay cannot serve, worded once. Client, manager and
727/// session all reject through here, so a user meets the same explanation
728/// wherever the request is stopped.
729pub mod replay_support {
730    use super::{ActionKind, CreateSessionParams};
731
732    /// `getTransaction`/`getSignatureStatuses`: nothing executed, so nothing is found.
733    pub const NO_TRANSACTIONS: &str =
734        "Create the session without replayAccountState to execute and persist transactions.";
735
736    pub const SEND_TRANSACTION: &str = "Use simulateTransaction to run it against the replayed state, or create the session \
737         without replayAccountState.";
738
739    pub const SEND_ACTION: &str = "Use kind: simulate, which runs against the replayed state without committing, post state \
740         directly with an action carrying accountOverrides and no transactions, or create the \
741         session without replayAccountState.";
742
743    /// States the operation, why replay cannot serve it, then the way forward.
744    pub fn unsupported(operation: &str, remedy: &str) -> String {
745        format!(
746            "{operation} is not supported while account state replay is enabled: replay \
747             reconstructs account state from the recorded delta bundle instead of executing \
748             transactions, so this session never executes or persists any. {remedy}"
749        )
750    }
751
752    /// The one rule decidable from the request alone, so it can be caught before
753    /// a session — or a runtime to host it — is ever created.
754    pub fn send_action_conflict(params: &CreateSessionParams) -> Option<String> {
755        (params.replay_account_state
756            && params
757                .actions
758                .iter()
759                .any(|action| action.kind == ActionKind::Send))
760        .then(|| unsupported("A scheduled action with kind: send", SEND_ACTION))
761    }
762}
763
764impl CreateSessionParams {
765    /// Whether a session needs the metis sidecar (reroute quotes through it).
766    pub fn requires_sidecar(&self) -> bool {
767        self.reroute_order_flow()
768    }
769
770    pub fn reroute_order_flow(&self) -> bool {
771        self.reroute.is_some()
772    }
773
774    /// The Override actions in [`Self::actions`], compiled into the slot-keyed schedule the
775    /// reroute layer consumes. Same-slot actions merge account-wise, later wins.
776    pub fn override_schedule(&self) -> Result<OverrideSchedule, String> {
777        decode_override_actions(&self.actions)
778    }
779
780    /// The mint pairs re-quoting is restricted to, if the run named any.
781    pub fn reroute_filter(&self) -> Option<&RerouteFilter> {
782        self.reroute.as_ref()?.filter.as_ref()
783    }
784}
785
786/// What counts as a "non-benign" divergence for the backtest session fail-fast
787/// behaviour (see `BacktestSessionOptions::fail_fast`).
788#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
789#[serde(rename_all = "kebab-case")]
790pub enum FailFastDivergenceKind {
791    /// Any divergence other than a benign log diff (the full `is_divergent()` set:
792    /// log mismatch, error mismatch, balance-diff mismatch).
793    #[default]
794    AnyNonBenign,
795    /// Only divergences on transactions that touch an account currently watched by an
796    /// account-diff subscription (subscribed directly, or owned by a subscribed program).
797    Tracked,
798}
799
800impl FailFastDivergenceKind {
801    /// Stable string form used for CLI/env-var serialization. Matches the serde
802    /// `kebab-case` representation.
803    pub fn as_str(self) -> &'static str {
804        match self {
805            Self::AnyNonBenign => "any-non-benign",
806            Self::Tracked => "tracked",
807        }
808    }
809
810    /// Inverse of [`Self::as_str`]; returns `None` for an unrecognized value.
811    pub fn from_str_opt(value: &str) -> Option<Self> {
812        match value {
813            "any-non-benign" => Some(Self::AnyNonBenign),
814            "tracked" => Some(Self::Tracked),
815            _ => None,
816        }
817    }
818}
819
820/// Account state modifications to apply.
821#[serde_with::serde_as]
822#[derive(Debug, Clone, Serialize, Deserialize, Default)]
823pub struct AccountModifications(
824    #[serde_as(as = "BTreeMap<serde_with::DisplayFromStr, _>")]
825    #[serde(default)]
826    pub BTreeMap<Address, AccountData>,
827);
828
829const TOKEN_PROGRAM_IDS: [Address; 2] = [spl_token_interface::ID, spl_token_2022_interface::ID];
830
831impl From<BTreeMap<Address, AccountData>> for AccountModifications {
832    fn from(value: BTreeMap<Address, AccountData>) -> Self {
833        Self(value)
834    }
835}
836
837/// Arguments used to continue an existing session.
838#[serde_with::serde_as]
839#[derive(Debug, Serialize, Deserialize)]
840#[serde(rename_all = "camelCase")]
841pub struct ContinueParams {
842    #[serde(default = "ContinueParams::default_advance_count")]
843    /// Number of blocks to advance before waiting.
844    pub advance_count: u64,
845    #[serde(default)]
846    /// Base64-encoded transactions to execute before advancing.
847    pub transactions: Vec<String>,
848    #[serde(default)]
849    /// Account state overrides to apply ahead of execution.
850    pub modify_account_states: AccountModifications,
851}
852
853impl Default for ContinueParams {
854    fn default() -> Self {
855        Self {
856            advance_count: Self::default_advance_count(),
857            transactions: Vec::new(),
858            modify_account_states: AccountModifications(BTreeMap::new()),
859        }
860    }
861}
862
863impl ContinueParams {
864    pub fn default_advance_count() -> u64 {
865        1
866    }
867}
868
869/// Payload emitted when a session halts at a caller-specified point.
870/// `batch_index` is `None` for block-boundary pauses and `Some(n)` when the
871/// session stopped *before* batch `n` within `slot` (no transaction from
872/// batch `n` has been applied). While paused, RPC reads against the session
873/// observe partial state up through batch `n - 1`.
874#[derive(Debug, Clone, Serialize, Deserialize)]
875#[serde(rename_all = "camelCase")]
876pub struct PausedEvent {
877    pub slot: u64,
878    #[serde(default, skip_serializing_if = "Option::is_none")]
879    pub batch_index: Option<u32>,
880}
881
882/// Payload emitted when the session has *discovered* an upcoming batch that
883/// matches one or more registered [`DiscoveryFilter`]s from session creation
884/// (for example, a batch containing a transaction that invokes a program of
885/// interest). The `(slot, batch_index)` pair can be fed directly to
886/// [`BacktestRequest::ContinueTo`] to pause immediately before the batch
887/// executes. After each `Continue` / `ContinueTo`, the session emits the
888/// next `DiscoveryBatchEvent` ahead of the next matching batch, enabling a
889/// reactive "pause on every discovery" driver loop.
890#[serde_with::serde_as]
891#[derive(Debug, Clone, Serialize, Deserialize)]
892#[serde(rename_all = "camelCase")]
893pub struct DiscoveryBatchEvent {
894    pub slot: u64,
895    pub batch_index: u32,
896    /// Filters that matched this batch (always non-empty).
897    pub matched: Vec<DiscoveryFilter>,
898    /// Encoded transactions in this batch that triggered the match. Each
899    /// entry carries the serialized `VersionedTransaction` bytes paired with
900    /// the encoding used.
901    pub transactions: Vec<EncodedBinary>,
902}
903
904/// Arguments used to step an existing session to a precise point.
905#[derive(Debug, Clone, Serialize, Deserialize)]
906#[serde(rename_all = "camelCase")]
907pub struct ContinueToParams {
908    /// Target slot to stop in (or at, if `batch_index` is `None`).
909    pub slot: u64,
910    /// Batch within the target slot at which to pause, **exclusive** — the
911    /// session halts immediately *before* batch `n` executes, so no
912    /// transaction in that batch has been applied yet. `None` runs the
913    /// whole slot, pausing at the block boundary. While paused, RPC reads
914    /// observe partial state up through batch `n - 1`.
915    #[serde(default)]
916    pub batch_index: Option<u32>,
917}
918
919/// Supported binary encodings for account/transaction payloads.
920#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
921#[serde(rename_all = "lowercase")]
922pub enum BinaryEncoding {
923    Base64,
924}
925
926impl BinaryEncoding {
927    pub fn encode(self, bytes: &[u8]) -> String {
928        match self {
929            Self::Base64 => BASE64.encode(bytes),
930        }
931    }
932
933    pub fn decode(self, data: &str) -> Result<Vec<u8>, Base64DecodeError> {
934        match self {
935            Self::Base64 => BASE64.decode(data),
936        }
937    }
938}
939
940/// A blob paired with the encoding needed to decode it.
941#[derive(Debug, Clone, Serialize, Deserialize)]
942#[serde(rename_all = "camelCase")]
943pub struct EncodedBinary {
944    /// Encoded payload.
945    pub data: String,
946    /// Encoding scheme used for the payload.
947    pub encoding: BinaryEncoding,
948}
949
950impl EncodedBinary {
951    pub fn new(data: String, encoding: BinaryEncoding) -> Self {
952        Self { data, encoding }
953    }
954
955    pub fn from_bytes(bytes: &[u8], encoding: BinaryEncoding) -> Self {
956        Self {
957            data: encoding.encode(bytes),
958            encoding,
959        }
960    }
961
962    pub fn decode(&self) -> Result<Vec<u8>, Base64DecodeError> {
963        self.encoding.decode(&self.data)
964    }
965}
966
967/// Account snapshot used to seed or modify state in a session.
968#[serde_with::serde_as]
969#[derive(Debug, Clone, Serialize, Deserialize)]
970#[serde(rename_all = "camelCase")]
971pub struct AccountData {
972    /// Account data bytes and encoding.
973    pub data: EncodedBinary,
974    /// Whether the account is executable.
975    pub executable: bool,
976    /// Lamport balance.
977    pub lamports: u64,
978    #[serde_as(as = "serde_with::DisplayFromStr")]
979    /// Account owner pubkey.
980    pub owner: Address,
981    /// Allocated space in bytes.
982    pub space: u64,
983}
984
985impl AccountData {
986    pub fn to_account(&self) -> Result<solana_account::Account, Base64DecodeError> {
987        Ok(solana_account::Account {
988            data: self.data.decode()?,
989            lamports: self.lamports,
990            owner: self.owner,
991            executable: self.executable,
992            rent_epoch: 0,
993        })
994    }
995}
996
997/// Responses returned over the backtest RPC channel.
998#[derive(Debug, Clone, Serialize, Deserialize)]
999#[serde(tag = "method", content = "params", rename_all = "camelCase")]
1000pub enum BacktestResponse {
1001    SessionCreated {
1002        session_id: SessionId,
1003        rpc_endpoint: String,
1004        #[serde(default, skip_serializing_if = "Option::is_none")]
1005        task_id: Option<String>,
1006    },
1007    SessionAttached {
1008        session_id: SessionId,
1009        rpc_endpoint: String,
1010        #[serde(default, skip_serializing_if = "Option::is_none")]
1011        task_id: Option<String>,
1012    },
1013    SessionsCreatedV2 {
1014        control_session_id: ControlSessionId,
1015        session_ids: Vec<SessionId>,
1016        #[serde(default)]
1017        task_ids: Vec<Option<String>>,
1018        /// Per-sub-session start/end slots, parallel to `session_ids`. The
1019        /// server's split is authoritative (a mid-bundle request anchors to the
1020        /// covering bundle), so the client binds each sub-session to its range
1021        /// from here rather than re-deriving the split locally.
1022        #[serde(default)]
1023        start_slots: Vec<u64>,
1024        #[serde(default)]
1025        end_slots: Vec<u64>,
1026    },
1027    ParallelSessionAttachedV2 {
1028        control_session_id: ControlSessionId,
1029        session_ids: Vec<SessionId>,
1030        #[serde(default)]
1031        task_ids: Vec<Option<String>>,
1032    },
1033    ReadyForContinue,
1034    SlotNotification(u64),
1035    Paused(PausedEvent),
1036    DiscoveryBatch(DiscoveryBatchEvent),
1037    Error(BacktestError),
1038    Success,
1039    Completed {
1040        /// Session summary with transaction statistics.
1041        /// The session always computes this summary, but management may omit it from
1042        /// client-facing responses unless `send_summary` was requested at session creation.
1043        #[serde(skip_serializing_if = "Option::is_none")]
1044        summary: Option<SessionSummary>,
1045    },
1046    Status {
1047        status: BacktestStatus,
1048    },
1049    SessionEventV2 {
1050        session_id: SessionId,
1051        seq_id: u64,
1052        event: SessionEventKind,
1053    },
1054}
1055
1056impl BacktestResponse {
1057    pub fn is_completed(&self) -> bool {
1058        matches!(self, BacktestResponse::Completed { .. })
1059    }
1060
1061    pub fn is_terminal(&self) -> bool {
1062        match self {
1063            BacktestResponse::Completed { .. } => true,
1064            BacktestResponse::Error(e) => e.is_terminal(),
1065            _ => false,
1066        }
1067    }
1068}
1069
1070impl From<BacktestStatus> for BacktestResponse {
1071    fn from(status: BacktestStatus) -> Self {
1072        Self::Status { status }
1073    }
1074}
1075
1076impl From<String> for BacktestResponse {
1077    fn from(message: String) -> Self {
1078        BacktestError::Internal { error: message }.into()
1079    }
1080}
1081
1082impl From<&str> for BacktestResponse {
1083    fn from(message: &str) -> Self {
1084        BacktestError::Internal {
1085            error: message.to_string(),
1086        }
1087        .into()
1088    }
1089}
1090
1091#[derive(Debug, Clone, Serialize, Deserialize)]
1092#[serde(tag = "method", content = "params", rename_all = "camelCase")]
1093pub enum SessionEventKind {
1094    ReadyForContinue,
1095    SlotNotification(u64),
1096    Paused(PausedEvent),
1097    DiscoveryBatch(DiscoveryBatchEvent),
1098    Error(BacktestError),
1099    Success,
1100    Completed {
1101        #[serde(skip_serializing_if = "Option::is_none")]
1102        summary: Option<SessionSummary>,
1103    },
1104    Status {
1105        status: BacktestStatus,
1106    },
1107}
1108
1109impl SessionEventKind {
1110    pub fn is_terminal(&self) -> bool {
1111        match self {
1112            Self::Completed { .. } => true,
1113            Self::Error(e) => e.is_terminal(),
1114            _ => false,
1115        }
1116    }
1117}
1118
1119/// Wire format wrapper for responses sent over the control websocket.
1120/// Sessions with `disconnect_timeout_secs > 0` use this to track client position.
1121#[derive(Debug, Clone, Serialize, Deserialize)]
1122#[serde(rename_all = "camelCase")]
1123pub struct SequencedResponse {
1124    pub seq_id: u64,
1125    #[serde(flatten)]
1126    pub response: BacktestResponse,
1127}
1128
1129/// High-level progress states during a `Continue` call.
1130#[derive(Debug, Clone, Serialize, Deserialize)]
1131#[serde(rename_all = "camelCase")]
1132pub enum BacktestStatus {
1133    /// Runtime startup is in progress.
1134    StartingRuntime,
1135    DecodedTransactions,
1136    AppliedAccountModifications,
1137    ReadyToExecuteUserTransactions,
1138    ExecutedUserTransactions,
1139    ExecutingBlockTransactions,
1140    ExecutedBlockTransactions,
1141    ProgramAccountsLoaded,
1142}
1143
1144impl std::fmt::Display for BacktestStatus {
1145    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1146        let s = match self {
1147            Self::StartingRuntime => "starting runtime",
1148            Self::DecodedTransactions => "decoded transactions",
1149            Self::AppliedAccountModifications => "applied account modifications",
1150            Self::ReadyToExecuteUserTransactions => "ready to execute user transactions",
1151            Self::ExecutedUserTransactions => "executed user transactions",
1152            Self::ExecutingBlockTransactions => "executing block transactions",
1153            Self::ExecutedBlockTransactions => "executed block transactions",
1154            Self::ProgramAccountsLoaded => "program accounts loaded",
1155        };
1156        f.write_str(s)
1157    }
1158}
1159
1160/// High-level funnel for a reroute-order-flow session: how many detected swaps made it from
1161/// re-quoting through simulation to a non-reverting execution. Per-swap economics and latency
1162/// are session-internal and land in the analysis DB.
1163#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
1164#[serde(rename_all = "camelCase")]
1165pub struct RerouteStatsReport {
1166    /// Requotable swaps detected in the replayed range, across the admitted venues.
1167    #[serde(default)]
1168    pub swaps_detected: u64,
1169    /// Detected swaps excluded before quoting — an unadmitted venue, an arbitrage cycle, a mint
1170    /// pair the filter does not name. Without it the gap between `swaps_detected` and
1171    /// `swaps_rerouted` reads as routing failures.
1172    #[serde(default)]
1173    pub swaps_filtered: u64,
1174    /// Swaps re-quoted through metis and queued for simulation.
1175    #[serde(default)]
1176    pub swaps_rerouted: u64,
1177    /// Swaps whose routed transaction was simulated.
1178    #[serde(default)]
1179    pub swaps_simulated: u64,
1180    /// Simulated swaps whose routed transaction executed without reverting.
1181    #[serde(default)]
1182    pub swaps_succeeded: u64,
1183    /// Detected swaps metis could not route; nothing was simulated for them.
1184    #[serde(default)]
1185    pub requote_failures: u64,
1186    /// Blocks replayed unchanged because the metis client was unavailable.
1187    #[serde(default)]
1188    pub metis_unavailable_blocks: u64,
1189    /// Quote attempts that could not be made because the metis client was unavailable, summed
1190    /// over every swap left queued by each such block. Each one is retried on the next block,
1191    /// so this counts retries forced rather than swaps lost.
1192    #[serde(default)]
1193    pub quote_attempts_unavailable: u64,
1194}
1195
1196/// Summary of transaction execution statistics for a completed backtest session.
1197#[derive(Debug, Clone, Default, Serialize, Deserialize)]
1198#[serde(rename_all = "camelCase")]
1199pub struct SessionSummary {
1200    /// Number of simulations where simulator outcome matched on-chain outcome
1201    /// (`true_success + true_failure`).
1202    pub correct_simulation: usize,
1203    /// Number of simulations where simulator outcome did not match on-chain outcome
1204    /// (`false_success + false_failure`).
1205    pub incorrect_simulation: usize,
1206    /// Number of transactions that had execution errors in simulation.
1207    pub execution_errors: usize,
1208    /// Number of transactions with different balance diffs.
1209    pub balance_diff: usize,
1210    /// Number of transactions with different log diffs.
1211    pub log_diff: usize,
1212    /// Transactions skipped because they could not be sanitized (malformed message or
1213    /// unresolvable address lookup table).
1214    #[serde(default)]
1215    pub unsanitizable: usize,
1216    /// Aggregate reroute metrics, populated only for reroute-order-flow sessions. Boxed to
1217    /// keep `SessionSummary` small in the many response enums that carry it.
1218    #[serde(default, skip_serializing_if = "Option::is_none")]
1219    pub reroute_stats: Option<Box<RerouteStatsReport>>,
1220}
1221
1222impl SessionSummary {
1223    /// Returns true if there were any execution deviations (errors or mismatched results).
1224    pub fn has_deviations(&self) -> bool {
1225        self.incorrect_simulation > 0 || self.execution_errors > 0 || self.balance_diff > 0
1226    }
1227
1228    /// Total number of transactions processed.
1229    pub fn total_transactions(&self) -> usize {
1230        self.correct_simulation
1231            + self.incorrect_simulation
1232            + self.execution_errors
1233            + self.balance_diff
1234            + self.log_diff
1235    }
1236
1237    /// Renders the compact "N incorrect simulations, N execution errors, ..." deviation list
1238    /// shared by console reporters. `balance_label` / `log_label` let each caller keep its own
1239    /// historical wording for the trailing two counts (e.g. "balance diffs" vs
1240    /// "different balances").
1241    pub fn deviation_details(&self, balance_label: &str, log_label: &str) -> String {
1242        format!(
1243            "{} incorrect simulations, {} execution errors, {} {balance_label}, {} {log_label}",
1244            self.incorrect_simulation, self.execution_errors, self.balance_diff, self.log_diff,
1245        )
1246    }
1247}
1248
1249impl std::fmt::Display for SessionSummary {
1250    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1251        let total = self.total_transactions();
1252        write!(
1253            f,
1254            "Session summary: {total} transactions\n\
1255             \x20  - {} correct simulation\n\
1256             \x20  - {} incorrect simulation\n\
1257             \x20  - {} execution errors\n\
1258             \x20  - {} unsanitizable\n\
1259             \x20  - {} balance diffs\n\
1260             \x20  - {} log diffs",
1261            self.correct_simulation,
1262            self.incorrect_simulation,
1263            self.execution_errors,
1264            self.unsanitizable,
1265            self.balance_diff,
1266            self.log_diff,
1267        )
1268    }
1269}
1270
1271/// Error variants surfaced to backtest RPC clients.
1272#[derive(Debug, Clone, Serialize, Deserialize)]
1273#[serde(rename_all = "camelCase")]
1274pub enum BacktestError {
1275    InvalidTransactionEncoding {
1276        index: usize,
1277        error: String,
1278    },
1279    InvalidTransactionFormat {
1280        index: usize,
1281        error: String,
1282    },
1283    InvalidAccountEncoding {
1284        address: String,
1285        encoding: BinaryEncoding,
1286        error: String,
1287    },
1288    InvalidAccountOwner {
1289        address: String,
1290        error: String,
1291    },
1292    InvalidAccountPubkey {
1293        address: String,
1294        error: String,
1295    },
1296    NoMoreBlocks,
1297    BundleNotFound {
1298        start_slot: u64,
1299        end_slot: u64,
1300    },
1301    AdvanceSlotFailed {
1302        slot: u64,
1303        error: String,
1304    },
1305    FinalizeSlotFailed {
1306        slot: u64,
1307        error: String,
1308    },
1309    InvalidRequest {
1310        error: String,
1311    },
1312    Internal {
1313        error: String,
1314    },
1315    InvalidBlockhashFormat {
1316        slot: u64,
1317        error: String,
1318    },
1319    InitializingSysvarsFailed {
1320        slot: u64,
1321        error: String,
1322    },
1323    ClerkError {
1324        error: String,
1325    },
1326    SimulationError {
1327        error: String,
1328    },
1329    SessionNotFound {
1330        session_id: String,
1331    },
1332    SessionOwnerMismatch,
1333    /// Session ownership is in transition (e.g. the previous manager is
1334    /// shutting down, or another attach raced this one). Clients should retry
1335    /// the attach within their reconnect budget; the route is expected to
1336    /// become claimable shortly.
1337    SessionOwnershipBusy {
1338        reason: String,
1339    },
1340    /// The requested range contains no produced blocks (every slot in it was
1341    /// skipped on-chain), so there is nothing to execute. A user error: the
1342    /// session cannot start and the range must be adjusted.
1343    // Appended last: the manager<->runtime hop is postcard, where variant
1344    // order is wire-significant across version skew.
1345    NoProducedBlocksInRange {
1346        start_slot: u64,
1347        end_slot: u64,
1348    },
1349}
1350
1351impl BacktestError {
1352    /// Errors the backtest cannot recover from mid-session — the single source
1353    /// of truth for terminal-error classification, shared by the response/event
1354    /// wrappers and the managed client.
1355    pub fn is_terminal(&self) -> bool {
1356        matches!(
1357            self,
1358            BacktestError::NoMoreBlocks
1359                | BacktestError::BundleNotFound { .. }
1360                | BacktestError::AdvanceSlotFailed { .. }
1361                | BacktestError::FinalizeSlotFailed { .. }
1362                | BacktestError::Internal { .. }
1363                | BacktestError::NoProducedBlocksInRange { .. }
1364        )
1365    }
1366}
1367
1368/// One contiguous block range available on the history clerk.
1369#[derive(Debug, Clone, Default, Serialize, Deserialize)]
1370pub struct AvailableRange {
1371    pub bundle_start_slot: u64,
1372    pub bundle_start_slot_utc: Option<String>,
1373    pub max_bundle_end_slot: Option<u64>,
1374    pub max_bundle_end_slot_utc: Option<String>,
1375    pub max_bundle_size: Option<u64>,
1376    /// What the range can be used for. `None` from a server that predates the
1377    /// field, where every listed range is [`BundleKind::Blocks`].
1378    #[serde(default)]
1379    pub kind: Option<BundleKind>,
1380    /// Who the range is available to. `None` from a server that predates the
1381    /// field.
1382    #[serde(default)]
1383    pub scope: Option<RangeScope>,
1384}
1385
1386/// Who a listed range is available to.
1387#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1388#[serde(rename_all = "snake_case")]
1389pub enum RangeScope {
1390    /// Available to every caller.
1391    Global,
1392    /// Built for the calling API key, and listed only for that key.
1393    User,
1394}
1395
1396impl RangeScope {
1397    /// Stable string form matching the serde `snake_case` wire representation.
1398    pub fn as_str(self) -> &'static str {
1399        match self {
1400            Self::Global => "global",
1401            Self::User => "user",
1402        }
1403    }
1404}
1405
1406/// What a bundle build produces.
1407#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
1408#[serde(rename_all = "snake_case")]
1409pub enum BundleKind {
1410    /// The block bundle a session replays transactions from.
1411    #[default]
1412    Blocks,
1413    /// An account-state bundle covering exactly the requested range, which
1414    /// later sessions replay instead of executing transactions. Builds the
1415    /// block bundle it needs first.
1416    AccountState,
1417}
1418
1419/// Why a replayed range has to begin where a recording does. Shared so the
1420/// splitter's refusal and the service's single-session refusal cannot drift
1421/// apart.
1422pub const REPLAY_RANGE_ADVICE: &str = "replayed state is resolved by the recording that begins \
1423     where the request does, so request a range that begins where one was recorded and ends \
1424     within it";
1425
1426impl BundleKind {
1427    /// What a caller is short of, named for a coverage error.
1428    fn coverage_noun(self) -> &'static str {
1429        match self {
1430            Self::Blocks => "bundle range",
1431            Self::AccountState => "account-state bundle you own",
1432        }
1433    }
1434
1435    /// [`Self::coverage_noun`] pluralised, for a message about several.
1436    fn coverage_nouns(self) -> &'static str {
1437        match self {
1438            Self::Blocks => "bundle ranges",
1439            Self::AccountState => "account-state bundles you own",
1440        }
1441    }
1442
1443    /// Stable string form matching the serde `snake_case` wire representation.
1444    pub fn as_str(self) -> &'static str {
1445        match self {
1446            Self::Blocks => "blocks",
1447            Self::AccountState => "account_state",
1448        }
1449    }
1450
1451    /// Inverse of [`Self::as_str`]; returns `None` for an unrecognized value.
1452    pub fn from_str_opt(value: &str) -> Option<Self> {
1453        match value {
1454            "blocks" => Some(Self::Blocks),
1455            "account_state" => Some(Self::AccountState),
1456            _ => None,
1457        }
1458    }
1459}
1460
1461/// Spelled for the reader, not the wire: `blocks` shows as `transactions`.
1462/// Use [`BundleKind::as_str`] where the wire spelling is required.
1463impl fmt::Display for BundleKind {
1464    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1465        let label = match self {
1466            Self::Blocks => "transactions",
1467            Self::AccountState => "account_state",
1468        };
1469        f.pad(label)
1470    }
1471}
1472
1473/// Request body of `POST /build-bundle`.
1474///
1475/// Construct via `..Default::default()` so a new field stays source-compatible.
1476#[derive(Debug, Clone, Default, Serialize, Deserialize)]
1477pub struct BundleBuildRequest {
1478    pub start_slot: u64,
1479    pub end_slot: u64,
1480    #[serde(default)]
1481    pub bundle_size: Option<u64>,
1482    /// Optional repeat key; a resubmit dedupes the quota charge. The same key
1483    /// under a different `kind` is a distinct request.
1484    #[serde(default)]
1485    pub idempotency_key: Option<String>,
1486    /// Absent on the wire means [`BundleKind::Blocks`].
1487    #[serde(default)]
1488    pub kind: BundleKind,
1489}
1490
1491#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1492#[serde(rename_all = "snake_case")]
1493pub enum BundleBuildStatus {
1494    /// The bundle was built and is ready to use.
1495    Completed,
1496    /// The build failed and will not produce a bundle.
1497    Failed,
1498    /// The build has not finished yet; poll the status endpoint again later.
1499    InProgress,
1500    /// The outcome of the build could not be determined; do not assume success
1501    /// or failure.
1502    NeedsInvestigation,
1503}
1504
1505impl BundleBuildStatus {
1506    /// Stable string form matching the serde `snake_case` wire representation.
1507    pub fn as_str(self) -> &'static str {
1508        match self {
1509            Self::Completed => "completed",
1510            Self::Failed => "failed",
1511            Self::InProgress => "in_progress",
1512            Self::NeedsInvestigation => "needs_investigation",
1513        }
1514    }
1515}
1516
1517/// Response of the management bundle-build endpoints.
1518#[derive(Debug, Clone, Serialize, Deserialize)]
1519pub struct BundleBuildStatusResponse {
1520    pub request_id: String,
1521    pub start_slot: u64,
1522    pub end_slot: u64,
1523    pub bundle_size: Option<u64>,
1524    pub status: BundleBuildStatus,
1525    /// When the request id was created (unix ms) — lets clients show a live ETA
1526    /// countdown by computing elapsed since this.
1527    pub created_at_unix_ms: u64,
1528    /// Absent on records predating the field, which are all `blocks`.
1529    #[serde(default)]
1530    pub kind: BundleKind,
1531}
1532
1533/// A bundle the caller owns and can use now. Ordered by slot range, then kind.
1534#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
1535pub struct CompletedBundle {
1536    pub start_slot: u64,
1537    pub end_slot: u64,
1538    pub kind: BundleKind,
1539    /// When the bundle became available (unix ms), where known.
1540    #[serde(default)]
1541    pub completed_at_unix_ms: Option<i64>,
1542    #[serde(default)]
1543    pub size_bytes: Option<i64>,
1544}
1545
1546/// Query of `DELETE /bundles`: the bundle to delete, by kind and exact range.
1547#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1548pub struct BundleDeleteRequest {
1549    pub start_slot: u64,
1550    pub end_slot: u64,
1551    /// Absent means [`BundleKind::Blocks`].
1552    #[serde(default)]
1553    pub kind: BundleKind,
1554}
1555
1556/// What `DELETE /bundles` removed.
1557#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1558pub struct BundleDeleteResponse {
1559    pub start_slot: u64,
1560    pub end_slot: u64,
1561    pub kind: BundleKind,
1562}
1563
1564/// Response of `GET /build-bundle`.
1565#[derive(Debug, Clone, Serialize, Deserialize)]
1566pub struct BundleBuildListResponse {
1567    /// The caller's own builds, newest first. A build belongs to the endpoint
1568    /// running it, so this covers only builds requested here.
1569    pub builds: Vec<BundleBuildStatusResponse>,
1570    /// The caller's own bundles, ascending by range, wherever they were built.
1571    /// Not a view of `builds`: one build can produce several bundles, and a
1572    /// bundle outlives the record of the build that made it. `None` when the
1573    /// listing was unavailable, as opposed to `Some([])` for owning nothing.
1574    #[serde(default)]
1575    pub bundles: Option<Vec<CompletedBundle>>,
1576}
1577
1578/// Response of the management `/snapshot-slot` endpoint. The chosen source and
1579/// import state are internal, so only the resolved slot is exposed.
1580#[derive(Debug, Clone, Serialize, Deserialize)]
1581pub struct SnapshotSlotResponse {
1582    pub snapshot_slot: u64,
1583}
1584
1585/// Split a user-requested `[start_slot, end_slot]` range across the available
1586/// bundle ranges, returning a list of contiguous, non-overlapping `(start, end)`
1587/// pairs — one per bundle that the server can serve as its own session.
1588///
1589/// Each emitted `start` is a real `bundle_start_slot` and each `end` is within
1590/// the `max_bundle_end_slot` of a bundle that begins *exactly* at that start
1591/// (the server requires `start_slot == bundle_start_slot` and caps `end_slot` at
1592/// that bundle's `max_bundle_end_slot`). Among all such gap-free splits we pick
1593/// the one with the most sub-ranges, i.e. the highest parallelism: a fine-grained
1594/// (e.g. 10k) bundle grid yields one session per fine bundle instead of
1595/// collapsing onto a coarser series that overlaps the same slots — even when the
1596/// coarse and fine bundles share a start slot. A coarser bundle is only ridden
1597/// where no finer grid continues the walk, which still lets us serve a request
1598/// the coarse data covers even across a hole in the fine grid.
1599///
1600/// `kind` picks which bundles the split rides; the walk itself is the same for
1601/// either, since a recording is read slot by slot exactly as a block bundle is.
1602/// An unlabelled range, from a server predating [`AvailableRange::kind`],
1603/// counts as [`BundleKind::Blocks`].
1604///
1605/// Returns an error if the requested start cannot anchor, or if no gap-free
1606/// split reaches `requested_end`.
1607pub fn split_range(
1608    ranges: &[AvailableRange],
1609    requested: SlotRange,
1610    kind: BundleKind,
1611) -> Result<Vec<(u64, u64)>, String> {
1612    let (requested_start, requested_end) = (requested.start(), requested.end());
1613
1614    // Every advertised end per bundle start. A start can carry several ends — a
1615    // 50k snapshot bundle and the 10k bundle built on the same snapshot share a
1616    // start slot but reach different ends — so we keep them all and let the walk
1617    // pick whichever maximises parallelism.
1618    // A one-slot bundle (`end == start`) counts: a series whose last split
1619    // begins on the requested end records exactly one slot.
1620    let ends_by_start = ranges
1621        .iter()
1622        .filter(|r| r.kind.unwrap_or_default() == kind)
1623        .filter_map(|r| {
1624            r.max_bundle_end_slot
1625                .filter(|&end| end >= r.bundle_start_slot)
1626                .map(|end| (r.bundle_start_slot, end))
1627        })
1628        .fold(
1629            BTreeMap::<u64, BTreeSet<u64>>::new(),
1630            |mut ends_by_start, (start, end)| {
1631                ends_by_start.entry(start).or_default().insert(end);
1632                ends_by_start
1633            },
1634        );
1635
1636    // Anchor a request that lands mid-bundle to the covering bundle's start —
1637    // the latest bundle beginning at or before `requested_start` that still
1638    // covers it. The server's `select_parallel_requests` shares this fn and
1639    // anchors the same way, so the client must too or it would falsely reject a
1640    // start that isn't itself a bundle boundary. An exact bundle-start request
1641    // anchors to itself. (Single sessions instead require an exact bundle start
1642    // for every kind: `evaluate_requested_range`.)
1643    let Some((&anchor_start, _)) = ends_by_start.range(..=requested_start).rfind(|(_, ends)| {
1644        ends.iter()
1645            .next_back()
1646            .is_some_and(|&end| end >= requested_start)
1647    }) else {
1648        return Err(format!(
1649            "start_slot {requested_start} is not covered by any available {}",
1650            kind.coverage_noun()
1651        ));
1652    };
1653
1654    // Walk the bundle starts from `requested_end` backwards, recording for each
1655    // start the gap-free split with the most sub-ranges (highest parallelism).
1656    // A sub-range ends within one of its start's bundles and the next sub-range
1657    // must begin on the very next slot, so a coarse bundle is ridden only when no
1658    // finer bundle continues the walk.
1659    let mut best_from: BTreeMap<u64, Vec<(u64, u64)>> = BTreeMap::new();
1660    for (&start, ends) in ends_by_start.range(anchor_start..=requested_end).rev() {
1661        let mut best: Option<Vec<(u64, u64)>> = None;
1662        for &end in ends {
1663            let candidate = if end >= requested_end {
1664                Some(vec![(start, requested_end)])
1665            } else {
1666                best_from.get(&(end + 1)).map(|rest| {
1667                    std::iter::once((start, end))
1668                        .chain(rest.iter().copied())
1669                        .collect()
1670                })
1671            };
1672            if let Some(candidate) = candidate
1673                && best.as_ref().is_none_or(|b| candidate.len() > b.len())
1674            {
1675                best = Some(candidate);
1676            }
1677        }
1678        if let Some(best) = best {
1679            best_from.insert(start, best);
1680        }
1681    }
1682
1683    best_from.remove(&anchor_start).ok_or_else(|| {
1684        // Point at the first uncovered slot when the bundles leave a hole; fall
1685        // back to a generic message when the range is covered but cannot be
1686        // tiled to bundle boundaries.
1687        let mut covered_to = anchor_start.saturating_sub(1);
1688        for (&start, ends) in ends_by_start.range(anchor_start..=requested_end) {
1689            if start > covered_to.saturating_add(1) {
1690                break;
1691            }
1692            if let Some(&end) = ends.iter().next_back() {
1693                covered_to = covered_to.max(end);
1694            }
1695        }
1696        if covered_to < requested_end {
1697            format!("gap in coverage at slot {}", covered_to + 1)
1698        } else {
1699            format!(
1700                "no gap-free split of [{requested_start}, {requested_end}] aligns with the available {}",
1701                kind.coverage_nouns()
1702            )
1703        }
1704    })
1705}
1706
1707impl From<BacktestError> for BacktestResponse {
1708    fn from(error: BacktestError) -> Self {
1709        Self::Error(error)
1710    }
1711}
1712
1713impl std::error::Error for BacktestError {}
1714
1715impl fmt::Display for BacktestError {
1716    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1717        match self {
1718            BacktestError::InvalidTransactionEncoding { index, error } => {
1719                write!(f, "invalid transaction encoding at index {index}: {error}")
1720            }
1721            BacktestError::InvalidTransactionFormat { index, error } => {
1722                write!(f, "invalid transaction format at index {index}: {error}")
1723            }
1724            BacktestError::InvalidAccountEncoding {
1725                address,
1726                encoding,
1727                error,
1728            } => write!(
1729                f,
1730                "invalid encoding for account {address} ({encoding:?}): {error}"
1731            ),
1732            BacktestError::InvalidAccountOwner { address, error } => {
1733                write!(f, "invalid owner for account {address}: {error}")
1734            }
1735            BacktestError::InvalidAccountPubkey { address, error } => {
1736                write!(f, "invalid account pubkey {address}: {error}")
1737            }
1738            BacktestError::NoMoreBlocks => write!(f, "no more blocks available"),
1739            BacktestError::BundleNotFound {
1740                start_slot,
1741                end_slot,
1742            } => write!(
1743                f,
1744                "block bundle not found for start_slot={start_slot}, end_slot={end_slot}"
1745            ),
1746            BacktestError::AdvanceSlotFailed { slot, error } => {
1747                write!(f, "failed to advance to slot {slot}: {error}")
1748            }
1749            BacktestError::FinalizeSlotFailed { slot, error } => {
1750                write!(f, "failed to finalize slot {slot}: {error}")
1751            }
1752            BacktestError::InvalidRequest { error } => write!(f, "invalid request: {error}"),
1753            BacktestError::Internal { error } => write!(f, "internal error: {error}"),
1754            BacktestError::InvalidBlockhashFormat { slot, error } => {
1755                write!(f, "invalid blockhash at slot {slot}: {error}")
1756            }
1757            BacktestError::InitializingSysvarsFailed { slot, error } => {
1758                write!(f, "failed to initialize sysvars at slot {slot}: {error}")
1759            }
1760            BacktestError::ClerkError { error } => write!(f, "clerk error: {error}"),
1761            BacktestError::SimulationError { error } => {
1762                write!(f, "simulation error: {error}")
1763            }
1764            BacktestError::SessionNotFound { session_id } => {
1765                write!(f, "session not found: {session_id}")
1766            }
1767            BacktestError::SessionOwnerMismatch => {
1768                write!(f, "session owner mismatch")
1769            }
1770            BacktestError::SessionOwnershipBusy { reason } => {
1771                write!(f, "session ownership busy: {reason}")
1772            }
1773            BacktestError::NoProducedBlocksInRange {
1774                start_slot,
1775                end_slot,
1776            } => write!(
1777                f,
1778                "no produced blocks in requested range {start_slot}..={end_slot}; \
1779                 slot(s) may be skipped on-chain"
1780            ),
1781        }
1782    }
1783}
1784
1785#[cfg(test)]
1786mod tests {
1787    use rstest::rstest;
1788    use solana_account::ReadableAccount;
1789
1790    use super::*;
1791
1792    fn address(byte: u8) -> Address {
1793        Address::from([byte; 32])
1794    }
1795
1796    #[rstest]
1797    #[case::one("jupiter", Some(&[SwapVenue::Jupiter][..]))]
1798    #[case::several("jupiter,okx", Some(&[SwapVenue::Jupiter, SwapVenue::Okx][..]))]
1799    #[case::spaced_and_uppercased(" Jupiter , OKX ", Some(&[SwapVenue::Jupiter, SwapVenue::Okx][..]))]
1800    #[case::repeats_collapse("okx,okx", Some(&[SwapVenue::Okx][..]))]
1801    #[case::trailing_separator("okx,", Some(&[SwapVenue::Okx][..]))]
1802    #[case::unknown("jupiter,bogus", None)]
1803    #[case::empty("", None)]
1804    #[case::separators_only(",,", None)]
1805    fn reroute_venues_parse_the_shapes_a_flag_carries(
1806        #[case] input: &str,
1807        #[case] expected: Option<&[SwapVenue]>,
1808    ) {
1809        assert_eq!(
1810            input.parse::<RerouteVenues>().ok(),
1811            expected.map(|venues| RerouteVenues::new(venues.iter().copied()))
1812        );
1813    }
1814
1815    /// `Display` renders `--reroute-venues`' default, so clap re-parses whatever it prints.
1816    #[test]
1817    fn reroute_venues_round_trip_through_their_rendering() {
1818        let venues = RerouteVenues::new([SwapVenue::Dflow, SwapVenue::Jupiter]);
1819
1820        assert_eq!(venues.to_string(), "jupiter,dflow");
1821        assert_eq!(venues.to_string().parse::<RerouteVenues>().unwrap(), venues);
1822        assert_eq!(RerouteVenues::default().to_string(), "jupiter");
1823    }
1824
1825    /// The API deserializes venues through the same parser the flag uses, so a spelling one
1826    /// entry point accepts is not a hard error at the other.
1827    #[test]
1828    fn venues_deserialize_through_the_same_parser_the_flag_uses() {
1829        let params: CreateSessionParams = serde_json::from_value(
1830            serde_json::json!({"startSlot": 1, "endSlot": 2, "reroute": {"venues": ["Jupiter", "OKX"]}}),
1831        )
1832        .expect("decode");
1833
1834        assert_eq!(
1835            params.reroute.expect("rerouting requested").venues,
1836            RerouteVenues::new([SwapVenue::Jupiter, SwapVenue::Okx])
1837        );
1838    }
1839
1840    /// A request that names no knob gets the narrow defaults, not the every-venue behaviour a
1841    /// client written before this change expected — the switch these defaults are here to make.
1842    #[test]
1843    fn params_without_the_new_reroute_fields_decode_to_the_narrow_defaults() {
1844        let params: CreateSessionParams = serde_json::from_value(
1845            serde_json::json!({"startSlot": 1, "endSlot": 2, "reroute": {}}),
1846        )
1847        .expect("decode");
1848
1849        let reroute = params.reroute.expect("rerouting requested");
1850        assert_eq!(reroute.venues, RerouteVenues::default());
1851        assert!(reroute.venues.contains(SwapVenue::Jupiter));
1852        assert!(!reroute.venues.contains(SwapVenue::Okx));
1853        assert!(!reroute.circular_arbs);
1854        assert!(!reroute.detect_failed_l1_swaps);
1855    }
1856
1857    /// Presence is the switch: a replay that will never quote must not reserve a metis sidecar.
1858    #[test]
1859    fn a_request_without_reroute_reserves_no_sidecar() {
1860        let params: CreateSessionParams =
1861            serde_json::from_value(serde_json::json!({"startSlot": 1, "endSlot": 2}))
1862                .expect("decode");
1863
1864        assert!(!params.requires_sidecar());
1865    }
1866
1867    /// A per-slot override schedule is the largest thing a client sends. Each
1868    /// action alone is incompressible enough to sit near the websocket message
1869    /// limit; the redundancy lives *across* actions, so the whole vector has to
1870    /// travel as one blob. Pins both the round trip and that size property.
1871    #[test]
1872    fn a_per_slot_action_schedule_compresses_across_actions() {
1873        let slots = 2_000u64;
1874        let actions = (0..slots)
1875            .map(|slot| ScheduledAction {
1876                anchor: ActionAnchor::AfterSlot { slots: vec![slot] },
1877                kind: ActionKind::Simulate,
1878                transactions: Vec::new(),
1879                account_overrides: AccountModifications(BTreeMap::from([(
1880                    address(7),
1881                    AccountData {
1882                        // Realistic pool account: 2 KiB that barely moves slot to slot.
1883                        data: EncodedBinary::new(
1884                            BASE64.encode(
1885                                [&slot.to_le_bytes()[..], &[0xAB; 2040][..]]
1886                                    .concat()
1887                                    .as_slice(),
1888                            ),
1889                            BinaryEncoding::Base64,
1890                        ),
1891                        executable: false,
1892                        lamports: 15_144_966,
1893                        owner: address(9),
1894                        space: 2048,
1895                    },
1896                )])),
1897                return_accounts: Vec::new(),
1898                label: None,
1899            })
1900            .collect::<Vec<_>>();
1901        let params = CreateSessionParams::builder()
1902            .start_slot(1)
1903            .end_slot(slots)
1904            .actions(actions.clone())
1905            .build();
1906
1907        let encoded = serde_json::to_string(&params).expect("serialize");
1908        let raw = serde_json::to_string(&actions).expect("serialize raw actions");
1909        assert!(
1910            encoded.len() * 8 < raw.len(),
1911            "expected the schedule to compress well; got {} bytes from {} raw",
1912            encoded.len(),
1913            raw.len()
1914        );
1915
1916        let decoded: CreateSessionParams = serde_json::from_str(&encoded).expect("round trip");
1917        assert_eq!(
1918            serde_json::to_string(&decoded.actions).expect("serialize decoded"),
1919            raw
1920        );
1921    }
1922
1923    /// Absent actions stay absent on the wire, so a session without a schedule
1924    /// is unchanged by the compression.
1925    #[test]
1926    fn an_empty_action_list_is_omitted_entirely() {
1927        let params = CreateSessionParams::builder()
1928            .start_slot(1)
1929            .end_slot(2)
1930            .build();
1931        let json = serde_json::to_value(&params).expect("serialize");
1932        assert!(json.get("actions").is_none(), "{json}");
1933        let decoded: CreateSessionParams = serde_json::from_value(json).expect("round trip");
1934        assert!(decoded.actions.is_empty());
1935    }
1936
1937    /// The conflict is `send` *and* replay: neither alone is a problem, and simulate runs
1938    /// fine under replay because it never commits.
1939    #[test]
1940    fn send_actions_conflict_with_replay_only_together() {
1941        let params = |replay, kind| {
1942            CreateSessionParams::builder()
1943                .start_slot(1)
1944                .end_slot(2)
1945                .replay_account_state(replay)
1946                .actions(vec![ScheduledAction {
1947                    anchor: ActionAnchor::BeforeSlot { slots: vec![1] },
1948                    kind,
1949                    transactions: vec!["dGVzdA==".to_string()],
1950                    account_overrides: Default::default(),
1951                    return_accounts: Default::default(),
1952                    label: None,
1953                }])
1954                .build()
1955        };
1956        assert!(replay_support::send_action_conflict(&params(false, ActionKind::Send)).is_none());
1957        assert!(
1958            replay_support::send_action_conflict(&params(true, ActionKind::Simulate)).is_none()
1959        );
1960
1961        let conflict = replay_support::send_action_conflict(&params(true, ActionKind::Send))
1962            .expect("send under replay conflicts");
1963        assert!(conflict.contains("kind: send"), "{conflict}");
1964        assert!(conflict.contains(replay_support::SEND_ACTION), "{conflict}");
1965    }
1966
1967    fn override_entry(owner: Address, executable: bool) -> AccountData {
1968        AccountData {
1969            data: EncodedBinary::from_bytes(&[1, 2, 3], BinaryEncoding::Base64),
1970            executable,
1971            lamports: 1,
1972            owner,
1973            space: 3,
1974        }
1975    }
1976
1977    fn before() -> ActionAnchor {
1978        ActionAnchor::BeforeSlot { slots: vec![1] }
1979    }
1980
1981    fn valid() -> ScheduledAction {
1982        override_action(before(), address(3), override_entry(address(4), false))
1983    }
1984
1985    fn override_action(
1986        anchor: ActionAnchor,
1987        account: Address,
1988        data: AccountData,
1989    ) -> ScheduledAction {
1990        ScheduledAction {
1991            anchor,
1992            kind: ActionKind::default(),
1993            transactions: Vec::new(),
1994            account_overrides: AccountModifications(BTreeMap::from([(account, data)])),
1995            return_accounts: Vec::new(),
1996            label: None,
1997        }
1998    }
1999
2000    #[test]
2001    fn override_action_round_trips() {
2002        let key = address(1);
2003        let entry = override_entry(address(2), false);
2004        let action = override_action(
2005            ActionAnchor::BeforeSlot { slots: vec![5, 9] },
2006            key,
2007            entry.clone(),
2008        );
2009        let json = serde_json::to_value(&action).unwrap();
2010        assert!(
2011            json["anchor"]["slots"].is_string(),
2012            "slots travel as a compressed blob: {json}"
2013        );
2014        let back: ScheduledAction = serde_json::from_value(json).unwrap();
2015        assert_eq!(back.anchor, ActionAnchor::BeforeSlot { slots: vec![5, 9] });
2016        assert!(back.is_override(), "no transactions makes it declarative");
2017        let survived = &back.account_overrides.0[&key];
2018        assert_eq!(survived.lamports, entry.lamports);
2019        assert_eq!(survived.data.data, entry.data.data);
2020    }
2021
2022    #[rstest]
2023    #[case::executable(override_action(before(), address(3), override_entry(address(10), true)))]
2024    #[case::program_owned(override_action(
2025        before(),
2026        address(3),
2027        override_entry(solana_account::PROGRAM_OWNERS[0], false),
2028    ))]
2029    #[case::token_owned(override_action(
2030        before(),
2031        address(3),
2032        override_entry(TOKEN_PROGRAM_IDS[0], false),
2033    ))]
2034    #[case::after_slot_anchor(override_action(
2035        ActionAnchor::AfterSlot { slots: vec![1] },
2036        address(3),
2037        override_entry(address(4), false),
2038    ))]
2039    #[case::nothing_to_post(ScheduledAction {
2040        account_overrides: AccountModifications::default(),
2041        ..valid()
2042    })]
2043    // Otherwise an action that meant to send would silently post overrides instead.
2044    #[case::send(ScheduledAction {
2045        kind: ActionKind::Send,
2046        ..valid()
2047    })]
2048    #[allow(deprecated, reason = "exercise the loader-owner rejection path")]
2049    fn override_actions_reject_invalid_accounts_and_shapes(#[case] action: ScheduledAction) {
2050        assert!(decode_override_actions(&[action]).is_err());
2051    }
2052
2053    /// The shapes the rejections above are variations of.
2054    #[test]
2055    fn override_actions_accept_a_static_and_a_computed_shape() {
2056        assert!(decode_override_actions(&[valid()]).is_ok());
2057        // Transactions make the contribution computed rather than static — still valid,
2058        // just not resolvable until the action runs.
2059        assert!(
2060            decode_override_actions(&[ScheduledAction {
2061                transactions: vec!["dGVzdA==".to_string()],
2062                ..valid()
2063            }])
2064            .is_ok()
2065        );
2066    }
2067
2068    #[test]
2069    fn override_actions_merge_account_wise_later_wins() {
2070        let entry = |lamports| AccountData {
2071            lamports,
2072            ..override_entry(address(9), false)
2073        };
2074        let slot_action = |slots, account, lamports| {
2075            override_action(ActionAnchor::BeforeSlot { slots }, account, entry(lamports))
2076        };
2077        let actions = [
2078            slot_action(vec![5, 7], address(1), 1),
2079            slot_action(vec![5], address(2), 2),
2080            slot_action(vec![5], address(1), 3),
2081            ScheduledAction {
2082                anchor: ActionAnchor::AfterSlot { slots: vec![5] },
2083                kind: ActionKind::Simulate,
2084                transactions: vec!["dGVzdA==".to_string()],
2085                account_overrides: AccountModifications::default(),
2086                return_accounts: Vec::new(),
2087                label: None,
2088            },
2089        ];
2090        let schedule = decode_override_actions(&actions).unwrap();
2091        assert_eq!(schedule.slots().collect::<Vec<_>>(), vec![5, 7]);
2092        let lamports_at = |key, slot| {
2093            schedule
2094                .active_for(&key, slot)
2095                .map(ReadableAccount::lamports)
2096        };
2097        assert_eq!(lamports_at(address(1), 5), Some(3));
2098        assert_eq!(lamports_at(address(2), 5), Some(2));
2099        assert_eq!(lamports_at(address(1), 7), Some(1));
2100    }
2101
2102    fn shared_account(lamports: u64) -> AccountSharedData {
2103        AccountSharedData::from(solana_account::Account {
2104            lamports,
2105            data: vec![1],
2106            owner: address(9),
2107            executable: false,
2108            rent_epoch: 0,
2109        })
2110    }
2111
2112    fn carry_forward_schedule() -> (Address, Address, OverrideSchedule) {
2113        let (a, b) = (address(1), address(2));
2114        let schedule = OverrideSchedule::from(BTreeMap::from([
2115            (10, HashMap::from([(a, shared_account(1))])),
2116            (
2117                20,
2118                HashMap::from([(a, shared_account(2)), (b, shared_account(9))]),
2119            ),
2120        ]));
2121        (a, b, schedule)
2122    }
2123
2124    #[test]
2125    fn override_schedule_resolves_carry_forward_per_account() {
2126        let (a, b, schedule) = carry_forward_schedule();
2127
2128        assert!(schedule.active_for(&a, 9).is_none());
2129        let cases = [
2130            (9, None, None),
2131            (10, Some(1), None),
2132            (15, Some(1), None),
2133            (20, Some(2), Some(9)),
2134            (u64::MAX, Some(2), Some(9)),
2135        ];
2136        for (slot, want_a, want_b) in cases {
2137            assert_eq!(
2138                schedule.active_for(&a, slot).map(ReadableAccount::lamports),
2139                want_a,
2140                "active_for a at {slot}"
2141            );
2142            let active = schedule.active_at(slot);
2143            assert_eq!(
2144                active.get(&a).map(ReadableAccount::lamports),
2145                want_a,
2146                "active_at a at {slot}"
2147            );
2148            assert_eq!(
2149                active.get(&b).map(ReadableAccount::lamports),
2150                want_b,
2151                "active_at b at {slot}"
2152            );
2153        }
2154    }
2155
2156    #[test]
2157    fn newly_active_yields_final_values_in_window() {
2158        let (a, b, schedule) = carry_forward_schedule();
2159
2160        let newly: HashMap<_, _> = schedule.newly_active(Some(10), 20).into_iter().collect();
2161        assert_eq!(newly.len(), 2);
2162        assert_eq!(newly[&a].lamports(), 2);
2163        assert_eq!(newly[&b].lamports(), 9);
2164        let spanning: HashMap<_, _> = schedule.newly_active(None, 20).into_iter().collect();
2165        assert_eq!(spanning[&a].lamports(), 2, "later slot wins");
2166        assert_eq!(schedule.newly_active(None, 15).len(), 1);
2167    }
2168
2169    #[test]
2170    fn reroute_filter_round_trips_camel_case() {
2171        let filter = RerouteFilter {
2172            pairs: BTreeSet::from([MintPair::new(address(1), address(2))]),
2173        };
2174        let json = serde_json::to_value(&filter).unwrap();
2175        assert!(
2176            json["pairs"][0][0].is_string() && json["pairs"][0][1].is_string(),
2177            "a pair serializes as two base58 strings: {json}"
2178        );
2179        let back: RerouteFilter = serde_json::from_value(json).unwrap();
2180        assert_eq!(back, filter);
2181    }
2182
2183    /// The set lookup is only direction-agnostic because construction normalizes, and
2184    /// values arriving over the wire have to be normalized on that path too.
2185    #[test]
2186    fn mint_pairs_are_normalized_however_they_are_built() {
2187        let (one, other) = (address(1), address(2));
2188        assert_eq!(MintPair::new(one, other), MintPair::new(other, one));
2189
2190        let reversed: MintPair =
2191            serde_json::from_value(serde_json::json!([other.to_string(), one.to_string()]))
2192                .unwrap();
2193        assert_eq!(reversed, MintPair::new(one, other));
2194    }
2195
2196    #[test]
2197    fn fail_fast_divergence_kind_str_round_trips() {
2198        for kind in [
2199            FailFastDivergenceKind::AnyNonBenign,
2200            FailFastDivergenceKind::Tracked,
2201        ] {
2202            assert_eq!(
2203                FailFastDivergenceKind::from_str_opt(kind.as_str()),
2204                Some(kind)
2205            );
2206        }
2207        assert_eq!(FailFastDivergenceKind::from_str_opt("nonsense"), None);
2208        assert_eq!(
2209            FailFastDivergenceKind::default(),
2210            FailFastDivergenceKind::AnyNonBenign
2211        );
2212    }
2213
2214    #[test]
2215    fn bundle_not_found_is_terminal_and_descriptive() {
2216        let error = BacktestError::BundleNotFound {
2217            start_slot: 100,
2218            end_slot: 600,
2219        };
2220
2221        assert!(error.is_terminal());
2222        assert_eq!(
2223            error.to_string(),
2224            "block bundle not found for start_slot=100, end_slot=600"
2225        );
2226    }
2227
2228    #[test]
2229    fn no_produced_blocks_is_terminal_and_descriptive() {
2230        let error = BacktestError::NoProducedBlocksInRange {
2231            start_slot: 100,
2232            end_slot: 100,
2233        };
2234
2235        assert!(error.is_terminal());
2236        // placement-e2e matches this phrase to classify the failure as
2237        // transient; keep it stable.
2238        assert_eq!(
2239            error.to_string(),
2240            "no produced blocks in requested range 100..=100; slot(s) may be skipped on-chain"
2241        );
2242    }
2243
2244    #[test]
2245    fn bundle_build_request_optional_fields_default_to_none() {
2246        let req: BundleBuildRequest =
2247            serde_json::from_str(r#"{"start_slot":1,"end_slot":2}"#).expect("parse");
2248        assert_eq!((req.start_slot, req.end_slot), (1, 2));
2249        assert_eq!(req.bundle_size, None);
2250        assert_eq!(req.idempotency_key, None);
2251    }
2252
2253    #[test]
2254    fn bundle_build_request_parses_optional_fields() {
2255        let req: BundleBuildRequest = serde_json::from_str(
2256            r#"{"start_slot":1,"end_slot":2,"bundle_size":500,"idempotency_key":"abc"}"#,
2257        )
2258        .expect("parse");
2259        assert_eq!(req.bundle_size, Some(500));
2260        assert_eq!(req.idempotency_key.as_deref(), Some("abc"));
2261    }
2262
2263    #[test]
2264    fn bundle_build_status_serde_round_trips_with_snake_case() {
2265        let cases = [
2266            (BundleBuildStatus::Completed, "\"completed\""),
2267            (BundleBuildStatus::Failed, "\"failed\""),
2268            (BundleBuildStatus::InProgress, "\"in_progress\""),
2269            (
2270                BundleBuildStatus::NeedsInvestigation,
2271                "\"needs_investigation\"",
2272            ),
2273        ];
2274        for (status, expected) in cases {
2275            assert_eq!(serde_json::to_string(&status).unwrap(), expected);
2276            assert_eq!(
2277                serde_json::from_str::<BundleBuildStatus>(expected).unwrap(),
2278                status
2279            );
2280            assert_eq!(status.as_str(), expected.trim_matches('"'));
2281        }
2282        assert!(serde_json::from_str::<BundleBuildStatus>("\"queued\"").is_err());
2283    }
2284
2285    #[test]
2286    fn bundle_build_request_without_kind_deserializes_as_blocks() {
2287        // An old client's payload must keep meaning a plain block build.
2288        let request: BundleBuildRequest =
2289            serde_json::from_str(r#"{"start_slot":100,"end_slot":200}"#).unwrap();
2290        assert_eq!(request.kind, BundleKind::Blocks);
2291    }
2292
2293    #[test]
2294    fn bundle_kind_round_trips_over_serde_and_as_str() {
2295        for kind in [BundleKind::Blocks, BundleKind::AccountState] {
2296            let json = serde_json::to_string(&kind).unwrap();
2297            assert_eq!(json, format!("\"{}\"", kind.as_str()));
2298            assert_eq!(serde_json::from_str::<BundleKind>(&json).unwrap(), kind);
2299            // The store round-trips through `as_str`, not serde, so the two
2300            // maps have to agree for a stored kind to read back.
2301            assert_eq!(BundleKind::from_str_opt(kind.as_str()), Some(kind));
2302        }
2303        assert!(serde_json::from_str::<BundleKind>("\"accountState\"").is_err());
2304        assert_eq!(BundleKind::from_str_opt("accountState"), None);
2305    }
2306
2307    #[test]
2308    fn bundle_build_request_carries_kind_through_serde() {
2309        let request = BundleBuildRequest {
2310            start_slot: 100,
2311            end_slot: 200,
2312            kind: BundleKind::AccountState,
2313            ..Default::default()
2314        };
2315        let json = serde_json::to_string(&request).unwrap();
2316        assert!(json.contains(r#""kind":"account_state""#), "{json}");
2317        let parsed: BundleBuildRequest = serde_json::from_str(&json).unwrap();
2318        assert_eq!(parsed.kind, BundleKind::AccountState);
2319    }
2320
2321    #[test]
2322    fn bundle_build_status_response_serializes_request_and_status() {
2323        let response = BundleBuildStatusResponse {
2324            request_id: "r".to_string(),
2325            start_slot: 100,
2326            end_slot: 200,
2327            bundle_size: Some(50),
2328            status: BundleBuildStatus::InProgress,
2329            created_at_unix_ms: 0,
2330            kind: BundleKind::Blocks,
2331        };
2332        let json = serde_json::to_value(&response).unwrap();
2333        assert_eq!(json["request_id"].as_str(), Some("r"));
2334        assert_eq!(json["start_slot"].as_u64(), Some(100));
2335        assert_eq!(json["bundle_size"].as_u64(), Some(50));
2336        assert_eq!(json["status"].as_str(), Some("in_progress"));
2337        assert_eq!(json["created_at_unix_ms"].as_u64(), Some(0));
2338    }
2339
2340    #[test]
2341    fn an_empty_bundles_list_is_distinct_from_an_absent_one() {
2342        // A server that predates the section sends no `bundles` at all.
2343        let absent: BundleBuildListResponse = serde_json::from_str(r#"{"builds":[]}"#).unwrap();
2344        let empty: BundleBuildListResponse =
2345            serde_json::from_str(r#"{"builds":[],"bundles":[]}"#).unwrap();
2346        assert_eq!(absent.bundles, None);
2347        assert_eq!(empty.bundles, Some(Vec::new()));
2348    }
2349
2350    #[test]
2351    fn completed_bundle_round_trips() {
2352        let bundle = CompletedBundle {
2353            start_slot: 100,
2354            end_slot: 200,
2355            kind: BundleKind::AccountState,
2356            completed_at_unix_ms: Some(1_700_000_000_000),
2357            size_bytes: Some(4_096),
2358        };
2359        let json = serde_json::to_string(&bundle).unwrap();
2360        assert!(json.contains(r#""kind":"account_state""#), "{json}");
2361        assert_eq!(
2362            serde_json::from_str::<CompletedBundle>(&json).unwrap(),
2363            bundle
2364        );
2365
2366        let sparse = r#"{"start_slot":1,"end_slot":2,"kind":"blocks"}"#;
2367        let parsed: CompletedBundle = serde_json::from_str(sparse).unwrap();
2368        assert_eq!(parsed.completed_at_unix_ms, None);
2369        assert_eq!(parsed.size_bytes, None);
2370    }
2371
2372    #[test]
2373    fn available_range_reads_a_server_that_predates_kind_and_scope() {
2374        // Deploys are not atomic: a new client must not fail on the old shape.
2375        let json = r#"{
2376            "bundle_start_slot": 100,
2377            "bundle_start_slot_utc": null,
2378            "max_bundle_end_slot": 200,
2379            "max_bundle_end_slot_utc": null,
2380            "max_bundle_size": 101
2381        }"#;
2382        let parsed: AvailableRange = serde_json::from_str(json).unwrap();
2383        assert_eq!(parsed.bundle_start_slot, 100);
2384        assert_eq!(parsed.kind, None);
2385        assert_eq!(parsed.scope, None);
2386    }
2387
2388    #[test]
2389    fn available_range_ignores_fields_it_does_not_know() {
2390        // The other direction: an old client against a newer server.
2391        let json = r#"{"bundle_start_slot": 1, "some_future_field": "x"}"#;
2392        assert_eq!(
2393            serde_json::from_str::<AvailableRange>(json)
2394                .unwrap()
2395                .bundle_start_slot,
2396            1
2397        );
2398    }
2399
2400    #[test]
2401    fn available_range_labels_round_trip() {
2402        let range = AvailableRange {
2403            bundle_start_slot: 100,
2404            max_bundle_end_slot: Some(200),
2405            kind: Some(BundleKind::AccountState),
2406            scope: Some(RangeScope::User),
2407            ..Default::default()
2408        };
2409        let json = serde_json::to_string(&range).unwrap();
2410        assert!(json.contains(r#""kind":"account_state""#), "{json}");
2411        assert!(json.contains(r#""scope":"user""#), "{json}");
2412        let parsed: AvailableRange = serde_json::from_str(&json).unwrap();
2413        assert_eq!(parsed.kind, Some(BundleKind::AccountState));
2414        assert_eq!(parsed.scope, Some(RangeScope::User));
2415    }
2416
2417    fn range(start: u64, end: u64) -> AvailableRange {
2418        AvailableRange {
2419            bundle_start_slot: start,
2420            max_bundle_end_slot: Some(end),
2421            ..Default::default()
2422        }
2423    }
2424
2425    /// Each case lists the available bundles, the requested `[start, end]`, and
2426    /// the expected split — `Some(_)` for an accepted plan, `None` when the range
2427    /// is unservable and `split_range` must error.
2428    #[rstest::rstest]
2429    #[case::single(vec![range(100, 300)], 100, 300, Some(vec![(100, 300)]))]
2430    #[case::multi(
2431        vec![range(100, 200), range(201, 300), range(301, 400)],
2432        100, 300, Some(vec![(100, 200), (201, 300)])
2433    )]
2434    // Smaller bundles nested inside a larger one don't bridge the gap, but the
2435    // range is still coverable by riding the coarse series.
2436    #[case::nested(
2437        vec![range(100, 500), range(110, 150), range(150, 190), range(501, 900)],
2438        100, 900, Some(vec![(100, 500), (501, 900)])
2439    )]
2440    // A fine grid (1k bundles standing in for the 10k grid) overlapped by a
2441    // coarser, differently-aligned series: ride the fine grid, never landing on
2442    // the coarse bundle's drifted start.
2443    #[case::prefers_finer_grid(
2444        vec![range(1_000, 1_999), range(1_500, 3_400), range(2_000, 2_999), range(3_000, 3_999)],
2445        1_000, 3_999, Some(vec![(1_000, 1_999), (2_000, 2_999), (3_000, 3_999)])
2446    )]
2447    // A 50k snapshot bundle and the 10k bundles built on it share start 100: ride
2448    // the 10k grid instead of the coarse end, which would strand the cursor on a
2449    // slot no bundle starts at.
2450    #[case::shared_start_prefers_finer(
2451        vec![range(100, 150), range(100, 120), range(121, 140), range(141, 160)],
2452        100, 160, Some(vec![(100, 120), (121, 140), (141, 160)])
2453    )]
2454    // A coarse bundle (100, 200) overlaps a finer pair covering the same span:
2455    // split_range maximises parallelism, riding the fine pair (two sessions)
2456    // rather than merging onto the coarse max end (one session).
2457    #[case::coarse_overlap_prefers_finer_pair(
2458        vec![range(100, 200), range(100, 150), range(151, 160)],
2459        100, 160, Some(vec![(100, 150), (151, 160)])
2460    )]
2461    // The fine grid has a hole (nothing starts at 141), but a coarse bundle
2462    // sharing the start spans it: serve the request off the coarse bundle rather
2463    // than reporting an unsupported range.
2464    #[case::falls_back_to_coarse(
2465        vec![range(100, 160), range(100, 120), range(121, 140)],
2466        100, 160, Some(vec![(100, 160)])
2467    )]
2468    // The last bundle overshoots the requested end and is clamped.
2469    #[case::clamps_final_bundle(vec![range(100, 199), range(200, 999)], 100, 450, Some(vec![(100, 199), (200, 450)]))]
2470    // A request landing mid-bundle anchors to the covering bundle's start
2471    // (matching the server) instead of erroring as "not covered".
2472    #[case::anchors_mid_bundle(vec![range(150, 350)], 200, 300, Some(vec![(150, 300)]))]
2473    #[case::anchors_then_continues(
2474        vec![range(150, 350), range(351, 600)],
2475        200, 600, Some(vec![(150, 350), (351, 600)])
2476    )]
2477    #[case::start_inside_bundle_anchors(vec![range(200, 400)], 300, 400, Some(vec![(200, 400)]))]
2478    // Errors: a start before any bundle's coverage has no covering bundle...
2479    #[case::start_before_first_bundle(vec![range(200, 400)], 100, 400, None)]
2480    // ...the end must be reachable...
2481    #[case::end_not_covered(vec![range(100, 200)], 100, 300, None)]
2482    #[case::gap_in_coverage(vec![range(100, 200), range(210, 300)], 100, 300, None)]
2483    // Per-API-key isolation: a user's finer bundles and a coarse GLOBAL bundle
2484    // sharing a start are fed as SEPARATE entries (never merged on snapshot_slot,
2485    // which would keep only the max end). split_range set-unions the ends and
2486    // rides the finer user grid — a regression guard against re-merging.
2487    #[case::user_and_global_ranges_not_collapsed(
2488        vec![range(100, 200), range(100, 150), range(151, 200)],
2489        100, 200, Some(vec![(100, 150), (151, 200)])
2490    )]
2491    fn split_range_cases(
2492        #[case] ranges: Vec<AvailableRange>,
2493        #[case] start: u64,
2494        #[case] end: u64,
2495        #[case] expected: Option<Vec<(u64, u64)>>,
2496    ) {
2497        match expected {
2498            Some(expected) => assert_eq!(
2499                split_range(
2500                    &ranges,
2501                    SlotRange::new_unchecked(start, end),
2502                    BundleKind::Blocks
2503                )
2504                .unwrap(),
2505                expected
2506            ),
2507            None => assert!(
2508                split_range(
2509                    &ranges,
2510                    SlotRange::new_unchecked(start, end),
2511                    BundleKind::Blocks
2512                )
2513                .is_err()
2514            ),
2515        }
2516    }
2517
2518    fn account_state_range(start: u64, end: u64) -> AvailableRange {
2519        AvailableRange {
2520            kind: Some(BundleKind::AccountState),
2521            ..range(start, end)
2522        }
2523    }
2524
2525    #[test]
2526    fn a_replay_split_rides_the_recordings_not_a_finer_block_grid() {
2527        // The finer block grid wins on piece count, which is precisely what would
2528        // hand a replay run sub-ranges nothing ever recorded.
2529        let ranges = vec![
2530            range(100, 149),
2531            range(150, 199),
2532            account_state_range(100, 199),
2533        ];
2534        let requested = SlotRange::new_unchecked(100, 199);
2535
2536        assert_eq!(
2537            split_range(&ranges, requested, BundleKind::Blocks).unwrap(),
2538            [(100, 149), (150, 199)]
2539        );
2540        assert_eq!(
2541            split_range(&ranges, requested, BundleKind::AccountState).unwrap(),
2542            [(100, 199)]
2543        );
2544    }
2545
2546    #[test]
2547    fn a_replay_split_tiles_consecutive_recordings() {
2548        let ranges = vec![account_state_range(100, 199), account_state_range(200, 299)];
2549        assert_eq!(
2550            split_range(
2551                &ranges,
2552                SlotRange::new_unchecked(100, 299),
2553                BundleKind::AccountState
2554            )
2555            .unwrap(),
2556            [(100, 199), (200, 299)]
2557        );
2558    }
2559
2560    #[test]
2561    fn a_replay_split_ends_short_of_a_recording() {
2562        // A recording is read slot by slot, so a session stops wherever asked
2563        // inside it — the same as a block session does inside its bundle.
2564        let requested = SlotRange::new_unchecked(100, 150);
2565        assert_eq!(
2566            split_range(&[range(100, 199)], requested, BundleKind::Blocks).unwrap(),
2567            [(100, 150)]
2568        );
2569        assert_eq!(
2570            split_range(
2571                &[account_state_range(100, 199)],
2572                requested,
2573                BundleKind::AccountState
2574            )
2575            .unwrap(),
2576            [(100, 150)]
2577        );
2578    }
2579
2580    #[test]
2581    fn a_replay_split_ends_short_of_its_last_recording() {
2582        // Only the last sub-range may stop mid-recording; the ones before it
2583        // still have to hand over on the very next slot.
2584        let ranges = vec![account_state_range(100, 199), account_state_range(200, 299)];
2585        assert_eq!(
2586            split_range(
2587                &ranges,
2588                SlotRange::new_unchecked(100, 250),
2589                BundleKind::AccountState
2590            )
2591            .unwrap(),
2592            [(100, 199), (200, 250)]
2593        );
2594    }
2595
2596    #[test]
2597    fn a_replay_split_tiles_a_one_slot_recording() {
2598        // A series whose last split begins on the requested end records exactly
2599        // one slot.
2600        let ranges = vec![account_state_range(100, 249), account_state_range(250, 250)];
2601        assert_eq!(
2602            split_range(
2603                &ranges,
2604                SlotRange::new_unchecked(100, 250),
2605                BundleKind::AccountState
2606            )
2607            .unwrap(),
2608            [(100, 249), (250, 250)]
2609        );
2610    }
2611
2612    #[test]
2613    fn a_replay_split_anchors_a_mid_recording_start() {
2614        // A mid-bundle start anchors to the bundle covering it, whichever kind
2615        // the walk rides — the two differ in the bundles they read, not in the
2616        // ranges they admit.
2617        let requested = SlotRange::new_unchecked(150, 199);
2618        assert_eq!(
2619            split_range(&[range(100, 199)], requested, BundleKind::Blocks).unwrap(),
2620            [(100, 199)]
2621        );
2622        assert_eq!(
2623            split_range(
2624                &[account_state_range(100, 199)],
2625                requested,
2626                BundleKind::AccountState
2627            )
2628            .unwrap(),
2629            [(100, 199)]
2630        );
2631    }
2632
2633    #[test]
2634    fn a_replay_split_refuses_a_start_no_recording_covers() {
2635        let error = split_range(
2636            &[account_state_range(200, 299)],
2637            SlotRange::new_unchecked(100, 299),
2638            BundleKind::AccountState,
2639        )
2640        .expect_err("no recording covers 100");
2641        assert!(
2642            error.contains("is not covered by any available account-state bundle"),
2643            "{error}"
2644        );
2645    }
2646
2647    #[test]
2648    fn a_replay_split_ignores_block_ranges() {
2649        let error = split_range(
2650            &[range(100, 199)],
2651            SlotRange::new_unchecked(100, 199),
2652            BundleKind::AccountState,
2653        )
2654        .expect_err("owns no recordings");
2655        assert!(error.contains("account-state bundle"), "{error}");
2656    }
2657
2658    #[test]
2659    fn blocks_mode_admits_ranges_from_a_server_that_predates_kind() {
2660        // A server predating the field omits it, and those ranges must still
2661        // split exactly as they did before the kind was consulted.
2662        let unlabelled = range(100, 199);
2663        assert_eq!(unlabelled.kind, None);
2664        assert_eq!(
2665            split_range(
2666                &[unlabelled],
2667                SlotRange::new_unchecked(100, 150),
2668                BundleKind::Blocks
2669            )
2670            .unwrap(),
2671            [(100, 150)]
2672        );
2673    }
2674
2675    /// Servable end per bundle start, as `split_range` sees it — every advertised
2676    /// end at or past its start, keyed by start.
2677    fn ends_by_start(ranges: &[AvailableRange]) -> BTreeMap<u64, BTreeSet<u64>> {
2678        let mut ends: BTreeMap<u64, BTreeSet<u64>> = BTreeMap::new();
2679        for r in ranges {
2680            if let Some(end) = r.max_bundle_end_slot
2681                && end >= r.bundle_start_slot
2682            {
2683                ends.entry(r.bundle_start_slot).or_default().insert(end);
2684            }
2685        }
2686        ends
2687    }
2688
2689    /// Independent spec for the optimum: the gap-free split of `[start, end]` with
2690    /// the most sub-ranges, found by exhaustively trying every advertised end at
2691    /// each cursor. `None` when no split aligns with the bundle starts.
2692    fn reference_max_split(
2693        ends: &BTreeMap<u64, BTreeSet<u64>>,
2694        cursor: u64,
2695        end: u64,
2696    ) -> Option<Vec<(u64, u64)>> {
2697        ends.get(&cursor)?
2698            .iter()
2699            .filter_map(|&bundle_end| {
2700                if bundle_end >= end {
2701                    Some(vec![(cursor, end)])
2702                } else {
2703                    reference_max_split(ends, bundle_end + 1, end).map(|mut rest| {
2704                        rest.insert(0, (cursor, bundle_end));
2705                        rest
2706                    })
2707                }
2708            })
2709            .max_by_key(Vec::len)
2710    }
2711
2712    /// Structure-independent check that a split is one the server can serve: it
2713    /// tiles `[start, end]` gap-free, every start is a real bundle start, and
2714    /// every end is within that start's largest advertised end.
2715    fn is_valid_split(
2716        split: &[(u64, u64)],
2717        ends: &BTreeMap<u64, BTreeSet<u64>>,
2718        start: u64,
2719        end: u64,
2720    ) -> bool {
2721        split.first().is_some_and(|&(s, _)| s == start)
2722            && split.last().is_some_and(|&(_, e)| e == end)
2723            && split.windows(2).all(|w| w[1].0 == w[0].1 + 1)
2724            && split.iter().all(|&(s, e)| {
2725                e >= s
2726                    && ends
2727                        .get(&s)
2728                        .and_then(|bundle_ends| bundle_ends.iter().next_back())
2729                        .is_some_and(|&max_end| e <= max_end)
2730            })
2731    }
2732
2733    /// Property test: across many randomized bundle layouts, `split_range` must
2734    /// return a *valid* split with the *maximum* number of sub-ranges whenever one
2735    /// exists, and error exactly when none does. Deterministic (fixed LCG seed) so
2736    /// a failure is reproducible.
2737    #[test]
2738    fn split_range_matches_reference() {
2739        let mut seed: u64 = 0x9E3779B97F4A7C15;
2740        let mut next = || {
2741            seed = seed
2742                .wrapping_mul(6364136223846793005)
2743                .wrapping_add(1442695040888963407);
2744            seed >> 33
2745        };
2746
2747        for _ in 0..50_000 {
2748            // A small slot universe makes shared starts, overlaps, nesting, and
2749            // gaps all common.
2750            let ranges: Vec<AvailableRange> = (0..next() % 6)
2751                .map(|_| {
2752                    let start = next() % 12;
2753                    range(start, start + next() % 6) // len 0 => filtered out
2754                })
2755                .collect();
2756            let start = next() % 12;
2757            let end = start + next() % 6; // sometimes start == end, sometimes unservable
2758
2759            let got = split_range(
2760                &ranges,
2761                SlotRange::new_unchecked(start, end),
2762                BundleKind::Blocks,
2763            );
2764            let ends = ends_by_start(&ranges);
2765            // split_range anchors a mid-bundle start to the covering bundle's
2766            // start, so the reference must walk from the same anchor.
2767            let anchor = ends
2768                .range(..=start)
2769                .rfind(|(_, e)| e.iter().next_back().is_some_and(|&x| x >= start))
2770                .map(|(&s, _)| s);
2771            let reference = anchor.and_then(|a| reference_max_split(&ends, a, end));
2772
2773            let layout: Vec<_> = ranges
2774                .iter()
2775                .map(|r| (r.bundle_start_slot, r.max_bundle_end_slot))
2776                .collect();
2777            match (&got, &reference) {
2778                (Ok(split), Some(best)) => {
2779                    assert!(
2780                        is_valid_split(split, &ends, anchor.unwrap(), end),
2781                        "invalid split {split:?} for {layout:?} [{start},{end}]"
2782                    );
2783                    assert_eq!(
2784                        split.len(),
2785                        best.len(),
2786                        "suboptimal split {split:?} vs {best:?} for {layout:?} [{start},{end}]"
2787                    );
2788                }
2789                (Err(_), None) => {}
2790                _ => panic!(
2791                    "disagreement: split_range={got:?}, reference={reference:?} for {layout:?} [{start},{end}]"
2792                ),
2793            }
2794        }
2795    }
2796}