fix: respawn panicked solver workers instead of losing them

fynd-core/src/algorithm/sim_guard.rs

@@ -2,9 +2,10 @@
 //!
 //! `ProtocolSim` implementations run third-party component math that can panic on degenerate
 //! inputs (e.g. a U256 division by zero when quoting an absurdly large amount). Left
-//! uncaught, such a panic unwinds through the solver worker thread and permanently kills
-//! it. Simulation calls made while solving therefore go through this guard, which turns
-//! a panic into a `SimulationError` so the algorithm skips the component and keeps solving.
+//! uncaught, such a panic unwinds out of the solve and ends the worker's session, which
+//! costs a respawn and drops every in-flight task on that thread. Simulation calls made
+//! while solving therefore go through this guard, which turns a panic into a
+//! `SimulationError` so the algorithm skips the component and keeps solving.
 
 use std::panic::{catch_unwind, AssertUnwindSafe};
 

fynd-core/src/algorithm/test_utils.rs

@@ -1,5 +1,7 @@
 //! Shared test utilities for algorithm tests.
 
+use std::{sync::Arc, time::Duration};
+
 use chrono::NaiveDateTime;
 use num_bigint::BigUint;
 use num_rational::BigRational;
@@ -22,10 +24,14 @@ use tycho_simulation::{
 };
 
 use crate::{
-    algorithm::most_liquid::DepthAndPrice,
-    feed::market_data::{MarketData, MarketState},
-    graph::{petgraph::PetgraphStableDiGraphManager, GraphManager, TopologyGraphManager},
-    types::{quote::OrderSide, BlockInfo, Order},
+    algorithm::{most_liquid::DepthAndPrice, Algorithm, AlgorithmError},
+    derived::{computation::ComputationRequirements, SharedDerivedDataRef},
+    feed::market_data::{MarketData, MarketState, StateLabel},
+    graph::{
+        petgraph::{PetgraphStableDiGraphManager, StableDiGraph},
+        GraphManager, TopologyGraphManager,
+    },
+    types::{quote::OrderSide, BlockInfo, Order, RouteResult},
 };
 
 /// Use amounts in wei scale (10^18) to exceed gas costs in tests.
@@ -640,6 +646,69 @@ pub fn market_read(market: &MarketData) -> crate::feed::market_data::MarketDataV
         .expect("lock should not be contested in test")
 }
 
