chore(agent): remove ractor from prod code paths

Signed-off-by: Joseph Livesey <[email protected]>

Author: suchapalaver

crates/tap-agent/Cargo.toml

@@ -10,7 +10,7 @@ path = "src/main.rs"
 
 [features]
 default = []
-test = ["dep:test-assets", "dep:rand"]
+test = ["dep:test-assets", "dep:rand", "dep:ractor"]
 profiling = ["profiler"]
 
 [dependencies]
@@ -43,7 +43,7 @@ graphql_client.workspace = true
 ruint.workspace = true
 futures-util.workspace = true
 jsonrpsee.workspace = true
-ractor.workspace = true
+ractor = { workspace = true, optional = true }
 tap_aggregator.workspace = true
 futures.workspace = true
 bon.workspace = true

crates/tap-agent/src/agent.rs

@@ -48,8 +48,7 @@ use sender_accounts_manager_task::SenderAccountsManagerTask;
 use tokio::task::JoinHandle;
 
 use crate::{
-    actor_migrate::LifecycleManager,
-    agent::sender_accounts_manager::{SenderAccountsManagerArgs, SenderAccountsManagerMessage},
+    actor_migrate::LifecycleManager, agent::sender_accounts_manager::SenderAccountsManagerMessage,
     database, CONFIG, EIP_712_DOMAIN,
 };
 
@@ -66,6 +65,9 @@ pub mod sender_accounts_manager_task;
 pub mod sender_allocation;
 /// Tokio task-based replacement for SenderAllocation actor
 pub mod sender_allocation_task;
+/// Comprehensive tests for tokio migration
+#[cfg(test)]
+mod test_tokio_migration;
 /// Unaggregated receipts containing total value and last id stored in the table
 pub mod unaggregated_receipts;
 
@@ -141,7 +143,7 @@ pub async fn start_agent() -> (TaskHandle<SenderAccountsManagerMessage>, JoinHan
         .await,
     ));
 
