[gql][1/n] cross-query consistency (#16635)

## Description 

Today, every query in a graphql request separately makes a query for the
available range. This means that every query is self-consistent (so if
it is paginated, you will continue paginating the state at the same
checkpoint as you started), but if you issue multiple queries at the
same time, each query in the graphql request may see different available
range values.

This can result in inconsistencies in the results of disjoint queries in
the same request:
```
query {
  alice: address(address: 0xA11CE) {
    objects { nodes { digest } }
  }
  
  bob: address(address: 0xB0B) {
    objects { nodes { digest } }
  }
}
```

The above query might witness the same object twice, or no times, based
on timing, if issued around the time that Alice transfers that object to
Bob.

This PR adds cross-query consistency by launching a background task that
periodically updates the available range, which gets passed to the
graphql service. Every top-level query in the request will inherit this
available range. Additionally, instead of making a separate fetch when
`Query.availableRange` is selected for, the resolver will simply return
the available range from the background task.

Subsequent PRs will change `checkpoint_viewed_at` from `Option<u64>` to
`u64` and simplify the logic around handling consistency for nested
fields.

## Test Plan 

Existing tests should pass. Because of the background task, we also need
to wait for the graphql service to "catch up" to the latest checkpoint

---
If your changes are not user-facing and do not break anything, you can
skip the following section. Otherwise, please briefly describe what has
changed under the Release Notes section.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

crates/sui-cluster-test/src/cluster.rs

@@ -253,6 +253,7 @@ impl Cluster for LocalNewCluster {
             start_graphql_server_with_fn_rpc(
                 graphql_connection_config.clone(),
                 Some(fullnode_url.clone()),
+                /* cancellation_token */ None,
             )
             .await;
         }

crates/sui-graphql-rpc/Cargo.toml

@@ -50,7 +50,8 @@ sui-types.workspace = true
 tap.workspace = true
 telemetry-subscribers.workspace = true
 tracing.workspace = true
-tokio.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread"] }
+tokio-util = { workspace = true, features = ["rt"] }
 toml.workspace = true
 tower.workspace = true
 tower-http.workspace = true

crates/sui-graphql-rpc/README.md

@@ -17,7 +17,7 @@ The order is important:
 
 ### Setting up local db
 
-Rpc 1.5 backs the graphql schema with a db based on IndexerV2 schema. To spin up a local db, follow the instructions at [sui-indexer](../sui-indexer/README.md) until "Running standalone indexer".
+The graphql service is backed by a db based on the db schema in [sui-indexer](../sui-indexer/src/schema.rs). To spin up a local db, follow the instructions at [sui-indexer](../sui-indexer/README.md) until "Running standalone indexer".
 
 If you have not created a db yet, you can do so as follows:
 ```sh
@@ -31,22 +31,36 @@ You should be able to refer to the db url now:
 With the new db, run the following commands (also under `sui/crates/sui-indexer`):
 
 ```sh
-diesel setup --database-url="<DATABASE_URL>" --migration-dir=migrations_v2
-diesel migration run --database-url="<DATABASE_URL>" --migration-dir=migrations_v2
+diesel setup --database-url="<DATABASE_URL>" --migration-dir=migrations
+diesel migration run --database-url="<DATABASE_URL>" --migration-dir=migrations
 ```
 
 ### Launching the server
 See [src/commands.rs](src/commands.rs) for all CLI options.
 
-```
-cargo run --bin sui-graphql-rpc start-server [--rpc-url] [--db-url] [--port] [--host] [--config]
+Example `.toml` config:
+```toml
+[limits]
+max-query-depth = 15
+max-query-nodes = 500
+max-output-nodes = 100000
+max-query-payload-size = 5000
+max-db-query-cost = 20000
+default-page-size = 5
+max-page-size = 10
+request-timeout-ms = 15000
+max-type-argument-depth = 16
+max-type-argument-width = 32
+max-type-nodes = 256
+max-move-value-depth = 128
+
+[background-tasks]
+watermark-update-ms=500
 ```
 
 This will build sui-graphql-rpc and start an IDE:
-
 ```
-Starting server...
-Launch GraphiQL IDE at: http://127.0.0.1:8000
+cargo run --bin sui-graphql-rpc start-server [--rpc-url] [--db-url] [--port] [--host] [--config]
 ```
 
 ### Launching the server w/ indexer

crates/sui-graphql-rpc/src/config.rs

@@ -42,6 +42,7 @@ pub(crate) const DEFAULT_SERVER_DB_URL: &str =
 pub(crate) const DEFAULT_SERVER_DB_POOL_SIZE: u32 = 3;
 pub(crate) const DEFAULT_SERVER_PROM_HOST: &str = "0.0.0.0";
 pub(crate) const DEFAULT_SERVER_PROM_PORT: u16 = 9184;
+pub(crate) const DEFAULT_WATERMARK_UPDATE_MS: u64 = 500;
 
 /// The combination of all configurations for the GraphQL service.
 #[derive(Serialize, Clone, Deserialize, Debug, Default)]
@@ -90,6 +91,9 @@ pub struct ServiceConfig {
 
     #[serde(default)]
     pub(crate) name_service: NameServiceConfig,
+
+    #[serde(default)]
+    pub(crate) background_tasks: BackgroundTasksConfig,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Copy)]
@@ -121,6 +125,13 @@ pub struct Limits {
     pub max_move_value_depth: u32,
 }
 
