wip: horizon migration exploration

Author: suchapalaver

Cargo.lock

@@ -82,13 +82,13 @@ sqlx = { version = "0.8.2", features = [
     "uuid",
 ], default-features = false }
 stdext = "0.3.3"
-tap_aggregator = { version = "0.5.5", default-features = false }
-tap_core = { version = "4.1.3", default-features = false }
-tap_graph = { version = "0.3.3", features = ["v2"] }
+tap_aggregator = { version = "0.5.6", 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 }
 test-with = "0.14.6"
-thegraph-core = { version = "0.15.0", features = [
+thegraph-core = { version = "0.15.0", git = "https://github.com/edgeandnode/toolshed", branch = "tmigone/collection-id", features = [
     "attestation",
     "alloy-eip712",
     "alloy-sol-types",

Cargo.toml

@@ -9,7 +9,10 @@ use tap_core::{
     },
     signed_message::SignatureBytes,
 };
-use thegraph_core::alloy::{dyn_abi::Eip712Domain, primitives::Address, signers::Signature};
+use thegraph_core::{
+    alloy::{dyn_abi::Eip712Domain, primitives::Address, signers::Signature},
+    CollectionId,
+};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TapReceipt {
@@ -70,13 +73,13 @@ impl Aggregate<TapReceipt> for tap_graph::v2::ReceiptAggregateVoucher {
             })
             .collect::<Result<_, _>>()
             .map_err(AggregationError::Other)?;
-        let allocation_id = receipts[0].message.allocation_id;
+        let collection_id = receipts[0].message.collection_id;
         let payer = receipts[0].message.payer;
         let data_service = receipts[0].message.data_service;
         let service_provider = receipts[0].message.service_provider;
 
         tap_graph::v2::ReceiptAggregateVoucher::aggregate_receipts(
-            allocation_id,
+            collection_id,
             payer,
             data_service,
             service_provider,
@@ -115,10 +118,17 @@ impl TapReceipt {
         }
     }
 
-    pub fn allocation_id(&self) -> Address {
+    pub fn allocation_id(&self) -> Option<Address> {
+        match self {
+            TapReceipt::V1(receipt) => Some(receipt.message.allocation_id),
+            _ => None,
+        }
+    }
+
+    pub fn collection_id(&self) -> Option<CollectionId> {
         match self {
-            TapReceipt::V1(receipt) => receipt.message.allocation_id,
-            TapReceipt::V2(receipt) => receipt.message.allocation_id,
+            TapReceipt::V1(_) => None,
+            TapReceipt::V2(receipt) => Some(receipt.message.collection_id.into()),
         }
     }
 

crates/indexer-receipt/src/lib.rs

@@ -29,16 +29,29 @@ impl Check<TapReceipt> for AllocationEligible {
         _: &tap_core::receipt::Context,
         receipt: &CheckingReceipt,
     ) -> CheckResult {
-        let allocation_id = receipt.signed_receipt().allocation_id();
-        if !self
-            .indexer_allocations
-            .borrow()
-            .contains_key(&allocation_id)
-        {
-            return Err(CheckError::Failed(anyhow!(
-                "Receipt allocation ID `{}` is not eligible for this indexer",
-                allocation_id
-            )));
+        if let Some(allocation_id) = receipt.signed_receipt().allocation_id() {
+            if !self
+                .indexer_allocations
+                .borrow()
+                .contains_key(&allocation_id)
+            {
+                return Err(CheckError::Failed(anyhow!(
+                    "Receipt allocation ID `{}` is not eligible for this indexer",
+                    allocation_id
+                )));
+            }
+        } else if let Some(collection_id) = receipt.signed_receipt().collection_id() {
+            let collection_id_address = collection_id.as_address();
+            if !self
+                .indexer_allocations
+                .borrow()
+                .contains_key(&collection_id_address)
+            {
+                return Err(CheckError::Failed(anyhow!(
+                    "Receipt collection ID `{}` is not eligible for this indexer",
+                    collection_id
+                )));
+            }
         }
         Ok(())
     }

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

@@ -164,7 +164,7 @@ impl InnerContext {
         let receipts_len = receipts.len();
         let mut signers = Vec::with_capacity(receipts_len);
         let mut signatures = Vec::with_capacity(receipts_len);
-        let mut allocation_ids = Vec::with_capacity(receipts_len);
+        let mut collection_ids = Vec::with_capacity(receipts_len);
         let mut payers = Vec::with_capacity(receipts_len);
         let mut data_services = Vec::with_capacity(receipts_len);
         let mut service_providers = Vec::with_capacity(receipts_len);
@@ -175,7 +175,7 @@ impl InnerContext {
         for receipt in receipts {
             signers.push(receipt.signer_address);
             signatures.push(receipt.signature);
-            allocation_ids.push(receipt.allocation_id);
+            collection_ids.push(receipt.collection_id);
             payers.push(receipt.payer);
             data_services.push(receipt.data_service);
             service_providers.push(receipt.service_provider);
@@ -187,7 +187,7 @@ impl InnerContext {
             r#"INSERT INTO tap_horizon_receipts (
                 signer_address,
                 signature,
-                allocation_id,
+                collection_id,
                 payer,
                 data_service,
                 service_provider,
@@ -207,7 +207,7 @@ impl InnerContext {
             )"#,
             &signers,
             &signatures,
-            &allocation_ids,
+            &collection_ids,
             &payers,
             &data_services,
             &service_providers,
@@ -328,7 +328,7 @@ impl DbReceiptV1 {
 pub struct DbReceiptV2 {
     signer_address: String,
     signature: Vec<u8>,
-    allocation_id: String,
+    collection_id: String,
     payer: String,
     data_service: String,
     service_provider: String,
@@ -342,7 +342,7 @@ impl DbReceiptV2 {
         receipt: &tap_graph::v2::SignedReceipt,
         separator: &Eip712Domain,
     ) -> anyhow::Result<Self> {
-        let allocation_id = receipt.message.allocation_id.encode_hex();
+        let collection_id = receipt.message.collection_id.encode_hex();
         let payer = receipt.message.payer.encode_hex();
         let data_service = receipt.message.data_service.encode_hex();
         let service_provider = receipt.message.service_provider.encode_hex();
@@ -360,7 +360,7 @@ impl DbReceiptV2 {
         let nonce = BigDecimal::from(receipt.message.nonce);
         let value = BigDecimal::from(BigInt::from(receipt.value()));
         Ok(Self {
-            allocation_id,
+            collection_id,
             payer,
             data_service,
             service_provider,

crates/service/src/tap/receipt_store.rs

@@ -25,10 +25,13 @@ 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,
+    },
+    CollectionId,
 };
 use tokio::{sync::watch::Receiver, task::JoinHandle};
 use tonic::transport::{Channel, Endpoint};
@@ -144,7 +147,7 @@ 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: value.message.allocationId,
+            allocation_id: CollectionId::from(value.message.collectionId).as_address(),
             value_aggregate: value.message.valueAggregate,
         }
     }
@@ -815,7 +818,7 @@ impl Actor for SenderAccount {
                         if config.horizon_enabled {
                             sqlx::query!(
                                 r#"
-                                    SELECT allocation_id, value_aggregate
+                                    SELECT collection_id, value_aggregate
                                     FROM tap_horizon_ravs
                                     WHERE payer = $1 AND last AND NOT final;
                                 "#,
@@ -825,7 +828,7 @@ impl Actor for SenderAccount {
                             .await
                             .expect("Should not fail to fetch from \"horizon\" scalar_tap_ravs")
                             .into_iter()
-                            .map(|record| (record.allocation_id, record.value_aggregate))
+                            .map(|record| (record.collection_id, record.value_aggregate))
                             .collect()
                         } else {
                             vec![]

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

@@ -700,13 +700,13 @@ impl State {
         let receipts_signer_allocations_in_db = sqlx::query!(
             r#"
                 WITH grouped AS (
-                    SELECT signer_address, allocation_id
+                    SELECT signer_address, collection_id
                     FROM tap_horizon_receipts
-                    GROUP BY signer_address, allocation_id
+                    GROUP BY signer_address, collection_id
                 )
                 SELECT 
                     signer_address,
-                    ARRAY_AGG(allocation_id) AS allocation_ids
+                    ARRAY_AGG(collection_id) AS allocation_ids
                 FROM grouped
                 GROUP BY signer_address
             "#
@@ -746,7 +746,7 @@ impl State {
             r#"
                 SELECT
                     payer,
-                    ARRAY_AGG(DISTINCT allocation_id) FILTER (WHERE NOT last) AS allocation_ids
+                    ARRAY_AGG(DISTINCT collection_id) FILTER (WHERE NOT last) AS allocation_ids
                 FROM tap_horizon_ravs
                 GROUP BY payer
             "#

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

@@ -818,7 +818,7 @@ where
         let reciepts_len = receipts.len();
         let mut reciepts_signers = Vec::with_capacity(reciepts_len);
         let mut encoded_signatures = Vec::with_capacity(reciepts_len);
-        let mut allocation_ids = Vec::with_capacity(reciepts_len);
+        let mut collection_ids = Vec::with_capacity(reciepts_len);
         let mut payers = Vec::with_capacity(reciepts_len);
         let mut data_services = Vec::with_capacity(reciepts_len);
         let mut service_providers = Vec::with_capacity(reciepts_len);
@@ -828,7 +828,7 @@ where
         let mut error_logs = Vec::with_capacity(reciepts_len);
 
         for (receipt, receipt_error) in receipts {
-            let allocation_id = receipt.message.allocation_id;
+            let collection_id = receipt.message.collection_id;
             let payer = receipt.message.payer;
             let data_service = receipt.message.data_service;
             let service_provider = receipt.message.service_provider;
@@ -841,13 +841,13 @@ where
                 })?;
             tracing::debug!(
                 "Receipt for allocation {} and signer {} failed reason: {}",
-                allocation_id.encode_hex(),
+                collection_id.encode_hex(),
                 receipt_signer.encode_hex(),
                 receipt_error
             );
             reciepts_signers.push(receipt_signer.encode_hex());
             encoded_signatures.push(encoded_signature);
-            allocation_ids.push(allocation_id.encode_hex());
+            collection_ids.push(collection_id.encode_hex());
             payers.push(payer.encode_hex());
             data_services.push(data_service.encode_hex());
             service_providers.push(service_provider.encode_hex());
@@ -860,7 +860,7 @@ where
             r#"INSERT INTO tap_horizon_receipts_invalid (
                 signer_address,
                 signature,
-                allocation_id,
+                collection_id,
                 payer,
                 data_service,
                 service_provider,
@@ -871,7 +871,7 @@ where
             ) SELECT * FROM UNNEST(
                 $1::CHAR(40)[],
                 $2::BYTEA[],
-                $3::CHAR(40)[],
+                $3::CHAR(64)[],
                 $4::CHAR(40)[],
                 $5::CHAR(40)[],
                 $6::CHAR(40)[],
@@ -882,7 +882,7 @@ where
             )"#,
             &reciepts_signers,
             &encoded_signatures,
-            &allocation_ids,
+            &collection_ids,
             &payers,
             &data_services,
             &service_providers,
@@ -1159,7 +1159,7 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
             FROM
                 tap_horizon_receipts_invalid
             WHERE
-                allocation_id = $1
+                collection_id = $1
                 AND signer_address IN (SELECT unnest($2::text[]))
             "#,
             self.allocation_id.encode_hex(),
@@ -1205,7 +1205,7 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
             FROM
                 tap_horizon_receipts
             WHERE
-                allocation_id = $1
+                collection_id = $1
                 AND service_provider = $2
                 AND id <= $3
                 AND signer_address IN (SELECT unnest($4::text[]))
@@ -1258,7 +1258,7 @@ impl DatabaseInteractions for SenderAllocationState<Horizon> {
                 UPDATE tap_horizon_ravs
                     SET last = true
                 WHERE 
-                    allocation_id = $1
+                    collection_id = $1
                     AND payer = $2
                     AND service_provider = $3
             "#,

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

@@ -155,6 +155,8 @@ impl NetworkVersion for Horizon {
 #[derive(Clone, bon::Builder)]
 pub struct TapAgentContext<T> {
     pgpool: PgPool,
+    /// For Legacy network: represents an allocation ID
+    /// For Horizon network: represents a collection ID (stored in collection_id database column)
     #[cfg_attr(test, builder(default = crate::test::ALLOCATION_ID_0))]
     allocation_id: Address,
     #[cfg_attr(test, builder(default = test_assets::TAP_SENDER.1))]

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

@@ -54,7 +54,7 @@ impl Check<TapReceipt> for AllocationId {
         _: &tap_core::receipt::Context,
         receipt: &CheckingReceipt,
     ) -> CheckResult {
-        let allocation_id = receipt.signed_receipt().allocation_id();
+        let allocation_id = receipt.signed_receipt().allocation_id().unwrap();
         // TODO: Remove the if block below? Each TAP Monitor is specific to an allocation
         // ID. So the receipts that are received here should already have been filtered by
         // allocation ID.