decouple transactional test runner from indexer and graphql flavor

crates/sui-graphql-e2e-tests/Cargo.toml

@@ -11,6 +11,8 @@ edition = "2021"
 workspace = true
 
 [dev-dependencies]
+anyhow.workspace = true
+async-trait.workspace = true
 telemetry-subscribers.workspace = true
 datatest-stable.workspace = true
 sui-graphql-rpc.workspace = true

crates/sui-graphql-e2e-tests/tests/tests.rs

@@ -3,12 +3,60 @@
 
 #![allow(unused_imports)]
 #![allow(unused_variables)]
-use std::{path::Path, sync::Arc};
+use async_trait::async_trait;
+use std::{path::Path, sync::Arc, time::Duration};
+use sui_graphql_rpc::test_infra::cluster::{serve_executor, ExecutorCluster};
 use sui_transactional_test_runner::{
-    run_test_impl,
+    args::SuiInitArgs,
+    create_adapter,
+    offchain_state::{OffchainStateReader, TestResponse},
+    run_tasks_with_adapter,
     test_adapter::{SuiTestAdapter, PRE_COMPILED},
 };
 
+pub struct OffchainReaderForAdapter {
+    cluster: Arc<ExecutorCluster>,
+}
+
+#[async_trait]
+impl OffchainStateReader for OffchainReaderForAdapter {
+    async fn wait_for_objects_snapshot_catchup(&self, base_timeout: Duration) {
+        self.cluster
+            .wait_for_objects_snapshot_catchup(base_timeout)
+            .await
+    }
+
+    async fn wait_for_checkpoint_catchup(&self, checkpoint: u64, base_timeout: Duration) {
+        self.cluster
+            .wait_for_checkpoint_catchup(checkpoint, base_timeout)
+            .await
+    }
+
+    async fn wait_for_pruned_checkpoint(&self, checkpoint: u64, base_timeout: Duration) {
+        self.cluster
+            .wait_for_checkpoint_pruned(checkpoint, base_timeout)
+            .await
+    }
+
+    async fn execute_graphql(
+        &self,
+        query: String,
+        show_usage: bool,
+    ) -> Result<TestResponse, anyhow::Error> {
+        let result = self
+            .cluster
+            .graphql_client
+            .execute_to_graphql(query, show_usage, vec![], vec![])
+            .await?;
+
+        Ok(TestResponse {
+            http_headers: Some(result.http_headers_without_date()),
+            response_body: result.response_body_json_pretty(),
+            service_version: result.graphql_version().ok(),
+        })
+    }
+}
+
 datatest_stable::harness!(
     run_test,
     "tests",
@@ -24,7 +72,35 @@ datatest_stable::harness!(
 async fn run_test(path: &Path) -> Result<(), Box<dyn std::error::Error>> {
     telemetry_subscribers::init_for_testing();
     if !cfg!(msim) {
-        run_test_impl::<SuiTestAdapter>(path, Some(Arc::new(PRE_COMPILED.clone()))).await?;
+        // start the adapter first to start the executor (simulacrum)
+        let (output, mut adapter) =
+            create_adapter::<SuiTestAdapter>(path, Some(Arc::new(PRE_COMPILED.clone()))).await?;
+
+        // In another crate like `sui-mvr-graphql-e2e-tests`, this would be the place to translate
+        // from `offchain_config` to something compatible with the indexer and graphql flavor of
+        // choice.
+        let offchain_config = adapter.offchain_config.as_ref().unwrap();
+
+        let cluster = serve_executor(
+            adapter.read_replica.as_ref().unwrap().clone(),
+            Some(offchain_config.snapshot_config.clone()),
+            offchain_config.retention_config.clone(),
+            offchain_config.data_ingestion_path.clone(),
+        )
+        .await;
+
+        let cluster_arc = Arc::new(cluster);
+
+        adapter.with_offchain_reader(Box::new(OffchainReaderForAdapter {
+            cluster: cluster_arc.clone(),
+        }));
+
+        run_tasks_with_adapter(path, adapter, output).await?;
+
+        match Arc::try_unwrap(cluster_arc) {
+            Ok(cluster) => cluster.cleanup_resources().await,
+            Err(_) => panic!("Still other Arc references!"),
+        }
     }
     Ok(())
 }

crates/sui-graphql-rpc/src/test_infra/cluster.rs

@@ -174,12 +174,16 @@ pub async fn serve_executor(
         .parse()
         .unwrap();
 
+    info!("Starting executor server on {}", executor_server_url);
+
     let executor_server_handle = tokio::spawn(async move {
         sui_rest_api::RestService::new_without_version(executor)
             .start_service(executor_server_url)
             .await;
     });
 
+    info!("spawned executor server");
+
     let snapshot_config = snapshot_config.unwrap_or_default();
 
     let (pg_store, pg_handle, _) = start_indexer_writer_for_testing(

crates/sui-transactional-test-runner/Cargo.toml

@@ -16,6 +16,7 @@ bcs.workspace = true
 bimap.workspace = true
 clap.workspace = true
 eyre.workspace = true
+http.workspace = true
 once_cell.workspace = true
 rand.workspace = true
 regex.workspace = true
@@ -25,6 +26,7 @@ tokio.workspace = true
 serde_json.workspace = true
 futures.workspace = true
 criterion.workspace = true
+tracing.workspace = true
 
 fastcrypto.workspace = true
 move-binary-format.workspace = true

crates/sui-transactional-test-runner/src/args.rs

@@ -1,6 +1,8 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::path::PathBuf;
+
 use crate::test_adapter::{FakeID, SuiTestAdapter};
 use anyhow::{bail, ensure};
 use clap;
@@ -73,6 +75,13 @@ pub struct SuiInitArgs {
     /// the indexer.
     #[clap(long = "epochs-to-keep")]
     pub epochs_to_keep: Option<u64>,
+    /// Dir for simulacrum to write checkpoint files to. To be passed to the offchain indexer and
+    /// reader.
+    #[clap(long)]
+    pub data_ingestion_path: Option<PathBuf>,
+    /// URL for the Sui REST API. To be passed to the offchain indexer and reader.
+    #[clap(long)]
+    pub rest_api_url: Option<String>,
 }
 
 #[derive(Debug, clap::Parser)]

crates/sui-transactional-test-runner/src/lib.rs

@@ -4,11 +4,14 @@
 //! This module contains the transactional test runner instantiation for the Sui adapter
 
 pub mod args;
+pub mod offchain_state;
 pub mod programmable_transaction_test_parser;
 mod simulator_persisted_store;
 pub mod test_adapter;
 
-pub use move_transactional_test_runner::framework::run_test_impl;
+pub use move_transactional_test_runner::framework::{
+    create_adapter, run_tasks_with_adapter, run_test_impl,
+};
 use rand::rngs::StdRng;
 use simulacrum::Simulacrum;
 use simulacrum::SimulatorStore;

crates/sui-transactional-test-runner/src/offchain_state.rs

@@ -0,0 +1,30 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use async_trait::async_trait;
+use std::time::Duration;
+
+pub struct TestResponse {
+    pub response_body: String,
+    pub http_headers: Option<http::HeaderMap>,
+    pub service_version: Option<String>,
+}
+
+/// Trait for interacting with the offchain state of the Sui network. To reduce test flakiness,
+/// these methods are used in the `RunGraphqlCommand` to stabilize the off-chain indexed state.
+#[async_trait]
+pub trait OffchainStateReader: Send + Sync + 'static {
+    /// Polls the objects snapshot table until it is within the allowed lag from the latest
+    /// checkpoint.
+    async fn wait_for_objects_snapshot_catchup(&self, base_timeout: Duration);
+    /// Polls the checkpoint table until the given checkpoint is committed.
+    async fn wait_for_checkpoint_catchup(&self, checkpoint: u64, base_timeout: Duration);
+    /// Polls the checkpoint table until the given checkpoint is pruned.
+    async fn wait_for_pruned_checkpoint(&self, checkpoint: u64, base_timeout: Duration);
+    /// Executes a GraphQL query and returns the response.
+    async fn execute_graphql(
+        &self,
+        query: String,
+        show_usage: bool,
+    ) -> Result<TestResponse, anyhow::Error>;
+}