[Horizon] Phase 4 gateway

Cargo.toml

@@ -18,6 +18,7 @@ axum = { version = "0.8.1", default-features = false, features = [
     "tokio",
     "http1",
 ] }
+base64 = "0.22.1"
 custom_debug = "0.6.1"
 faster-hex = "0.10.0"
 futures = "0.3"
@@ -46,8 +47,8 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0.116", features = ["raw_value"] }
 serde_with = "3.8.1"
 snmalloc-rs = "0.3"
-tap_graph = "0.3.3"
-thegraph-core = { version = "0.15", features = [
+tap_graph = { version = "0.3.4", features = ["v2"] }
+thegraph-core = { version = "0.15.1", features = [
     "alloy-contract",
     "alloy-signer-local",
     "attestation",

README.md

@@ -101,7 +101,7 @@ client query. Indexer fees are clamped to a maximum of the gateway's budget.
 
 For an overview of TAP see https://github.com/semiotic-ai/timeline-aggregation-protocol.
 
-The gateway acts as a TAP sender, where each indexer request is sent with a TAP receipt. The gateway
+This is a horizon-ready gateway that generates TAP v2 receipts exclusively. The gateway acts as a TAP sender, where each indexer request is sent with a TAP v2 receipt. The gateway
 operator is expected to run 2 additional services:
 
 - [tap-aggregator](https://github.com/semiotic-ai/timeline-aggregation-protocol/tree/main/tap_aggregator):
@@ -113,7 +113,7 @@ operator is expected to run 2 additional services:
 The gateway operator is also expected to manage at least 2 wallets:
 
 - sender: requires ETH for transaction gas and GRT to allocate into TAP escrow balances for paying indexers
-- authorized signer: used by the gateway and tap-aggregator to sign receipts and RAVs
+- authorized signer: used by the gateway and tap-aggregator to sign v2 receipts and RAVs
 
 ## operational notes
 

src/client_query.rs

@@ -285,7 +285,10 @@ async fn run_indexer_queries(
             let min_fee = *(min_fee.0 * grt_per_usd * one_grt) / selections.len() as f64;
             let indexer_fee = selection.fee.as_f64() * budget as f64;
             let fee = indexer_fee.max(min_fee) as u128;
-            let receipt = match ctx.receipt_signer.create_receipt(largest_allocation, fee) {
+            let receipt = match ctx
+                .receipt_signer
+                .create_receipt(largest_allocation, *indexer, fee)
+            {
                 Ok(receipt) => receipt,
                 Err(err) => {
                     tracing::error!(?indexer, %deployment, error=?err, "failed to create receipt");
@@ -682,7 +685,7 @@ pub async fn handle_indexer_query(
     let fee = *(ctx.budgeter.query_fees_target.0 * grt_per_usd * one_grt) as u128;
 
     let allocation = indexing.largest_allocation;
-    let receipt = match ctx.receipt_signer.create_receipt(allocation, fee) {
+    let receipt = match ctx.receipt_signer.create_receipt(allocation, *indexer, fee) {
         Ok(receipt) => receipt,
         Err(err) => {
             return Err(Error::Internal(anyhow!("failed to create receipt: {err}")));

src/config.rs

@@ -46,8 +46,11 @@ pub struct Config {
     /// Minimum indexer-service version that will receive queries
     #[serde_as(as = "DisplayFromStr")]
     pub min_indexer_version: Version,
-    /// Indexers used to query the network subgraph
+    /// Trusted indexers that can serve the network subgraph for free
     pub trusted_indexers: Vec<TrustedIndexer>,
+    /// Maximum acceptable lag (in seconds) for network subgraph responses (default: 120)
+    #[serde(default = "default_network_subgraph_max_lag_seconds")]
+    pub network_subgraph_max_lag_seconds: u64,
     /// Check payment state of client (disable for testnets)
     pub payment_required: bool,
     /// public API port
@@ -58,6 +61,13 @@ pub struct Config {
     #[serde(deserialize_with = "deserialize_not_nan_f64")]
     pub query_fees_target: NotNan<f64>,
     pub receipts: Receipts,
+    /// Address for the Subgraph Service
+    pub subgraph_service: Address,
+}
+
+/// Default network subgraph max lag threshold (120 seconds)
+fn default_network_subgraph_max_lag_seconds() -> u64 {
+    120
 }
 
 /// Deserialize a `NotNan<f64>` from a `f64` and return an error if the value is NaN.

src/indexer_client.rs

@@ -47,7 +47,7 @@ impl IndexerClient {
         query: &str,
     ) -> Result<IndexerResponse, IndexerError> {
         let (auth_key, auth_value) = match auth {
-            IndexerAuth::Paid(receipt, _) => ("Tap-Receipt", receipt.serialize()),
+            IndexerAuth::Paid(receipt, _) => ("tap-receipt", receipt.serialize()),
             IndexerAuth::Free(token) => (AUTHORIZATION.as_str(), format!("Bearer {token}")),
         };
 

src/main.rs

@@ -112,6 +112,7 @@ async fn main() {
         indexers: conf.trusted_indexers,
         latest_block: None,
         page_size: 500,
+        max_lag_seconds: conf.network_subgraph_max_lag_seconds,
     };
     let indexer_host_blocklist = match &conf.ip_blocker_db {
         Some(path) => {
@@ -136,6 +137,7 @@ async fn main() {
         receipt_signer,
         conf.receipts.chain_id,
         conf.receipts.verifier,
+        conf.subgraph_service,
     )));
 
     let auth_service = init_auth_service(

src/network/subgraph_client.rs

@@ -94,7 +94,7 @@ pub mod types {
 #[serde_as]
 #[derive(Clone, CustomDebug, Deserialize)]
 pub struct TrustedIndexer {
-    /// network subgraph endpoint
+    /// Complete network subgraph endpoint URL (e.g., http://indexer:7601/subgraphs/id/Qmc2Cb...)
     #[debug(with = std::fmt::Display::fmt)]
     #[serde_as(as = "serde_with::DisplayFromStr")]
     pub url: Url,
@@ -125,6 +125,7 @@ pub struct Client {
     pub indexers: Vec<TrustedIndexer>,
     pub page_size: usize,
     pub latest_block: Option<Block>,
+    pub max_lag_seconds: u64,
 }
 
 impl Client {
@@ -271,7 +272,8 @@ impl Client {
                     "response block before latest",
                 );
                 ensure!(
-                    (unix_timestamp() / 1_000).saturating_sub(block.timestamp) < 120,
+                    (unix_timestamp() / 1_000).saturating_sub(block.timestamp)
+                        < self.max_lag_seconds,
                     "response too far behind",
                 );
                 query_block = Some(block);

src/receipts.rs

@@ -1,74 +1,138 @@
 use std::time::SystemTime;
 
-use rand::RngCore;
-use tap_graph::{Receipt as TapReceipt, SignedReceipt};
+use base64::{Engine as _, prelude::BASE64_STANDARD};
+use prost::Message as _;
+use rand::RngCore as _;
 use thegraph_core::{
-    AllocationId,
+    AllocationId, CollectionId,
     alloy::{
         dyn_abi::Eip712Domain,
         primitives::{Address, U256},
         signers::local::PrivateKeySigner,
     },
 };
 
-pub struct Receipt(SignedReceipt);
+#[derive(Debug, Clone)]
+pub struct Receipt(pub tap_graph::v2::SignedReceipt);
 
 impl Receipt {
     pub fn value(&self) -> u128 {
         self.0.message.value
     }
 
     pub fn allocation(&self) -> Address {
-        self.0.message.allocation_id
+        // TAP v2 receipts use collection ids which are 32 bytes.
+        // For the Subgraph Service these are 20 byte allocation ids with zero padding.
+        CollectionId::from(self.0.message.collection_id).as_address()
     }
 
     pub fn serialize(&self) -> String {
-        serde_json::to_string(&self.0).unwrap()
+        #[derive(prost::Message)]
+        struct ReceiptMessage {
+            #[prost(bytes, tag = "1")]
+            collection_id: Vec<u8>,
+            #[prost(bytes, tag = "2")]
+            payer: Vec<u8>,
+            #[prost(bytes, tag = "3")]
+            data_service: Vec<u8>,
+            #[prost(bytes, tag = "4")]
+            service_provider: Vec<u8>,
+            #[prost(uint64, tag = "5")]
+            timestamp_ns: u64,
+            #[prost(uint64, tag = "6")]
+            nonce: u64,
+            #[prost(message, optional, tag = "7")]
+            value: Option<Uint128>,
+        }
+        #[derive(prost::Message)]
+        struct Uint128 {
+            #[prost(uint64, tag = "1")]
+            high: u64,
+            #[prost(uint64, tag = "2")]
+            low: u64,
+        }
+        #[derive(prost::Message)]
+        struct SignedReceipt {
+            #[prost(message, optional, tag = "1")]
+            message: Option<ReceiptMessage>,
+            #[prost(bytes, tag = "2")]
+            signature: Vec<u8>,
+        }
+        let receipt_message = ReceiptMessage {
+            collection_id: self.0.message.collection_id.to_vec(),
+            payer: self.0.message.payer.to_vec(),
+            data_service: self.0.message.data_service.to_vec(),
+            service_provider: self.0.message.service_provider.to_vec(),
+            timestamp_ns: self.0.message.timestamp_ns,
+            nonce: self.0.message.nonce,
+            value: Some(Uint128 {
+                high: (self.0.message.value >> 64) as u64,
+                low: self.0.message.value as u64,
+            }),
+        };
+        let signature_bytes = self.0.signature.as_bytes().to_vec();
+
+        let signed_receipt = SignedReceipt {
+            message: Some(receipt_message),
+            signature: signature_bytes,
+        };
+
+        BASE64_STANDARD.encode(signed_receipt.encode_to_vec())
     }
 }
 
 pub struct ReceiptSigner {
     signer: PrivateKeySigner,
     domain: Eip712Domain,
+    data_service: Address,
 }
 
 impl ReceiptSigner {
-    pub fn new(signer: PrivateKeySigner, chain_id: U256, verifying_contract: Address) -> Self {
+    pub fn new(
+        signer: PrivateKeySigner,
+        chain_id: U256,
+        verifying_contract: Address,
+        data_service: Address,
+    ) -> Self {
+        let domain = Eip712Domain {
+            name: Some("GraphTallyCollector".into()),
+            version: Some("1".into()),
+            chain_id: Some(chain_id),
+            verifying_contract: Some(verifying_contract),
+            salt: None,
+        };
         Self {
             signer,
-            domain: Eip712Domain {
-                name: Some("TAP".into()),
-                version: Some("1".into()),
-                chain_id: Some(chain_id),
-                verifying_contract: Some(verifying_contract),
-                salt: None,
-            },
+            domain,
+            data_service,
         }
     }
 
-    pub fn create_receipt(&self, allocation: AllocationId, fee: u128) -> anyhow::Result<Receipt> {
-        // Nonce generated with CSPRNG (ChaCha12), to avoid collision with receipts generated by
-        // other gateway processes.
-        // See https://docs.rs/rand/latest/rand/rngs/index.html#our-generators.
+    pub fn create_receipt(
+        &self,
+        allocation: AllocationId,
+        indexer: Address,
+        fee: u128,
+    ) -> anyhow::Result<Receipt> {
         let nonce = rand::rng().next_u64();
-
         let timestamp_ns = SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)
             .unwrap()
             .as_nanos()
             .try_into()
             .map_err(|_| anyhow::anyhow!("failed to convert timestamp to ns"))?;
-
-        let receipt = TapReceipt {
-            allocation_id: allocation.0.0.into(),
+        let receipt = tap_graph::v2::Receipt {
+            collection_id: CollectionId::from(allocation).0.into(),
+            payer: self.signer.address(),
+            data_service: self.data_service,
+            service_provider: indexer,
             timestamp_ns,
             nonce,
             value: fee,
         };
-        let signed = SignedReceipt::new(&self.domain, receipt, &self.signer)
-            .map_err(|e| anyhow::anyhow!("failed to sign receipt: {:?}", e))?;
-
-        Ok(Receipt(signed))
+        tap_graph::v2::SignedReceipt::new(&self.domain, receipt, &self.signer)
+            .map(Receipt)
+            .map_err(|e| anyhow::anyhow!("failed to sign v2 receipt: {:?}", e))
     }
 }
 
@@ -81,25 +145,50 @@ mod tests {
 
     use super::*;
 
-    #[test]
-    fn create_receipt() {
-        //* Given
+    fn create_test_signer() -> ReceiptSigner {
         let secret_key = PrivateKeySigner::from_slice(&[0xcd; 32]).expect("invalid secret key");
-        let signer = ReceiptSigner::new(
+        ReceiptSigner::new(
             secret_key,
             1.try_into().expect("invalid chain id"),
             address!("177b557b12f22bb17a9d73dcc994d978dd6f5f89"),
-        );
+            address!("2222222222222222222222222222222222222222"),
+        )
+    }
 
+    #[test]
+    fn create_v2_receipt() {
+        let signer = create_test_signer();
         let allocation = allocation_id!("89b23fea4e46d40e8a4c6cca723e2a03fdd4bec2");
         let fee = 1000;
 
-        //* When
-        let res = signer.create_receipt(allocation, fee);
+        let receipt = signer
+            .create_receipt(
+                allocation,
+                address!("3333333333333333333333333333333333333333"),
+                fee,
+            )
+            .expect("failed to create v2 receipt");
+
+        assert_eq!(receipt.value(), fee);
+        assert_eq!(AllocationId::from(receipt.allocation()), allocation);
+    }
+
+    #[test]
+    fn test_receipt_serialization() {
+        let signer = create_test_signer();
+        let allocation = allocation_id!("89b23fea4e46d40e8a4c6cca723e2a03fdd4bec2");
+        let fee = 1000;
 
-        //* Then
-        let receipt = res.expect("failed to create tap receipt");
+        let receipt = signer
+            .create_receipt(
+                allocation,
+                address!("3333333333333333333333333333333333333333"),
+                fee,
+            )
+            .expect("failed to create v2 receipt");
 
+        let serialized = receipt.serialize();
+        assert!(!serialized.is_empty());
         assert_eq!(receipt.value(), fee);
     }
 }