wip: horizon migration exploration

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

Author: suchapalaver

Cargo.toml

@@ -83,7 +83,7 @@ sqlx = { version = "0.8.2", features = [
 ], default-features = false }
 stdext = "0.3.3"
 tap_aggregator = { version = "0.5.6", default-features = false }
-tap_core = { version = "4.1.3", default-features = false }
+tap_core = { version = "4.1.4", default-features = false }
 tap_graph = { version = "0.3.4", features = ["v2"] }
 tempfile = "3.8.0"
 test-log = { version = "0.2.12", default-features = false }

crates/profiler/src/lib.rs

@@ -1,13 +1,17 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::fs::{self, File};
-use std::io::Write;
-use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::Arc;
-use std::thread;
-use std::time::{Duration, SystemTime};
+use std::{
+    fs::{self, File},
+    io::Write,
+    path::{Path, PathBuf},
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    thread,
+    time::{Duration, SystemTime},
+};
 
 use chrono::{DateTime, Utc};
 use pprof::protos::Message;

crates/service/src/tap/checks/allocation_eligible.rs

@@ -6,8 +6,7 @@ use std::collections::HashMap;
 use anyhow::anyhow;
 use indexer_allocation::Allocation;
 use tap_core::receipt::checks::{Check, CheckError, CheckResult};
-use thegraph_core::alloy::primitives::Address;
-use thegraph_core::CollectionId;
+use thegraph_core::{alloy::primitives::Address, CollectionId};
 use tokio::sync::watch::Receiver;
 
 use crate::tap::{CheckingReceipt, TapReceipt};
