Update fisherman challenge structure (#249)

- Related PR: edgeandnode/network-services#60

Author: Theodus

graph-gateway/src/fisherman_client.rs

@@ -1,10 +1,10 @@
 use crate::indexer_client::Attestation;
 use async_trait::async_trait;
+use indexer_selection::Indexing;
 use prelude::*;
 use reqwest;
 use serde::Deserialize;
 use serde_json::json;
-use std::error::Error;
 
 #[derive(Clone, Copy, Debug, Deserialize)]
 pub enum ChallengeOutcome {
@@ -19,9 +19,10 @@ pub enum ChallengeOutcome {
 pub trait FishermanInterface {
     async fn challenge(
         &self,
-        indexer: &Address,
+        indexing: &Indexing,
         allocation: &Address,
         indexer_query: &str,
+        indexer_response: &str,
         attestation: &Attestation,
     ) -> ChallengeOutcome;
 }
@@ -36,13 +37,20 @@ pub struct FishermanClient {
 impl FishermanInterface for FishermanClient {
     async fn challenge(
         &self,
-        indexer: &Address,
+        indexing: &Indexing,
         allocation: &Address,
         indexer_query: &str,
+        indexer_response: &str,
         attestation: &Attestation,
     ) -> ChallengeOutcome {
         match self
-            .send_challenge(indexer, allocation, indexer_query, attestation)
+            .send_challenge(
+                indexing,
+                allocation,
+                indexer_query,
+                indexer_response,
+                attestation,
+            )
             .await
         {
             Ok(outcome) => outcome,
@@ -61,22 +69,25 @@ impl FishermanClient {
 
     async fn send_challenge(
         &self,
-        indexer: &Address,
+        indexing: &Indexing,
         allocation: &Address,
         indexer_query: &str,
+        indexer_response: &str,
         attestation: &Attestation,
-    ) -> Result<ChallengeOutcome, Box<dyn Error>> {
+    ) -> anyhow::Result<ChallengeOutcome> {
         let challenge = serde_json::to_string(&json!({
             "jsonrpc": "2.0",
             "id": 0,
             "method": "challenge",
             "params": {
-                "readOperation": indexer_query,
-                "allocationID": allocation.to_string(),
                 "attestation": serde_json::to_value(attestation)?,
+                "subgraphDeploymentID": format!("0x{}", hex::encode(&indexing.deployment.0)),
+                "allocationID": allocation.to_string(),
+                "query": indexer_query,
+                "response": indexer_response,
             },
         }))?;
-        tracing::trace!(%indexer, %challenge);
+        tracing::trace!(?indexing, %challenge);
         self.client
             .post(self.url.0.clone())
             .header("Content-Type", "application/json")

graph-gateway/src/indexer_client.rs

@@ -4,6 +4,7 @@ use indexer_selection::Selection;
 use prelude::*;
 use reqwest;
 use serde::{Deserialize, Serialize};
+use std::sync::Arc;
 
 #[async_trait]
 pub trait IndexerInterface {
@@ -18,7 +19,7 @@ pub trait IndexerInterface {
 #[derive(Clone, Debug)]
 pub struct IndexerResponse {
     pub status: u16,
-    pub payload: String,
+    pub payload: Arc<String>,
     pub attestation: Option<Attestation>,
 }
 
@@ -108,7 +109,7 @@ impl IndexerInterface for IndexerClient {
         };
         Ok(IndexerResponse {
             status: response_status.as_u16(),
-            payload: graphql_response,
+            payload: Arc::new(graphql_response),
             attestation: payload.attestation,
         })
     }

graph-gateway/src/main.rs

@@ -731,7 +731,7 @@ async fn handle_subgraph_query_inner(
     Ok(HttpResponseBuilder::new(StatusCode::OK)
         .insert_header(header::ContentType::json())
         .insert_header(("Graph-Attestation", attestation))
-        .body(&response.payload))
+        .body(response.payload.as_ref()))
 }
 
 pub fn graphql_error_response<S: ToString>(message: S) -> HttpResponse {

graph-gateway/src/query_engine.rs

@@ -564,6 +564,7 @@ where
                 selection.indexing.clone(),
                 result.allocation.clone(),
                 result.query.clone(),
+                response.payload.clone(),
                 attestation.clone(),
             );
         }
@@ -576,6 +577,7 @@ where
         indexing: Indexing,
         allocation: Address,
         indexer_query: Arc<String>,
+        indexer_response: Arc<String>,
         attestation: Attestation,
     ) {
         let fisherman = match &self.fisherman_client {
@@ -585,7 +587,13 @@ where
         let observations = self.observations.clone();
         tokio::spawn(async move {
             let outcome = fisherman
-                .challenge(&indexing.indexer, &allocation, &indexer_query, &attestation)
+                .challenge(
+                    &indexing,
+                    &allocation,
+                    &indexer_query,
+                    &indexer_response,
+                    &attestation,
+                )
                 .await;
             tracing::trace!(?outcome);
             let penalty = match outcome {