evm_oracle_state/pending/
source.rs

1//! Pluggable transport sessions for speculative oracle candidates.
2
3use std::{future::Future, pin::Pin, time::SystemTime};
4
5use tokio::{sync::watch, task::JoinHandle};
6
7use super::{PendingOracleEvent, PendingOracleRuntime, PendingOracleSource, PendingOracleSourceId};
8
9/// Owned future returned by a pending candidate source.
10pub type PendingOracleSourceFuture =
11    Pin<Box<dyn Future<Output = Result<(), PendingOracleSourceError>> + Send + 'static>>;
12
13/// Description of one concrete pending transport connection.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub struct PendingOracleSourceDescriptor {
16    /// Transport family used by the source.
17    pub source: PendingOracleSource,
18    /// Stable identifier for this provider, relay, or caller-owned stream.
19    pub id: PendingOracleSourceId,
20}
21
22impl PendingOracleSourceDescriptor {
23    /// Construct a source descriptor.
24    pub fn new(source: PendingOracleSource, id: PendingOracleSourceId) -> Self {
25        Self { source, id }
26    }
27}
28
29/// Connection state for one concrete pending source.
30#[non_exhaustive]
31#[derive(Clone, Copy, Debug, PartialEq, Eq)]
32pub enum PendingOracleSourceState {
33    /// A source task has started and is establishing its transport.
34    Connecting,
35    /// The source reports that its transport is ready.
36    Ready,
37    /// The source ended with a transport or decoding failure.
38    Degraded,
39    /// The source stopped normally or was shut down.
40    Stopped,
41}
42
43/// Latest observable health of one pending source.
44#[derive(Clone, Debug, PartialEq, Eq)]
45pub struct PendingOracleSourceHealth {
46    /// Source family and concrete identifier.
47    pub descriptor: PendingOracleSourceDescriptor,
48    /// Current connection state.
49    pub state: PendingOracleSourceState,
50    /// Most recent transport message, including unrelated pending transactions.
51    pub last_transport_message_at: Option<SystemTime>,
52    /// Most recent candidate submitted to the pending decoder.
53    pub last_candidate_at: Option<SystemTime>,
54    /// Number of coverage interruptions reported by this source session.
55    pub coverage_gap_count: u64,
56    /// Most recent source error.
57    pub last_error: Option<String>,
58}
59
60impl PendingOracleSourceHealth {
61    fn connecting(descriptor: PendingOracleSourceDescriptor) -> Self {
62        Self {
63            descriptor,
64            state: PendingOracleSourceState::Connecting,
65            last_transport_message_at: None,
66            last_candidate_at: None,
67            coverage_gap_count: 0,
68            last_error: None,
69        }
70    }
71}
72
73/// Explicit interval during which a pending source could not provide coverage.
74#[derive(Clone, Debug, PartialEq, Eq)]
75pub struct PendingOracleCoverageGap {
76    /// Source whose coverage was interrupted.
77    pub descriptor: PendingOracleSourceDescriptor,
78    /// Time at which the runtime recorded the gap.
79    pub observed_at: SystemTime,
80    /// Provider- or relay-facing reason.
81    pub reason: String,
82}
83
84/// Failure to start or run a pending source.
85#[non_exhaustive]
86#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
87pub enum PendingOracleSourceError {
88    /// Pending updates were not enabled on the runtime.
89    #[error("pending oracle updates are not enabled on this runtime")]
90    PendingUpdatesDisabled,
91    /// The source family was not enabled in [`super::PendingOracleConfig`].
92    #[error("pending source {transport:?} is not enabled")]
93    SourceDisabled {
94        /// Disabled transport family.
95        transport: PendingOracleSource,
96    },
97    /// A live source session already owns this stable source id.
98    #[error("pending source id {id} is already running")]
99    DuplicateSource {
100        /// Conflicting stable source identifier.
101        id: super::PendingOracleSourceId,
102    },
103    /// No Tokio runtime was available to own the source task.
104    #[error("a Tokio runtime is required to start a pending oracle source")]
105    RuntimeUnavailable,
106    /// The source connection or protocol failed.
107    #[error("pending source transport failed: {0}")]
108    Transport(String),
109    /// The source task panicked or was cancelled unexpectedly.
110    #[error("pending source task failed: {0}")]
111    Task(String),
112}
113
114/// Asynchronous producer of full pending oracle transaction candidates.
115pub trait PendingOracleCandidateSource: Send + 'static {
116    /// Return the source family and stable concrete identifier.
117    fn descriptor(&self) -> PendingOracleSourceDescriptor;
118
119    /// Run until the source ends or the shutdown receiver changes to `true`.
120    fn run(
121        self: Box<Self>,
122        sink: PendingOracleSourceSink,
123        shutdown: watch::Receiver<bool>,
124    ) -> PendingOracleSourceFuture;
125}
126
127/// Restricted ingestion and health handle supplied to a pending source.
128#[derive(Clone, Debug)]
129pub struct PendingOracleSourceSink {
130    runtime: PendingOracleRuntime,
131    descriptor: PendingOracleSourceDescriptor,
132}
133
134impl PendingOracleSourceSink {
135    pub(crate) fn new(
136        runtime: PendingOracleRuntime,
137        descriptor: PendingOracleSourceDescriptor,
138    ) -> Self {
139        Self {
140            runtime,
141            descriptor,
142        }
143    }
144
145    /// Report that the underlying source transport is ready.
146    pub fn ready(&self) {
147        self.runtime
148            .update_source_health(&self.descriptor, |health| {
149                health.state = PendingOracleSourceState::Ready;
150                health.last_error = None;
151            });
152    }
153
154    /// Record receipt of any source message, including unrelated transactions.
155    pub fn transport_message(&self) {
156        self.runtime
157            .mutate_source_health(&self.descriptor, |health| {
158                health.last_transport_message_at = Some(SystemTime::now());
159            });
160    }
161
162    /// Record that the source submitted a candidate to an oracle adapter.
163    pub fn candidate(&self) {
164        self.runtime
165            .mutate_source_health(&self.descriptor, |health| {
166                health.last_candidate_at = Some(SystemTime::now());
167            });
168    }
169
170    /// Report a transport interval that cannot be reconstructed reliably.
171    pub fn coverage_gap(&self, reason: impl Into<String>) {
172        let reason = reason.into();
173        self.runtime
174            .update_source_health(&self.descriptor, |health| {
175                health.state = PendingOracleSourceState::Degraded;
176                health.coverage_gap_count = health.coverage_gap_count.saturating_add(1);
177                health.last_error = Some(reason.clone());
178            });
179        self.runtime
180            .publisher()
181            .publish(PendingOracleEvent::CoverageGap(PendingOracleCoverageGap {
182                descriptor: self.descriptor.clone(),
183                observed_at: SystemTime::now(),
184                reason,
185            }));
186    }
187
188    /// Report that a degraded source is attempting to reconnect.
189    pub fn reconnecting(&self) {
190        self.runtime
191            .update_source_health(&self.descriptor, |health| {
192                health.state = PendingOracleSourceState::Connecting;
193            });
194    }
195
196    /// Borrow the pending runtime used to decode and route candidates.
197    pub fn runtime(&self) -> &PendingOracleRuntime {
198        &self.runtime
199    }
200
201    /// Return the source descriptor associated with this sink.
202    pub fn descriptor(&self) -> &PendingOracleSourceDescriptor {
203        &self.descriptor
204    }
205
206    pub(crate) fn connecting(&self) {
207        self.runtime
208            .insert_source_health(PendingOracleSourceHealth::connecting(
209                self.descriptor.clone(),
210            ));
211    }
212
213    pub(crate) fn stopped(&self) {
214        self.runtime
215            .update_source_health(&self.descriptor, |health| {
216                health.state = PendingOracleSourceState::Stopped;
217            });
218    }
219
220    pub(crate) fn failed(&self, error: &PendingOracleSourceError) {
221        let reason = error.to_string();
222        self.runtime
223            .update_source_health(&self.descriptor, |health| {
224                health.state = PendingOracleSourceState::Degraded;
225                health.coverage_gap_count = health.coverage_gap_count.saturating_add(1);
226                health.last_error = Some(reason.clone());
227            });
228        self.runtime
229            .publisher()
230            .publish(PendingOracleEvent::CoverageGap(PendingOracleCoverageGap {
231                descriptor: self.descriptor.clone(),
232                observed_at: SystemTime::now(),
233                reason,
234            }));
235    }
236}
237
238/// Runtime-owned source task. Dropping the session shuts the source down.
239#[derive(Debug)]
240pub struct PendingOracleSourceSession {
241    descriptor: PendingOracleSourceDescriptor,
242    shutdown: watch::Sender<bool>,
243    task: Option<JoinHandle<Result<(), PendingOracleSourceError>>>,
244}
245
246impl PendingOracleSourceSession {
247    pub(crate) fn new(
248        descriptor: PendingOracleSourceDescriptor,
249        shutdown: watch::Sender<bool>,
250        task: JoinHandle<Result<(), PendingOracleSourceError>>,
251    ) -> Self {
252        Self {
253            descriptor,
254            shutdown,
255            task: Some(task),
256        }
257    }
258
259    /// Return the source owned by this session.
260    pub fn descriptor(&self) -> &PendingOracleSourceDescriptor {
261        &self.descriptor
262    }
263
264    /// Request graceful source shutdown.
265    pub fn stop(&self) {
266        let _ = self.shutdown.send(true);
267    }
268
269    /// Wait for the source task to finish.
270    pub async fn join(mut self) -> Result<(), PendingOracleSourceError> {
271        let Some(task) = self.task.take() else {
272            return Ok(());
273        };
274        task.await
275            .map_err(|error| PendingOracleSourceError::Task(error.to_string()))?
276    }
277}
278
279impl Drop for PendingOracleSourceSession {
280    fn drop(&mut self) {
281        let _ = self.shutdown.send(true);
282        if let Some(task) = self.task.take() {
283            task.abort();
284        }
285    }
286}
evm_oracle_state/pending/source.rs

evm_oracle_state/pending/
source.rs
