[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,12 @@ 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,
+                selection.data.largest_allocation_is_legacy,
+            ) {
                 Ok(receipt) => receipt,
                 Err(err) => {
                     tracing::error!(?indexer, %deployment, error=?err, "failed to create receipt");
@@ -303,7 +308,11 @@ async fn run_indexer_queries(
                     let start_time = Instant::now();
                     // URL checked: ref df8e647b-1e6e-422a-8846-dc9ee7e0dcc2
                     let deployment_url = url.join(&format!("subgraphs/id/{deployment}")).unwrap();
-                    let auth = IndexerAuth::Paid(&receipt, ctx.attestation_domain);
+                    let auth = IndexerAuth::Paid(
+                        &receipt,
+                        ctx.attestation_domain,
+                        ctx.legacy_attestation_domain,
+                    );
                     let result = indexer_client
                         .query_indexer(deployment_url, auth, &indexer_query)
                         .in_current_span()
@@ -464,6 +473,7 @@ struct CandidateMetadata {
     #[debug(with = std::fmt::Display::fmt)]
     url: Url,
     largest_allocation: AllocationId,
+    largest_allocation_is_legacy: bool,
 }
 
 /// Given a list of indexings, build a list of candidates that are within the required block range
@@ -572,6 +582,7 @@ fn build_candidates_list(
                 deployment,
                 url: indexing.indexer.url.clone(),
                 largest_allocation: indexing.largest_allocation,
+                largest_allocation_is_legacy: indexing.largest_allocation_is_legacy,
             },
             perf: perf.response,
             fee: Normalized::new(indexing.fee as f64 / budget as f64).unwrap_or(Normalized::ONE),
@@ -682,7 +693,12 @@ 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,
+        indexing.largest_allocation_is_legacy,
+    ) {
         Ok(receipt) => receipt,
         Err(err) => {
             return Err(Error::Internal(anyhow!("failed to create receipt: {err}")));
@@ -695,7 +711,11 @@ pub async fn handle_indexer_query(
         .url
         .join(&format!("subgraphs/id/{deployment}"))
         .unwrap();
-    let indexer_auth = IndexerAuth::Paid(&receipt, ctx.attestation_domain);
+    let indexer_auth = IndexerAuth::Paid(
+        &receipt,
+        ctx.attestation_domain,
+        ctx.legacy_attestation_domain,
+    );
 
     let indexer_start_time = Instant::now();
     let result = ctx

src/client_query/context.rs

@@ -18,5 +18,6 @@ pub struct Context {
     pub network: NetworkService,
     pub indexing_perf: IndexingPerformance,
     pub attestation_domain: &'static Eip712Domain,
+    pub legacy_attestation_domain: &'static Eip712Domain,
     pub reporter: mpsc::UnboundedSender<reports::ClientRequest>,
 }

src/config.rs

@@ -46,7 +46,7 @@ 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")]
@@ -61,6 +61,8 @@ 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)
@@ -128,6 +130,7 @@ pub enum BlocklistEntry {
 pub struct AttestationConfig {
     pub chain_id: String,
     pub dispute_manager: Address,
+    pub legacy_dispute_manager: Address,
 }
 
 /// The exchange rate provider.
@@ -183,10 +186,14 @@ impl From<KafkaConfig> for rdkafka::config::ClientConfig {
 pub struct Receipts {
     /// TAP verifier contract chain
     pub chain_id: U256,
+    /// TAP payer address
+    pub payer: Address,
     /// TAP signer key
     pub signer: B256,
-    /// TAP verifier contract address
+    /// TAP verifier contract address (v2)
     pub verifier: Address,
+    /// Legacy TAP verifier contract address (v1)
+    pub legacy_verifier: Address,
 }
 
 /// Load the configuration from a JSON file.

src/indexer_client.rs

@@ -35,7 +35,7 @@ pub struct IndexerClient {
 }
 
 pub enum IndexerAuth<'a> {
-    Paid(&'a Receipt, &'a Eip712Domain),
+    Paid(&'a Receipt, &'a Eip712Domain, &'a Eip712Domain),
     Free(&'a str),
 }
 
@@ -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}")),
         };
 
@@ -113,18 +113,26 @@ impl IndexerClient {
             return Err(BadResponse(format!("unattestable response: {error}")));
         }
 
-        if let IndexerAuth::Paid(receipt, attestation_domain) = auth {
+        if let IndexerAuth::Paid(receipt, attestation_domain, legacy_attestation_domain) = auth {
             match &payload.attestation {
                 Some(attestation) => {
                     let allocation = receipt.allocation();
-                    if let Err(err) = attestation::verify(
+                    if let Err(legacy_err) = attestation::verify(
+                        legacy_attestation_domain,
+                        attestation,
+                        &allocation,
+                        query,
+                        &original_response,
+                    ) && let Err(err) = attestation::verify(
                         attestation_domain,
                         attestation,
                         &allocation,
                         query,
                         &original_response,
                     ) {
-                        return Err(BadResponse(format!("bad attestation: {err}")));
+                        return Err(BadResponse(format!(
+                            "bad attestation: {legacy_err} - {err}"
+                        )));
                     }
                 }
                 None => {

src/main.rs

@@ -104,6 +104,15 @@ async fn main() {
             conf.attestations.dispute_manager,
         )));
 
+    let legacy_attestation_domain: &'static Eip712Domain =
+        Box::leak(Box::new(attestation::eip712_domain(
+            conf.attestations
+                .chain_id
+                .parse::<ChainId>()
+                .expect("failed to parse attestation domain chain_id"),
+            conf.attestations.legacy_dispute_manager,
+        )));
+
     let indexer_client = IndexerClient {
         client: http_client.clone(),
     };
@@ -134,9 +143,12 @@ async fn main() {
     network.wait_until_ready().await;
 
     let receipt_signer: &'static ReceiptSigner = Box::leak(Box::new(ReceiptSigner::new(
+        conf.receipts.payer,
         receipt_signer,
         conf.receipts.chain_id,
         conf.receipts.verifier,
+        conf.receipts.legacy_verifier,
+        conf.subgraph_service,
     )));
 
     let auth_service = init_auth_service(
@@ -171,6 +183,7 @@ async fn main() {
         indexing_perf,
         network,
         attestation_domain,
+        legacy_attestation_domain,
         reporter,
     };
 

src/network/indexer_processing.rs

@@ -54,6 +54,8 @@ pub struct IndexerInfo<I> {
 pub struct IndexingRawInfo {
     /// The largest allocation.
     pub largest_allocation: AllocationId,
+    /// Whether the largest allocation is a legacy allocation
+    pub largest_allocation_is_legacy: bool,
 }
 
 /// Internal representation of the fetched indexer's indexing information.
@@ -65,6 +67,9 @@ pub struct IndexingInfo<P, C> {
     /// The largest allocation.
     pub largest_allocation: AllocationId,
 
+    /// Whether the largest allocation is a legacy allocation
+    pub largest_allocation_is_legacy: bool,
+
     /// The indexing progress information
     ///
     /// See [`IndexingProgress`] for more information.
@@ -78,6 +83,7 @@ impl From<IndexingRawInfo> for IndexingInfo<(), ()> {
     fn from(raw: IndexingRawInfo) -> Self {
         Self {
             largest_allocation: raw.largest_allocation,
+            largest_allocation_is_legacy: raw.largest_allocation_is_legacy,
             progress: (),
             fee: (),
         }
@@ -93,6 +99,7 @@ impl IndexingInfo<(), ()> {
     ) -> IndexingInfo<IndexingProgress, ()> {
         IndexingInfo {
             largest_allocation: self.largest_allocation,
+            largest_allocation_is_legacy: self.largest_allocation_is_legacy,
             progress,
             fee: self.fee,
         }
@@ -105,6 +112,7 @@ impl IndexingInfo<IndexingProgress, ()> {
     fn with_fee(self, fee: u128) -> IndexingInfo<IndexingProgress, u128> {
         IndexingInfo {
             largest_allocation: self.largest_allocation,
+            largest_allocation_is_legacy: self.largest_allocation_is_legacy,
             progress: self.progress,
             fee,
         }

src/network/pre_processing.rs

@@ -18,7 +18,7 @@ pub fn into_internal_indexers_raw_info<'a>(
 ) -> HashMap<IndexerId, IndexerRawInfo> {
     let mut indexer_indexing_largest_allocation: HashMap<
         (IndexerId, DeploymentId),
-        (AllocationId, u128),
+        (AllocationId, u128, bool),
     > = HashMap::new();
 
     data.flat_map(|subgraph| {
@@ -53,32 +53,42 @@ pub fn into_internal_indexers_raw_info<'a>(
             };
 
             // Update the indexer's indexings largest allocations table
-            let indexing_largest_allocation = match indexer_indexing_largest_allocation
-                .entry((indexer_id, deployment_id))
-            {
-                Entry::Vacant(entry) => {
-                    entry.insert((allocation.id, allocation.allocated_tokens));
-                    allocation.id
-                }
-                Entry::Occupied(entry) => {
-                    let (largest_allocation_address, largest_allocation_amount) = entry.into_mut();
-                    if allocation.allocated_tokens > *largest_allocation_amount {
-                        *largest_allocation_address = allocation.id;
-                        *largest_allocation_amount = allocation.allocated_tokens;
+            let (indexing_largest_allocation, indexing_largest_allocation_is_legacy) =
+                match indexer_indexing_largest_allocation.entry((indexer_id, deployment_id)) {
+                    Entry::Vacant(entry) => {
+                        entry.insert((
+                            allocation.id,
+                            allocation.allocated_tokens,
+                            allocation.is_legacy,
+                        ));
+                        (allocation.id, allocation.is_legacy)
                     }
-                    *largest_allocation_address
-                }
-            };
+                    Entry::Occupied(entry) => {
+                        let (
+                            largest_allocation_address,
+                            largest_allocation_amount,
+                            largest_allocation_is_legacy,
+                        ) = entry.into_mut();
+                        if allocation.allocated_tokens > *largest_allocation_amount {
+                            *largest_allocation_address = allocation.id;
+                            *largest_allocation_amount = allocation.allocated_tokens;
+                            *largest_allocation_is_legacy = allocation.is_legacy;
+                        }
+                        (*largest_allocation_address, *largest_allocation_is_legacy)
+                    }
+                };
 
             // Update the indexer's indexings info
             let indexing = indexer
                 .indexings
                 .entry(deployment_id)
                 .or_insert(IndexingRawInfo {
                     largest_allocation: allocation.id,
+                    largest_allocation_is_legacy: allocation.is_legacy,
                 });
 
             indexing.largest_allocation = indexing_largest_allocation;
+            indexing.largest_allocation_is_legacy = indexing_largest_allocation_is_legacy;
         }
 
         acc

src/network/snapshot.rs

@@ -36,6 +36,8 @@ pub struct Indexing {
     /// This is, among all allocations associated with the indexer and deployment, the address
     /// with the largest amount of allocated tokens.
     pub largest_allocation: AllocationId,
+    /// Whether the largest allocation is a legacy allocation
+    pub largest_allocation_is_legacy: bool,
     /// The indexer
     pub indexer: Arc<Indexer>,
     /// The indexing progress.
@@ -365,6 +367,7 @@ fn construct_indexings_table_row(
     let indexing = Indexing {
         id: indexing_id,
         largest_allocation: indexing_largest_allocation_addr,
+        largest_allocation_is_legacy: indexing_info.largest_allocation_is_legacy,
         indexer: Arc::clone(indexer),
         progress: IndexingProgress {
             latest_block: indexing_progress.latest_block,

src/network/subgraph_client.rs

@@ -77,6 +77,7 @@ pub mod types {
         pub id: AllocationId,
         #[serde_as(as = "serde_with::DisplayFromStr")]
         pub allocated_tokens: u128,
+        pub is_legacy: bool,
         pub indexer: Indexer,
     }
 
@@ -94,7 +95,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,
@@ -180,6 +181,7 @@ impl Client {
                             ) {
                                 id
                                 allocatedTokens
+                                isLegacy
                                 indexer {
                                     id
                                     url