fix: v1 processing fix allocation type

Author: neithanmo

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

@@ -11,6 +11,7 @@ use std::{
 use anyhow::Context;
 use bigdecimal::{num_bigint::ToBigInt, ToPrimitive};
 use futures::{stream, StreamExt};
+use indexer_allocation::Allocation;
 use indexer_monitor::{EscrowAccounts, SubgraphClient};
 use indexer_query::{
     closed_allocations::{self, ClosedAllocations},
@@ -274,8 +275,9 @@ pub struct SenderAccountArgs {
     pub sender_id: Address,
     /// Watcher that returns a list of escrow accounts for current indexer
     pub escrow_accounts: Receiver<EscrowAccounts>,
-    /// Watcher that returns a set of open and recently closed allocation ids
-    pub indexer_allocations: Receiver<HashSet<AllocationId>>,
+    /// Raw watcher of open and recently closed allocations from Network Subgraph
+    /// We normalize per-sender to the correct variant (Legacy/Horizon)
+    pub indexer_allocations: Receiver<HashMap<Address, Allocation>>,
     /// SubgraphClient of the escrow subgraph
     pub escrow_subgraph: &'static SubgraphClient,
     /// SubgraphClient of the network subgraph
@@ -828,16 +830,38 @@ impl Actor for SenderAccount {
             sender_type,
         }: Self::Arguments,
     ) -> Result<Self::State, ActorProcessingErr> {
+        // Normalize raw allocation addresses to the correct variant for this sender_type
         let myself_clone = myself.clone();
-        watch_pipe(indexer_allocations, move |allocation_ids| {
-            let allocation_ids = allocation_ids.clone();
-            // Update the allocation_ids
-            myself_clone
-                .cast(SenderAccountMessage::UpdateAllocationIds(allocation_ids))
-                .unwrap_or_else(|e| {
-                    tracing::error!(error=?e, "Error while updating allocation_ids");
-                });
-            async {}
+        let sender_type_for_log = sender_type; // copy for move into closure
+        watch_pipe(indexer_allocations, move |alloc_map| {
+            let raw_count = alloc_map.len();
+            // Extract addresses and normalize based on sender_type
+            let normalized: HashSet<AllocationId> = alloc_map
+                .keys()
+                .cloned()
+                .map(|addr| match sender_type_for_log {
+                    SenderType::Legacy => AllocationId::Legacy(AllocationIdCore::from(addr)),
+                    SenderType::Horizon => AllocationId::Horizon(CollectionId::from(addr)),
+                })
+                .collect();
+
+            tracing::info!(
+                sender = %sender_id,
+                ?sender_type_for_log,
+                raw_count,
+                normalized_count = normalized.len(),
+                "indexer_allocations update: normalizing allocations for sender_type",
+            );
+
+            // Forward normalized set to the actor to update allocations
+            let myself = myself_clone.clone();
+            async move {
+                myself
+                    .cast(SenderAccountMessage::UpdateAllocationIds(normalized))
+                    .unwrap_or_else(|e| {
+                        tracing::error!(error=?e, "Error while updating allocation_ids");
+                    });
+            }
         });
 
         let myself_clone = myself.clone();

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

@@ -304,7 +304,8 @@ pub struct State {
     domain_separator: Eip712Domain,
     domain_separator_v2: Eip712Domain,
     pgpool: PgPool,
-    indexer_allocations: Receiver<HashSet<AllocationId>>,
+    // Raw allocation watcher (address -> Allocation). Normalized per-sender later.
+    indexer_allocations: Receiver<HashMap<Address, Allocation>>,
     /// Watcher containing the escrow accounts for v1
     escrow_accounts_v1: Receiver<EscrowAccounts>,
     /// Watcher containing the escrow accounts for v2
@@ -341,48 +342,12 @@ impl Actor for SenderAccountsManager {
             prefix,
         }: Self::Arguments,
     ) -> Result<Self::State, ActorProcessingErr> {
-        // we can use config.tap_mode.is_horizon() directly
-        // because agent_start method does the horizon smart contract/subgraph validation
-        // and updates the passed in config accordingly
-        let is_horizon_active = config.tap_mode.is_horizon();
-        if is_horizon_active {
-            tracing::info!(is_horizon = true, "Allocations will use Horizon type");
-        } else {
-            tracing::info!(is_horizon = false, "Allocations will use Legacy type");
-        }
-
-        let indexer_allocations = map_watcher(indexer_allocations, move |allocation_id| {
-            let allocations: HashSet<_> = allocation_id
-                .keys()
-                .cloned()
-                .map(|addr| {
-                    if is_horizon_active {
-                        AllocationId::Horizon(CollectionId::from(addr))
-                    } else {
-                        AllocationId::Legacy(AllocationIdCore::from(addr))
-                    }
-                })
-                .collect();
-
-            tracing::info!(
-                "Allocation watcher update: {} allocations mapped to {} format",
-                allocations.len(),
-                if is_horizon_active {
-                    "Horizon"
-                } else {
-                    "Legacy"
-                }
-            );
-            for alloc in &allocations {
-                tracing::debug!(
-                    "Mapped allocation: {} (address: {})",
-                    alloc,
-                    alloc.address()
-                );
-            }
-
-            allocations
-        });
+        // Do not pre-map allocations globally. We keep the raw watcher and
+        // normalize per SenderAccount based on its sender_type (Legacy/Horizon).
+        tracing::info!(
+            horizon_active = %config.tap_mode.is_horizon(),
+            "Using raw indexer_allocations watcher; normalization happens per sender"
+        );
         // we need two connections because each one will listen to different notify events
         let pglistener_v1 = PgListener::connect_with(&pgpool.clone()).await.unwrap();
 
@@ -1420,7 +1385,7 @@ mod tests {
                 new_receipts_watcher_handle_v1: None,
                 new_receipts_watcher_handle_v2: None,
                 pgpool,
-                indexer_allocations: watch::channel(HashSet::new()).1,
+                indexer_allocations: watch::channel(HashMap::new()).1,
                 escrow_accounts_v1: watch::channel(escrow_accounts.clone()).1,
                 escrow_accounts_v2: watch::channel(escrow_accounts).1,
                 escrow_subgraph: get_subgraph_client().await,

crates/tap-agent/src/test.rs

@@ -13,6 +13,7 @@ use actors::TestableActor;
 use anyhow::anyhow;
 use bigdecimal::num_bigint::BigInt;
 use indexer_config;
+use indexer_allocation::Allocation;
 use indexer_monitor::{DeploymentDetails, EscrowAccounts, SubgraphClient};
 use indexer_receipt::TapReceipt;
 use ractor::{concurrency::JoinHandle, Actor, ActorRef};
@@ -104,7 +105,7 @@ pub fn get_sender_account_config() -> &'static SenderAccountConfig {
 #[bon::builder]
 pub async fn create_sender_account(
     pgpool: PgPool,
-    #[builder(default = HashSet::new())] initial_allocation: HashSet<AllocationId>,
+    #[builder(default = HashMap::new())] initial_allocations_raw: HashMap<Address, Allocation>,
     #[builder(default = TRIGGER_VALUE)] rav_request_trigger_value: u128,
     #[builder(default = TRIGGER_VALUE)] max_amount_willing_to_lose_grt: u128,
     escrow_subgraph_endpoint: Option<&str>,
@@ -174,7 +175,7 @@ pub async fn create_sender_account(
         pgpool,
         sender_id: SENDER.1,
         escrow_accounts: escrow_accounts_rx,
-        indexer_allocations: watch::channel(initial_allocation).1,
+        indexer_allocations: watch::channel(initial_allocations_raw).1,
         escrow_subgraph,
         network_subgraph,
         domain_separator: TAP_EIP712_DOMAIN_SEPARATOR.clone(),