@@ -32,7 +31,9 @@ impl Check<TapReceipt> for AllocationEligible {
     ) -> CheckResult {
         let allocation_id = match receipt.signed_receipt() {
             TapReceipt::V1(receipt) => receipt.message.allocation_id,
-            TapReceipt::V2(receipt) => CollectionId::from(receipt.message.collection_id).as_address(),
+            TapReceipt::V2(receipt) => {
+                CollectionId::from(receipt.message.collection_id).as_address()
+            }
         };
         if !self
             .indexer_allocations

crates/service/src/tap/receipt_store.rs

@@ -7,7 +7,10 @@ use itertools::{Either, Itertools};
 use sqlx::{types::BigDecimal, PgPool};
 use tap_core::{manager::adapters::ReceiptStore, receipt::WithValueAndTimestamp};
 use thegraph_core::alloy::{hex::ToHexExt, sol_types::Eip712Domain};
-use tokio::{sync::mpsc::Receiver, sync::oneshot::Sender as OneShotSender, task::JoinHandle};
+use tokio::{
+    sync::{mpsc::Receiver, oneshot::Sender as OneShotSender},
+    task::JoinHandle,
+};
 use tokio_util::sync::CancellationToken;
 
 use super::{AdapterError, CheckingReceipt, IndexerTapContext, TapReceipt};

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

@@ -25,12 +25,14 @@ use tap_aggregator::grpc::{
     v1::tap_aggregator_client::TapAggregatorClient as AggregatorV1,
     v2::tap_aggregator_client::TapAggregatorClient as AggregatorV2,
 };
-use thegraph_core::alloy::{
-    hex::ToHexExt,
-    primitives::{Address, U256},
-    sol_types::Eip712Domain,
+use thegraph_core::{
+    alloy::{
+        hex::ToHexExt,
+        primitives::{Address, U256},
+        sol_types::Eip712Domain,
+    },
+    AllocationId as AllocationIdCore, CollectionId,
 };
-use thegraph_core::{AllocationId as AllocationIdCore, CollectionId};
 use tokio::{sync::watch::Receiver, task::JoinHandle};
 use tonic::transport::{Channel, Endpoint};
 use tracing::Level;
@@ -145,7 +147,8 @@ impl From<tap_graph::SignedRav> for RavInformation {
 impl From<&tap_graph::v2::SignedRav> for RavInformation {
     fn from(value: &tap_graph::v2::SignedRav) -> Self {
         RavInformation {
-            allocation_id: AllocationIdCore::from(CollectionId::from(value.message.collectionId)).into_inner(),
+            allocation_id: AllocationIdCore::from(CollectionId::from(value.message.collectionId))
+                .into_inner(),
             value_aggregate: value.message.valueAggregate,
         }
     }
@@ -214,9 +217,9 @@ pub enum SenderAccountMessage {
     /// as well as requesting the underlaying allocation rav request
     ///
     /// Custom behavior is defined in [ReceiptFees]
-    UpdateReceiptFees(Address, ReceiptFees),
+    UpdateReceiptFees(AllocationId, ReceiptFees),
     /// Updates the counter for invalid receipts and verify to deny sender
-    UpdateInvalidReceiptFees(Address, UnaggregatedReceipts),
+    UpdateInvalidReceiptFees(AllocationId, UnaggregatedReceipts),
     /// Update rav tracker
     UpdateRav(RavInformation),
     #[cfg(test)]
@@ -475,7 +478,7 @@ impl State {
                     .build();
 
                 SenderAllocation::<Horizon>::spawn_linked(
-                    Some(self.format_sender_allocation(&id)),
+                    Some(self.format_sender_allocation(&id.as_address())),
                     SenderAllocation::default(),
                     args,
                     sender_account_ref.get_cell(),
@@ -1061,7 +1064,7 @@ impl Actor for SenderAccount {
 
                 state
                     .invalid_receipts_tracker
-                    .update(allocation_id, unaggregated_fees.value);
+                    .update(allocation_id.address(), unaggregated_fees.value);
 
                 // invalid receipts can't go down
                 let should_deny = !state.denied && state.deny_condition_reached();
@@ -1098,7 +1101,7 @@ impl Actor for SenderAccount {
                         // add new value
                         state
                             .sender_fee_tracker
-                            .add(allocation_id, value, timestamp_ns);
+                            .add(allocation_id.address(), value, timestamp_ns);
 
                         SENDER_FEE_TRACKER
                             .with_label_values(&[&state.sender.to_string()])
@@ -1111,16 +1114,16 @@ impl Actor for SenderAccount {
                             .set(
                                 state
                                     .sender_fee_tracker
-                                    .get_total_fee_for_allocation(&allocation_id)
+                                    .get_total_fee_for_allocation(&allocation_id.address())
                                     .map(|fee| fee.value)
                                     .unwrap_or_default() as f64,
                             );
                     }
                     ReceiptFees::RavRequestResponse(fees, rav_result) => {
-                        state.finalize_rav_request(allocation_id, (fees, rav_result));
+                        state.finalize_rav_request(allocation_id.address(), (fees, rav_result));
                     }
                     ReceiptFees::UpdateValue(unaggregated_fees) => {
-                        state.update_sender_fee(allocation_id, unaggregated_fees);
+                        state.update_sender_fee(allocation_id.address(), unaggregated_fees);
                     }
                     ReceiptFees::Retry => {}
                 }
@@ -1138,8 +1141,10 @@ impl Actor for SenderAccount {
                     let total_fee_outside_buffer = state.sender_fee_tracker.get_ravable_total_fee();
                     let total_counter_for_allocation = state
                         .sender_fee_tracker
-                        .get_count_outside_buffer_for_allocation(&allocation_id);
-                    let can_trigger_rav = state.sender_fee_tracker.can_trigger_rav(allocation_id);
+                        .get_count_outside_buffer_for_allocation(&allocation_id.address());
+                    let can_trigger_rav = state
+                        .sender_fee_tracker
+                        .can_trigger_rav(allocation_id.address());
                     let counter_greater_receipt_limit = total_counter_for_allocation
                         >= state.config.rav_request_receipt_limit
                         && can_trigger_rav;
@@ -1159,7 +1164,9 @@ impl Actor for SenderAccount {
                             %allocation_id,
                             "Total counter greater than the receipt limit per rav. Triggering RAV request"
                         );
-                        state.rav_request_for_allocation(allocation_id).await
+                        state
+                            .rav_request_for_allocation(allocation_id.address())
+                            .await
                     } else {
                         Ok(())
                     };
@@ -1368,7 +1375,7 @@ impl Actor for SenderAccount {
 
                 // check for deny conditions
                 let _ = myself.cast(SenderAccountMessage::UpdateReceiptFees(
-                    allocation_id,
+                    AllocationId::Legacy(AllocationIdCore::from(allocation_id)),
                     ReceiptFees::Retry,
                 ));
 
@@ -1472,7 +1479,10 @@ pub mod tests {
         flush_messages, pgpool, ALLOCATION_ID_0, ALLOCATION_ID_1, TAP_SENDER as SENDER,
         TAP_SIGNER as SIGNER,
     };
-    use thegraph_core::alloy::{hex::ToHexExt, primitives::U256};
+    use thegraph_core::{
+        alloy::{hex::ToHexExt, primitives::U256},
+        AllocationId as AllocationIdCore,
+    };
     use tokio::sync::mpsc;
     use wiremock::{
         matchers::{body_string_contains, method},
@@ -1566,7 +1576,9 @@ pub mod tests {
             .call()
             .await;
 
-        let allocation_ids = HashSet::from_iter([AllocationId::Legacy(ALLOCATION_ID_0)]);
+        let allocation_ids = HashSet::from_iter([AllocationId::Legacy(AllocationIdCore::from(
+            ALLOCATION_ID_0,
+        ))]);
         // we expect it to create a sender allocation
         sender_account
             .cast(SenderAccountMessage::UpdateAllocationIds(
@@ -1660,7 +1672,7 @@ pub mod tests {
         // we expect it to create a sender allocation
         sender_account
             .cast(SenderAccountMessage::NewAllocationId(AllocationId::Legacy(
-                ALLOCATION_ID_0,
+                AllocationIdCore::from(ALLOCATION_ID_0),
             )))
             .unwrap();
 
@@ -1674,9 +1686,11 @@ pub mod tests {
         // nothing should change because we already created
         sender_account
             .cast(SenderAccountMessage::UpdateAllocationIds(
-                vec![AllocationId::Legacy(ALLOCATION_ID_0)]
-                    .into_iter()
-                    .collect(),
+                vec![AllocationId::Legacy(AllocationIdCore::from(
+                    ALLOCATION_ID_0,
+                ))]
+                .into_iter()
+                .collect(),
             ))
             .unwrap();
 
@@ -1754,7 +1768,7 @@ pub mod tests {
         basic_sender_account
             .sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::NewReceipt(TRIGGER_VALUE - 1, get_current_timestamp_u64_ns()),
             ))
             .unwrap();
@@ -1784,7 +1798,7 @@ pub mod tests {
         basic_sender_account
             .sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::NewReceipt(TRIGGER_VALUE, get_current_timestamp_u64_ns()),
             ))
             .unwrap();
@@ -1798,7 +1812,7 @@ pub mod tests {
         basic_sender_account
             .sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::Retry,
             ))
             .unwrap();
@@ -1827,7 +1841,7 @@ pub mod tests {
 
         sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::NewReceipt(1, get_current_timestamp_u64_ns()),
             ))
             .unwrap();
@@ -1837,7 +1851,7 @@ pub mod tests {
 
         sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::NewReceipt(1, get_current_timestamp_u64_ns()),
             ))
             .unwrap();
@@ -1848,7 +1862,7 @@ pub mod tests {
 
         sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::Retry,
             ))
             .unwrap();
@@ -1867,9 +1881,11 @@ pub mod tests {
         let (sender_account, _, prefix, _) = create_sender_account()
             .pgpool(pgpool)
             .initial_allocation(
-                vec![AllocationId::Legacy(ALLOCATION_ID_0)]
-                    .into_iter()
-                    .collect(),
+                vec![AllocationId::Legacy(AllocationIdCore::from(
+                    ALLOCATION_ID_0,
+                ))]
+                .into_iter()
+                .collect(),
             )
             .escrow_subgraph_endpoint(&mock_escrow_subgraph.uri())
             .call()
@@ -1954,7 +1970,7 @@ pub mod tests {
 
         sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::NewReceipt(TRIGGER_VALUE, get_current_timestamp_u64_ns()),
             ))
             .unwrap();
@@ -1990,7 +2006,7 @@ pub mod tests {
             ($value:expr) => {
                 sender_account
                     .cast(SenderAccountMessage::UpdateReceiptFees(
-                        ALLOCATION_ID_0,
+                        AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                         ReceiptFees::UpdateValue(UnaggregatedReceipts {
                             value: $value,
                             last_id: 11,
@@ -2007,7 +2023,7 @@ pub mod tests {
             ($value:expr) => {
                 sender_account
                     .cast(SenderAccountMessage::UpdateInvalidReceiptFees(
-                        ALLOCATION_ID_0,
+                        AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                         UnaggregatedReceipts {
                             value: $value,
                             last_id: 11,
@@ -2143,7 +2159,7 @@ pub mod tests {
             ($value:expr) => {
                 sender_account
                     .cast(SenderAccountMessage::UpdateReceiptFees(
-                        ALLOCATION_ID_0,
+                        AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                         ReceiptFees::UpdateValue(UnaggregatedReceipts {
                             value: $value,
                             last_id: 11,
@@ -2433,17 +2449,17 @@ pub mod tests {
         // set retry
         sender_account
             .cast(SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(AllocationIdCore::from(ALLOCATION_ID_0)),
                 ReceiptFees::NewReceipt(TRIGGER_VALUE, get_current_timestamp_u64_ns()),
             ))
             .unwrap();
         let msg = msg_receiver.recv().await.expect("Channel failed");
         assert!(matches!(
             msg,
             SenderAccountMessage::UpdateReceiptFees(
-                ALLOCATION_ID_0,
+                AllocationId::Legacy(allocation_id),
                 ReceiptFees::NewReceipt(TRIGGER_VALUE, _)
-            )
+            ) if allocation_id == AllocationIdCore::from(ALLOCATION_ID_0)
         ));
 
         let deny = call!(sender_account, SenderAccountMessage::GetDeny).unwrap();