feat(tap-aggregator): add v2 endpoint

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

Author: suchapalaver

Cargo.toml

@@ -50,7 +50,7 @@ tap_eip712_message = { version = "0.2.1", path = "tap_eip712_message" }
 tap_core = { version = "4.1.2", path = "tap_core" }
 tap_graph = { version = "0.3.2", path = "tap_graph", features = ["v2"] }
 tap_receipt = { version = "1.1.2", path = "tap_receipt" }
-thegraph-core = "0.15.0"
+thegraph-core = "0.15.1"
 thiserror = "2.0.12"
 tokio = { version = "1.44.2", features = ["macros", "signal"] }
 tonic = { version = "0.13.0", features = ["transport", "zstd"] }

tap_aggregator/Cargo.toml

@@ -7,6 +7,10 @@ repository.workspace = true
 readme = "README.md"
 description = "A JSON-RPC service for the Timeline Aggregation Protocol that lets clients request an aggregate receipt from a list of individual receipts."
 
+[features]
+default = ["v2"]
+v2 = ["tap_graph/v2"]
+
 [[bin]]
 name = "tap_aggregator"
 path = "src/main.rs"

tap_aggregator/src/lib.rs

@@ -7,4 +7,6 @@ pub mod error_codes;
 pub mod grpc;
 pub mod jsonrpsee_helpers;
 pub mod metrics;
+pub mod protocol_mode;
+pub mod receipt_classifier;
 pub mod server;

tap_aggregator/src/protocol_mode.rs

@@ -0,0 +1,29 @@
+// Copyright 2024 The Graph Foundation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum ProtocolMode {
+    /// Pre-horizon: V1 receipts accepted, V1 aggregation used
+    Legacy,
+    /// Post-horizon: V2 for new receipts, V1 only for legacy receipt aggregation
+    Horizon,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum ReceiptType {
+    /// V1 receipt created before horizon activation (legacy, still needs aggregation)
+    LegacyV1,
+    /// V2 receipt (collection-based, created after horizon activation)
+    V2,
+}

tap_aggregator/src/receipt_classifier.rs

@@ -0,0 +1,88 @@
+// Copyright 2024 The Graph Foundation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use anyhow::Result;
+
+use crate::protocol_mode::ProtocolMode;
+
+/// Validate that a batch of v1 receipts is valid for legacy processing
+pub fn validate_v1_receipt_batch<T>(receipts: &[T]) -> Result<ProtocolMode> {
+    if receipts.is_empty() {
+        return Err(anyhow::anyhow!("Cannot aggregate empty receipt batch"));
+    }
+    // All v1 receipts use legacy mode
+    Ok(ProtocolMode::Legacy)
+}
+
+/// Validate that a batch of v2 receipts is valid for horizon processing
+#[cfg(feature = "v2")]
+pub fn validate_v2_receipt_batch<T>(receipts: &[T]) -> Result<ProtocolMode> {
+    if receipts.is_empty() {
+        return Err(anyhow::anyhow!("Cannot aggregate empty receipt batch"));
+    }
+    // All v2 receipts use horizon mode
+    Ok(ProtocolMode::Horizon)
+}
+
+#[cfg(test)]
+mod tests {
+    use tap_graph::Receipt;
+
+    use super::*;
+
+    #[test]
+    fn test_validate_v1_batch() {
+        use thegraph_core::alloy::primitives::Address;
+        let receipt = Receipt::new(Address::ZERO, 100).unwrap();
+        let receipts = vec![receipt];
+        assert_eq!(
+            validate_v1_receipt_batch(&receipts).unwrap(),
+            ProtocolMode::Legacy
+        );
+    }
+
+    #[test]
+    fn test_validate_v1_empty_batch_fails() {
+        let receipts: Vec<Receipt> = vec![];
+        assert!(validate_v1_receipt_batch(&receipts).is_err());
+    }
+
+    #[cfg(feature = "v2")]
+    #[test]
+    fn test_validate_v2_batch() {
+        use tap_graph::v2;
+        use thegraph_core::alloy::primitives::{Address, FixedBytes};
+        let receipt = v2::Receipt::new(
+            FixedBytes::ZERO,
+            Address::ZERO,
+            Address::ZERO,
+            Address::ZERO,
+            100,
+        )
+        .unwrap();
+        let receipts = vec![receipt];
+        assert_eq!(
+            validate_v2_receipt_batch(&receipts).unwrap(),
+            ProtocolMode::Horizon
+        );
+    }
+
+    #[cfg(feature = "v2")]
+    #[test]
+    fn test_validate_v2_empty_batch_fails() {
+        use tap_graph::v2;
+        let receipts: Vec<v2::Receipt> = vec![];
+        assert!(validate_v2_receipt_batch(&receipts).is_err());
+    }
+}

