fix(tap-agent): enforce Horizon bucket isolation for RAV/receipt queries

  - Use full Horizon identity (payer, data_service, service_provider,
  collection_id) in reads to prevent cross-bucket reads.
  - sender_account.rs: last non-final RAVs lookup now filters by
  service_provider and data_service.
  - sender_accounts_manager.rs: non-final RAV aggregation constrained to the
  configured bucket via WHERE data_service = $1 AND service_provider = $2.
  - sender_allocation.rs (Horizon):
      - calculate_invalid_receipts_fee() now filters by collection_id, payer,
  service_provider, and data_service.
      - calculate_fee_until_last_id() now filters by collection_id, payer,
  service_provider, and data_service (and updated SQLx arg indices).
  - Rationale: Aligns with tap_horizon_ravs PK, matches Horizon’s identity
  model, avoids mixing state across different data services/providers, and
  complements the service-side DataServiceCheck which only guards ingestion.

Author: neithanmo

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

@@ -275,7 +275,7 @@ pub struct SenderAccountArgs {
     /// Watcher that returns a list of escrow accounts for current indexer
     pub escrow_accounts: Receiver<EscrowAccounts>,
     /// Watcher of normalized allocation IDs (Legacy/Horizon) for this sender type
-    pub indexer_allocations: Receiver<HashSet<AllocationId>>, 
+    pub indexer_allocations: Receiver<HashSet<AllocationId>>,
     /// SubgraphClient of the escrow subgraph
     pub escrow_subgraph: &'static SubgraphClient,
     /// SubgraphClient of the network subgraph
@@ -888,9 +888,18 @@ impl Actor for SenderAccount {
                                 r#"
                                     SELECT collection_id, value_aggregate
                                     FROM tap_horizon_ravs
-                                    WHERE payer = $1 AND last AND NOT final;
+                                    WHERE payer = $1
+                                    AND service_provider = $2
+                                    AND data_service = $3
+                                    AND last AND NOT final;
                                 "#,
                                 sender_id.encode_hex(),
+                                // service_provider is the indexer address; data_service comes from TapMode config
+                                config.indexer_address.encode_hex(),
+                                config
+                                    .tap_mode
+                                    .require_subgraph_service_address()
+                                    .encode_hex(),
                             )
                             .fetch_all(&pgpool)
                             .await

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

@@ -935,8 +935,16 @@ impl State {
                     payer,
                     ARRAY_AGG(DISTINCT collection_id) FILTER (WHERE NOT last) AS allocation_ids
                 FROM tap_horizon_ravs
+                WHERE data_service = $1 AND service_provider = $2
                 GROUP BY payer
-            "#
+            "#,
+            // Constrain to our Horizon bucket to avoid conflating RAVs across services/providers
+            self
+                .config
+                .tap_mode
+                .require_subgraph_service_address()
+                .encode_hex(),
+            self.config.indexer_address.encode_hex()
         )
         .fetch_all(&self.pgpool)
         .await

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

@@ -1277,13 +1277,19 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
                         WHERE timestamp_ns BETWEEN $1 AND $2
                         AND collection_id = $3
                         AND service_provider = $4
-                        AND signer_address IN (SELECT unnest($5::text[]));
+                        AND payer = $5
+                        AND data_service = $6
+                        AND signer_address IN (SELECT unnest($7::text[]));
                     "#,
             BigDecimal::from(min_timestamp),
             BigDecimal::from(max_timestamp),
             // self.allocation_id is already a CollectionId in Horizon state
             self.allocation_id.encode_hex(),
             self.indexer_address.encode_hex(),
+            self.sender.encode_hex(),
+            self.data_service
+                .expect("data_service should be available in Horizon mode")
+                .encode_hex(),
             &signers,
         )
         .execute(&self.pgpool)
@@ -1305,10 +1311,18 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
                 tap_horizon_receipts_invalid
             WHERE
                 collection_id = $1
-                AND signer_address IN (SELECT unnest($2::text[]))
+                AND service_provider = $2
+                AND payer = $3
+                AND data_service = $4
+                AND signer_address IN (SELECT unnest($5::text[]))
             "#,
             // self.allocation_id is already a CollectionId in Horizon state
             self.allocation_id.encode_hex(),
+            self.indexer_address.encode_hex(),
+            self.sender.encode_hex(),
+            self.data_service
+                .expect("data_service should be available in Horizon mode")
+                .encode_hex(),
             &signers
         )
         .fetch_one(&self.pgpool)
@@ -1353,13 +1367,19 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
             WHERE
                 collection_id = $1
                 AND service_provider = $2
-                AND id <= $3
-                AND signer_address IN (SELECT unnest($4::text[]))
-                AND timestamp_ns > $5
+                AND payer = $3
+                AND data_service = $4
+                AND id <= $5
+                AND signer_address IN (SELECT unnest($6::text[]))
+                AND timestamp_ns > $7
             "#,
             // self.allocation_id is already a CollectionId in Horizon state
             self.allocation_id.encode_hex(),
             self.indexer_address.encode_hex(),
+            self.sender.encode_hex(),
+            self.data_service
+                .expect("data_service should be available in Horizon mode")
+                .encode_hex(),
             last_id,
             &signers,
             BigDecimal::from(
@@ -1399,7 +1419,7 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
             allocation_id = %self.allocation_id,
             "Marking rav as last!",
         );
-        // TODO add service_provider filter
+
         let updated_rows = sqlx::query!(
             r#"
                 UPDATE tap_horizon_ravs