fix: legacy receipt release

Author: Theodus

graph-gateway/src/client_query.rs

@@ -26,6 +26,7 @@ use gateway_framework::{
     indexing::Indexing,
     metrics::{with_metric, METRICS},
     network::{discovery::Status, indexing_performance::Snapshot},
+    scalar::ReceiptStatus,
     topology::network::{Deployment, GraphNetwork, Manifest, Subgraph},
 };
 use headers::ContentType;
@@ -413,14 +414,12 @@ async fn run_indexer_queries(
                     indexer,
                     deployment,
                     url: url.to_string(),
-                    allocation: receipt.allocation(),
+                    receipt,
                     subgraph_chain,
                     result,
                     response_time_ms,
                     seconds_behind,
                     blocks_behind,
-                    legacy_scalar,
-                    fee,
                     request: indexer_query,
                 };
                 tx.try_send(report).unwrap();
@@ -438,6 +437,21 @@ async fn run_indexer_queries(
                     indexer_errors.insert(report.indexer, err.clone());
                 }
             }
+
+            let receipt_status = match &report.result {
+                Ok(_) => ReceiptStatus::Success,
+                Err(IndexerError::Timeout) => ReceiptStatus::Unknown,
+                Err(_) => ReceiptStatus::Failure,
+            };
+            ctx.receipt_signer
+                .record_receipt(
+                    report.indexer,
+                    report.deployment,
+                    &report.receipt,
+                    receipt_status,
+                )
+                .await;
+
             indexer_requests.push(report);
         }
 
@@ -465,7 +479,10 @@ async fn run_indexer_queries(
         Err(Error::BadIndexers(IndexerErrors(indexer_errors)))
     };
 
-    let total_fees_grt: f64 = indexer_requests.iter().map(|i| i.fee as f64 * 1e-18).sum();
+    let total_fees_grt: f64 = indexer_requests
+        .iter()
+        .map(|i| i.receipt.grt_value() as f64 * 1e-18)
+        .sum();
     let total_fees_usd = USD(NotNan::new(total_fees_grt / *grt_per_usd).unwrap());
     let _ = ctx.budgeter.feedback.send(total_fees_usd);
 
@@ -505,12 +522,12 @@ async fn run_indexer_queries(
         tracing::info!(
             indexer = ?indexer_request.indexer,
             deployment = %indexer_request.deployment,
-            allocation = ?indexer_request.allocation,
+            allocation = ?indexer_request.receipt.allocation(),
             url = indexer_request.url,
             result = ?indexer_request.result.as_ref().map(|_| ()),
             response_time_ms = indexer_request.response_time_ms,
             seconds_behind = indexer_request.seconds_behind,
-            fee = indexer_request.fee as f64 * 1e-18,
+            fee = indexer_request.receipt.grt_value() as f64 * 1e-18,
             "indexer_request"
         );
         tracing::trace!(indexer_request = indexer_request.request);

graph-gateway/src/reports.rs

@@ -1,7 +1,10 @@
 use alloy_primitives::Address;
 use anyhow::{anyhow, Context};
 use gateway_common::utils::timestamp::unix_timestamp;
-use gateway_framework::errors::{self, IndexerError};
+use gateway_framework::{
+    errors::{self, IndexerError},
+    scalar::ScalarReceipt,
+};
 use ordered_float::NotNan;
 use prost::Message;
 use serde_json::json;
@@ -25,14 +28,12 @@ pub struct IndexerRequest {
     pub indexer: Address,
     pub url: String,
     pub deployment: DeploymentId,
-    pub allocation: Address,
+    pub receipt: ScalarReceipt,
     pub subgraph_chain: String,
     pub result: Result<IndexerResponse, errors::IndexerError>,
     pub response_time_ms: u16,
     pub seconds_behind: u32,
     pub blocks_behind: u64, // TODO: rm
-    pub legacy_scalar: bool,
-    pub fee: u128,
     pub request: String,
 }
 
@@ -87,7 +88,7 @@ impl Reporter {
         let total_fees_grt: f64 = client_request
             .indexer_requests
             .iter()
-            .map(|i| i.fee as f64 * 1e-18)
+            .map(|i| i.receipt.grt_value() as f64 * 1e-18)
             .sum();
         let total_fees_usd: f64 = total_fees_grt / *client_request.grt_per_usd;
 
@@ -198,7 +199,7 @@ impl Reporter {
                         "api_key": &client_request.api_key,
                         "fee": total_fees_grt as f32,
                         "response_time_ms": indexer_request.response_time_ms,
-                        "allocation": &indexer_request.allocation,
+                        "allocation": &indexer_request.receipt.allocation(),
                         "indexer_errors": indexer_errors,
                         "status": indexer_request.result.as_ref().map(|_| "200 OK".into()).unwrap_or_else(|err| err.to_string()),
                         "status_code": legacy_status_code,
@@ -218,12 +219,12 @@ impl Reporter {
                 "network": &indexer_request.subgraph_chain,
                 "indexer": &indexer_request.indexer,
                 "url": &indexer_request.url,
-                "fee": (indexer_request.fee as f64 * 1e-18) as f32,
-                "legacy_scalar": indexer_request.legacy_scalar,
+                "fee": (indexer_request.receipt.grt_value() as f64 * 1e-18) as f32,
+                "legacy_scalar": matches!(&indexer_request.receipt, ScalarReceipt::Legacy(_, _)),
                 "utility": 1.0,
                 "blocks_behind": indexer_request.blocks_behind,
                 "response_time_ms": indexer_request.response_time_ms,
-                "allocation": &indexer_request.allocation,
+                "allocation": &indexer_request.receipt.allocation(),
                 "indexer_errors": indexer_errors,
                 "status": indexer_request.result.as_ref().map(|_| "200 OK".into()).unwrap_or_else(|err| err.to_string()),
                 "status_code": legacy_status_code,
@@ -250,7 +251,7 @@ impl Reporter {
                 AttestationProtobuf {
                     request: Some(indexer_request.request).filter(|r| r.len() <= MAX_PAYLOAD_BYTES),
                     response: Some(original_response).filter(|r| r.len() <= MAX_PAYLOAD_BYTES),
-                    allocation: indexer_request.allocation.0 .0.into(),
+                    allocation: indexer_request.receipt.allocation().0 .0.into(),
                     subgraph_deployment: attestation.deployment.0.into(),
                     request_cid: attestation.request_cid.0.into(),
                     response_cid: attestation.response_cid.0.into(),