tap_aggregator/src/server.rs

@@ -94,6 +94,17 @@ pub trait Rpc {
         receipts: Vec<Eip712SignedMessage<Receipt>>,
         previous_rav: Option<Eip712SignedMessage<ReceiptAggregateVoucher>>,
     ) -> JsonRpcResult<Eip712SignedMessage<ReceiptAggregateVoucher>>;
+
+    /// Aggregates the given v2 receipts into a v2 receipt aggregate voucher.
+    /// Uses the Horizon protocol for collection-based aggregation.
+    #[cfg(feature = "v2")]
+    #[method(name = "aggregate_receipts_v2")]
+    fn aggregate_receipts_v2(
+        &self,
+        api_version: String,
+        receipts: Vec<Eip712SignedMessage<tap_graph::v2::Receipt>>,
+        previous_rav: Option<Eip712SignedMessage<tap_graph::v2::ReceiptAggregateVoucher>>,
+    ) -> JsonRpcResult<Eip712SignedMessage<tap_graph::v2::ReceiptAggregateVoucher>>;
 }
 
 #[derive(Clone)]
@@ -141,6 +152,8 @@ fn aggregate_receipts_(
     receipts: Vec<Eip712SignedMessage<Receipt>>,
     previous_rav: Option<Eip712SignedMessage<ReceiptAggregateVoucher>>,
 ) -> JsonRpcResult<Eip712SignedMessage<ReceiptAggregateVoucher>> {
+    use crate::receipt_classifier::validate_v1_receipt_batch;
+
     // Return an error if the API version is not supported.
     let api_version = match parse_api_version(api_version.as_str()) {
         Ok(v) => v,
@@ -157,16 +170,85 @@ fn aggregate_receipts_(
         DEPRECATION_WARNING_COUNT.inc();
     }
 
-    let res = match api_version {
-        TapRpcApiVersion::V0_0 => aggregator::v1::check_and_aggregate_receipts(
-            domain_separator,
-            &receipts,
-            previous_rav,
-            wallet,
-            accepted_addresses,
-        ),
+    // This endpoint handles v1 receipts for legacy aggregation
+    // V2 receipts are handled through the aggregate_receipts_v2 endpoint
+    if let Err(e) = validate_v1_receipt_batch(&receipts) {
+        return Err(jsonrpsee::types::ErrorObject::owned(
+            JsonRpcErrorCode::Aggregation as i32,
+            e.to_string(),
+            None::<()>,
+        ));
+    }
+
+    log::debug!("Processing V1 receipts");
+
+    // Execute v1 aggregation
+    let res = aggregator::v1::check_and_aggregate_receipts(
+        domain_separator,
+        &receipts,
+        previous_rav,
+        wallet,
+        accepted_addresses,
+    );
+
+    // Handle aggregation error
+    match res {
+        Ok(res) => Ok(JsonRpcResponse::warn(res, warnings)),
+        Err(e) => Err(jsonrpsee::types::ErrorObject::owned(
+            JsonRpcErrorCode::Aggregation as i32,
+            e.to_string(),
+            None::<()>,
+        )),
+    }
+}
+
+#[cfg(feature = "v2")]
+fn aggregate_receipts_v2_(
+    api_version: String,
+    wallet: &PrivateKeySigner,
+    accepted_addresses: &HashSet<Address>,
+    domain_separator: &Eip712Domain,
+    receipts: Vec<Eip712SignedMessage<tap_graph::v2::Receipt>>,
+    previous_rav: Option<Eip712SignedMessage<tap_graph::v2::ReceiptAggregateVoucher>>,
+) -> JsonRpcResult<Eip712SignedMessage<tap_graph::v2::ReceiptAggregateVoucher>> {
+    use crate::receipt_classifier::validate_v2_receipt_batch;
+
+    // Return an error if the API version is not supported.
+    let api_version = match parse_api_version(api_version.as_str()) {
+        Ok(v) => v,
+        Err(e) => {
+            VERSION_ERROR_COUNT.inc();
+            return Err(e);
+        }
     };
 
+    // Add a warning if the API version is to be deprecated.
+    let mut warnings: Vec<JsonRpcWarning> = Vec::new();
+    if let Some(w) = check_api_version_deprecation(&api_version) {
+        warnings.push(w);
+        DEPRECATION_WARNING_COUNT.inc();
+    }
+
+    // Validate v2 receipt batch for horizon processing
+    if let Err(e) = validate_v2_receipt_batch(&receipts) {
+        return Err(jsonrpsee::types::ErrorObject::owned(
+            JsonRpcErrorCode::Aggregation as i32,
+            e.to_string(),
+            None::<()>,
+        ));
+    }
+
+    log::debug!("Processing V2 receipts with Horizon protocol");
+
+    // Execute v2 aggregation
+    let res = aggregator::v2::check_and_aggregate_receipts(
+        domain_separator,
+        &receipts,
+        previous_rav,
+        wallet,
+        accepted_addresses,
+    );
+
     // Handle aggregation error
     match res {
         Ok(res) => Ok(JsonRpcResponse::warn(res, warnings)),
@@ -337,6 +419,47 @@ impl RpcServer for RpcImpl {
             }
         }
     }
+
+    #[cfg(feature = "v2")]
+    fn aggregate_receipts_v2(
+        &self,
+        api_version: String,
+        receipts: Vec<Eip712SignedMessage<tap_graph::v2::Receipt>>,
+        previous_rav: Option<Eip712SignedMessage<tap_graph::v2::ReceiptAggregateVoucher>>,
+    ) -> JsonRpcResult<Eip712SignedMessage<tap_graph::v2::ReceiptAggregateVoucher>> {
+        // Values for Prometheus metrics
+        let receipts_grt: u128 = receipts.iter().map(|r| r.message.value).sum();
+        let receipts_count: u64 = receipts.len() as u64;
+
+        match aggregate_receipts_v2_(
+            api_version,
+            &self.wallet,
+            &self.accepted_addresses,
+            &self.domain_separator,
+            receipts,
+            previous_rav,
+        ) {
+            Ok(res) => {
+                TOTAL_GRT_AGGREGATED.inc_by(receipts_grt as f64);
+                TOTAL_AGGREGATED_RECEIPTS.inc_by(receipts_count);
+                AGGREGATION_SUCCESS_COUNTER.inc();
+                if let Some(kafka) = &self.kafka {
+                    // V2 RAVs use collectionId instead of allocationId
+                    produce_kafka_records(
+                        kafka,
+                        &self.wallet.address(),
+                        &res.data.message.collectionId,
+                        res.data.message.valueAggregate,
+                    );
+                }
+                Ok(res)
+            }
+            Err(e) => {
+                AGGREGATION_FAILURE_COUNTER.inc();
+                Err(e)
+            }
+        }
+    }
 }
 
 #[allow(clippy::too_many_arguments)]