graphql-alt: Fetch latest Epoch

## Description

Support calling `Query.epoch` without an ID to fetch the latest epoch.

## Test plan

Updated E2E tests:

```
sui$ cargo nextest run         \
  -p sui-indexer-alt-e2e-tests \
  -- graphql/epochs
```

Author: amnn

crates/sui-indexer-alt-e2e-tests/tests/graphql/epochs/query.move

@@ -15,6 +15,8 @@
 
 //# run-graphql
 {
+  latest: epoch { ...E }
+
   e0: epoch(epochId: 0) { ...E }
   e1: epoch(epochId: 1) { ...E }
   e2: epoch(epochId: 2) { ...E }
@@ -31,16 +33,21 @@ fragment E on Epoch {
 }
 
 //# run-graphql
-{ # This checkpoint is half way through the epoch, so the epoch should return
-  # its starting information, but not its ending information.
+{ # This checkpoint is half way through an earlier epoch, which should be
+  # reflected in the latest epoch we get a start time for and an end time for.
   checkpoint(sequenceNumber: 2) {
     query {
-      epoch(epochId: 1) {
-        epochId
-        referenceGasPrice
-        startTimestamp
-        endTimestamp
-      }
+      latest: epoch { ...E }
+      e0: epoch(epochId: 0) { ...E }
+      e1: epoch(epochId: 1) { ...E }
+      e2: epoch(epochId: 2) { ...E }
     }
   }
 }
+
+fragment E on Epoch {
+  epochId
+  referenceGasPrice
+  startTimestamp
+  endTimestamp
+}

crates/sui-indexer-alt-e2e-tests/tests/graphql/epochs/query.snap

@@ -18,10 +18,16 @@ task 5, line 14:
 //# advance-epoch
 Epoch advanced: 2
 
-task 6, lines 16-31:
+task 6, lines 16-33:
 //# run-graphql
 Response: {
   "data": {
+    "latest": {
+      "epochId": 2,
+      "referenceGasPrice": "1000",
+      "startTimestamp": "1970-01-01T00:00:00.444Z",
+      "endTimestamp": null
+    },
     "e0": {
       "epochId": 0,
       "referenceGasPrice": "1000",
@@ -44,18 +50,31 @@ Response: {
   }
 }
 
-task 7, lines 33-46:
+task 7, lines 35-53:
 //# run-graphql
 Response: {
   "data": {
     "checkpoint": {
       "query": {
-        "epoch": {
+        "latest": {
+          "epochId": 1,
+          "referenceGasPrice": "1000",
+          "startTimestamp": "1970-01-01T00:00:00.123Z",
+          "endTimestamp": null
+        },
+        "e0": {
+          "epochId": 0,
+          "referenceGasPrice": "1000",
+          "startTimestamp": "1970-01-01T00:00:00Z",
+          "endTimestamp": "1970-01-01T00:00:00.123Z"
+        },
+        "e1": {
           "epochId": 1,
           "referenceGasPrice": "1000",
           "startTimestamp": "1970-01-01T00:00:00.123Z",
           "endTimestamp": null
-        }
+        },
+        "e2": null
       }
     }
   }

crates/sui-indexer-alt-graphql/schema.graphql

@@ -373,11 +373,11 @@ type Query {
 	"""
 	checkpoint(sequenceNumber: UInt53): Checkpoint
 	"""
-	Fetch an epoch by its ID.
+	Fetch an epoch by its ID, or fetch the latest epoch if no ID is provided.
 	
 	Returns `null` if the epoch does not exist yet, or was pruned.
 	"""
-	epoch(epochId: UInt53!): Epoch
+	epoch(epochId: UInt53): Epoch
 	"""
 	Fetch checkpoints by their sequence numbers.
 	

crates/sui-indexer-alt-graphql/src/api/query.rs

@@ -52,10 +52,14 @@ impl Query {
         Ok(Checkpoint::with_sequence_number(scope, sequence_number))
     }
 
-    /// Fetch an epoch by its ID.
+    /// Fetch an epoch by its ID, or fetch the latest epoch if no ID is provided.
     ///
     /// Returns `null` if the epoch does not exist yet, or was pruned.
-    async fn epoch(&self, ctx: &Context<'_>, epoch_id: UInt53) -> Result<Option<Epoch>, RpcError> {
+    async fn epoch(
+        &self,
+        ctx: &Context<'_>,
+        epoch_id: Option<UInt53>,
+    ) -> Result<Option<Epoch>, RpcError> {
         let scope = self.scope(ctx)?;
         Epoch::fetch(ctx, scope, epoch_id).await
     }
@@ -86,7 +90,7 @@ impl Query {
         let scope = self.scope(ctx)?;
         let epochs = keys
             .into_iter()
-            .map(|k| Epoch::fetch(ctx, scope.clone(), k));
+            .map(|k| Epoch::fetch(ctx, scope.clone(), Some(k)));
 
         try_join_all(epochs).await
     }

crates/sui-indexer-alt-graphql/src/api/types/epoch.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use anyhow::Context as _;
 use async_graphql::{dataloader::DataLoader, Context, Object};
 use sui_indexer_alt_reader::{
-    epochs::{EpochEndKey, EpochStartKey},
+    epochs::{CheckpointBoundedEpochStartKey, EpochEndKey, EpochStartKey},
     pg_reader::PgReader,
 };
 use sui_indexer_alt_schema::epochs::{StoredEpochEnd, StoredEpochStart};
@@ -58,7 +58,7 @@ impl Epoch {
 
     #[graphql(flatten)]
     async fn start(&self, ctx: &Context<'_>) -> Result<EpochStart, RpcError> {
-        self.start.fetch(ctx, self.epoch_id).await
+        self.start.fetch(ctx, Some(self.epoch_id)).await
     }
 
     #[graphql(flatten)]
@@ -143,14 +143,18 @@ impl Epoch {
     }
 
     /// Load the epoch from the store, and return it fully inflated (with contents already
-    /// fetched). Returns `None` if the epoch does not exist (or started after the checkpoint being
-    /// viewed).
+    /// fetched). If `epoch_id` is provided, the epoch with that ID is loaded. Otherwise, the
+    /// latest epoch for the current checkpoint is loaded.
+    ///
+    /// Returns `None` if the epoch does not exist (or started after the checkpoint being viewed).
     pub(crate) async fn fetch(
         ctx: &Context<'_>,
         scope: Scope,
-        epoch_id: UInt53,
+        epoch_id: Option<UInt53>,
     ) -> Result<Option<Self>, RpcError> {
-        let start = EpochStart::empty(scope).fetch(ctx, epoch_id.into()).await?;
+        let start = EpochStart::empty(scope)
+            .fetch(ctx, epoch_id.map(|id| id.into()))
+            .await?;
 
         let Some(contents) = &start.contents else {
             return Ok(None);
@@ -175,17 +179,22 @@ impl EpochStart {
     /// otherwise attempts to fetch from the store. The resulting value may still have an empty
     /// contents field, because it could not be found in the store, or the epoch started after the
     /// checkpoint being viewed.
-    async fn fetch(&self, ctx: &Context<'_>, epoch_id: u64) -> Result<Self, RpcError> {
+    async fn fetch(&self, ctx: &Context<'_>, epoch_id: Option<u64>) -> Result<Self, RpcError> {
         if self.contents.is_some() {
             return Ok(self.clone());
         }
 
         let pg_loader: &Arc<DataLoader<PgReader>> = ctx.data()?;
-        let Some(stored) = pg_loader
-            .load_one(EpochStartKey(epoch_id))
-            .await
-            .context("Failed to fetch epoch start information")?
-        else {
+
+        let load = if let Some(id) = epoch_id {
+            pg_loader.load_one(EpochStartKey(id)).await
+        } else {
+            let cp = self.scope.checkpoint_viewed_at();
+            pg_loader.load_one(CheckpointBoundedEpochStartKey(cp)).await
+        }
+        .context("Failed to fetch epoch start information")?;
+
+        let Some(stored) = load else {
             return Ok(self.clone());
         };
 

crates/sui-indexer-alt-graphql/src/snapshots/sui_indexer_alt_graphql__tests__schema.graphql.snap

@@ -377,11 +377,11 @@ type Query {
 	"""
 	checkpoint(sequenceNumber: UInt53): Checkpoint
 	"""
-	Fetch an epoch by its ID.
+	Fetch an epoch by its ID, or fetch the latest epoch if no ID is provided.
 
 	Returns `null` if the epoch does not exist yet, or was pruned.
 	"""
-	epoch(epochId: UInt53!): Epoch
+	epoch(epochId: UInt53): Epoch
 	"""
 	Fetch checkpoints by their sequence numbers.
 

crates/sui-indexer-alt-graphql/src/snapshots/sui_indexer_alt_graphql__tests__staging.graphql.snap

@@ -377,11 +377,11 @@ type Query {
 	"""
 	checkpoint(sequenceNumber: UInt53): Checkpoint
 	"""
-	Fetch an epoch by its ID.
+	Fetch an epoch by its ID, or fetch the latest epoch if no ID is provided.
 
 	Returns `null` if the epoch does not exist yet, or was pruned.
 	"""
-	epoch(epochId: UInt53!): Epoch
+	epoch(epochId: UInt53): Epoch
 	"""
 	Fetch checkpoints by their sequence numbers.
 

crates/sui-indexer-alt-graphql/staging.graphql

@@ -373,11 +373,11 @@ type Query {
 	"""
 	checkpoint(sequenceNumber: UInt53): Checkpoint
 	"""
-	Fetch an epoch by its ID.
+	Fetch an epoch by its ID, or fetch the latest epoch if no ID is provided.
 	
 	Returns `null` if the epoch does not exist yet, or was pruned.
 	"""
-	epoch(epochId: UInt53!): Epoch
+	epoch(epochId: UInt53): Epoch
 	"""
 	Fetch checkpoints by their sequence numbers.
 	

crates/sui-indexer-alt-reader/src/epochs.rs

@@ -1,10 +1,16 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{BTreeMap, HashMap},
+    sync::Arc,
+};
 
 use async_graphql::dataloader::Loader;
-use diesel::{ExpressionMethods, QueryDsl};
+use diesel::{
+    sql_types::{Array, BigInt},
+    ExpressionMethods, QueryDsl,
+};
 use sui_indexer_alt_schema::{
     epochs::{StoredEpochEnd, StoredEpochStart},
     schema::{kv_epoch_ends, kv_epoch_starts},
@@ -16,6 +22,10 @@ use crate::{error::Error, pg_reader::PgReader};
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct EpochStartKey(pub u64);
 
+/// Key for fetching information about the latest epoch to have started as of a given checkpoint.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct CheckpointBoundedEpochStartKey(pub u64);
+
 /// Key for fetching information about the end of an epoch (which must already be finished).
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct EpochEndKey(pub u64);
@@ -50,6 +60,69 @@ impl Loader<EpochStartKey> for PgReader {
     }
 }
 
+#[async_trait::async_trait]
+impl Loader<CheckpointBoundedEpochStartKey> for PgReader {
+    type Value = StoredEpochStart;
+    type Error = Arc<Error>;
+
+    async fn load(
+        &self,
+        keys: &[CheckpointBoundedEpochStartKey],
+    ) -> Result<HashMap<CheckpointBoundedEpochStartKey, Self::Value>, Self::Error> {
+        if keys.is_empty() {
+            return Ok(HashMap::new());
+        }
+
+        let mut conn = self.connect().await.map_err(Arc::new)?;
+
+        let cps: Vec<_> = keys.iter().map(|e| e.0 as i64).collect();
+        let query = diesel::sql_query(
+            r#"
+                SELECT
+                    v.*
+                FROM (
+                    SELECT UNNEST($1) cp_sequence_number
+                ) k
+                CROSS JOIN LATERAL (
+                    SELECT
+                        epoch,
+                        protocol_version,
+                        cp_lo,
+                        start_timestamp_ms,
+                        reference_gas_price,
+                        system_state
+                    FROM
+                        kv_epoch_starts
+                    WHERE
+                        kv_epoch_starts.cp_lo <= k.cp_sequence_number
+                    ORDER BY
+                        kv_epoch_starts.cp_lo DESC
+                    LIMIT
+                        1
+                ) v
+            "#,
+        )
+        .bind::<Array<BigInt>, _>(cps);
+
+        let stored_epochs: Vec<StoredEpochStart> = conn.results(query).await.map_err(Arc::new)?;
+
+        // A single data loader request may contain multiple keys for the same epoch. Store them in
+        // an ordered map, so that we can find the latest version for each key.
+        let cp_to_stored: BTreeMap<_, _> = stored_epochs
+            .into_iter()
+            .map(|epoch| (epoch.cp_lo as u64, epoch))
+            .collect();
+
+        Ok(keys
+            .iter()
+            .filter_map(|key| {
+                let stored = cp_to_stored.range(..=key.0).last()?.1;
+                Some((*key, stored.clone()))
+            })
+            .collect())
+    }
+}
+
 #[async_trait::async_trait]
 impl Loader<EpochEndKey> for PgReader {
     type Value = StoredEpochEnd;

crates/sui-indexer-alt-schema/src/epochs.rs

@@ -27,7 +27,7 @@ pub struct StoredEpochEnd {
     pub epoch_commitments: Vec<u8>,
 }
 
-#[derive(Insertable, Queryable, Debug, Clone, FieldCount)]
+#[derive(Insertable, Queryable, QueryableByName, Debug, Clone, FieldCount)]
 #[diesel(table_name = kv_epoch_starts)]
 pub struct StoredEpochStart {
     pub epoch: i64,