+// ==================== Stub Algorithm ====================
+
+/// Solve behavior injected into a [`StubAlgorithm`].
+type StubSolve = dyn Fn(&Order) -> Result<RouteResult, AlgorithmError> + Send + Sync;
+
+/// Test algorithm whose solve behavior is injected as a closure.
+///
+/// The closure runs on every `find_best_route` call, so one stub covers the whole range of
+/// worker tests: always failing, panicking on a marker order, or returning a hand-built route.
+#[derive(Clone)]
+pub struct StubAlgorithm {
+    solve: Arc<StubSolve>,
+    requirements: ComputationRequirements,
+    timeout: Duration,
+}
+
+impl StubAlgorithm {
+    /// Creates a stub that answers every solve with `solve`.
+    pub fn returning(
+        solve: impl Fn(&Order) -> Result<RouteResult, AlgorithmError> + Send + Sync + 'static,
+    ) -> Self {
+        Self {
+            solve: Arc::new(solve),
+            requirements: ComputationRequirements::none(),
+            timeout: Duration::from_secs(1),
+        }
+    }
+
+    /// Override the derived-data requirements the worker gates on.
+    pub fn with_requirements(mut self, requirements: ComputationRequirements) -> Self {
+        self.requirements = requirements;
+        self
+    }
+}
+
+impl Algorithm for StubAlgorithm {
+    type GraphType = StableDiGraph<DepthAndPrice>;
+    type GraphManager = PetgraphStableDiGraphManager<DepthAndPrice>;
+
+    fn name(&self) -> &str {
+        "stub"
+    }
+
+    async fn find_best_route(
+        &self,
+        _graph: &Self::GraphType,
+        _market: MarketData,
+        _label: Option<StateLabel>,
+        _derived: Option<SharedDerivedDataRef>,
+        order: &Order,
+    ) -> Result<RouteResult, AlgorithmError> {
+        (self.solve)(order)
+    }
+
+    fn computation_requirements(&self) -> ComputationRequirements {
+        self.requirements.clone()
+    }
+
+    fn timeout(&self) -> Duration {
+        self.timeout
+    }
+}
+
 /// Common fixtures for tests.
 pub mod fixtures {
     use super::*;

fynd-core/src/derived/tracker.rs

@@ -9,6 +9,7 @@ use rustc_hash::FxHashSet;
 use super::{
     computation::{ComputationId, ComputationRequirements},
     events::DerivedDataEvent,
+    store::DerivedData,
 };
 
 /// Tracks which derived data computations are ready based on freshness requirements.
@@ -220,6 +221,34 @@ impl ReadinessTracker {
     pub fn current_block(&self) -> Option<u64> {
         self.current_block
     }
+
+    /// Marks `allow_stale` computations that already have an output in `store` as computed.
+    ///
+    /// A worker rebuilt mid-run (respawn after a panic) starts with an empty tracker while the
+    /// shared store already holds results; without seeding, `allow_stale` requirements would
+    /// fail tasks as not-ready until the next broadcast event arrives.
+    ///
+    /// `require_fresh` computations are not seeded: a stored output was computed for whatever
+    /// block the store held it at, not necessarily the worker's current block, so treating it as
+    /// fresh could mark a `require_fresh` requirement ready with outdated data. Fresh readiness
+    /// is established the normal way, from the next `ComputationComplete` event.
+    ///
+    /// Only `ever_computed` is touched. The block a stored output carries can be well behind the
+    /// chain head, so adopting it as `current_block` would make the tracker drop a
+    /// `ComputationFailed` for the real head block and wait out the algorithm timeout instead of
+    /// failing fast.
+    pub fn seed_from_store(&mut self, store: &DerivedData) {
+        for id in self
+            .requirements
+            .stale_requirements()
+            .iter()
+            .copied()
+        {
+            if store.output_block(id).is_some() {
+                self.ever_computed.insert(id);
+            }
+        }
+    }
 }
 
 #[cfg(test)]
@@ -531,6 +560,27 @@ mod tests {
         assert!(tracker.is_ready());
     }
 
+    #[test]
+    fn seed_from_store_satisfies_stale_requirements() {
+        let mut tracker = ReadinessTracker::new(stale_requirements(&["token_prices"]));
+        assert!(!tracker.is_ready());
+
+        let mut store = DerivedData::new();
+        store.set_output("token_prices", 42u32, 5);
+        tracker.seed_from_store(&store);
+
+        assert!(tracker.is_ready());
+    }
+
+    #[test]
+    fn seed_from_store_ignores_missing_outputs() {
+        let mut tracker = ReadinessTracker::new(stale_requirements(&["token_prices"]));
+
+        tracker.seed_from_store(&DerivedData::new());
+
+        assert!(!tracker.is_ready());
+    }
+
     #[test]
     fn missing_returns_unready_set() {
         let requirements = ComputationRequirements::none()

fynd-core/src/worker_pool/mod.rs

@@ -1,6 +1,7 @@
 pub mod pool;
 mod price_impact;
 pub mod registry;
+pub(crate) mod supervisor;
 pub(crate) mod task_queue;
 pub(crate) mod worker;
 

fynd-core/src/worker_pool/pool.rs

@@ -8,7 +8,10 @@
 //! Worker pools can use either a built-in algorithm (by name via [`WorkerPoolBuilder::algorithm`])
 //! or a custom [`Algorithm`](crate::algorithm::Algorithm) implementation (via
 //! [`WorkerPoolBuilder::with_algorithm`]).
-use std::thread::JoinHandle;
+use std::{
+    sync::{Arc, Once},
+    thread::JoinHandle,
+};
 
 use tokio::sync::broadcast;
 use tracing::{error, info};
@@ -28,6 +31,7 @@ use crate::{
             spawn_workers_generic, AlgorithmSpawner, SpawnWorkersParams, UnknownAlgorithmError,
             DEFAULT_ALGORITHM,
         },
+        supervisor::RespawnPolicy,
         task_queue::{TaskQueue, TaskQueueConfig, TaskQueueHandle},
     },
     worker_pool_router::LiquidityScope,
@@ -141,6 +145,18 @@ impl WorkerPool {
             liquidity_scope,
             exclude_protocols,
             fallback_fee_tiers,
+            respawn_policy: RespawnPolicy::default(),
+            on_worker_gave_up: Arc::new(|| {
+                // A deterministic panic hits every worker at once, so the whole pool gives up
+                // within milliseconds. `exit` is not reentrant across threads — concurrent
+                // callers can deadlock in the atexit handlers and hang the process, which is
+                // exactly what this exit exists to prevent.
+                static EXIT_ONCE: Once = Once::new();
+                EXIT_ONCE.call_once(|| {
+                    error!("a solver worker gave up after repeated panics; exiting so the orchestrator restarts the process");
+                    std::process::exit(1);
+                });
+            }),
         };
         let workers = config.spawner.spawn(params)?;