-    let indexer_allocations = indexer_allocations(
+    let _indexer_allocations = indexer_allocations(
         network_subgraph,
         *indexer_address,
         *network_sync_interval,
@@ -224,7 +226,7 @@ pub async fn start_agent() -> (TaskHandle<SenderAccountsManagerMessage>, JoinHan
     };
 
     // In both modes we need both watchers for the hybrid processing
-    let (escrow_accounts_v1_final, escrow_accounts_v2_final) = if is_horizon_enabled {
+    let (_escrow_accounts_v1_final, _escrow_accounts_v2_final) = if is_horizon_enabled {
         tracing::info!("TAP Agent: Horizon migration mode - processing existing V1 receipts and new V2 receipts");
         (escrow_accounts_v1, escrow_accounts_v2)
     } else {
@@ -238,19 +240,6 @@ pub async fn start_agent() -> (TaskHandle<SenderAccountsManagerMessage>, JoinHan
         config
     }));
 
-    let _args = SenderAccountsManagerArgs {
-        config,
-        domain_separator: EIP_712_DOMAIN.clone(),
-        pgpool: pgpool.clone(),
-        indexer_allocations,
-        escrow_accounts_v1: escrow_accounts_v1_final,
-        escrow_accounts_v2: escrow_accounts_v2_final,
-        escrow_subgraph,
-        network_subgraph,
-        sender_aggregator_endpoints: sender_aggregator_endpoints.clone(),
-        prefix: None,
-    };
-
     // 🎯 TOKIO MIGRATION: Using SenderAccountsManagerTask instead of ractor SenderAccountsManager
     let lifecycle = LifecycleManager::new();
 

crates/tap-agent/src/agent/sender_account.rs

@@ -40,10 +40,11 @@ use tracing::Level;
 
 use super::{
     sender_accounts_manager::{AllocationId, SenderType},
-    sender_allocation::{
-        AllocationConfig, SenderAllocation, SenderAllocationArgs, SenderAllocationMessage,
-    },
+    sender_allocation::{AllocationConfig, SenderAllocationMessage},
 };
+
+#[cfg(any(test, feature = "test"))]
+use super::sender_allocation::{SenderAllocation, SenderAllocationArgs};
 use crate::{
     adaptative_concurrency::AdaptiveLimiter,
     agent::unaggregated_receipts::UnaggregatedReceipts,
@@ -516,6 +517,7 @@ impl SenderAccountConfig {
     }
 }
 
+#[cfg(any(test, feature = "test"))]
 impl State {
     /// Spawn a sender allocation given the allocation_id
     ///
@@ -1609,6 +1611,7 @@ impl Actor for SenderAccount {
     }
 }
 
+#[cfg(any(test, feature = "test"))]
 impl SenderAccount {
     /// Deny sender by giving `sender` [Address]
     pub async fn deny_sender(sender_type: SenderType, pool: &PgPool, sender: Address) {

crates/tap-agent/src/agent/sender_accounts_manager.rs

@@ -15,6 +15,7 @@ use indexer_allocation::Allocation;
 use indexer_monitor::{EscrowAccounts, SubgraphClient};
 use indexer_watcher::{map_watcher, watch_pipe};
 use prometheus::{register_counter_vec, CounterVec};
+#[cfg(any(test, feature = "test"))]
 use ractor::{Actor, ActorCell, ActorProcessingErr, ActorRef, SupervisionEvent};
 use reqwest::Url;
 use serde::Deserialize;
@@ -25,9 +26,9 @@ use thegraph_core::{
 };
 use tokio::{select, sync::watch::Receiver};
 
-use super::sender_account::{
-    SenderAccount, SenderAccountArgs, SenderAccountConfig, SenderAccountMessage,
-};
+#[cfg(any(test, feature = "test"))]
+use super::sender_account::{SenderAccount, SenderAccountArgs};
+use super::sender_account::{SenderAccountConfig, SenderAccountMessage};
 use crate::agent::sender_allocation::SenderAllocationMessage;
 
 static RECEIPTS_CREATED: LazyLock<CounterVec> = LazyLock::new(|| {
@@ -144,6 +145,7 @@ impl NewReceiptNotification {
     }
 }
 
+#[cfg(any(test, feature = "test"))]
 /// Manager Actor
 #[derive(Debug, Clone)]
 pub struct SenderAccountsManager;
@@ -223,6 +225,7 @@ pub enum SenderAccountsManagerMessage {
     UpdateSenderAccountsV2(HashSet<Address>),
 }
 
+#[cfg(any(test, feature = "test"))]
 /// Arguments received in startup while spawing [SenderAccount] actor
 pub struct SenderAccountsManagerArgs {
     /// Config forwarded to [SenderAccount]
@@ -253,6 +256,7 @@ pub struct SenderAccountsManagerArgs {
 ///
 /// This is a separate instance that makes it easier to have mutable
 /// reference, for more information check ractor library
+#[cfg(any(test, feature = "test"))]
 pub struct State {
     sender_ids_v1: HashSet<Address>,
     sender_ids_v2: HashSet<Address>,
@@ -274,6 +278,7 @@ pub struct State {
 }
 
 #[async_trait::async_trait]
+#[cfg(any(test, feature = "test"))]
 impl Actor for SenderAccountsManager {
     type Msg = SenderAccountsManagerMessage;
     type State = State;
@@ -614,6 +619,7 @@ impl Actor for SenderAccountsManager {
     }
 }
 
+#[cfg(any(test, feature = "test"))]
 impl State {
     fn format_sender_account(&self, sender: &Address, sender_type: SenderType) -> String {
         let mut sender_allocation_id = String::new();
@@ -636,6 +642,7 @@ impl State {
     ///
     /// In case there's an error creating it, deny so it
     /// can no longer send queries
+    #[cfg(any(test, feature = "test"))]
     async fn create_or_deny_sender(
         &self,
         supervisor: ActorCell,
@@ -661,6 +668,7 @@ impl State {
     /// It takes the current [SenderAccountsManager] cell to use it
     /// as supervisor, sender address and a list of initial allocations
     ///
+    #[cfg(any(test, feature = "test"))]
     async fn create_sender_account(
         &self,
         supervisor: ActorCell,
@@ -922,6 +930,7 @@ impl State {
     ///
     /// Fails if the provided sender_id is not present
     /// in the sender_aggregator_endpoints map
+    #[cfg(any(test, feature = "test"))]
     fn new_sender_account_args(
         &self,
         sender_id: &Address,
@@ -958,6 +967,7 @@ impl State {
 
 /// Continuously listens for new receipt notifications from Postgres and forwards them to the
 /// corresponding SenderAccount.
+#[cfg(any(test, feature = "test"))]
 #[bon::builder]
 async fn new_receipts_watcher(
     actor_cell: ActorCell,
@@ -1080,6 +1090,7 @@ async fn new_receipts_watcher(
 /// After a request to create allocation, we don't need to do anything
 /// since the startup script is going to recalculate the receipt in the
 /// database
+#[cfg(any(test, feature = "test"))]
 async fn handle_notification(
     new_receipt_notification: NewReceiptNotification,
     escrow_accounts_rx: Receiver<EscrowAccounts>,

crates/tap-agent/src/agent/sender_allocation.rs

@@ -13,6 +13,7 @@ use bigdecimal::{num_bigint::BigInt, ToPrimitive};
 use indexer_monitor::{EscrowAccounts, SubgraphClient};
 use itertools::{Either, Itertools};
 use prometheus::{register_counter_vec, register_histogram_vec, CounterVec, HistogramVec};
+#[cfg(any(test, feature = "test"))]
 use ractor::{Actor, ActorProcessingErr, ActorRef};
 use sqlx::{types::BigDecimal, PgPool};
 use tap_core::{
@@ -117,14 +118,17 @@ type TapManager<T> = tap_core::manager::Manager<TapAgentContext<T>, TapReceipt>;
 ///
 /// T is used in SenderAllocationState<T> and SenderAllocationArgs<T> to store the
 /// correct Rav type and the correct aggregator client
+#[cfg(any(test, feature = "test"))]
 pub struct SenderAllocation<T>(PhantomData<T>);
+#[cfg(any(test, feature = "test"))]
 impl<T: NetworkVersion> Default for SenderAllocation<T> {
     fn default() -> Self {
         Self(PhantomData)
     }
 }
 
 /// State for [SenderAllocation] actor
+#[cfg(any(test, feature = "test"))]
 pub struct SenderAllocationState<T: NetworkVersion> {
     /// Sum of all receipt fees for the current allocation
     unaggregated_fees: UnaggregatedReceipts,
@@ -192,6 +196,7 @@ impl AllocationConfig {
 }
 
 /// Arguments used to initialize [SenderAllocation]
+#[cfg(any(test, feature = "test"))]
 #[derive(bon::Builder)]
 pub struct SenderAllocationArgs<T: NetworkVersion> {
     /// Database connection
@@ -245,6 +250,7 @@ pub enum SenderAllocationMessage {
 /// We use some bounds so [TapAgentContext] implements all parts needed for the given
 /// [crate::tap::context::NetworkVersion]
 #[async_trait::async_trait]
+#[cfg(any(test, feature = "test"))]
 impl<T> Actor for SenderAllocation<T>
 where
     SenderAllocationState<T>: DatabaseInteractions,
@@ -438,6 +444,7 @@ where
 
 /// We use some bounds so [TapAgentContext] implements all parts needed for the given
 /// [crate::tap::context::NetworkVersion]
+#[cfg(any(test, feature = "test"))]
 impl<T> SenderAllocationState<T>
 where
     T: NetworkVersion,
@@ -958,6 +965,7 @@ pub trait DatabaseInteractions {
     fn mark_rav_last(&self) -> impl Future<Output = anyhow::Result<()>> + Send;
 }
 
+#[cfg(any(test, feature = "test"))]
 impl DatabaseInteractions for SenderAllocationState<Legacy> {
     async fn delete_receipts_between(
         &self,
@@ -1119,6 +1127,7 @@ impl DatabaseInteractions for SenderAllocationState<Legacy> {
     }
 }
 
+#[cfg(any(test, feature = "test"))]
 impl DatabaseInteractions for SenderAllocationState<Horizon> {
     async fn delete_receipts_between(
         &self,