fynd_core/derived/
computation.rs

1//! Core computation trait and types.
2
3use std::collections::HashSet;
4
5use async_trait::async_trait;
6
7use super::{
8    error::ComputationError,
9    manager::{ChangedComponents, SharedDerivedDataRef},
10};
11use crate::feed::market_data::SharedMarketDataRef;
12
13/// Unique identifier for a computation type.
14///
15/// Used for event discrimination, storage keys, and readiness tracking.
16pub type ComputationId = &'static str;
17
18/// Error when building computation requirements.
19#[derive(Debug, Clone, thiserror::Error)]
20#[error("conflicting requirement: '{id}' cannot be both fresh and stale")]
21pub struct RequirementConflict {
22    /// The computation ID that was added with conflicting freshness.
23    pub(crate) id: ComputationId,
24}
25
26impl RequirementConflict {
27    /// Returns the conflicting computation ID.
28    pub fn id(&self) -> ComputationId {
29        self.id
30    }
31}
32
33/// Requirements for derived data computations.
34///
35/// Each algorithm declares which computations it needs and their freshness requirements:
36///
37/// - `require_fresh`: Data must be from the current block (same block as SharedMarketData). Workers
38///   wait for these computations to complete for the current block before solving.
39///
40/// - `allow_stale`: Data can be from any past block, as long as it has been computed at least once.
41///   Workers only check that the data exists, not that it's from the current block.
42///
43///
44/// # Example
45///
46/// ```ignore
47/// // Token prices don't change much block-to-block, stale is fine
48/// ComputationRequirements::none()
49///     .expect_stale("token_prices")?
50///
51/// // Spot prices must be fresh for accurate routing
52/// ComputationRequirements::none()
53///     .expect_fresh("spot_prices")?
54/// ```
55#[derive(Debug, Clone, Default)]
56pub struct ComputationRequirements {
57    /// Computations that must be from the current block.
58    pub(crate) require_fresh: HashSet<ComputationId>,
59    /// Computations that can use data from any past block.
60    ///
61    /// TODO: Stale data can be dangerous if stale for too long. In the future, associate staleness
62    /// to a block limit might be implemented.
63    pub(crate) allow_stale: HashSet<ComputationId>,
64}
65
66impl ComputationRequirements {
67    /// Returns the set of computations that require fresh data.
68    pub fn fresh_requirements(&self) -> &HashSet<ComputationId> {
69        &self.require_fresh
70    }
71
72    /// Returns the set of computations that allow stale data.
73    pub fn stale_requirements(&self) -> &HashSet<ComputationId> {
74        &self.allow_stale
75    }
76
77    /// Creates empty requirements (no derived data needed).
78    pub fn none() -> Self {
79        Self::default()
80    }
81
82    /// Builder method to add a computation that requires fresh data (current block).
83    ///
84    /// # Errors
85    ///
86    /// Returns `RequirementConflict` if the same ID is already in `allow_stale`.
87    pub fn require_fresh(mut self, id: ComputationId) -> Result<Self, RequirementConflict> {
88        if self.allow_stale.contains(&id) {
89            return Err(RequirementConflict { id });
90        }
91        self.require_fresh.insert(id);
92        Ok(self)
93    }
94
95    /// Builder method to add a computation that allows stale data (any past block).
96    ///
97    /// # Errors
98    ///
99    /// Returns `RequirementConflict` if the same ID is already in `require_fresh`.
100    pub fn allow_stale(mut self, id: ComputationId) -> Result<Self, RequirementConflict> {
101        if self.require_fresh.contains(&id) {
102            return Err(RequirementConflict { id });
103        }
104        self.allow_stale.insert(id);
105        Ok(self)
106    }
107
108    /// Returns true if there are any requirements.
109    pub fn has_requirements(&self) -> bool {
110        !self.require_fresh.is_empty() || !self.allow_stale.is_empty()
111    }
112
113    /// Returns true if the given computation is required (fresh or stale).
114    pub fn is_required(&self, id: ComputationId) -> bool {
115        self.require_fresh.contains(&id) || self.allow_stale.contains(&id)
116    }
117}
118
119/// Trait for derived data computations.
120///
121/// Implement this trait to define a new type of derived data that can be
122/// computed from market data.
123///
124/// # Design
125///
126/// - No `dependencies()` method - execution order is hardcoded in `ComputationManager`
127/// - Typed `DerivedDataStore` - access previous results via `store.token_prices()` etc.
128/// - Each computation is explicitly added to `ComputationManager`
129/// - Computations receive `Arc<RwLock<>>` references and acquire locks as needed, allowing early
130///   release and granular locking strategies
131///
132/// # Example
133///
134/// ```ignore
135/// pub struct TokenPriceComputation {
136///     gas_token: Address,
137/// }
138///
139/// #[async_trait]
140/// impl DerivedComputation for TokenPriceComputation {
141///     type Output = TokenPrices;
142///     const ID: ComputationId = "token_prices";
143///
144///     async fn compute(
145///         &self,
146///         market: &SharedMarketDataRef,
147///         store: &SharedDerivedDataRef,
148///         changed: &ChangedComponents,
149///     ) -> Result<Self::Output, ComputationError> {
150///         if changed.is_full_recompute {
151///             // Full recompute: process all components
152///         } else {
153///             // Incremental: only process changed components
154///         }
155///     }
156/// }
157/// ```
158#[async_trait]
159pub trait DerivedComputation: Send + Sync + 'static {
160    /// The output type produced by this computation.
161    ///
162    /// Must be `Clone` for storage retrieval and `Send + Sync` for thread safety.
163    type Output: Clone + Send + Sync + 'static;
164
165    /// Unique identifier for this computation.
166    ///
167    /// Used for event discrimination, storage keys, and readiness tracking.
168    const ID: ComputationId;
169
170    /// Computes the derived data from market state.
171    ///
172    /// # Arguments
173    ///
174    /// * `market` - Reference to shared market data (computation acquires lock as needed)
175    /// * `store` - Reference to derived data store (computation acquires lock as needed)
176    /// * `changed` - Information about which components changed, enabling incremental computation
177    ///
178    /// # Returns
179    ///
180    /// The computed output, or an error if computation failed.
181    ///
182    /// # Incremental Computation
183    ///
184    /// Implementations should use `changed` to only recompute data affected by the changes:
185    /// - `changed.is_full_recompute` - If true, recompute everything (startup/lag recovery)
186    /// - `changed.added` - New components to compute
187    /// - `changed.removed` - Components to remove from results
188    /// - `changed.updated` - Components whose state changed
189    ///
190    /// # Lock Management
191    ///
192    /// Computations should acquire locks only when needed and release them as early
193    /// as possible to minimize contention. Use `.read().await` for async lock acquisition.
194    // TODO: Support Partial Failures, including IDs for which computation failed.
195    async fn compute(
196        &self,
197        market: &SharedMarketDataRef,
198        store: &SharedDerivedDataRef,
199        changed: &ChangedComponents,
200    ) -> Result<Self::Output, ComputationError>;
201}
fynd_core/derived/computation.rs

fynd_core/derived/
computation.rs
