From 1900b3c504adfc0795152ea4eaec5b8d8cd79968 Mon Sep 17 00:00:00 2001 From: zizou0x <111426680+zizou0x@users.noreply.github.com> Date: Thu, 23 Jul 2026 12:58:15 +0000 Subject: [PATCH 01/13] fix: respawn panicked solver workers instead of losing them A panic while solving previously unwound through the worker thread and killed it silently; the pool only noticed at shutdown join(). Once every worker had hit a poison request, the pool answered 100% no_route until the pod was recreated. Each worker thread now runs sessions in a catch_unwind loop: a panic is logged at ERROR with the panic message, counted in the worker_pool_worker_panics_total metric, and the worker is respawned after a short backoff. Clean shutdowns still exit the thread, including shutdown signals arriving while a worker is mid-respawn. Co-Authored-By: Claude Fable 5 --- fynd-core/src/worker_pool/registry.rs | 237 +++++++++++++++++++++++--- 1 file changed, 210 insertions(+), 27 deletions(-) diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index 5f9a2f28..56de23c4 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -12,12 +12,14 @@ //! 3. Add the algorithm name to `AVAILABLE_ALGORITHMS` use std::{ + panic::{catch_unwind, AssertUnwindSafe}, sync::Arc, thread::{self, JoinHandle}, + time::Duration, }; use tokio::sync::broadcast; -use tracing::info; +use tracing::{error, info}; use crate::{ algorithm::{ @@ -39,6 +41,10 @@ pub(crate) const AVAILABLE_ALGORITHMS: &[&str] = /// Default algorithm to use if none specified. pub(crate) const DEFAULT_ALGORITHM: &str = "most_liquid"; +/// Pause before respawning a panicked worker, so a panic during worker +/// initialization cannot turn the respawn loop into a hot spin. +const RESPAWN_BACKOFF: Duration = Duration::from_millis(100); + /// Parameters for spawning workers. pub(crate) struct SpawnWorkersParams { /// Algorithm name (e.g., "most_liquid") — used for thread naming and logging. @@ -190,7 +196,10 @@ where let event_rx = params.event_rx.resubscribe(); let derived_event_rx = params.derived_event_rx.resubscribe(); let algorithm_config = params.algorithm_config.clone(); - let shutdown_rx = params.shutdown_tx.subscribe(); + // Subscribed before the thread starts so shutdown signals sent at any point, + // including while the worker is recovering from a panic, are never missed. + let mut shutdown_rx = params.shutdown_tx.subscribe(); + let shutdown_tx = params.shutdown_tx.clone(); let algorithm_name = params.algorithm.clone(); let pool_name = params.pool_name.clone(); let factory = factory.clone(); @@ -201,30 +210,85 @@ where let handle = thread::Builder::new() .name(format!("{}-worker-{}", algorithm_name, worker_id)) .spawn(move || { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("failed to create tokio runtime"); - - rt.block_on(async move { - let algorithm = factory(algorithm_config); - - let mut worker = SolverWorker::new( - market_data, - derived_data, - algorithm, - worker_id, - pool_name, - ) - .with_liquidity_scope(liquidity_scope) - .with_exclude_protocols(exclude_protocols) - .with_fallback_fee_tiers(fallback_fee_tiers); - - worker.initialize_graph().await; - worker - .run(event_rx, derived_event_rx, task_rx, shutdown_rx) - .await; - }); + // Run worker sessions until clean shutdown. A panic while solving (e.g. pool + // math dividing by zero) only ends the current session: it is reported loudly + // and the worker is respawned, instead of silently losing the thread until + // every worker in the pool is dead. + loop { + // Fresh receivers for this session; the previous session's receivers are + // consumed (or poisoned) when it panics. + let session_event_rx = event_rx.resubscribe(); + let session_derived_event_rx = derived_event_rx.resubscribe(); + let session_shutdown_rx = shutdown_tx.subscribe(); + + // A shutdown sent while no session was listening (e.g. mid-respawn) is + // buffered in the receiver created before the thread started. + match shutdown_rx.try_recv() { + Err(broadcast::error::TryRecvError::Empty) => {} + // Received a shutdown, or the pool dropped the sender. + _ => break, + } + + let session = catch_unwind(AssertUnwindSafe(|| { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("failed to create tokio runtime"); + + rt.block_on(async { + let algorithm = factory(algorithm_config.clone()); + + let mut worker = SolverWorker::new( + market_data.clone(), + Arc::clone(&derived_data), + algorithm, + worker_id, + pool_name.clone(), + ) + .with_liquidity_scope(liquidity_scope) + .with_exclude_protocols(exclude_protocols.clone()) + .with_fallback_fee_tiers(fallback_fee_tiers.clone()); + + worker.initialize_graph().await; + worker + .run( + session_event_rx, + session_derived_event_rx, + task_rx.clone(), + session_shutdown_rx, + ) + .await; + }); + })); + + match session { + Ok(()) => break, + Err(panic_payload) => { + let panic_message = panic_payload + .downcast_ref::<&str>() + .copied() + .or_else(|| { + panic_payload + .downcast_ref::() + .map(String::as_str) + }) + .unwrap_or(""); + error!( + pool = %pool_name, + algorithm = %algorithm_name, + worker_id, + panic = %panic_message, + "worker thread panicked; respawning worker" + ); + metrics::counter!( + "worker_pool_worker_panics_total", + "pool" => pool_name.clone() + ) + .increment(1); + thread::sleep(RESPAWN_BACKOFF); + } + } + } }) .expect("failed to spawn worker thread"); @@ -276,8 +340,22 @@ fn spawn_water_fill_workers(params: SpawnWorkersParams) -> Vec> { mod tests { use std::time::Duration; + use num_bigint::BigUint; + use tokio::sync::oneshot; + use uuid::Uuid; + use super::*; - use crate::{derived::DerivedData, feed::market_data::MarketData}; + use crate::{ + algorithm::{ + most_liquid::DepthAndPrice, + test_utils::{order, setup_market_weighted, token}, + Algorithm, AlgorithmError, + }, + derived::{computation::ComputationRequirements, DerivedData}, + feed::market_data::{MarketData, StateLabel}, + graph::petgraph::{PetgraphStableDiGraphManager, StableDiGraph}, + types::{quote::OrderSide, Order, RouteResult, SolveError}, + }; fn make_params(algorithm: &str, num_workers: usize) -> SpawnWorkersParams { let (_task_tx, task_rx) = async_channel::bounded(10); @@ -423,6 +501,111 @@ mod tests { let _ = shutdown_tx.send(()); } + /// Amount marking the order whose solve panics in [`PanicOnPoisonAlgorithm`]. + const POISON_AMOUNT: u64 = 666; + + /// Algorithm that panics while solving the poison order and returns an error + /// otherwise. Used to verify that a panicking task does not kill the worker. + #[derive(Clone)] + struct PanicOnPoisonAlgorithm; + + impl Algorithm for PanicOnPoisonAlgorithm { + type GraphType = StableDiGraph; + type GraphManager = PetgraphStableDiGraphManager; + + fn name(&self) -> &str { + "panic_on_poison" + } + + async fn find_best_route( + &self, + _graph: &Self::GraphType, + _market: MarketData, + _label: Option, + _derived: Option, + order: &Order, + ) -> Result { + if order.amount() == &BigUint::from(POISON_AMOUNT) { + panic!("poison order"); + } + Err(AlgorithmError::Other("no route in mock".to_string())) + } + + fn computation_requirements(&self) -> ComputationRequirements { + ComputationRequirements::none() + } + + fn timeout(&self) -> Duration { + Duration::from_secs(1) + } + } + + #[tokio::test] + async fn worker_respawns_after_panic_and_processes_next_task() { + let (market, _) = setup_market_weighted(vec![]); + let derived_data = DerivedData::new_shared(); + let (task_tx, task_rx) = async_channel::bounded(10); + let (event_tx, _) = broadcast::channel::(10); + let (derived_event_tx, _) = broadcast::channel(10); + let (shutdown_tx, _) = broadcast::channel(1); + + let params = SpawnWorkersParams { + algorithm: "panic_on_poison".to_string(), + pool_name: "test_pool".to_string(), + num_workers: 1, + algorithm_config: AlgorithmConfig::default(), + task_rx, + market_data: market, + derived_data, + event_rx: event_tx.subscribe(), + derived_event_rx: derived_event_tx.subscribe(), + shutdown_tx: shutdown_tx.clone(), + liquidity_scope: LiquidityScope::default(), + exclude_protocols: Vec::new(), + fallback_fee_tiers: SharedFeeTiers::default(), + }; + let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; + let workers = spawn_workers_generic(params, &factory); + + let token_a = token(0x01, "A"); + let token_b = token(0x02, "B"); + + // The poison task panics mid-solve; its response channel is dropped. + let (poison_tx, poison_rx) = oneshot::channel(); + let poison_order = order(&token_a, &token_b, POISON_AMOUNT as u128, OrderSide::Sell); + task_tx + .send(SolveTask::new(Uuid::new_v4(), poison_order, poison_tx)) + .await + .unwrap(); + let _ = poison_rx.await; + + // The worker must come back and answer the next task. + let (ok_tx, ok_rx) = oneshot::channel(); + let normal_order = order(&token_a, &token_b, 100, OrderSide::Sell); + task_tx + .send(SolveTask::new(Uuid::new_v4(), normal_order, ok_tx)) + .await + .unwrap(); + let response = tokio::time::timeout(Duration::from_secs(5), ok_rx) + .await + .expect("worker should respawn after the panic and process the next task") + .expect("worker should respond to the task"); + match response { + Err(SolveError::AlgorithmError(msg)) => { + assert!(msg.contains("no route in mock"), "unexpected message: {msg}"); + } + other => panic!("expected AlgorithmError from mock, got {other:?}"), + } + + let _ = shutdown_tx.send(()); + drop(task_tx); + for handle in workers { + handle + .join() + .expect("worker thread should shut down cleanly"); + } + } + #[test] fn test_registry_spawns_path_frank_wolfe() { let (shutdown_tx, _) = broadcast::channel(1); From 92c4d7bd96da0c20b84029891db74d315ced56d7 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 14:53:31 -0300 Subject: [PATCH 02/13] fix: let workers exit when the pool drops the shutdown sender --- fynd-core/src/worker_pool/registry.rs | 42 +++++++++++++++++++++++++-- 1 file changed, 40 insertions(+), 2 deletions(-) diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index 56de23c4..c08b2bf3 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -199,7 +199,6 @@ where // Subscribed before the thread starts so shutdown signals sent at any point, // including while the worker is recovering from a panic, are never missed. let mut shutdown_rx = params.shutdown_tx.subscribe(); - let shutdown_tx = params.shutdown_tx.clone(); let algorithm_name = params.algorithm.clone(); let pool_name = params.pool_name.clone(); let factory = factory.clone(); @@ -219,7 +218,7 @@ where // consumed (or poisoned) when it panics. let session_event_rx = event_rx.resubscribe(); let session_derived_event_rx = derived_event_rx.resubscribe(); - let session_shutdown_rx = shutdown_tx.subscribe(); + let session_shutdown_rx = shutdown_rx.resubscribe(); // A shutdown sent while no session was listening (e.g. mid-respawn) is // buffered in the receiver created before the thread started. @@ -606,6 +605,45 @@ mod tests { } } + #[tokio::test] + async fn workers_exit_when_pool_drops_shutdown_sender() { + let (market, _) = setup_market_weighted(vec![]); + let (_task_tx, task_rx) = async_channel::bounded::(10); + let (event_tx, _) = broadcast::channel::(10); + let (derived_event_tx, _) = broadcast::channel(10); + let (shutdown_tx, _) = broadcast::channel(1); + + let params = SpawnWorkersParams { + algorithm: "panic_on_poison".to_string(), + pool_name: "test_pool".to_string(), + num_workers: 1, + algorithm_config: AlgorithmConfig::default(), + task_rx, + market_data: market, + derived_data: DerivedData::new_shared(), + event_rx: event_tx.subscribe(), + derived_event_rx: derived_event_tx.subscribe(), + shutdown_tx, + liquidity_scope: LiquidityScope::default(), + exclude_protocols: Vec::new(), + fallback_fee_tiers: SharedFeeTiers::default(), + }; + let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; + let workers = spawn_workers_generic(params, &factory); + // `params` (holding the only shutdown sender) is consumed and dropped above. + + for handle in workers { + tokio::time::timeout( + Duration::from_secs(5), + tokio::task::spawn_blocking(move || handle.join()), + ) + .await + .expect("worker should exit when the shutdown sender drops") + .unwrap() + .expect("worker thread should exit cleanly"); + } + } + #[test] fn test_registry_spawns_path_frank_wolfe() { let (shutdown_tx, _) = broadcast::channel(1); From ecc05bb0df5373595d4afbeb3ba57ba9c485abf9 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:00:23 -0300 Subject: [PATCH 03/13] feat: add respawn policy with capped backoff and give-up --- fynd-core/src/worker_pool/mod.rs | 1 + fynd-core/src/worker_pool/supervisor.rs | 122 ++++++++++++++++++++++++ 2 files changed, 123 insertions(+) create mode 100644 fynd-core/src/worker_pool/supervisor.rs diff --git a/fynd-core/src/worker_pool/mod.rs b/fynd-core/src/worker_pool/mod.rs index e9b146b4..39d9fa36 100644 --- a/fynd-core/src/worker_pool/mod.rs +++ b/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; diff --git a/fynd-core/src/worker_pool/supervisor.rs b/fynd-core/src/worker_pool/supervisor.rs new file mode 100644 index 00000000..d0838da9 --- /dev/null +++ b/fynd-core/src/worker_pool/supervisor.rs @@ -0,0 +1,122 @@ +//! Worker-session supervision: respawn policy and (from Task 4) the session loop. + +use std::time::Duration; + +/// Retry policy for respawning a panicked worker. +/// +/// Backoff doubles per consecutive failure up to `max_backoff` (the same +/// doubling-with-cap shape as the Rust client's `RetryConfig`). After +/// `max_attempts` consecutive failures the worker gives up. A session that +/// lives at least `stable_session` resets the budget, so spaced transient +/// panics respawn indefinitely while deterministic failures stop fast. +#[derive(Clone, Copy, Debug)] +pub(crate) struct RespawnPolicy { + pub initial_backoff: Duration, + pub max_backoff: Duration, + pub max_attempts: u32, + pub stable_session: Duration, +} + +impl Default for RespawnPolicy { + fn default() -> Self { + Self { + initial_backoff: Duration::from_millis(100), + max_backoff: Duration::from_secs(2), + max_attempts: 10, + stable_session: Duration::from_secs(600), + } + } +} + +/// What the supervision loop does after a session panicked. +#[derive(Debug, PartialEq)] +pub(crate) enum FailureAction { + /// Sleep this long, then respawn the worker. + Retry(Duration), + /// Stop respawning this worker. + GiveUp, +} + +/// Tracks consecutive session failures against a [`RespawnPolicy`]. +pub(crate) struct RespawnState { + policy: RespawnPolicy, + consecutive_failures: u32, + next_backoff: Duration, +} + +impl RespawnState { + pub(crate) fn new(policy: RespawnPolicy) -> Self { + Self { policy, consecutive_failures: 0, next_backoff: policy.initial_backoff } + } + + /// Records a panicked session that lived `session_lived` and decides the next action. + pub(crate) fn on_failure(&mut self, session_lived: Duration) -> FailureAction { + if session_lived >= self.policy.stable_session { + self.consecutive_failures = 0; + self.next_backoff = self.policy.initial_backoff; + } + self.consecutive_failures += 1; + if self.consecutive_failures >= self.policy.max_attempts { + return FailureAction::GiveUp; + } + let delay = self.next_backoff; + self.next_backoff = (self.next_backoff * 2).min(self.policy.max_backoff); + FailureAction::Retry(delay) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn fast_policy() -> RespawnPolicy { + RespawnPolicy { + initial_backoff: Duration::from_millis(100), + max_backoff: Duration::from_millis(400), + max_attempts: 3, + stable_session: Duration::from_secs(60), + } + } + + #[test] + fn backoff_doubles_up_to_the_cap() { + let mut state = RespawnState::new(fast_policy()); + let lived = Duration::from_millis(1); + assert_eq!(state.on_failure(lived), FailureAction::Retry(Duration::from_millis(100))); + assert_eq!(state.on_failure(lived), FailureAction::Retry(Duration::from_millis(200))); + assert_eq!(state.on_failure(lived), FailureAction::GiveUp); + } + + #[test] + fn gives_up_after_max_attempts() { + let mut state = RespawnState::new(RespawnPolicy { max_attempts: 1, ..fast_policy() }); + assert_eq!(state.on_failure(Duration::from_millis(1)), FailureAction::GiveUp); + } + + #[test] + fn stable_session_resets_the_budget() { + let mut state = RespawnState::new(fast_policy()); + let rapid = Duration::from_millis(1); + state.on_failure(rapid); + state.on_failure(rapid); + // A session that lived past the stability threshold resets attempts and backoff. + assert_eq!( + state.on_failure(Duration::from_secs(61)), + FailureAction::Retry(Duration::from_millis(100)) + ); + } + + #[test] + fn backoff_cap_bounds_the_delay() { + let mut state = RespawnState::new(RespawnPolicy { max_attempts: 10, ..fast_policy() }); + let rapid = Duration::from_millis(1); + let mut last = Duration::ZERO; + for _ in 0..5 { + if let FailureAction::Retry(d) = state.on_failure(rapid) { + last = d; + } + } + assert_eq!(last, Duration::from_millis(400)); + } +} From e7957325dd4bae8c471450beddca4f9b1bbb21e0 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:08:30 -0300 Subject: [PATCH 04/13] feat: give up respawning after repeated rapid panics and exit the process --- fynd-core/src/worker_pool/pool.rs | 8 +- fynd-core/src/worker_pool/registry.rs | 162 ++++++++---------------- fynd-core/src/worker_pool/supervisor.rs | 154 +++++++++++++++++++++- 3 files changed, 208 insertions(+), 116 deletions(-) diff --git a/fynd-core/src/worker_pool/pool.rs b/fynd-core/src/worker_pool/pool.rs index eaa2b701..ceefb183 100644 --- a/fynd-core/src/worker_pool/pool.rs +++ b/fynd-core/src/worker_pool/pool.rs @@ -8,7 +8,7 @@ //! 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, thread::JoinHandle}; use tokio::sync::broadcast; use tracing::{error, info}; @@ -28,6 +28,7 @@ use crate::{ spawn_workers_generic, AlgorithmSpawner, SpawnWorkersParams, UnknownAlgorithmError, DEFAULT_ALGORITHM, }, + supervisor::RespawnPolicy, task_queue::{TaskQueue, TaskQueueConfig, TaskQueueHandle}, }, worker_pool_router::LiquidityScope, @@ -141,6 +142,11 @@ impl WorkerPool { liquidity_scope, exclude_protocols, fallback_fee_tiers, + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| { + 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)?; diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index c08b2bf3..1b10bf92 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -12,14 +12,12 @@ //! 3. Add the algorithm name to `AVAILABLE_ALGORITHMS` use std::{ - panic::{catch_unwind, AssertUnwindSafe}, sync::Arc, thread::{self, JoinHandle}, - time::Duration, }; use tokio::sync::broadcast; -use tracing::{error, info}; +use tracing::info; use crate::{ algorithm::{ @@ -30,7 +28,7 @@ use crate::{ feed::{events::MarketEvent, market_data::MarketData}, propamm_fallback::SharedFeeTiers, types::internal::SolveTask, - worker_pool::worker::SolverWorker, + worker_pool::supervisor::{RespawnPolicy, WorkerContext}, worker_pool_router::LiquidityScope, }; @@ -41,10 +39,6 @@ pub(crate) const AVAILABLE_ALGORITHMS: &[&str] = /// Default algorithm to use if none specified. pub(crate) const DEFAULT_ALGORITHM: &str = "most_liquid"; -/// Pause before respawning a panicked worker, so a panic during worker -/// initialization cannot turn the respawn loop into a hot spin. -const RESPAWN_BACKOFF: Duration = Duration::from_millis(100); - /// Parameters for spawning workers. pub(crate) struct SpawnWorkersParams { /// Algorithm name (e.g., "most_liquid") — used for thread naming and logging. @@ -73,6 +67,10 @@ pub(crate) struct SpawnWorkersParams { pub exclude_protocols: Vec, /// PropAMMRouter fee tiers, shared with the fetcher that refreshes them. pub fallback_fee_tiers: SharedFeeTiers, + /// Retry policy for respawning panicked workers. + pub respawn_policy: RespawnPolicy, + /// Called when a worker gives up respawning. + pub on_worker_gave_up: Arc, } /// Error returned when algorithm registration fails. @@ -169,13 +167,11 @@ impl AlgorithmSpawner { /// Generic worker spawning logic. /// -/// This handles the common parts of spawning workers: -/// - Creating threads with proper names -/// - Setting up tokio runtimes -/// - Initializing graphs and running worker loops -/// -/// The `factory` closure is called once per worker to create the algorithm instance. -/// It is borrowed rather than consumed, so callers (including type-erased spawner closures) +/// Each worker thread runs sessions in a loop (see [`WorkerContext::run_sessions`]): +/// a panic ends the current session and the worker is respawned after a backoff, +/// giving up after repeated rapid failures. The `factory` closure is called at +/// every session (re)start, so it must tolerate repeated calls. It is borrowed +/// rather than consumed, so callers (including type-erased spawner closures) /// can call this function without giving up ownership of the factory. pub(crate) fn spawn_workers_generic( params: SpawnWorkersParams, @@ -190,105 +186,31 @@ where let mut workers = Vec::with_capacity(params.num_workers); for worker_id in 0..params.num_workers { - let task_rx = params.task_rx.clone(); - let market_data = params.market_data.clone(); - let derived_data = Arc::clone(¶ms.derived_data); - let event_rx = params.event_rx.resubscribe(); - let derived_event_rx = params.derived_event_rx.resubscribe(); - let algorithm_config = params.algorithm_config.clone(); - // Subscribed before the thread starts so shutdown signals sent at any point, - // including while the worker is recovering from a panic, are never missed. - let mut shutdown_rx = params.shutdown_tx.subscribe(); - let algorithm_name = params.algorithm.clone(); - let pool_name = params.pool_name.clone(); - let factory = factory.clone(); - let liquidity_scope = params.liquidity_scope; - let exclude_protocols = params.exclude_protocols.clone(); - let fallback_fee_tiers = params.fallback_fee_tiers.clone(); + let ctx = WorkerContext { + worker_id, + algorithm_name: params.algorithm.clone(), + pool_name: params.pool_name.clone(), + factory: factory.clone(), + algorithm_config: params.algorithm_config.clone(), + market_data: params.market_data.clone(), + derived_data: Arc::clone(¶ms.derived_data), + task_rx: params.task_rx.clone(), + event_rx: params.event_rx.resubscribe(), + derived_event_rx: params.derived_event_rx.resubscribe(), + // Subscribed before the thread starts so shutdown signals sent at any + // point, including while the worker is recovering from a panic, are + // never missed. + shutdown_rx: params.shutdown_tx.subscribe(), + liquidity_scope: params.liquidity_scope, + exclude_protocols: params.exclude_protocols.clone(), + fallback_fee_tiers: params.fallback_fee_tiers.clone(), + respawn_policy: params.respawn_policy, + on_worker_gave_up: Arc::clone(¶ms.on_worker_gave_up), + }; let handle = thread::Builder::new() - .name(format!("{}-worker-{}", algorithm_name, worker_id)) - .spawn(move || { - // Run worker sessions until clean shutdown. A panic while solving (e.g. pool - // math dividing by zero) only ends the current session: it is reported loudly - // and the worker is respawned, instead of silently losing the thread until - // every worker in the pool is dead. - loop { - // Fresh receivers for this session; the previous session's receivers are - // consumed (or poisoned) when it panics. - let session_event_rx = event_rx.resubscribe(); - let session_derived_event_rx = derived_event_rx.resubscribe(); - let session_shutdown_rx = shutdown_rx.resubscribe(); - - // A shutdown sent while no session was listening (e.g. mid-respawn) is - // buffered in the receiver created before the thread started. - match shutdown_rx.try_recv() { - Err(broadcast::error::TryRecvError::Empty) => {} - // Received a shutdown, or the pool dropped the sender. - _ => break, - } - - let session = catch_unwind(AssertUnwindSafe(|| { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("failed to create tokio runtime"); - - rt.block_on(async { - let algorithm = factory(algorithm_config.clone()); - - let mut worker = SolverWorker::new( - market_data.clone(), - Arc::clone(&derived_data), - algorithm, - worker_id, - pool_name.clone(), - ) - .with_liquidity_scope(liquidity_scope) - .with_exclude_protocols(exclude_protocols.clone()) - .with_fallback_fee_tiers(fallback_fee_tiers.clone()); - - worker.initialize_graph().await; - worker - .run( - session_event_rx, - session_derived_event_rx, - task_rx.clone(), - session_shutdown_rx, - ) - .await; - }); - })); - - match session { - Ok(()) => break, - Err(panic_payload) => { - let panic_message = panic_payload - .downcast_ref::<&str>() - .copied() - .or_else(|| { - panic_payload - .downcast_ref::() - .map(String::as_str) - }) - .unwrap_or(""); - error!( - pool = %pool_name, - algorithm = %algorithm_name, - worker_id, - panic = %panic_message, - "worker thread panicked; respawning worker" - ); - metrics::counter!( - "worker_pool_worker_panics_total", - "pool" => pool_name.clone() - ) - .increment(1); - thread::sleep(RESPAWN_BACKOFF); - } - } - } - }) + .name(format!("{}-worker-{}", params.algorithm, worker_id)) + .spawn(move || ctx.run_sessions()) .expect("failed to spawn worker thread"); workers.push(handle); @@ -377,6 +299,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), } } @@ -418,6 +342,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }; let workers = @@ -462,6 +388,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }); assert!(registry_err.is_err()); @@ -492,6 +420,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }); assert!(workers.is_ok()); @@ -562,6 +492,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }; let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; let workers = spawn_workers_generic(params, &factory); @@ -627,6 +559,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }; let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; let workers = spawn_workers_generic(params, &factory); @@ -667,6 +601,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }; let workers = @@ -701,6 +637,8 @@ mod tests { liquidity_scope: LiquidityScope::default(), exclude_protocols: Vec::new(), fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy::default(), + on_worker_gave_up: Arc::new(|| {}), }; let workers = diff --git a/fynd-core/src/worker_pool/supervisor.rs b/fynd-core/src/worker_pool/supervisor.rs index d0838da9..24c9313a 100644 --- a/fynd-core/src/worker_pool/supervisor.rs +++ b/fynd-core/src/worker_pool/supervisor.rs @@ -1,6 +1,24 @@ -//! Worker-session supervision: respawn policy and (from Task 4) the session loop. +//! Worker-session supervision: the respawn policy and the per-thread session loop. -use std::time::Duration; +use std::{ + panic::{catch_unwind, AssertUnwindSafe}, + sync::Arc, + thread, + time::{Duration, Instant}, +}; + +use tokio::sync::broadcast; +use tracing::error; + +use crate::{ + algorithm::AlgorithmConfig, + derived::{events::DerivedDataEvent, SharedDerivedDataRef}, + feed::{events::MarketEvent, market_data::MarketData}, + propamm_fallback::SharedFeeTiers, + types::internal::SolveTask, + worker_pool::worker::SolverWorker, + worker_pool_router::LiquidityScope, +}; /// Retry policy for respawning a panicked worker. /// @@ -65,11 +83,141 @@ impl RespawnState { } } +/// Everything one worker thread needs to run sessions until shutdown or give-up. +pub(crate) struct WorkerContext +where + A: crate::algorithm::Algorithm + 'static, + A::GraphManager: + crate::feed::events::MarketEventHandler + crate::graph::EdgeWeightUpdaterWithDerived, + F: Fn(AlgorithmConfig) -> A + Send + Sync + 'static, +{ + pub worker_id: usize, + pub algorithm_name: String, + pub pool_name: String, + pub factory: F, + pub algorithm_config: AlgorithmConfig, + pub market_data: MarketData, + pub derived_data: SharedDerivedDataRef, + pub task_rx: async_channel::Receiver, + pub event_rx: broadcast::Receiver, + pub derived_event_rx: broadcast::Receiver, + pub shutdown_rx: broadcast::Receiver<()>, + pub liquidity_scope: LiquidityScope, + pub exclude_protocols: Vec, + pub fallback_fee_tiers: SharedFeeTiers, + pub respawn_policy: RespawnPolicy, + pub on_worker_gave_up: Arc, +} + +impl WorkerContext +where + A: crate::algorithm::Algorithm + 'static, + A::GraphManager: + crate::feed::events::MarketEventHandler + crate::graph::EdgeWeightUpdaterWithDerived, + F: Fn(AlgorithmConfig) -> A + Send + Sync + 'static, +{ + /// Runs worker sessions until clean shutdown or give-up. + /// + /// Panics (e.g. pool math dividing by zero) are contained to the current + /// session so one bad task cannot permanently kill the worker thread. + pub(crate) fn run_sessions(mut self) { + let mut respawn = RespawnState::new(self.respawn_policy); + loop { + // Fresh receivers for this session; the previous session's receivers + // were moved into it and dropped when it ended. + let session_event_rx = self.event_rx.resubscribe(); + let session_derived_event_rx = self.derived_event_rx.resubscribe(); + let session_shutdown_rx = self.shutdown_rx.resubscribe(); + + // A shutdown sent while no session was listening (e.g. mid-respawn) is + // buffered in the receiver created before the thread started. + match self.shutdown_rx.try_recv() { + Err(broadcast::error::TryRecvError::Empty) => {} + // Received a shutdown, or the pool dropped the sender. + _ => break, + } + + let session_started = Instant::now(); + let session_result = catch_unwind(AssertUnwindSafe(|| { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("failed to create tokio runtime"); + + rt.block_on(async { + let algorithm = (self.factory)(self.algorithm_config.clone()); + + let mut worker = SolverWorker::new( + self.market_data.clone(), + Arc::clone(&self.derived_data), + algorithm, + self.worker_id, + self.pool_name.clone(), + ) + .with_liquidity_scope(self.liquidity_scope) + .with_exclude_protocols(self.exclude_protocols.clone()) + .with_fallback_fee_tiers(self.fallback_fee_tiers.clone()); + + worker.initialize_graph().await; + worker + .run( + session_event_rx, + session_derived_event_rx, + self.task_rx.clone(), + session_shutdown_rx, + ) + .await; + }); + })); + + match session_result { + Ok(()) => break, + Err(panic_payload) => { + let panic_message = panic_payload + .downcast_ref::<&str>() + .copied() + .or_else(|| { + panic_payload + .downcast_ref::() + .map(String::as_str) + }) + .unwrap_or(""); + error!( + pool = %self.pool_name, + algorithm = %self.algorithm_name, + worker_id = self.worker_id, + panic = %panic_message, + "worker thread panicked; respawning worker" + ); + metrics::counter!( + "worker_pool_worker_panics_total", + "pool" => self.pool_name.clone() + ) + .increment(1); + match respawn.on_failure(session_started.elapsed()) { + FailureAction::Retry(delay) => thread::sleep(delay), + FailureAction::GiveUp => { + error!( + pool = %self.pool_name, + worker_id = self.worker_id, + "worker gave up after repeated rapid panics" + ); + (self.on_worker_gave_up)(); + break; + } + } + } + } + } + } +} + #[cfg(test)] mod tests { - use super::*; use std::time::Duration; + use super::*; + fn fast_policy() -> RespawnPolicy { RespawnPolicy { initial_backoff: Duration::from_millis(100), From 8f07bb542264021e51844c9d4c50326079826d05 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:17:06 -0300 Subject: [PATCH 05/13] test: cover give-up and shutdown sent mid-respawn --- fynd-core/src/worker_pool/registry.rs | 117 ++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index 1b10bf92..f115238b 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -578,6 +578,123 @@ mod tests { } } + #[tokio::test] + async fn worker_gives_up_after_repeated_rapid_panics() { + let (market, _) = setup_market_weighted(vec![]); + let (_task_tx, task_rx) = async_channel::bounded::(10); + let (event_tx, _) = broadcast::channel::(10); + let (derived_event_tx, _) = broadcast::channel(10); + let (shutdown_tx, _) = broadcast::channel(1); + let gave_up = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let gave_up_flag = Arc::clone(&gave_up); + + let params = SpawnWorkersParams { + algorithm: "panic_on_poison".to_string(), + pool_name: "test_pool".to_string(), + num_workers: 1, + algorithm_config: AlgorithmConfig::default(), + task_rx, + market_data: market, + derived_data: DerivedData::new_shared(), + event_rx: event_tx.subscribe(), + derived_event_rx: derived_event_tx.subscribe(), + shutdown_tx: shutdown_tx.clone(), + liquidity_scope: LiquidityScope::default(), + exclude_protocols: Vec::new(), + fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy { + initial_backoff: Duration::from_millis(1), + max_backoff: Duration::from_millis(2), + max_attempts: 3, + stable_session: Duration::from_secs(60), + }, + on_worker_gave_up: Arc::new(move || { + gave_up_flag.store(true, std::sync::atomic::Ordering::SeqCst) + }), + }; + let factory = |_config: AlgorithmConfig| -> PanicOnPoisonAlgorithm { + panic!("deterministic init panic") + }; + let workers = spawn_workers_generic(params, &factory); + + for handle in workers { + tokio::time::timeout( + Duration::from_secs(5), + tokio::task::spawn_blocking(move || handle.join()), + ) + .await + .expect("worker should give up instead of retrying forever") + .unwrap() + .expect("worker thread should exit cleanly after giving up"); + } + assert!(gave_up.load(std::sync::atomic::Ordering::SeqCst)); + } + + #[tokio::test] + async fn shutdown_sent_mid_respawn_is_not_lost() { + let (market, _) = setup_market_weighted(vec![]); + let derived_data = DerivedData::new_shared(); + let (task_tx, task_rx) = async_channel::bounded(10); + let (event_tx, _) = broadcast::channel::(10); + let (derived_event_tx, _) = broadcast::channel(10); + let (shutdown_tx, _) = broadcast::channel(1); + + let params = SpawnWorkersParams { + algorithm: "panic_on_poison".to_string(), + pool_name: "test_pool".to_string(), + num_workers: 1, + algorithm_config: AlgorithmConfig::default(), + task_rx, + market_data: market, + derived_data, + event_rx: event_tx.subscribe(), + derived_event_rx: derived_event_tx.subscribe(), + shutdown_tx: shutdown_tx.clone(), + liquidity_scope: LiquidityScope::default(), + exclude_protocols: Vec::new(), + fallback_fee_tiers: SharedFeeTiers::default(), + respawn_policy: RespawnPolicy { + initial_backoff: Duration::from_millis(500), + ..RespawnPolicy::default() + }, + on_worker_gave_up: Arc::new(|| {}), + }; + let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; + let workers = spawn_workers_generic(params, &factory); + + let token_a = token(0x01, "A"); + let token_b = token(0x02, "B"); + + // The poison task panics mid-solve; its response channel is dropped. The worker + // now sleeps through its 500ms backoff with no session listening for shutdown. + let (poison_tx, poison_rx) = oneshot::channel(); + let poison_order = order(&token_a, &token_b, POISON_AMOUNT as u128, OrderSide::Sell); + task_tx + .send(SolveTask::new(Uuid::new_v4(), poison_order, poison_tx)) + .await + .unwrap(); + assert!(poison_rx.await.is_err()); + + // Sent while the worker is mid-respawn (no session receiver listening yet). The + // buffered `shutdown_rx.try_recv()` pre-check must still catch it. + shutdown_tx.send(()).unwrap(); + + for handle in workers { + tokio::time::timeout( + Duration::from_secs(5), + tokio::task::spawn_blocking(move || handle.join()), + ) + .await + .expect("buffered shutdown sent mid-respawn should not be lost") + .unwrap() + .expect("worker thread should shut down cleanly"); + } + + // Keep the sender alive until after the join so the worker exits via the + // buffered shutdown signal, not because the task channel closed. + drop(task_tx); + } + #[test] fn test_registry_spawns_path_frank_wolfe() { let (shutdown_tx, _) = broadcast::channel(1); From 08852f751857029b2127a1fecee5f57fbc3c1599 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:25:44 -0300 Subject: [PATCH 06/13] fix: seed respawned worker readiness from the shared derived store --- fynd-core/src/derived/tracker.rs | 46 +++++++++++++++++++++++++++++ fynd-core/src/worker_pool/worker.rs | 9 ++++++ 2 files changed, 55 insertions(+) diff --git a/fynd-core/src/derived/tracker.rs b/fynd-core/src/derived/tracker.rs index b2b13000..c2e21846 100644 --- a/fynd-core/src/derived/tracker.rs +++ b/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,30 @@ impl ReadinessTracker { pub fn current_block(&self) -> Option { 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. + pub fn seed_from_store(&mut self, store: &DerivedData) { + let stale_ids: Vec = self + .requirements + .stale_requirements() + .iter() + .copied() + .collect(); + for id in stale_ids { + if let Some(block) = store.output_block(id) { + self.on_computation_complete(id, block); + } + } + } } #[cfg(test)] @@ -531,6 +556,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() diff --git a/fynd-core/src/worker_pool/worker.rs b/fynd-core/src/worker_pool/worker.rs index e2f013e4..0bc84d1c 100644 --- a/fynd-core/src/worker_pool/worker.rs +++ b/fynd-core/src/worker_pool/worker.rs @@ -624,6 +624,15 @@ where ) where A::GraphManager: EdgeWeightUpdaterWithDerived, { + // A respawned worker (fresh session after a panic) starts with an empty tracker while + // the shared store already holds results from before the panic; seed it so allow_stale + // requirements don't wait for the next broadcast event to become ready again. + let derived_data = Arc::clone(&self.derived_data); + let store = derived_data.read().await; + self.readiness_tracker + .seed_from_store(&store); + drop(store); + info!(self.worker_id, "worker started"); // Once the derived-data channel closes, its recv() returns Closed instantly on every From fe9708aa4bee2f24d71329f9d0fff45e151762c9 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:34:09 -0300 Subject: [PATCH 07/13] test: tighten the poison-task assertions and naming --- fynd-core/src/worker_pool/registry.rs | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index f115238b..2dfb3a71 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -431,10 +431,11 @@ mod tests { } /// Amount marking the order whose solve panics in [`PanicOnPoisonAlgorithm`]. - const POISON_AMOUNT: u64 = 666; + const POISON_AMOUNT: u128 = 666; /// Algorithm that panics while solving the poison order and returns an error - /// otherwise. Used to verify that a panicking task does not kill the worker. + /// otherwise. Used to verify that a panicking task does not permanently lose the + /// worker: the worker respawns. #[derive(Clone)] struct PanicOnPoisonAlgorithm; @@ -503,21 +504,24 @@ mod tests { // The poison task panics mid-solve; its response channel is dropped. let (poison_tx, poison_rx) = oneshot::channel(); - let poison_order = order(&token_a, &token_b, POISON_AMOUNT as u128, OrderSide::Sell); + let poison_order = order(&token_a, &token_b, POISON_AMOUNT, OrderSide::Sell); task_tx .send(SolveTask::new(Uuid::new_v4(), poison_order, poison_tx)) .await .unwrap(); - let _ = poison_rx.await; + tokio::time::timeout(Duration::from_secs(5), poison_rx) + .await + .expect("poison task should be picked up") + .expect_err("poison task must panic, not respond"); // The worker must come back and answer the next task. - let (ok_tx, ok_rx) = oneshot::channel(); + let (normal_tx, normal_rx) = oneshot::channel(); let normal_order = order(&token_a, &token_b, 100, OrderSide::Sell); task_tx - .send(SolveTask::new(Uuid::new_v4(), normal_order, ok_tx)) + .send(SolveTask::new(Uuid::new_v4(), normal_order, normal_tx)) .await .unwrap(); - let response = tokio::time::timeout(Duration::from_secs(5), ok_rx) + let response = tokio::time::timeout(Duration::from_secs(5), normal_rx) .await .expect("worker should respawn after the panic and process the next task") .expect("worker should respond to the task"); From c54a10d3537f40082a0a6f4010e3b006f9503345 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:39:05 -0300 Subject: [PATCH 08/13] fix: remove redundant u128 cast on POISON_AMOUNT --- fynd-core/src/worker_pool/registry.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index 2dfb3a71..6dd8e49b 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -672,7 +672,7 @@ mod tests { // The poison task panics mid-solve; its response channel is dropped. The worker // now sleeps through its 500ms backoff with no session listening for shutdown. let (poison_tx, poison_rx) = oneshot::channel(); - let poison_order = order(&token_a, &token_b, POISON_AMOUNT as u128, OrderSide::Sell); + let poison_order = order(&token_a, &token_b, POISON_AMOUNT, OrderSide::Sell); task_tx .send(SolveTask::new(Uuid::new_v4(), poison_order, poison_tx)) .await From cbdedb622ebadbd01ab0fbf8365c7fe5df2f3ea5 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 15:46:35 -0300 Subject: [PATCH 09/13] test: share one configurable stub algorithm across worker tests --- fynd-core/src/algorithm/test_utils.rs | 83 ++++++++++++++++- fynd-core/src/worker_pool/registry.rs | 61 ++++--------- fynd-core/src/worker_pool/worker.rs | 124 +++++++------------------- 3 files changed, 127 insertions(+), 141 deletions(-) diff --git a/fynd-core/src/algorithm/test_utils.rs b/fynd-core/src/algorithm/test_utils.rs index 32d98107..4e80112d 100644 --- a/fynd-core/src/algorithm/test_utils.rs +++ b/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,75 @@ 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 + 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, + requirements: ComputationRequirements, + timeout: Duration, +} + +impl StubAlgorithm { + /// Creates a stub that answers every solve with `solve`. + pub fn returning( + solve: impl Fn(&Order) -> Result + 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 + } + + /// Override the solve timeout. + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } +} + +impl Algorithm for StubAlgorithm { + type GraphType = StableDiGraph; + type GraphManager = PetgraphStableDiGraphManager; + + fn name(&self) -> &str { + "stub" + } + + async fn find_best_route( + &self, + _graph: &Self::GraphType, + _market: MarketData, + _label: Option, + _derived: Option, + order: &Order, + ) -> Result { + (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::*; diff --git a/fynd-core/src/worker_pool/registry.rs b/fynd-core/src/worker_pool/registry.rs index 6dd8e49b..43f927b1 100644 --- a/fynd-core/src/worker_pool/registry.rs +++ b/fynd-core/src/worker_pool/registry.rs @@ -268,14 +268,12 @@ mod tests { use super::*; use crate::{ algorithm::{ - most_liquid::DepthAndPrice, - test_utils::{order, setup_market_weighted, token}, - Algorithm, AlgorithmError, + test_utils::{order, setup_market_weighted, token, StubAlgorithm}, + AlgorithmError, }, - derived::{computation::ComputationRequirements, DerivedData}, - feed::market_data::{MarketData, StateLabel}, - graph::petgraph::{PetgraphStableDiGraphManager, StableDiGraph}, - types::{quote::OrderSide, Order, RouteResult, SolveError}, + derived::DerivedData, + feed::market_data::MarketData, + types::{quote::OrderSide, SolveError}, }; fn make_params(algorithm: &str, num_workers: usize) -> SpawnWorkersParams { @@ -430,44 +428,18 @@ mod tests { let _ = shutdown_tx.send(()); } - /// Amount marking the order whose solve panics in [`PanicOnPoisonAlgorithm`]. + /// Amount marking the order whose solve panics in [`panic_on_poison`]. const POISON_AMOUNT: u128 = 666; - /// Algorithm that panics while solving the poison order and returns an error - /// otherwise. Used to verify that a panicking task does not permanently lose the - /// worker: the worker respawns. - #[derive(Clone)] - struct PanicOnPoisonAlgorithm; - - impl Algorithm for PanicOnPoisonAlgorithm { - type GraphType = StableDiGraph; - type GraphManager = PetgraphStableDiGraphManager; - - fn name(&self) -> &str { - "panic_on_poison" - } - - async fn find_best_route( - &self, - _graph: &Self::GraphType, - _market: MarketData, - _label: Option, - _derived: Option, - order: &Order, - ) -> Result { + /// Stub that panics while solving the poison order and returns an error otherwise. Used to + /// verify that a panicking task does not permanently lose the worker: the worker respawns. + fn panic_on_poison(_config: AlgorithmConfig) -> StubAlgorithm { + StubAlgorithm::returning(|order| { if order.amount() == &BigUint::from(POISON_AMOUNT) { panic!("poison order"); } Err(AlgorithmError::Other("no route in mock".to_string())) - } - - fn computation_requirements(&self) -> ComputationRequirements { - ComputationRequirements::none() - } - - fn timeout(&self) -> Duration { - Duration::from_secs(1) - } + }) } #[tokio::test] @@ -496,7 +468,7 @@ mod tests { respawn_policy: RespawnPolicy::default(), on_worker_gave_up: Arc::new(|| {}), }; - let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; + let factory = panic_on_poison; let workers = spawn_workers_generic(params, &factory); let token_a = token(0x01, "A"); @@ -566,7 +538,7 @@ mod tests { respawn_policy: RespawnPolicy::default(), on_worker_gave_up: Arc::new(|| {}), }; - let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; + let factory = panic_on_poison; let workers = spawn_workers_generic(params, &factory); // `params` (holding the only shutdown sender) is consumed and dropped above. @@ -616,9 +588,8 @@ mod tests { gave_up_flag.store(true, std::sync::atomic::Ordering::SeqCst) }), }; - let factory = |_config: AlgorithmConfig| -> PanicOnPoisonAlgorithm { - panic!("deterministic init panic") - }; + let factory = + |_config: AlgorithmConfig| -> StubAlgorithm { panic!("deterministic init panic") }; let workers = spawn_workers_generic(params, &factory); for handle in workers { @@ -663,7 +634,7 @@ mod tests { }, on_worker_gave_up: Arc::new(|| {}), }; - let factory = |_config: AlgorithmConfig| PanicOnPoisonAlgorithm; + let factory = panic_on_poison; let workers = spawn_workers_generic(params, &factory); let token_a = token(0x01, "A"); diff --git a/fynd-core/src/worker_pool/worker.rs b/fynd-core/src/worker_pool/worker.rs index 0bc84d1c..e984969d 100644 --- a/fynd-core/src/worker_pool/worker.rs +++ b/fynd-core/src/worker_pool/worker.rs @@ -826,7 +826,7 @@ mod tests { most_liquid::DepthAndPrice, test_utils::{ component, component_with_protocol, order, setup_market_weighted, token, - MockProtocolSim, + MockProtocolSim, StubAlgorithm, }, }, derived::{ @@ -840,73 +840,16 @@ mod tests { AlgorithmError, }; - /// A minimal mock algorithm for testing the worker. - /// Uses DepthAndPrice as the edge weight type to satisfy trait bounds. - struct MockAlgorithm { - requirements: ComputationRequirements, - timeout: Duration, + /// A stub whose solve always fails, for tests that exercise the worker around the solve + /// rather than the route it produces. + fn failing_algorithm() -> StubAlgorithm { + StubAlgorithm::returning(|_order| Err(AlgorithmError::Other("not implemented".to_string()))) } - impl MockAlgorithm { - fn new() -> Self { - Self { requirements: ComputationRequirements::none(), timeout: Duration::from_secs(1) } - } - - fn with_requirements(mut self, requirements: ComputationRequirements) -> Self { - self.requirements = requirements; - self - } - } - - impl Algorithm for MockAlgorithm { - type GraphType = StableDiGraph; - type GraphManager = PetgraphStableDiGraphManager; - - fn name(&self) -> &str { - "mock" - } - - async fn find_best_route( - &self, - _graph: &Self::GraphType, - _market: MarketData, - _label: Option, - _derived: Option, - _order: &Order, - ) -> Result { - Err(crate::AlgorithmError::Other("not implemented".to_string())) - } - - fn computation_requirements(&self) -> ComputationRequirements { - self.requirements.clone() - } - - fn timeout(&self) -> Duration { - self.timeout - } - } - - /// Mock algorithm that returns a structurally invalid route (two disconnected swaps). - /// Used to verify the worker rejects invalid routes regardless of which algorithm produced - /// them. - struct InvalidRouteAlgorithm; - - impl Algorithm for InvalidRouteAlgorithm { - type GraphType = StableDiGraph; - type GraphManager = PetgraphStableDiGraphManager; - - fn name(&self) -> &str { - "invalid_route_mock" - } - - async fn find_best_route( - &self, - _graph: &Self::GraphType, - _market: MarketData, - _label: Option, - _derived: Option, - _order: &Order, - ) -> Result { + /// A stub that returns a structurally invalid route (two disconnected swaps). Used to verify + /// the worker rejects invalid routes regardless of which algorithm produced them. + fn invalid_route_algorithm() -> StubAlgorithm { + StubAlgorithm::returning(|_order| { let token_a = token(0x01, "A"); let token_b = token(0x02, "B"); let token_c = token(0x03, "C"); @@ -938,23 +881,20 @@ mod tests { let route = Route::new(vec![swap_ab, swap_cd], FxHashMap::default()).expect("non-empty route"); Ok(RouteResult::new(route, num_bigint::BigInt::from(0), BigUint::from(1u64))) - } - - fn computation_requirements(&self) -> ComputationRequirements { - ComputationRequirements::none() - } - - fn timeout(&self) -> Duration { - Duration::from_secs(1) - } + }) } #[tokio::test] async fn test_quote_rejects_invalid_route() { let (market, _) = setup_market_weighted(vec![]); let derived = DerivedData::new_shared(); - let mut worker = - SolverWorker::new(market, derived, InvalidRouteAlgorithm, 0, "test_pool".to_string()); + let mut worker = SolverWorker::new( + market, + derived, + invalid_route_algorithm(), + 0, + "test_pool".to_string(), + ); let token_a = token(0x01, "A"); let token_b = token(0x02, "B"); @@ -1147,7 +1087,7 @@ mod tests { let (market, _) = setup_market_weighted(vec![]); let derived = DerivedData::new_shared(); - let algorithm = MockAlgorithm::new(); + let algorithm = failing_algorithm(); let worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Should return immediately since there are no requirements @@ -1165,7 +1105,7 @@ mod tests { let requirements = ComputationRequirements::none() .allow_stale(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Mark as ready by handling a completion event @@ -1192,7 +1132,7 @@ mod tests { let requirements = ComputationRequirements::none() .require_fresh(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Should timeout since no events are received @@ -1218,7 +1158,7 @@ mod tests { let requirements = ComputationRequirements::none() .require_fresh(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Clone the notify handle to simulate the main loop notifying @@ -1250,7 +1190,7 @@ mod tests { let worker = SolverWorker::new( market, DerivedData::new_shared(), - MockAlgorithm::new(), + failing_algorithm(), 0, "test_pool".to_string(), ) @@ -1273,7 +1213,7 @@ mod tests { let worker = SolverWorker::new( market, DerivedData::new_shared(), - MockAlgorithm::new(), + failing_algorithm(), 0, "test_pool".to_string(), ); @@ -1296,7 +1236,7 @@ mod tests { let worker = SolverWorker::new( market, DerivedData::new_shared(), - MockAlgorithm::new(), + failing_algorithm(), 0, "test_pool".to_string(), ); @@ -1316,7 +1256,7 @@ mod tests { let worker = SolverWorker::new( market, DerivedData::new_shared(), - MockAlgorithm::new(), + failing_algorithm(), 0, "test_pool".to_string(), ) @@ -1334,7 +1274,7 @@ mod tests { let worker = SolverWorker::new( market, DerivedData::new_shared(), - MockAlgorithm::new(), + failing_algorithm(), 0, "test_pool".to_string(), ); @@ -1353,7 +1293,7 @@ mod tests { let requirements = ComputationRequirements::none() .require_fresh(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Clone the notify handle and get a reference to the tracker @@ -1396,7 +1336,7 @@ mod tests { let requirements = ComputationRequirements::none() .allow_stale(TokenGasPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); let notify = worker.ready_notify.clone(); @@ -1441,7 +1381,7 @@ mod tests { let requirements = ComputationRequirements::none() .require_fresh(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Mark the current block and record a failure for spot_prices @@ -1489,7 +1429,7 @@ mod tests { let requirements = ComputationRequirements::none() .require_fresh(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Mark the current block and record a failure for spot_prices @@ -1531,7 +1471,7 @@ mod tests { let requirements = ComputationRequirements::none() .require_fresh(SpotPriceComputation::ID) .unwrap(); - let algorithm = MockAlgorithm::new().with_requirements(requirements); + let algorithm = failing_algorithm().with_requirements(requirements); let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); // Create channels @@ -1609,7 +1549,7 @@ mod tests { let (market, _) = setup_market_weighted(vec![]); let derived = DerivedData::new_shared(); let mut worker = - SolverWorker::new(market, derived, MockAlgorithm::new(), 0, "test_pool".to_string()); + SolverWorker::new(market, derived, failing_algorithm(), 0, "test_pool".to_string()); let (_event_tx, event_rx) = broadcast::channel::(16); let (derived_tx, derived_rx) = broadcast::channel::(16); From 0d4c313fa87e6d7ef742df33e289d3e3bb4b5c95 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 16:07:07 -0300 Subject: [PATCH 10/13] fix: exit exactly once when several workers give up together --- fynd-core/src/worker_pool/pool.rs | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/fynd-core/src/worker_pool/pool.rs b/fynd-core/src/worker_pool/pool.rs index ceefb183..433942d1 100644 --- a/fynd-core/src/worker_pool/pool.rs +++ b/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::{sync::Arc, thread::JoinHandle}; +use std::{ + sync::{Arc, Once}, + thread::JoinHandle, +}; use tokio::sync::broadcast; use tracing::{error, info}; @@ -144,8 +147,15 @@ impl WorkerPool { fallback_fee_tiers, respawn_policy: RespawnPolicy::default(), on_worker_gave_up: Arc::new(|| { - error!("a solver worker gave up after repeated panics; exiting so the orchestrator restarts the process"); - std::process::exit(1); + // 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)?; From 78da62515a111eb4e071235f19981e9c36a72b97 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 16:07:07 -0300 Subject: [PATCH 11/13] fix: seed stale readiness without moving the tracker's block --- fynd-core/src/derived/tracker.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/fynd-core/src/derived/tracker.rs b/fynd-core/src/derived/tracker.rs index c2e21846..1d96e70b 100644 --- a/fynd-core/src/derived/tracker.rs +++ b/fynd-core/src/derived/tracker.rs @@ -232,16 +232,20 @@ impl ReadinessTracker { /// 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) { - let stale_ids: Vec = self + for id in self .requirements .stale_requirements() .iter() .copied() - .collect(); - for id in stale_ids { - if let Some(block) = store.output_block(id) { - self.on_computation_complete(id, block); + { + if store.output_block(id).is_some() { + self.ever_computed.insert(id); } } } From 181bb5b38bc5e8b4730408ddef8d2b40029a3314 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 16:07:07 -0300 Subject: [PATCH 12/13] chore: correct panic docs, log wording, test names and dead code --- fynd-core/src/algorithm/sim_guard.rs | 7 ++++--- fynd-core/src/algorithm/test_utils.rs | 6 ------ fynd-core/src/worker_pool/supervisor.rs | 15 ++++++++------- fynd-core/src/worker_pool/worker.rs | 3 +-- 4 files changed, 13 insertions(+), 18 deletions(-) diff --git a/fynd-core/src/algorithm/sim_guard.rs b/fynd-core/src/algorithm/sim_guard.rs index 08cb263e..4486ec1e 100644 --- a/fynd-core/src/algorithm/sim_guard.rs +++ b/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}; diff --git a/fynd-core/src/algorithm/test_utils.rs b/fynd-core/src/algorithm/test_utils.rs index 4e80112d..ddb634b5 100644 --- a/fynd-core/src/algorithm/test_utils.rs +++ b/fynd-core/src/algorithm/test_utils.rs @@ -679,12 +679,6 @@ impl StubAlgorithm { self.requirements = requirements; self } - - /// Override the solve timeout. - pub fn with_timeout(mut self, timeout: Duration) -> Self { - self.timeout = timeout; - self - } } impl Algorithm for StubAlgorithm { diff --git a/fynd-core/src/worker_pool/supervisor.rs b/fynd-core/src/worker_pool/supervisor.rs index 24c9313a..d5032fa6 100644 --- a/fynd-core/src/worker_pool/supervisor.rs +++ b/fynd-core/src/worker_pool/supervisor.rs @@ -118,8 +118,9 @@ where { /// Runs worker sessions until clean shutdown or give-up. /// - /// Panics (e.g. pool math dividing by zero) are contained to the current - /// session so one bad task cannot permanently kill the worker thread. + /// Panics (e.g. algorithm or graph arithmetic overflowing) are contained to + /// the current session so one bad task cannot permanently kill the worker + /// thread. pub(crate) fn run_sessions(mut self) { let mut respawn = RespawnState::new(self.respawn_policy); loop { @@ -187,7 +188,7 @@ where algorithm = %self.algorithm_name, worker_id = self.worker_id, panic = %panic_message, - "worker thread panicked; respawning worker" + "worker session panicked" ); metrics::counter!( "worker_pool_worker_panics_total", @@ -228,7 +229,7 @@ mod tests { } #[test] - fn backoff_doubles_up_to_the_cap() { + fn test_backoff_doubles_up_to_the_cap() { let mut state = RespawnState::new(fast_policy()); let lived = Duration::from_millis(1); assert_eq!(state.on_failure(lived), FailureAction::Retry(Duration::from_millis(100))); @@ -237,13 +238,13 @@ mod tests { } #[test] - fn gives_up_after_max_attempts() { + fn test_gives_up_after_max_attempts() { let mut state = RespawnState::new(RespawnPolicy { max_attempts: 1, ..fast_policy() }); assert_eq!(state.on_failure(Duration::from_millis(1)), FailureAction::GiveUp); } #[test] - fn stable_session_resets_the_budget() { + fn test_stable_session_resets_the_budget() { let mut state = RespawnState::new(fast_policy()); let rapid = Duration::from_millis(1); state.on_failure(rapid); @@ -256,7 +257,7 @@ mod tests { } #[test] - fn backoff_cap_bounds_the_delay() { + fn test_backoff_cap_bounds_the_delay() { let mut state = RespawnState::new(RespawnPolicy { max_attempts: 10, ..fast_policy() }); let rapid = Duration::from_millis(1); let mut last = Duration::ZERO; diff --git a/fynd-core/src/worker_pool/worker.rs b/fynd-core/src/worker_pool/worker.rs index e984969d..56abd871 100644 --- a/fynd-core/src/worker_pool/worker.rs +++ b/fynd-core/src/worker_pool/worker.rs @@ -627,8 +627,7 @@ where // A respawned worker (fresh session after a panic) starts with an empty tracker while // the shared store already holds results from before the panic; seed it so allow_stale // requirements don't wait for the next broadcast event to become ready again. - let derived_data = Arc::clone(&self.derived_data); - let store = derived_data.read().await; + let store = self.derived_data.read().await; self.readiness_tracker .seed_from_store(&store); drop(store); From b20231f5247e0db17eefd4537905395652c704d2 Mon Sep 17 00:00:00 2001 From: Bruno Eidam Guerios Date: Thu, 3 Sep 2026 16:07:31 -0300 Subject: [PATCH 13/13] test: cover readiness seeding in the worker run loop --- fynd-core/src/worker_pool/worker.rs | 60 +++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/fynd-core/src/worker_pool/worker.rs b/fynd-core/src/worker_pool/worker.rs index 56abd871..246794da 100644 --- a/fynd-core/src/worker_pool/worker.rs +++ b/fynd-core/src/worker_pool/worker.rs @@ -1507,6 +1507,66 @@ mod tests { .expect("worker task should not panic"); } + /// A worker respawned after a panic starts with an empty readiness tracker, and no derived + /// event is replayed to it. It must answer an `allow_stale` task from what the shared store + /// already holds instead of waiting out the algorithm timeout. + #[tokio::test] + async fn worker_answers_stale_task_seeded_from_the_store() { + let (market, _) = setup_market_weighted(vec![]); + let derived = DerivedData::new_shared(); + derived + .write() + .await + .set_spot_prices(Default::default(), vec![], 7, true); + + let requirements = ComputationRequirements::none() + .allow_stale(SpotPriceComputation::ID) + .unwrap(); + let algorithm = failing_algorithm().with_requirements(requirements); + let mut worker = SolverWorker::new(market, derived, algorithm, 0, "test_pool".to_string()); + + let (_event_tx, event_rx) = broadcast::channel::(16); + let (_derived_tx, derived_rx) = broadcast::channel::(16); + let (task_tx, task_rx) = async_channel::bounded::(16); + let (shutdown_tx, shutdown_rx) = broadcast::channel::<()>(1); + + let handle = tokio::spawn(async move { + worker + .run(event_rx, derived_rx, task_rx, shutdown_rx) + .await; + }); + + let token_a = token(0x01, "A"); + let token_b = token(0x02, "B"); + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + task_tx + .send(SolveTask::new( + uuid::Uuid::new_v4(), + order(&token_a, &token_b, 100, OrderSide::Sell), + response_tx, + )) + .await + .expect("worker should be receiving tasks"); + + let result = tokio::time::timeout(Duration::from_secs(5), response_rx) + .await + .expect("worker should answer the task") + .expect("worker should not drop the responder"); + + // The stub always fails to solve; reaching the solve at all is what proves readiness was + // seeded. Without seeding the worker returns NotReady after the algorithm timeout. + assert!( + matches!(result, Err(SolveError::AlgorithmError(_))), + "expected the seeded worker to solve, got {result:?}" + ); + + let _ = shutdown_tx.send(()); + tokio::time::timeout(Duration::from_secs(1), handle) + .await + .expect("worker should shutdown") + .expect("worker task should not panic"); + } + /// Captures log output for assertions, shared between the subscriber and the test. #[derive(Clone, Default)] struct SharedLogBuffer(std::sync::Arc>>);