wip: enable horizon

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

Author: suchapalaver

tap_aggregator/Cargo.toml

@@ -41,3 +41,8 @@ tonic-build.workspace = true
 [dev-dependencies]
 rand.workspace = true
 rstest.workspace = true
+jsonrpsee = { workspace = true, features = ["http-client"] }
+
+[features]
+default = ["v2"]
+v2 = []

tap_aggregator/src/server.rs

@@ -14,6 +14,9 @@ use lazy_static::lazy_static;
 use log::{error, info};
 use prometheus::{register_counter, register_int_counter, Counter, IntCounter};
 use tap_core::signed_message::Eip712SignedMessage;
+#[cfg(feature = "v2")]
+use tap_graph::v2::{Receipt, ReceiptAggregateVoucher};
+#[cfg(not(feature = "v2"))]
 use tap_graph::{Receipt, ReceiptAggregateVoucher, SignedReceipt};
 use thegraph_core::alloy::{
     dyn_abi::Eip712Domain, primitives::Address, signers::local::PrivateKeySigner,
@@ -159,13 +162,28 @@ fn aggregate_receipts_(
     }
 
     let res = match api_version {
-        TapRpcApiVersion::V0_0 => aggregator::v1::check_and_aggregate_receipts(
-            domain_separator,
-            &receipts,
-            previous_rav,
-            wallet,
-            accepted_addresses,
-        ),
+        TapRpcApiVersion::V0_0 => {
+            #[cfg(feature = "v2")]
+            {
+                aggregator::v2::check_and_aggregate_receipts(
+                    domain_separator,
+                    &receipts,
+                    previous_rav,
+                    wallet,
+                    accepted_addresses,
+                )
+            }
+            #[cfg(not(feature = "v2"))]
+            {
+                aggregator::v1::check_and_aggregate_receipts(
+                    domain_separator,
+                    &receipts,
+                    previous_rav,
+                    wallet,
+                    accepted_addresses,
+                )
+            }
+        }
     };
 
     // Handle aggregation error
@@ -186,7 +204,9 @@ impl v1::tap_aggregator_server::TapAggregator for RpcImpl {
         request: Request<v1::RavRequest>,
     ) -> Result<Response<v1::RavResponse>, Status> {
         let rav_request = request.into_inner();
-        let receipts: Vec<SignedReceipt> = rav_request
+
+        // Always use v1 aggregation for v1 gRPC calls to maintain compatibility
+        let receipts: Vec<tap_graph::SignedReceipt> = rav_request
             .receipts
             .into_iter()
             .map(TryFrom::try_from)
@@ -506,6 +526,9 @@ mod tests {
     use jsonrpsee::{core::client::ClientT, http_client::HttpClientBuilder, rpc_params};
     use rstest::*;
     use tap_core::{signed_message::Eip712SignedMessage, tap_eip712_domain};
+    #[cfg(feature = "v2")]
+    use tap_graph::v2::{Receipt, ReceiptAggregateVoucher};
+    #[cfg(not(feature = "v2"))]
     use tap_graph::{Receipt, ReceiptAggregateVoucher};
     use thegraph_core::alloy::{
         dyn_abi::Eip712Domain, primitives::Address, signers::local::PrivateKeySigner,
@@ -535,6 +558,21 @@ mod tests {
         ]
     }
 
+    #[fixture]
+    fn payer() -> Address {
+        Address::from_str("0xabababababababababababababababababababab").unwrap()
+    }
+
+    #[fixture]
+    fn data_service() -> Address {
+        Address::from_str("0xdeaddeaddeaddeaddeaddeaddeaddeaddeaddead").unwrap()
+    }
+
+    #[fixture]
+    fn service_provider() -> Address {
+        Address::from_str("0xbeefbeefbeefbeefbeefbeefbeefbeefbeefbeef").unwrap()
+    }
+
     #[fixture]
     fn domain_separator() -> Eip712Domain {
         tap_eip712_domain(1, Address::from([0x11u8; 20]))
@@ -602,6 +640,9 @@ mod tests {
         http_response_size_limit: u32,
         http_max_concurrent_connections: u32,
         allocation_ids: Vec<Address>,
+        payer: Address,
+        data_service: Address,
+        service_provider: Address,
         #[case] values: Vec<u128>,
         #[values("0.0")] api_version: &str,
         #[values(0, 1, 2)] random_seed: u64,
@@ -643,6 +684,16 @@ mod tests {
             receipts.push(
                 Eip712SignedMessage::new(
                     &domain_separator,
+                    #[cfg(feature = "v2")]
+                    Receipt::new(
+                        allocation_ids[0],
+                        payer,
+                        data_service,
+                        service_provider,
+                        value,
+                    )
+                    .unwrap(),
+                    #[cfg(not(feature = "v2"))]
                     Receipt::new(allocation_ids[0], value).unwrap(),
                     &all_wallets.choose(&mut rng).unwrap().wallet,
                 )
@@ -662,9 +713,25 @@ mod tests {
 
         let remote_rav = res.data;
 
-        let local_rav =
-            ReceiptAggregateVoucher::aggregate_receipts(allocation_ids[0], &receipts, None)
-                .unwrap();
+        let local_rav = {
+            #[cfg(feature = "v2")]
+            {
+                ReceiptAggregateVoucher::aggregate_receipts(
+                    allocation_ids[0],
+                    payer,
+                    data_service,
+                    service_provider,
+                    &receipts,
+                    None,
+                )
+                .unwrap()
+            }
+            #[cfg(not(feature = "v2"))]
+            {
+                ReceiptAggregateVoucher::aggregate_receipts(allocation_ids[0], &receipts, None)
+                    .unwrap()
+            }
+        };
 
         assert!(remote_rav.message.allocationId == local_rav.allocationId);
         assert!(remote_rav.message.timestampNs == local_rav.timestampNs);
@@ -685,6 +752,9 @@ mod tests {
         http_response_size_limit: u32,
         http_max_concurrent_connections: u32,
         allocation_ids: Vec<Address>,
+        payer: Address,
+        data_service: Address,
+        service_provider: Address,
         #[case] values: Vec<u128>,
         #[values("0.0")] api_version: &str,
         #[values(0, 1, 2, 3, 4)] random_seed: u64,
@@ -726,6 +796,16 @@ mod tests {
             receipts.push(
                 Eip712SignedMessage::new(
                     &domain_separator,
+                    #[cfg(feature = "v2")]
+                    Receipt::new(
+                        allocation_ids[0],
+                        payer,
+                        data_service,
+                        service_provider,
+                        value,
+                    )
+                    .unwrap(),
+                    #[cfg(not(feature = "v2"))]
                     Receipt::new(allocation_ids[0], value).unwrap(),
                     &all_wallets.choose(&mut rng).unwrap().wallet,
                 )
@@ -734,12 +814,29 @@ mod tests {
         }
 
         // Create previous RAV from first half of receipts locally
-        let prev_rav = ReceiptAggregateVoucher::aggregate_receipts(
-            allocation_ids[0],
-            &receipts[0..receipts.len() / 2],
-            None,
-        )
-        .unwrap();
+        let prev_rav = {
+            #[cfg(feature = "v2")]
+            {
+                ReceiptAggregateVoucher::aggregate_receipts(
+                    allocation_ids[0],
+                    payer,
+                    data_service,
+                    service_provider,
+                    &receipts[0..receipts.len() / 2],
+                    None,
+                )
+                .unwrap()
+            }
+            #[cfg(not(feature = "v2"))]
+            {
+                ReceiptAggregateVoucher::aggregate_receipts(
+                    allocation_ids[0],
+                    &receipts[0..receipts.len() / 2],
+                    None,
+                )
+                .unwrap()
+            }
+        };
         let signed_prev_rav = Eip712SignedMessage::new(
             &domain_separator,
             prev_rav,
@@ -775,6 +872,9 @@ mod tests {
         http_response_size_limit: u32,
         http_max_concurrent_connections: u32,
         allocation_ids: Vec<Address>,
+        payer: Address,
+        data_service: Address,
+        service_provider: Address,
     ) {
         // The keys that will be used to sign the new RAVs
         let keys_main = keys();
@@ -801,6 +901,9 @@ mod tests {
         // Create receipts
         let receipts = vec![Eip712SignedMessage::new(
             &domain_separator,
+            #[cfg(feature = "v2")]
+            Receipt::new(allocation_ids[0], payer, data_service, service_provider, 42).unwrap(),
+            #[cfg(not(feature = "v2"))]
             Receipt::new(allocation_ids[0], 42).unwrap(),
             &keys_main.wallet,
         )
@@ -857,6 +960,9 @@ mod tests {
         http_response_size_limit: u32,
         http_max_concurrent_connections: u32,
         allocation_ids: Vec<Address>,
+        payer: Address,
+        data_service: Address,
+        service_provider: Address,
         #[values("0.0")] api_version: &str,
     ) {
         // The keys that will be used to sign the new RAVs
@@ -869,6 +975,10 @@ mod tests {
         // Number of receipts that is just above the number that would fit within the
         // request size limit. This value is hard-coded here because it supports the
         // maximum number of receipts per aggregate value we wrote in the spec / docs.
+        // Reduced for v2 receipts which are larger due to additional fields
+        #[cfg(feature = "v2")]
+        let number_of_receipts_to_exceed_limit = 200;
+        #[cfg(not(feature = "v2"))]
         let number_of_receipts_to_exceed_limit = 300;
 
         // Start the JSON-RPC server.
@@ -896,6 +1006,16 @@ mod tests {
             receipts.push(
                 Eip712SignedMessage::new(
                     &domain_separator,
+                    #[cfg(feature = "v2")]
+                    Receipt::new(
+                        allocation_ids[0],
+                        payer,
+                        data_service,
+                        service_provider,
+                        u128::MAX / 1000,
+                    )
+                    .unwrap(),
+                    #[cfg(not(feature = "v2"))]
                     Receipt::new(allocation_ids[0], u128::MAX / 1000).unwrap(),
                     &keys_main.wallet,
                 )

tap_aggregator/tests/aggregate_test.rs

@@ -10,6 +10,9 @@ use tap_aggregator::{
     server,
 };
 use tap_core::{signed_message::Eip712SignedMessage, tap_eip712_domain};
+#[cfg(feature = "v2")]
+use tap_graph::v2::{Receipt, ReceiptAggregateVoucher};
+#[cfg(not(feature = "v2"))]
 use tap_graph::{Receipt, ReceiptAggregateVoucher};
 use thegraph_core::alloy::{primitives::Address, signers::local::PrivateKeySigner};
 use tonic::codec::CompressionEncoding;
@@ -47,24 +50,45 @@ async fn aggregation_test() {
         .send_compressed(CompressionEncoding::Zstd);
 
     let allocation_id = Address::from_str("0xabababababababababababababababababababab").unwrap();
+    let payer = Address::from_str("0xabababababababababababababababababababab").unwrap();
+    let data_service = Address::from_str("0xdeaddeaddeaddeaddeaddeaddeaddeaddeaddead").unwrap();
+    let service_provider = Address::from_str("0xbeefbeefbeefbeefbeefbeefbeefbeefbeefbeef").unwrap();
 
-    // Create receipts
-    let mut receipts = Vec::new();
-    for value in 50..60 {
-        receipts.push(
-            Eip712SignedMessage::new(
-                &domain_separator,
-                Receipt::new(allocation_id, value).unwrap(),
-                &wallet,
-            )
-            .unwrap(),
-        );
+    // Use a fixed timestamp to ensure both v1 and v2 receipts have the same timestamps
+    let fixed_timestamp = 1700000000000000000u64; // Fixed timestamp in nanoseconds
+
+    // Create v1 receipts for gRPC v1 compatibility
+    let mut v1_receipts = Vec::new();
+    for (i, value) in (50..60).enumerate() {
+        let mut receipt = tap_graph::Receipt::new(allocation_id, value).unwrap();
+        receipt.timestamp_ns = fixed_timestamp + i as u64; // Ensure increasing timestamps
+        v1_receipts.push(Eip712SignedMessage::new(&domain_separator, receipt, &wallet).unwrap());
     }
 
-    let rav_request = RavRequest::new(receipts.clone(), None);
+    let rav_request = RavRequest::new(v1_receipts, None);
     let res = client.aggregate_receipts(rav_request).await.unwrap();
     let signed_rav: tap_graph::SignedRav = res.into_inner().signed_rav().unwrap();
 
+    // Create v2 receipts for JSON-RPC API with the same timestamps
+    let mut v2_receipts = Vec::new();
+    for (i, value) in (50..60).enumerate() {
+        #[cfg(feature = "v2")]
+        {
+            let mut receipt =
+                Receipt::new(allocation_id, payer, data_service, service_provider, value).unwrap();
+            receipt.timestamp_ns = fixed_timestamp + i as u64; // Same timestamps as v1
+            v2_receipts
+                .push(Eip712SignedMessage::new(&domain_separator, receipt, &wallet).unwrap());
+        }
+        #[cfg(not(feature = "v2"))]
+        {
+            let mut receipt = Receipt::new(allocation_id, value).unwrap();
+            receipt.timestamp_ns = fixed_timestamp + i as u64;
+            v2_receipts
+                .push(Eip712SignedMessage::new(&domain_separator, receipt, &wallet).unwrap());
+        }
+    }
+
     let sender_aggregator = HttpClientBuilder::default().build(&endpoint).unwrap();
 
     let previous_rav: Option<tap_graph::SignedRav> = None;
@@ -74,13 +98,22 @@ async fn aggregation_test() {
             "aggregate_receipts",
             rpc_params!(
                 "0.0", // TODO: Set the version in a smarter place.
-                receipts,
+                v2_receipts,
                 previous_rav
             ),
         )
         .await
         .unwrap();
     let response = response.data;
-    assert_eq!(signed_rav, response);
+    // Compare the core fields since the types might differ between v1 and v2
+    assert_eq!(
+        signed_rav.message.allocationId,
+        response.message.allocationId
+    );
+    assert_eq!(signed_rav.message.timestampNs, response.message.timestampNs);
+    assert_eq!(
+        signed_rav.message.valueAggregate,
+        response.message.valueAggregate
+    );
     join_handle.abort();
 }

tap_aggregator/tests/aggregate_v1_and_v2.rs

@@ -68,6 +68,9 @@ async fn aggregation_test() {
     let rav_request = ReqV1::new(receipts.clone(), None);
     let res = client.aggregate_receipts(rav_request).await;
 
+    if res.is_err() {
+        println!("V1 gRPC Error: {:?}", res.as_ref().err());
+    }
     assert!(res.is_ok());
 
     let mut client = ClientV2::connect(endpoint.clone())

tap_core/Cargo.toml

@@ -25,5 +25,6 @@ rstest.workspace = true
 serde_json.workspace = true
 
 [features]
-default = ["in_memory"]
+default = ["in_memory", "v2"]
 in_memory = ["dep:tap_graph"]
+v2 = []