+#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Copy)]
+#[serde(rename_all = "kebab-case")]
+pub struct BackgroundTasksConfig {
+    #[serde(default)]
+    pub watermark_update_ms: u64,
+}
+
 #[derive(Debug)]
 pub struct Version(pub &'static str);
 
@@ -323,6 +334,13 @@ impl ServiceConfig {
     pub fn read(contents: &str) -> Result<Self, toml::de::Error> {
         toml::de::from_str::<Self>(contents)
     }
+
+    pub fn test_defaults() -> Self {
+        Self {
+            background_tasks: BackgroundTasksConfig::test_defaults(),
+            ..Default::default()
+        }
+    }
 }
 
 impl Limits {
@@ -345,6 +363,14 @@ impl Ide {
     }
 }
 
+impl BackgroundTasksConfig {
+    pub fn test_defaults() -> Self {
+        Self {
+            watermark_update_ms: 100, // Set to 100ms for testing
+        }
+    }
+}
+
 impl Default for Ide {
     fn default() -> Self {
         Self {
@@ -400,6 +426,14 @@ impl Default for InternalFeatureConfig {
     }
 }
 
+impl Default for BackgroundTasksConfig {
+    fn default() -> Self {
+        Self {
+            watermark_update_ms: DEFAULT_WATERMARK_UPDATE_MS,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/sui-graphql-rpc/src/consistency.rs

@@ -43,6 +43,11 @@ pub(crate) struct ConsistentNamedCursor {
     pub c: u64,
 }
 
+/// The high checkpoint watermark stamped on each GraphQL request. This is used to ensure
+/// cross-query consistency.
+#[derive(Clone, Copy)]
+pub(crate) struct CheckpointViewedAt(pub u64);
+
 /// Trait for cursors that have a checkpoint sequence number associated with them.
 pub(crate) trait Checkpointed: CursorType {
     fn checkpoint_viewed_at(&self) -> u64;

crates/sui-graphql-rpc/src/context_data/db_data_provider.rs

@@ -64,14 +64,6 @@ impl PgManager {
             .map_err(|e| Error::Internal(format!("{e}")))
     }
 
-    pub(crate) async fn available_range(&self) -> Result<(u64, u64), Error> {
-        Ok(self
-            .inner
-            .spawn_blocking(|this| this.get_consistent_read_range())
-            .await
-            .map(|(start, end)| (start as u64, end as u64))?)
-    }
-
     /// If no epoch was requested or if the epoch requested is in progress,
     /// returns the latest sui system state.
     pub(crate) async fn fetch_sui_system_state(

crates/sui-graphql-rpc/src/main.rs

@@ -11,6 +11,8 @@ use sui_graphql_rpc::config::{
 };
 use sui_graphql_rpc::server::builder::export_schema;
 use sui_graphql_rpc::server::graphiql_server::start_graphiql_server;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
 
 // WARNING!!!
 //
@@ -86,6 +88,8 @@ async fn main() {
             let _guard = telemetry_subscribers::TelemetryConfig::new()
                 .with_env()
                 .init();
+            let tracker = TaskTracker::new();
+            let cancellation_token = CancellationToken::new();
 
             println!("Starting server...");
             let server_config = ServerConfig {
@@ -96,9 +100,31 @@ async fn main() {
                 ..ServerConfig::default()
             };
 
-            start_graphiql_server(&server_config, &VERSION)
-                .await
-                .unwrap();
+            let cancellation_token_clone = cancellation_token.clone();
+            let graphql_service_handle = tracker.spawn(async move {
+                start_graphiql_server(&server_config, &VERSION, cancellation_token_clone)
+                    .await
+                    .unwrap();
+            });
+
+            // Wait for shutdown signal
+            tokio::select! {
+                result = graphql_service_handle => {
+                    if let Err(e) = result {
+                        println!("GraphQL service crashed or exited with error: {:?}", e);
+                    }
+                }
+                _ = tokio::signal::ctrl_c() => {
+                    println!("Ctrl+C signal received.");
+                },
+            }
+
+            println!("Shutting down...");
+
+            // Send shutdown signal to application
+            cancellation_token.cancel();
+            tracker.close();
+            tracker.wait().await;
         }
     }
 }