sof_tx/submit/
types.rs

1//! Shared submission types, errors, and transport traits.
2
3use std::{
4    collections::{HashMap, VecDeque},
5    hash::{Hash, Hasher},
6    sync::{
7        Arc,
8        atomic::{AtomicU64, Ordering},
9    },
10    time::{Duration, SystemTime},
11};
12
13use async_trait::async_trait;
14use serde::{Deserialize, Serialize};
15use sof_types::SignatureBytes;
16use thiserror::Error;
17
18use crate::{providers::LeaderTarget, routing::RoutingPolicy};
19
20/// Legacy runtime submit presets.
21#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
22pub enum SubmitMode {
23    /// Submit only through JSON-RPC.
24    RpcOnly,
25    /// Submit only through Jito block engine.
26    JitoOnly,
27    /// Submit only through direct leader/validator targets.
28    DirectOnly,
29    /// Submit direct first, then RPC fallback on failure.
30    Hybrid,
31}
32
33/// One concrete submit route.
34#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, Serialize, Deserialize)]
35pub enum SubmitRoute {
36    /// JSON-RPC submission.
37    Rpc,
38    /// Jito block-engine submission.
39    Jito,
40    /// Direct leader/validator submission.
41    Direct,
42}
43
44/// Route execution policy for one submission attempt.
45#[derive(Debug, Clone, Copy, Eq, PartialEq, Default, Serialize, Deserialize)]
46pub enum SubmitStrategy {
47    /// Execute routes in order and stop at the first accepted route.
48    #[default]
49    OrderedFallback,
50    /// Execute all configured routes at the same time.
51    AllAtOnce,
52}
53
54/// Route-plan based submission configuration.
55#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
56pub struct SubmitPlan {
57    /// Configured submit routes.
58    pub routes: Vec<SubmitRoute>,
59    /// Route execution strategy.
60    pub strategy: SubmitStrategy,
61}
62
63impl SubmitPlan {
64    /// Creates one normalized plan.
65    #[must_use]
66    pub fn new(routes: Vec<SubmitRoute>, strategy: SubmitStrategy) -> Self {
67        Self { routes, strategy }.into_normalized()
68    }
69
70    /// Normalizes this plan in place while preserving route order.
71    #[must_use]
72    pub fn into_normalized(mut self) -> Self {
73        let mut seen_rpc = false;
74        let mut seen_jito = false;
75        let mut seen_direct = false;
76        self.routes.retain(|route| match route {
77            SubmitRoute::Rpc if seen_rpc => false,
78            SubmitRoute::Rpc => {
79                seen_rpc = true;
80                true
81            }
82            SubmitRoute::Jito if seen_jito => false,
83            SubmitRoute::Jito => {
84                seen_jito = true;
85                true
86            }
87            SubmitRoute::Direct if seen_direct => false,
88            SubmitRoute::Direct => {
89                seen_direct = true;
90                true
91            }
92        });
93        self
94    }
95
96    /// Returns a normalized clone of this plan.
97    #[must_use]
98    pub fn normalized(&self) -> Self {
99        self.clone().into_normalized()
100    }
101
102    /// Builds a single-route RPC fallback plan.
103    #[must_use]
104    pub fn rpc_only() -> Self {
105        Self::new(vec![SubmitRoute::Rpc], SubmitStrategy::OrderedFallback)
106    }
107
108    /// Builds a single-route Jito plan.
109    #[must_use]
110    pub fn jito_only() -> Self {
111        Self::new(vec![SubmitRoute::Jito], SubmitStrategy::OrderedFallback)
112    }
113
114    /// Builds a single-route direct plan.
115    #[must_use]
116    pub fn direct_only() -> Self {
117        Self::new(vec![SubmitRoute::Direct], SubmitStrategy::OrderedFallback)
118    }
119
120    /// Builds one custom ordered-fallback plan.
121    #[must_use]
122    pub fn ordered(routes: Vec<SubmitRoute>) -> Self {
123        Self::new(routes, SubmitStrategy::OrderedFallback)
124    }
125
126    /// Builds the legacy direct-then-RPC fallback plan.
127    #[must_use]
128    pub fn hybrid() -> Self {
129        Self::ordered(vec![SubmitRoute::Direct, SubmitRoute::Rpc])
130    }
131
132    /// Builds one concurrent all-route plan.
133    #[must_use]
134    pub fn all_at_once(routes: Vec<SubmitRoute>) -> Self {
135        Self::new(routes, SubmitStrategy::AllAtOnce)
136    }
137
138    /// Returns the matching legacy preset when this plan is one exact legacy shape.
139    #[must_use]
140    pub fn legacy_mode(&self) -> Option<SubmitMode> {
141        match (self.strategy, self.routes.as_slice()) {
142            (SubmitStrategy::OrderedFallback, [SubmitRoute::Rpc]) => Some(SubmitMode::RpcOnly),
143            (SubmitStrategy::OrderedFallback, [SubmitRoute::Jito]) => Some(SubmitMode::JitoOnly),
144            (SubmitStrategy::OrderedFallback, [SubmitRoute::Direct]) => {
145                Some(SubmitMode::DirectOnly)
146            }
147            (SubmitStrategy::OrderedFallback, [SubmitRoute::Direct, SubmitRoute::Rpc]) => {
148                Some(SubmitMode::Hybrid)
149            }
150            _ => None,
151        }
152    }
153}
154
155impl Default for SubmitPlan {
156    fn default() -> Self {
157        Self::rpc_only()
158    }
159}
160
161impl From<SubmitMode> for SubmitPlan {
162    fn from(value: SubmitMode) -> Self {
163        match value {
164            SubmitMode::RpcOnly => Self::rpc_only(),
165            SubmitMode::JitoOnly => Self::jito_only(),
166            SubmitMode::DirectOnly => Self::direct_only(),
167            SubmitMode::Hybrid => Self::hybrid(),
168        }
169    }
170}
171
172/// Reliability profile for direct and hybrid submission behavior.
173#[derive(Debug, Clone, Copy, Eq, PartialEq, Default, Serialize, Deserialize)]
174pub enum SubmitReliability {
175    /// Fastest path with minimal retrying.
176    LowLatency,
177    /// Balanced latency and retry behavior.
178    #[default]
179    Balanced,
180    /// Aggressive retrying before giving up.
181    HighReliability,
182}
183
184/// Signed transaction payload variants accepted by submit APIs.
185#[derive(Debug, Clone, Eq, PartialEq)]
186pub enum SignedTx {
187    /// Bincode-serialized `VersionedTransaction` bytes.
188    VersionedTransactionBytes(Vec<u8>),
189    /// Wire-format transaction bytes.
190    WireTransactionBytes(Vec<u8>),
191}
192
193/// RPC submit tuning.
194#[derive(Debug, Clone, Eq, PartialEq)]
195pub struct RpcSubmitConfig {
196    /// Skip preflight simulation when true.
197    pub skip_preflight: bool,
198    /// Optional preflight commitment string.
199    pub preflight_commitment: Option<String>,
200}
201
202impl Default for RpcSubmitConfig {
203    fn default() -> Self {
204        Self {
205            skip_preflight: true,
206            preflight_commitment: None,
207        }
208    }
209}
210
211/// Jito block-engine submit tuning.
212#[derive(Debug, Clone, Eq, PartialEq, Default)]
213pub struct JitoSubmitConfig {
214    /// Enables revert protection by sending through bundle-only mode.
215    ///
216    /// This applies to the JSON-RPC transaction path. The gRPC transport is bundle-based and
217    /// therefore always behaves as a bundle submission.
218    pub bundle_only: bool,
219}
220
221/// Successful Jito submission metadata.
222#[derive(Debug, Clone, Eq, PartialEq, Default)]
223pub struct JitoSubmitResponse {
224    /// Jito JSON-RPC returned transaction signature, when available.
225    pub transaction_signature: Option<String>,
226    /// Jito gRPC returned bundle UUID, when available.
227    pub bundle_id: Option<String>,
228}
229
230/// Direct submit tuning.
231#[derive(Debug, Clone, Eq, PartialEq)]
232pub struct DirectSubmitConfig {
233    /// Per-target send timeout.
234    pub per_target_timeout: Duration,
235    /// Global send budget for one submission.
236    pub global_timeout: Duration,
237    /// Number of rounds to iterate across selected direct targets.
238    pub direct_target_rounds: usize,
239    /// Number of direct-only submit attempts (target selection can refresh per attempt).
240    pub direct_submit_attempts: usize,
241    /// Number of direct submit attempts in `Hybrid` mode before RPC fallback.
242    pub hybrid_direct_attempts: usize,
243    /// Delay between direct rebroadcast attempts/rounds (Agave-like pacing).
244    pub rebroadcast_interval: Duration,
245    /// Enables Agave-style post-ack rebroadcast persistence for direct submits.
246    pub agave_rebroadcast_enabled: bool,
247    /// Maximum time budget for background rebroadcast persistence.
248    pub agave_rebroadcast_window: Duration,
249    /// Delay between background rebroadcast cycles.
250    pub agave_rebroadcast_interval: Duration,
251    /// When true, `Hybrid` mode broadcasts to RPC even after direct send succeeds.
252    pub hybrid_rpc_broadcast: bool,
253    /// Enables latency-aware ordering of direct targets before submit.
254    pub latency_aware_targeting: bool,
255    /// Timeout used for per-target TCP latency probes.
256    pub latency_probe_timeout: Duration,
257    /// Optional extra TCP port to probe (in addition to target TPU port).
258    pub latency_probe_port: Option<u16>,
259    /// Max number of targets to probe per submission.
260    pub latency_probe_max_targets: usize,
261}
262
263impl DirectSubmitConfig {
264    /// Builds a direct-submit config from a reliability profile.
265    #[must_use]
266    pub const fn from_reliability(reliability: SubmitReliability) -> Self {
267        match reliability {
268            SubmitReliability::LowLatency => Self {
269                per_target_timeout: Duration::from_millis(200),
270                global_timeout: Duration::from_millis(1_200),
271                direct_target_rounds: 3,
272                direct_submit_attempts: 3,
273                hybrid_direct_attempts: 2,
274                rebroadcast_interval: Duration::from_millis(90),
275                agave_rebroadcast_enabled: true,
276                agave_rebroadcast_window: Duration::from_secs(30),
277                agave_rebroadcast_interval: Duration::from_millis(700),
278                hybrid_rpc_broadcast: false,
279                latency_aware_targeting: true,
280                latency_probe_timeout: Duration::from_millis(80),
281                latency_probe_port: Some(8899),
282                latency_probe_max_targets: 128,
283            },
284            SubmitReliability::Balanced => Self {
285                per_target_timeout: Duration::from_millis(300),
286                global_timeout: Duration::from_millis(1_800),
287                direct_target_rounds: 4,
288                direct_submit_attempts: 4,
289                hybrid_direct_attempts: 3,
290                rebroadcast_interval: Duration::from_millis(110),
291                agave_rebroadcast_enabled: true,
292                agave_rebroadcast_window: Duration::from_secs(45),
293                agave_rebroadcast_interval: Duration::from_millis(800),
294                hybrid_rpc_broadcast: true,
295                latency_aware_targeting: true,
296                latency_probe_timeout: Duration::from_millis(120),
297                latency_probe_port: Some(8899),
298                latency_probe_max_targets: 128,
299            },
300            SubmitReliability::HighReliability => Self {
301                per_target_timeout: Duration::from_millis(450),
302                global_timeout: Duration::from_millis(3_200),
303                direct_target_rounds: 6,
304                direct_submit_attempts: 5,
305                hybrid_direct_attempts: 4,
306                rebroadcast_interval: Duration::from_millis(140),
307                agave_rebroadcast_enabled: true,
308                agave_rebroadcast_window: Duration::from_secs(70),
309                agave_rebroadcast_interval: Duration::from_millis(900),
310                hybrid_rpc_broadcast: true,
311                latency_aware_targeting: true,
312                latency_probe_timeout: Duration::from_millis(160),
313                latency_probe_port: Some(8899),
314                latency_probe_max_targets: 128,
315            },
316        }
317    }
318
319    /// Returns this config with minimum valid retry counters.
320    #[must_use]
321    pub const fn normalized(self) -> Self {
322        let per_target_timeout = if self.per_target_timeout.is_zero() {
323            Duration::from_millis(1)
324        } else {
325            self.per_target_timeout
326        };
327        let global_timeout = if self.global_timeout.is_zero() {
328            Duration::from_millis(1)
329        } else {
330            self.global_timeout
331        };
332        let direct_target_rounds = if self.direct_target_rounds == 0 {
333            1
334        } else {
335            self.direct_target_rounds
336        };
337        let direct_submit_attempts = if self.direct_submit_attempts == 0 {
338            1
339        } else {
340            self.direct_submit_attempts
341        };
342        let hybrid_direct_attempts = if self.hybrid_direct_attempts == 0 {
343            1
344        } else {
345            self.hybrid_direct_attempts
346        };
347        let latency_probe_max_targets = if self.latency_probe_max_targets == 0 {
348            1
349        } else {
350            self.latency_probe_max_targets
351        };
352        let rebroadcast_interval = if self.rebroadcast_interval.is_zero() {
353            Duration::from_millis(1)
354        } else {
355            self.rebroadcast_interval
356        };
357        let agave_rebroadcast_interval = if self.agave_rebroadcast_interval.is_zero() {
358            Duration::from_millis(1)
359        } else {
360            self.agave_rebroadcast_interval
361        };
362        let latency_probe_timeout = if self.latency_probe_timeout.is_zero() {
363            Duration::from_millis(1)
364        } else {
365            self.latency_probe_timeout
366        };
367        Self {
368            per_target_timeout,
369            global_timeout,
370            direct_target_rounds,
371            direct_submit_attempts,
372            hybrid_direct_attempts,
373            rebroadcast_interval,
374            agave_rebroadcast_enabled: self.agave_rebroadcast_enabled,
375            agave_rebroadcast_window: self.agave_rebroadcast_window,
376            agave_rebroadcast_interval,
377            hybrid_rpc_broadcast: self.hybrid_rpc_broadcast,
378            latency_aware_targeting: self.latency_aware_targeting,
379            latency_probe_timeout,
380            latency_probe_port: self.latency_probe_port,
381            latency_probe_max_targets,
382        }
383    }
384}
385
386impl Default for DirectSubmitConfig {
387    fn default() -> Self {
388        Self::from_reliability(SubmitReliability::default())
389    }
390}
391
392/// Low-level transport errors surfaced by submit backends.
393#[derive(Debug, Error, Clone, Eq, PartialEq)]
394pub enum SubmitTransportError {
395    /// Invalid transport configuration.
396    #[error("transport configuration invalid: {message}")]
397    Config {
398        /// Human-readable description.
399        message: String,
400    },
401    /// Transport operation failed.
402    #[error("transport failure: {message}")]
403    Failure {
404        /// Human-readable description.
405        message: String,
406    },
407}
408
409/// Submission-level errors.
410#[derive(Debug, Error)]
411pub enum SubmitError {
412    /// No blockhash available for unsigned submit path.
413    #[error("blockhash provider returned no recent blockhash")]
414    MissingRecentBlockhash,
415    /// Signed bytes could not be decoded into a transaction.
416    #[error("failed to decode signed transaction bytes: {source}")]
417    DecodeSignedBytes {
418        /// Bincode decode error.
419        source: Box<bincode::ErrorKind>,
420    },
421    /// Duplicate signature was suppressed by dedupe window.
422    #[error("duplicate signature suppressed by dedupe window")]
423    DuplicateSignature,
424    /// RPC mode requested but no RPC transport was configured.
425    #[error("rpc transport is not configured")]
426    MissingRpcTransport,
427    /// Jito mode requested but no Jito transport was configured.
428    #[error("jito transport is not configured")]
429    MissingJitoTransport,
430    /// Direct mode requested but no direct transport was configured.
431    #[error("direct transport is not configured")]
432    MissingDirectTransport,
433    /// No direct targets resolved from routing inputs.
434    #[error("no direct targets resolved from leader/backups")]
435    NoDirectTargets,
436    /// Direct transport failure.
437    #[error("direct submit failed: {source}")]
438    Direct {
439        /// Direct transport error.
440        source: SubmitTransportError,
441    },
442    /// RPC transport failure.
443    #[error("rpc submit failed: {source}")]
444    Rpc {
445        /// RPC transport error.
446        source: SubmitTransportError,
447    },
448    /// Jito transport failure.
449    #[error("jito submit failed: {source}")]
450    Jito {
451        /// Jito transport error.
452        source: SubmitTransportError,
453    },
454    /// Internal synchronization failure.
455    #[error("internal synchronization failure: {message}")]
456    InternalSync {
457        /// Synchronization error details.
458        message: String,
459    },
460    /// Submit attempt was rejected by the toxic-flow guard.
461    #[error("submission rejected by toxic-flow guard: {reason}")]
462    ToxicFlow {
463        /// Structured reason for the rejection.
464        reason: TxToxicFlowRejectionReason,
465    },
466}
467
468/// Summary of a successful submission.
469#[derive(Debug, Clone, Eq, PartialEq)]
470pub struct SubmitResult {
471    /// Signature parsed from submitted transaction bytes.
472    pub signature: Option<SignatureBytes>,
473    /// Route plan selected by caller.
474    pub plan: SubmitPlan,
475    /// Legacy preset used by caller when one exact preset was selected.
476    pub legacy_mode: Option<SubmitMode>,
477    /// First successful route observed before this submit call returned.
478    pub first_success_route: Option<SubmitRoute>,
479    ///
480    /// Later background accepts from concurrently configured routes are counted by built-in
481    /// telemetry and are best-effort through [`TxSubmitOutcomeReporter`], including any
482    /// route-specific acceptance metadata, rather than retroactively mutating this synchronous
483    /// result. Reporter delivery failures are reflected in
484    /// [`TxToxicFlowTelemetrySnapshot::reporter_outcomes_dropped`] and
485    /// [`TxToxicFlowTelemetrySnapshot::reporter_outcomes_unavailable`].
486    pub successful_routes: Vec<SubmitRoute>,
487    /// Target chosen by direct path when applicable.
488    pub direct_target: Option<LeaderTarget>,
489    /// RPC-returned signature string when RPC path succeeded before return.
490    pub rpc_signature: Option<String>,
491    /// Jito block-engine returned transaction signature when the JSON-RPC path succeeded before
492    /// return.
493    pub jito_signature: Option<String>,
494    /// Jito block-engine returned bundle UUID when the gRPC bundle path succeeded before return.
495    pub jito_bundle_id: Option<String>,
496    /// True when one later ordered-fallback route accepted the submit before return.
497    pub used_fallback_route: bool,
498    /// Number of direct targets selected for submit attempt that succeeded.
499    pub selected_target_count: usize,
500    /// Number of unique validator identities in selected direct targets.
501    pub selected_identity_count: usize,
502}
503
504/// Coarse toxic-flow quality used by submit guards.
505#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
506pub enum TxFlowSafetyQuality {
507    /// Required inputs are present, coherent, and safe to use.
508    Stable,
509    /// Inputs exist but have not reached a stable confirmation boundary yet.
510    Provisional,
511    /// Inputs exist but the current branch still carries material reorg risk.
512    ReorgRisk,
513    /// Inputs are present but stale.
514    Stale,
515    /// Inputs are present but mutually inconsistent.
516    Degraded,
517    /// Required inputs are still missing.
518    IncompleteControlPlane,
519}
520
521/// One concrete toxic-flow issue reported by a submit guard source.
522#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
523pub enum TxFlowSafetyIssue {
524    /// Submit source is still recovering from replay/checkpoint continuity.
525    ReplayRecoveryPending,
526    /// Control-plane inputs are still missing.
527    MissingControlPlane,
528    /// Control-plane inputs are stale.
529    StaleControlPlane,
530    /// Control-plane inputs are inconsistent.
531    DegradedControlPlane,
532    /// Current branch carries reorg risk.
533    ReorgRisk,
534    /// Current branch is still provisional.
535    Provisional,
536}
537
538/// Current toxic-flow safety snapshot exposed by one submit guard source.
539#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
540pub struct TxFlowSafetySnapshot {
541    /// Coarse quality classification for the current control-plane view.
542    pub quality: TxFlowSafetyQuality,
543    /// Concrete issues behind `quality`.
544    pub issues: Vec<TxFlowSafetyIssue>,
545    /// Current upstream state version when known.
546    pub current_state_version: Option<u64>,
547    /// True when replay recovery is still pending.
548    pub replay_recovery_pending: bool,
549}
550
551impl TxFlowSafetySnapshot {
552    /// Returns true when the current snapshot is strategy-safe.
553    #[must_use]
554    pub const fn is_safe(&self) -> bool {
555        matches!(self.quality, TxFlowSafetyQuality::Stable) && !self.replay_recovery_pending
556    }
557}
558
559/// Dynamic source of toxic-flow safety state for submit guards.
560pub trait TxFlowSafetySource: Send + Sync {
561    /// Returns the latest toxic-flow safety snapshot.
562    fn toxic_flow_snapshot(&self) -> TxFlowSafetySnapshot;
563}
564
565/// One key used to suppress repeated submission attempts for the same opportunity.
566#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
567pub enum TxSubmitSuppressionKey {
568    /// Suppress by signature.
569    Signature(SignatureBytes),
570    /// Suppress by opaque opportunity identifier.
571    Opportunity([u8; 32]),
572    /// Suppress by hashed account set identifier.
573    AccountSet([u8; 32]),
574    /// Suppress by slot-window key.
575    SlotWindow {
576        /// Slot associated with the opportunity.
577        slot: u64,
578        /// Window width used to group nearby opportunities.
579        window: u64,
580    },
581}
582
583impl Hash for TxSubmitSuppressionKey {
584    fn hash<H: Hasher>(&self, state: &mut H) {
585        match self {
586            Self::Signature(signature) => {
587                0_u8.hash(state);
588                signature.as_array().hash(state);
589            }
590            Self::Opportunity(key) => {
591                1_u8.hash(state);
592                key.hash(state);
593            }
594            Self::AccountSet(key) => {
595                2_u8.hash(state);
596                key.hash(state);
597            }
598            Self::SlotWindow { slot, window } => {
599                3_u8.hash(state);
600                slot.hash(state);
601                window.hash(state);
602            }
603        }
604    }
605}
606
607/// Call-site context used by toxic-flow guards.
608#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
609pub struct TxSubmitContext {
610    /// Additional suppression keys to apply before submit.
611    pub suppression_keys: Vec<TxSubmitSuppressionKey>,
612    /// State version used when the submit decision was made.
613    pub decision_state_version: Option<u64>,
614    /// Timestamp when the opportunity or decision was created.
615    pub opportunity_created_at: Option<SystemTime>,
616}
617
618/// Policy controlling toxic-flow submit rejection.
619#[derive(Debug, Clone, Eq, PartialEq)]
620pub struct TxSubmitGuardPolicy {
621    /// Reject when the flow-safety source is not `Stable`.
622    pub require_stable_control_plane: bool,
623    /// Reject when the flow-safety source reports replay recovery pending.
624    pub reject_on_replay_recovery_pending: bool,
625    /// Maximum allowed drift between decision and current state versions.
626    pub max_state_version_drift: Option<u64>,
627    /// Maximum allowed age for one opportunity before submit.
628    pub max_opportunity_age: Option<Duration>,
629    /// TTL applied to built-in suppression keys.
630    pub suppression_ttl: Duration,
631}
632
633impl Default for TxSubmitGuardPolicy {
634    fn default() -> Self {
635        Self {
636            require_stable_control_plane: true,
637            reject_on_replay_recovery_pending: true,
638            max_state_version_drift: Some(4),
639            max_opportunity_age: Some(Duration::from_millis(750)),
640            suppression_ttl: Duration::from_millis(750),
641        }
642    }
643}
644
645/// Concrete reason one submit attempt was rejected before transport.
646#[derive(Debug, Error, Clone, Eq, PartialEq, Serialize, Deserialize)]
647pub enum TxToxicFlowRejectionReason {
648    /// Control-plane quality is not safe enough.
649    #[error("control-plane quality {quality:?} is not safe for submit")]
650    UnsafeControlPlane {
651        /// Current quality observed from the guard source.
652        quality: TxFlowSafetyQuality,
653    },
654    /// Replay recovery is still pending.
655    #[error("submit source is still recovering replay continuity")]
656    ReplayRecoveryPending,
657    /// One suppression key is still active.
658    #[error("submit suppressed by active key")]
659    Suppressed,
660    /// State version drift exceeded policy.
661    #[error("state version drift {drift} exceeded maximum {max_allowed}")]
662    StateDrift {
663        /// Observed drift between decision and current state versions.
664        drift: u64,
665        /// Maximum drift allowed by policy.
666        max_allowed: u64,
667    },
668    /// Opportunity age exceeded policy.
669    #[error("opportunity age {age_ms}ms exceeded maximum {max_allowed_ms}ms")]
670    OpportunityStale {
671        /// Observed age in milliseconds.
672        age_ms: u64,
673        /// Maximum age allowed in milliseconds.
674        max_allowed_ms: u64,
675    },
676}
677
678/// Final or immediate outcome classification for one submit attempt.
679#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
680pub enum TxSubmitOutcomeKind {
681    /// Direct path accepted the transaction.
682    DirectAccepted,
683    /// RPC path accepted the transaction.
684    RpcAccepted,
685    /// Jito block-engine accepted the transaction.
686    JitoAccepted,
687    /// Transaction landed on chain.
688    Landed,
689    /// Transaction expired before landing.
690    Expired,
691    /// Transaction was dropped before landing.
692    Dropped,
693    /// Route missed the intended leader window.
694    LeaderMissed,
695    /// Submit used a stale blockhash.
696    BlockhashStale,
697    /// Selected route was unhealthy.
698    UnhealthyRoute,
699    /// Submit was rejected due to stale inputs.
700    RejectedDueToStaleness,
701    /// Submit was rejected due to reorg risk.
702    RejectedDueToReorgRisk,
703    /// Submit was rejected due to state drift.
704    RejectedDueToStateDrift,
705    /// Submit was rejected by replay recovery pending.
706    RejectedDueToReplayRecovery,
707    /// Submit was suppressed by a built-in key.
708    Suppressed,
709}
710
711/// Structured outcome record for toxic-flow telemetry/reporting.
712///
713/// Route-level accepts may carry richer metadata than the synchronous [`SubmitResult`] when one
714/// different route already returned first. Consumers that care about later accepts should observe
715/// this surface rather than expecting the original [`SubmitResult`] to mutate.
716#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
717pub struct TxSubmitOutcome {
718    /// Outcome classification.
719    pub kind: TxSubmitOutcomeKind,
720    /// Transaction signature when available.
721    pub signature: Option<SignatureBytes>,
722    /// Concrete accepted route when the outcome came from one route-level success.
723    pub route: Option<SubmitRoute>,
724    /// Route plan selected for the submit attempt.
725    pub plan: SubmitPlan,
726    /// Legacy preset used for the submit attempt when applicable.
727    pub legacy_mode: Option<SubmitMode>,
728    /// RPC-returned signature metadata when the RPC route accepted.
729    pub rpc_signature: Option<String>,
730    /// Jito-returned transaction signature metadata when the Jito route accepted.
731    pub jito_signature: Option<String>,
732    /// Jito-returned bundle UUID when the gRPC bundle route accepted.
733    pub jito_bundle_id: Option<String>,
734    /// Current state version at outcome time when known.
735    pub state_version: Option<u64>,
736    /// Opportunity age in milliseconds when known.
737    pub opportunity_age_ms: Option<u64>,
738}
739
740/// Callback surface for external outcome sinks.
741///
742/// `TxSubmitClient` delivers these callbacks asynchronously through one bounded FIFO dispatcher per
743/// reporter instance, shared across clients that use that same reporter. That keeps the reporter
744/// off the synchronous submit hot path while preserving outcome order for events that are
745/// successfully queued. Under sustained pressure or dispatcher startup failure, reporter delivery
746/// is best-effort and built-in telemetry remains the authoritative always-inline outcome counter
747/// surface.
748pub trait TxSubmitOutcomeReporter: Send + Sync {
749    /// Records one structured outcome.
750    fn record_outcome(&self, outcome: &TxSubmitOutcome);
751}
752
753/// Snapshot of built-in toxic-flow counters collected by [`TxSubmitClient`](crate::submit::TxSubmitClient).
754#[derive(Debug, Clone, Copy, Default, Eq, PartialEq, Serialize, Deserialize)]
755pub struct TxToxicFlowTelemetrySnapshot {
756    /// Number of external reporter outcomes dropped because the reporter queue was full.
757    pub reporter_outcomes_dropped: u64,
758    /// Number of outcomes that could not reach the external reporter because the worker was unavailable.
759    pub reporter_outcomes_unavailable: u64,
760    /// Number of direct-route accepts observed.
761    pub direct_accepted: u64,
762    /// Number of RPC-route accepts observed.
763    pub rpc_accepted: u64,
764    /// Number of Jito-route accepts observed.
765    pub jito_accepted: u64,
766    /// Number of submit attempts rejected due to stale inputs.
767    pub rejected_due_to_staleness: u64,
768    /// Number of submit attempts rejected due to reorg risk.
769    pub rejected_due_to_reorg_risk: u64,
770    /// Number of submit attempts rejected due to state drift.
771    pub rejected_due_to_state_drift: u64,
772    /// Number of accepted submits emitted while the source was stale.
773    pub submit_on_stale_blockhash: u64,
774    /// Number of route misses classified as leader misses.
775    pub leader_route_miss_rate: u64,
776    /// Most recent opportunity age seen at submit time.
777    pub opportunity_age_at_send_ms: Option<u64>,
778    /// Number of submits rejected while replay recovery was pending.
779    pub rejected_due_to_replay_recovery: u64,
780    /// Number of submits suppressed by a built-in key.
781    pub suppressed_submissions: u64,
782}
783
784/// One cache-line-aligned atomic counter used by hot telemetry updates.
785#[derive(Debug, Default)]
786#[repr(align(64))]
787struct CacheAlignedAtomicU64(AtomicU64);
788
789impl CacheAlignedAtomicU64 {
790    /// Loads the current counter value.
791    fn load(&self, ordering: Ordering) -> u64 {
792        self.0.load(ordering)
793    }
794
795    /// Stores a new value and returns the previous one.
796    fn swap(&self, value: u64, ordering: Ordering) -> u64 {
797        self.0.swap(value, ordering)
798    }
799
800    /// Increments the counter and returns the previous value.
801    fn fetch_add(&self, value: u64, ordering: Ordering) -> u64 {
802        self.0.fetch_add(value, ordering)
803    }
804}
805
806/// In-memory telemetry counters for toxic-flow outcomes.
807#[derive(Debug, Default)]
808pub struct TxToxicFlowTelemetry {
809    /// Number of external reporter outcomes dropped because the reporter queue was full.
810    reporter_outcomes_dropped: CacheAlignedAtomicU64,
811    /// Number of outcomes that could not reach the external reporter because the worker was unavailable.
812    reporter_outcomes_unavailable: CacheAlignedAtomicU64,
813    /// Number of direct-route accepts.
814    direct_accepted: CacheAlignedAtomicU64,
815    /// Number of RPC-route accepts.
816    rpc_accepted: CacheAlignedAtomicU64,
817    /// Number of Jito-route accepts.
818    jito_accepted: CacheAlignedAtomicU64,
819    /// Number of stale-input rejections.
820    rejected_due_to_staleness: CacheAlignedAtomicU64,
821    /// Number of reorg-risk rejections.
822    rejected_due_to_reorg_risk: CacheAlignedAtomicU64,
823    /// Number of state-drift rejections.
824    rejected_due_to_state_drift: CacheAlignedAtomicU64,
825    /// Number of accepted submits observed with stale blockhash state.
826    submit_on_stale_blockhash: CacheAlignedAtomicU64,
827    /// Number of leader-route misses.
828    leader_route_miss_rate: CacheAlignedAtomicU64,
829    /// Last opportunity age seen by the client.
830    opportunity_age_at_send_ms: CacheAlignedAtomicU64,
831    /// Number of replay-recovery rejections.
832    rejected_due_to_replay_recovery: CacheAlignedAtomicU64,
833    /// Number of suppressed submissions.
834    suppressed_submissions: CacheAlignedAtomicU64,
835}
836
837impl TxToxicFlowTelemetry {
838    /// Returns a shareable telemetry sink.
839    #[must_use]
840    pub fn shared() -> Arc<Self> {
841        Arc::new(Self::default())
842    }
843
844    /// Records one structured outcome.
845    pub fn record(&self, outcome: &TxSubmitOutcome) {
846        if let Some(age_ms) = outcome.opportunity_age_ms {
847            let _ = self
848                .opportunity_age_at_send_ms
849                .swap(age_ms, Ordering::Relaxed);
850        }
851        match outcome.kind {
852            TxSubmitOutcomeKind::DirectAccepted => {
853                let _ = self.direct_accepted.fetch_add(1, Ordering::Relaxed);
854            }
855            TxSubmitOutcomeKind::RpcAccepted => {
856                let _ = self.rpc_accepted.fetch_add(1, Ordering::Relaxed);
857            }
858            TxSubmitOutcomeKind::JitoAccepted => {
859                let _ = self.jito_accepted.fetch_add(1, Ordering::Relaxed);
860            }
861            TxSubmitOutcomeKind::RejectedDueToStaleness => {
862                let _ = self
863                    .rejected_due_to_staleness
864                    .fetch_add(1, Ordering::Relaxed);
865            }
866            TxSubmitOutcomeKind::RejectedDueToReorgRisk => {
867                let _ = self
868                    .rejected_due_to_reorg_risk
869                    .fetch_add(1, Ordering::Relaxed);
870            }
871            TxSubmitOutcomeKind::RejectedDueToStateDrift => {
872                let _ = self
873                    .rejected_due_to_state_drift
874                    .fetch_add(1, Ordering::Relaxed);
875            }
876            TxSubmitOutcomeKind::RejectedDueToReplayRecovery => {
877                let _ = self
878                    .rejected_due_to_replay_recovery
879                    .fetch_add(1, Ordering::Relaxed);
880            }
881            TxSubmitOutcomeKind::Suppressed => {
882                let _ = self.suppressed_submissions.fetch_add(1, Ordering::Relaxed);
883            }
884            TxSubmitOutcomeKind::LeaderMissed => {
885                let _ = self.leader_route_miss_rate.fetch_add(1, Ordering::Relaxed);
886            }
887            TxSubmitOutcomeKind::Landed
888            | TxSubmitOutcomeKind::Expired
889            | TxSubmitOutcomeKind::Dropped
890            | TxSubmitOutcomeKind::UnhealthyRoute => {}
891            TxSubmitOutcomeKind::BlockhashStale => {
892                let _ = self
893                    .submit_on_stale_blockhash
894                    .fetch_add(1, Ordering::Relaxed);
895            }
896        }
897    }
898
899    /// Records one dropped external-reporter outcome.
900    pub(crate) fn record_reporter_drop(&self) {
901        let _ = self
902            .reporter_outcomes_dropped
903            .fetch_add(1, Ordering::Relaxed);
904    }
905
906    /// Records one unavailable external-reporter outcome.
907    pub(crate) fn record_reporter_unavailable(&self) {
908        let _ = self
909            .reporter_outcomes_unavailable
910            .fetch_add(1, Ordering::Relaxed);
911    }
912
913    /// Returns the current telemetry snapshot.
914    #[must_use]
915    pub fn snapshot(&self) -> TxToxicFlowTelemetrySnapshot {
916        let age_ms = self.opportunity_age_at_send_ms.load(Ordering::Relaxed);
917        TxToxicFlowTelemetrySnapshot {
918            reporter_outcomes_dropped: self.reporter_outcomes_dropped.load(Ordering::Relaxed),
919            reporter_outcomes_unavailable: self
920                .reporter_outcomes_unavailable
921                .load(Ordering::Relaxed),
922            direct_accepted: self.direct_accepted.load(Ordering::Relaxed),
923            rpc_accepted: self.rpc_accepted.load(Ordering::Relaxed),
924            jito_accepted: self.jito_accepted.load(Ordering::Relaxed),
925            rejected_due_to_staleness: self.rejected_due_to_staleness.load(Ordering::Relaxed),
926            rejected_due_to_reorg_risk: self.rejected_due_to_reorg_risk.load(Ordering::Relaxed),
927            rejected_due_to_state_drift: self.rejected_due_to_state_drift.load(Ordering::Relaxed),
928            submit_on_stale_blockhash: self.submit_on_stale_blockhash.load(Ordering::Relaxed),
929            leader_route_miss_rate: self.leader_route_miss_rate.load(Ordering::Relaxed),
930            opportunity_age_at_send_ms: if age_ms == 0 { None } else { Some(age_ms) },
931            rejected_due_to_replay_recovery: self
932                .rejected_due_to_replay_recovery
933                .load(Ordering::Relaxed),
934            suppressed_submissions: self.suppressed_submissions.load(Ordering::Relaxed),
935        }
936    }
937}
938
939impl TxSubmitOutcomeReporter for TxToxicFlowTelemetry {
940    fn record_outcome(&self, outcome: &TxSubmitOutcome) {
941        self.record(outcome);
942    }
943}
944
945/// Internal suppression map used by the submit client.
946#[derive(Debug, Default)]
947pub(crate) struct TxSuppressionCache {
948    /// Active suppression entries keyed by opportunity identity.
949    entries: HashMap<TxSubmitSuppressionKey, SystemTime>,
950    /// Insertion order for eviction, including stale superseded timestamps.
951    order: VecDeque<(TxSubmitSuppressionKey, SystemTime)>,
952}
953
954impl TxSuppressionCache {
955    /// Returns true when at least one key is still active inside `ttl`.
956    pub(crate) fn is_suppressed(
957        &mut self,
958        keys: &[TxSubmitSuppressionKey],
959        now: SystemTime,
960        ttl: Duration,
961    ) -> bool {
962        self.evict_expired(now, ttl);
963        keys.iter().any(|key| self.entries.contains_key(key))
964    }
965
966    /// Inserts all provided suppression keys with the current timestamp.
967    pub(crate) fn insert_all(&mut self, keys: &[TxSubmitSuppressionKey], now: SystemTime) {
968        for key in keys {
969            let _ = self.entries.insert(key.clone(), now);
970            self.order.push_back((key.clone(), now));
971        }
972    }
973
974    /// Removes entries older than the current TTL window.
975    fn evict_expired(&mut self, now: SystemTime, ttl: Duration) {
976        while let Some((_, front_inserted_at)) = self.order.front() {
977            let still_live = now
978                .duration_since(*front_inserted_at)
979                .map(|elapsed| elapsed <= ttl)
980                .unwrap_or(false);
981            if still_live {
982                break;
983            }
984            let Some((key, queued_inserted_at)) = self.order.pop_front() else {
985                break;
986            };
987            if self.entries.get(&key) == Some(&queued_inserted_at) {
988                let _ = self.entries.remove(&key);
989            }
990        }
991    }
992}
993
994/// RPC transport interface.
995#[async_trait]
996pub trait RpcSubmitTransport: Send + Sync {
997    /// Submits transaction bytes to RPC and returns signature string.
998    async fn submit_rpc(
999        &self,
1000        tx_bytes: &[u8],
1001        config: &RpcSubmitConfig,
1002    ) -> Result<String, SubmitTransportError>;
1003}
1004
1005/// Jito transport interface.
1006#[async_trait]
1007pub trait JitoSubmitTransport: Send + Sync {
1008    /// Submits transaction bytes to Jito block engine and returns Jito-specific acceptance data.
1009    async fn submit_jito(
1010        &self,
1011        tx_bytes: &[u8],
1012        config: &JitoSubmitConfig,
1013    ) -> Result<JitoSubmitResponse, SubmitTransportError>;
1014}
1015
1016/// Direct transport interface.
1017#[async_trait]
1018pub trait DirectSubmitTransport: Send + Sync {
1019    /// Submits transaction bytes to direct targets and returns the first successful target.
1020    async fn submit_direct(
1021        &self,
1022        tx_bytes: &[u8],
1023        targets: &[LeaderTarget],
1024        policy: RoutingPolicy,
1025        config: &DirectSubmitConfig,
1026    ) -> Result<LeaderTarget, SubmitTransportError>;
1027}
1028
1029#[cfg(test)]
1030mod tests {
1031    use super::*;
1032    use std::{hint::black_box, time::Instant};
1033
1034    use sof_support::bench::profile_iterations;
1035
1036    #[test]
1037    #[ignore = "profiling fixture for submit suppression cache churn"]
1038    fn suppression_cache_profile_fixture() {
1039        let iterations = profile_iterations(50_000);
1040        let ttl = Duration::from_millis(750);
1041        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1);
1042        let keys = (0_u8..64)
1043            .map(|value| TxSubmitSuppressionKey::Opportunity([value; 32]))
1044            .collect::<Vec<_>>();
1045        let mut cache = TxSuppressionCache::default();
1046
1047        let started = Instant::now();
1048        for (iteration, key) in keys.iter().cycle().take(iterations).enumerate() {
1049            let now = base + Duration::from_millis(u64::try_from(iteration % 2_000).unwrap_or(0));
1050            cache.insert_all(std::slice::from_ref(key), now);
1051            black_box(cache.is_suppressed(std::slice::from_ref(key), now, ttl));
1052        }
1053        let elapsed = started.elapsed();
1054        let avg_ns_per_iteration = elapsed.as_nanos() / u128::try_from(iterations).unwrap_or(1);
1055        let avg_us_per_iteration = avg_ns_per_iteration as f64 / 1_000.0;
1056
1057        eprintln!(
1058            "suppression_cache_profile_fixture iterations={} elapsed_us={} avg_ns_per_iteration={} avg_us_per_iteration={:.3} entries={}",
1059            iterations,
1060            elapsed.as_micros(),
1061            avg_ns_per_iteration,
1062            avg_us_per_iteration,
1063            cache.entries.len(),
1064        );
1065    }
1066
1067    #[test]
1068    fn suppression_cache_keeps_refreshed_entry_live() {
1069        let mut cache = TxSuppressionCache::default();
1070        let key = TxSubmitSuppressionKey::Opportunity([7_u8; 32]);
1071        let ttl = Duration::from_millis(750);
1072        let first_inserted_at = SystemTime::UNIX_EPOCH + Duration::from_secs(1);
1073        let refreshed_at = first_inserted_at + Duration::from_millis(500);
1074
1075        cache.insert_all(std::slice::from_ref(&key), first_inserted_at);
1076        cache.insert_all(std::slice::from_ref(&key), refreshed_at);
1077
1078        assert!(cache.is_suppressed(
1079            std::slice::from_ref(&key),
1080            refreshed_at + Duration::from_millis(100),
1081            ttl,
1082        ));
1083        assert!(!cache.is_suppressed(
1084            std::slice::from_ref(&key),
1085            refreshed_at + ttl + Duration::from_millis(1),
1086            ttl,
1087        ));
1088    }
1089
1090    #[test]
1091    fn direct_submit_config_clamps_zero_timeouts() {
1092        let normalized = DirectSubmitConfig {
1093            per_target_timeout: Duration::ZERO,
1094            global_timeout: Duration::ZERO,
1095            direct_target_rounds: 1,
1096            direct_submit_attempts: 1,
1097            hybrid_direct_attempts: 1,
1098            rebroadcast_interval: Duration::from_millis(5),
1099            agave_rebroadcast_enabled: false,
1100            agave_rebroadcast_window: Duration::ZERO,
1101            agave_rebroadcast_interval: Duration::from_millis(5),
1102            hybrid_rpc_broadcast: false,
1103            latency_aware_targeting: true,
1104            latency_probe_timeout: Duration::ZERO,
1105            latency_probe_port: None,
1106            latency_probe_max_targets: 1,
1107        }
1108        .normalized();
1109
1110        assert_eq!(normalized.per_target_timeout, Duration::from_millis(1));
1111        assert_eq!(normalized.global_timeout, Duration::from_millis(1));
1112        assert_eq!(normalized.latency_probe_timeout, Duration::from_millis(1));
1113    }
1114}
sof_tx/submit/types.rs

sof_tx/submit/
types.rs
