sui-indexer-alt-reader: chunk oversized LedgerGrpcReader batch requests (#27516)

## Description

`LedgerGrpcReader` (used by GraphQL and jsonrpc for ledger gRPC/KV-RPC
access) built a single unchunked
`BatchGetTransactions`/`BatchGetObjects` request per `DataLoader` batch,
which could exceed the ledger service's own 200/1000 batch-size caps and
surface as an opaque gRPC error.

Introduces a `ChunkedLoader<K>` trait: `LedgerGrpcReader` implements it
per key type with `chunk_size()` (the limit) and `load_chunk()` (the
actual gRPC call); a provided `load_chunked()` method splits keys into
chunks, dispatches `load_chunk` concurrently per chunk, and merges the
results. A single generic `impl<K> Loader<K> for LedgerGrpcReader`
(where `Self: ChunkedLoader<K>`) forwards to `load_chunked`, covering
every chunked key type at once. `CheckpointKey`'s loader already fans
out per-key rather than batching, so it keeps its own ordinary
`Loader<CheckpointKey>` impl and never implements `ChunkedLoader`.

## Test plan

New chunking tests in `sui-indexer-alt-reader`
(`events`/`transactions`/`objects`) construct oversized key sets against
a mock ledger service and assert requests are split to the batch limit.

## Release notes

- [ ] Nothing user-facing changed.

Author: evan-wall-mysten

Cargo.lock

@@ -46,3 +46,6 @@ sui-rpc.workspace = true
 sui-sdk-types.workspace = true
 sui-sql-macro.workspace = true
 sui-types.workspace = true
+
+[dev-dependencies]
+tokio-stream.workspace = true

crates/sui-indexer-alt-reader/Cargo.toml

@@ -23,7 +23,9 @@ use sui_types::effects::TransactionEvents;
 
 use crate::bigtable_reader::BigtableReader;
 use crate::error::Error;
+use crate::ledger_grpc_reader::ChunkedLoader;
 use crate::ledger_grpc_reader::LedgerGrpcReader;
+use crate::ledger_grpc_reader::MAX_BATCH_GET_TRANSACTIONS;
 use crate::pg_reader::PgReader;
 
 /// Key for fetching transaction events contents (Events, TimestampMs) by digest.
@@ -102,18 +104,18 @@ impl Loader<TransactionEventsKey> for BigtableReader {
 }
 
 #[async_trait::async_trait]
-impl Loader<TransactionEventsKey> for LedgerGrpcReader {
+impl ChunkedLoader<TransactionEventsKey> for LedgerGrpcReader {
     type Value = TransactionEventsData;
     type Error = Error;
 
-    async fn load(
+    fn chunk_size(&self) -> usize {
+        MAX_BATCH_GET_TRANSACTIONS
+    }
+
+    async fn load_chunk(
         &self,
         keys: &[TransactionEventsKey],
-    ) -> Result<HashMap<TransactionEventsKey, Self::Value>, Self::Error> {
-        if keys.is_empty() {
-            return Ok(HashMap::new());
-        }
-
+    ) -> Result<HashMap<TransactionEventsKey, TransactionEventsData>, Error> {
         let digests = keys.iter().map(|key| key.0.to_string()).collect();
 
         let mut request = proto::BatchGetTransactionsRequest::default();
@@ -171,3 +173,28 @@ impl Loader<TransactionEventsKey> for LedgerGrpcReader {
             .map_err(Error::from)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ledger_grpc_reader::test_support::assert_chunked;
+    use crate::ledger_grpc_reader::test_support::mock_reader;
+
+    #[tokio::test]
+    async fn load_chunks_oversized_batches() {
+        let (reader, mock, server) = mock_reader().await;
+        let limit = MAX_BATCH_GET_TRANSACTIONS;
+
+        let keys: Vec<TransactionEventsKey> = (0..limit + 50)
+            .map(|_| TransactionEventsKey(TransactionDigest::random()))
+            .collect();
+
+        let result = reader.load(&keys).await.expect("load should succeed");
+        assert!(result.is_empty());
+
+        let expected: Vec<String> = keys.iter().map(|key| key.0.to_string()).collect();
+        assert_chunked(mock.transaction_batches(), limit, &expected);
+
+        server.abort();
+    }
+}

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

@@ -1,10 +1,14 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::collections::HashMap;
+use std::hash::Hash;
 use std::time::Duration;
 
 use anyhow::Context;
 use async_graphql::dataloader::DataLoader;
+use async_graphql::dataloader::Loader;
+use futures::future::try_join_all;
 use prometheus::Registry;
 use sui_rpc::Client;
 use sui_rpc::proto::sui::rpc::v2 as grpc;
@@ -49,6 +53,68 @@ pub struct LedgerGrpcReader {
     timeout: Option<Duration>,
 }
 
+/// Maximum number of transaction digests `LedgerGrpcReader` will put in a single
+/// `BatchGetTransactions` call, matching the ledger gRPC/KV-RPC service's own hard cap.
+pub(crate) const MAX_BATCH_GET_TRANSACTIONS: usize = 200;
+
+/// Maximum number of object keys `LedgerGrpcReader` will put in a single `BatchGetObjects`
+/// call, matching the ledger gRPC/KV-RPC service's own hard cap.
+pub(crate) const MAX_BATCH_GET_OBJECTS: usize = 1000;
+
+/// Implemented by `LedgerGrpcReader` for each key type whose `Loader::load` needs to stay under
+/// the ledger service's batch-size limit — `DataLoader`'s own `max_batch_size` is only a dispatch
+/// trigger, not a hard cap on how many keys reach a single `Loader::load` call. `load_chunk`
+/// supplies the raw, single-chunk fetch and `chunk_size` supplies the limit; `load_chunked`'s
+/// default body splits `keys` into chunks, dispatches `load_chunk` concurrently per chunk, and
+/// merges the results.
+///
+/// `pub`, not `pub(crate)`: it's referenced through `LedgerGrpcReader`'s blanket `Loader<K>` impl
+/// below (`type Value = <Self as ChunkedLoader<K>>::Value`), and that impl is part of
+/// `LedgerGrpcReader`'s public interface since both the type and `Loader` are public.
+#[async_trait::async_trait]
+pub trait ChunkedLoader<K>
+where
+    K: Send + Sync + Hash + Eq + Clone + 'static,
+{
+    type Value: Send + Sync + Clone + 'static;
+    type Error: Send + Sync + Clone + 'static;
+
+    fn chunk_size(&self) -> usize;
+
+    async fn load_chunk(&self, keys: &[K]) -> Result<HashMap<K, Self::Value>, Self::Error>;
+
+    async fn load_chunked(&self, keys: &[K]) -> Result<HashMap<K, Self::Value>, Self::Error>
+    where
+        Self: Sync,
+    {
+        let limit = self.chunk_size();
+
+        let mut results = HashMap::new();
+        for batch in try_join_all(keys.chunks(limit).map(|chunk| self.load_chunk(chunk))).await? {
+            results.extend(batch);
+        }
+        Ok(results)
+    }
+}
+
+/// Covers every key type `LedgerGrpcReader` implements [`ChunkedLoader`] for. Coherent because
+/// `Self` (`LedgerGrpcReader`) is a concrete local type — only `K` is generic — unlike a bare
+/// `impl<K, T: ChunkedLoader<K>> Loader<K> for T`, which the orphan rules reject since neither
+/// `Loader` (foreign) nor `T` (an uncovered generic) is local.
+#[async_trait::async_trait]
+impl<K> Loader<K> for LedgerGrpcReader
+where
+    K: Send + Sync + Hash + Eq + Clone + 'static,
+    Self: ChunkedLoader<K>,
+{
+    type Value = <Self as ChunkedLoader<K>>::Value;
+    type Error = <Self as ChunkedLoader<K>>::Error;
+
+    async fn load(&self, keys: &[K]) -> Result<HashMap<K, Self::Value>, Self::Error> {
+        self.load_chunked(keys).await
+    }
+}
+
 impl LedgerGrpcArgs {
     pub fn new(
         statement_timeout_ms: Option<u64>,
@@ -90,7 +156,7 @@ impl LedgerGrpcReader {
         Ok(Self { client, timeout })
     }
 
-    pub fn as_data_loader(&self) -> DataLoader<Self> {
+    pub(crate) fn as_data_loader(&self) -> DataLoader<Self> {
         DataLoader::new(self.clone(), tokio::spawn)
     }
 
@@ -207,3 +273,126 @@ impl Default for LedgerGrpcArgs {
         }
     }
 }
+
+#[cfg(test)]
+pub(crate) mod test_support {
+    use std::net::SocketAddr;
+    use std::sync::Arc;
+    use std::sync::Mutex;
+    use std::time::Duration;
+
+    use prometheus::Registry;
+    use sui_rpc::proto::sui::rpc::v2::BatchGetObjectsRequest;
+    use sui_rpc::proto::sui::rpc::v2::BatchGetObjectsResponse;
+    use sui_rpc::proto::sui::rpc::v2::BatchGetTransactionsRequest;
+    use sui_rpc::proto::sui::rpc::v2::BatchGetTransactionsResponse;
+    use sui_rpc::proto::sui::rpc::v2::GetObjectRequest;
+    use sui_rpc::proto::sui::rpc::v2::ledger_service_server::LedgerService;
+    use sui_rpc::proto::sui::rpc::v2::ledger_service_server::LedgerServiceServer;
+    use tokio::net::TcpListener;
+    use tokio::task::JoinHandle;
+    use tokio_stream::wrappers::TcpListenerStream;
+    use tonic::Request;
+    use tonic::Response;
+    use tonic::Status;
+
+    use super::LedgerGrpcArgs;
+    use super::LedgerGrpcReader;
+
+    /// Starts a [`MockLedgerServer`] and constructs a [`LedgerGrpcReader`]
+    /// pointed at it. Shared by every loader's chunking test.
+    pub(crate) async fn mock_reader() -> (LedgerGrpcReader, MockLedgerServer, JoinHandle<()>) {
+        let mock = MockLedgerServer::new();
+        let (addr, server) = mock.start().await.expect("start mock ledger service");
+        let reader = LedgerGrpcReader::new(
+            format!("http://{addr}").parse().unwrap(),
+            LedgerGrpcArgs::default(),
+            None,
+            &Registry::new(),
+        )
+        .await
+        .expect("construct LedgerGrpcReader");
+        (reader, mock, server)
+    }
+
+    /// Asserts that `batches` (the digest lists recorded by
+    /// [`MockLedgerServer`] across however many gRPC calls a loader made) is
+    /// chunked to `limit`, and that their union reconstructs `expected`
+    /// exactly, with none lost or duplicated.
+    pub(crate) fn assert_chunked(batches: Vec<Vec<String>>, limit: usize, expected: &[String]) {
+        assert_eq!(batches.len(), expected.len().div_ceil(limit));
+        assert!(batches.iter().all(|batch| batch.len() <= limit));
+
+        let mut requested: Vec<String> = batches.into_iter().flatten().collect();
+        requested.sort();
+        let mut expected = expected.to_vec();
+        expected.sort();
+        assert_eq!(requested, expected);
+    }
+
+    /// A minimal in-process `LedgerService` that records the shape of every
+    /// `BatchGetTransactions`/`BatchGetObjects` call it receives and responds
+    /// with an empty result set. Every other RPC falls back to the generated
+    /// trait's default `unimplemented` behavior, which is all the chunking
+    /// tests that use this need.
+    #[derive(Clone, Default)]
+    pub(crate) struct MockLedgerServer {
+        transaction_batches: Arc<Mutex<Vec<Vec<String>>>>,
+        object_batches: Arc<Mutex<Vec<Vec<GetObjectRequest>>>>,
+    }
+
+    impl MockLedgerServer {
+        pub(crate) fn new() -> Self {
+            Self::default()
+        }
+
+        pub(crate) async fn start(&self) -> anyhow::Result<(SocketAddr, JoinHandle<()>)> {
+            let listener = TcpListener::bind("127.0.0.1:0").await?;
+            let addr = listener.local_addr()?;
+            let mock = self.clone();
+            let handle = tokio::spawn(async move {
+                let incoming = TcpListenerStream::new(listener);
+                tonic::transport::Server::builder()
+                    .add_service(LedgerServiceServer::new(mock))
+                    .serve_with_incoming(incoming)
+                    .await
+                    .ok();
+            });
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            Ok((addr, handle))
+        }
+
+        pub(crate) fn transaction_batches(&self) -> Vec<Vec<String>> {
+            self.transaction_batches.lock().unwrap().clone()
+        }
+
+        pub(crate) fn object_batches(&self) -> Vec<Vec<GetObjectRequest>> {
+            self.object_batches.lock().unwrap().clone()
+        }
+    }
+
+    #[tonic::async_trait]
+    impl LedgerService for MockLedgerServer {
+        async fn batch_get_transactions(
+            &self,
+            request: Request<BatchGetTransactionsRequest>,
+        ) -> Result<Response<BatchGetTransactionsResponse>, Status> {
+            self.transaction_batches
+                .lock()
+                .unwrap()
+                .push(request.into_inner().digests);
+            Ok(Response::new(BatchGetTransactionsResponse::default()))
+        }
+
+        async fn batch_get_objects(
+            &self,
+            request: Request<BatchGetObjectsRequest>,
+        ) -> Result<Response<BatchGetObjectsResponse>, Status> {
+            self.object_batches
+                .lock()
+                .unwrap()
+                .push(request.into_inner().requests);
+            Ok(Response::new(BatchGetObjectsResponse::default()))
+        }
+    }
+}

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

@@ -19,7 +19,9 @@ use sui_types::storage::ObjectKey;
 
 use crate::bigtable_reader::BigtableReader;
 use crate::error::Error;
+use crate::ledger_grpc_reader::ChunkedLoader;
 use crate::ledger_grpc_reader::LedgerGrpcReader;
+use crate::ledger_grpc_reader::MAX_BATCH_GET_OBJECTS;
 use crate::pg_reader::PgReader;
 
 /// Key for fetching the contents a particular version of an object.
@@ -103,18 +105,18 @@ impl Loader<VersionedObjectKey> for BigtableReader {
 }
 
 #[async_trait::async_trait]
-impl Loader<VersionedObjectKey> for LedgerGrpcReader {
+impl ChunkedLoader<VersionedObjectKey> for LedgerGrpcReader {
     type Value = Object;
     type Error = Error;
 
-    async fn load(
+    fn chunk_size(&self) -> usize {
+        MAX_BATCH_GET_OBJECTS
+    }
+
+    async fn load_chunk(
         &self,
         keys: &[VersionedObjectKey],
     ) -> Result<HashMap<VersionedObjectKey, Object>, Error> {
-        if keys.is_empty() {
-            return Ok(HashMap::new());
-        }
-
         let requests = keys
             .iter()
             .map(|key| {
@@ -145,3 +147,47 @@ impl Loader<VersionedObjectKey> for LedgerGrpcReader {
         Ok(results)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use sui_sdk_types::Address;
+    use sui_types::base_types::ObjectID;
+
+    use super::*;
+    use crate::ledger_grpc_reader::test_support::mock_reader;
+
+    #[tokio::test]
+    async fn load_chunks_oversized_batches() {
+        let (reader, mock, server) = mock_reader().await;
+        let limit = MAX_BATCH_GET_OBJECTS;
+
+        let keys: Vec<VersionedObjectKey> = (0..limit + 50)
+            .map(|i| VersionedObjectKey(ObjectID::random(), i as u64))
+            .collect();
+
+        let result = reader.load(&keys).await.expect("load should succeed");
+        assert!(result.is_empty());
+
+        let batches = mock.object_batches();
+        assert_eq!(batches.len(), 2);
+        assert!(batches.iter().all(|batch| batch.len() <= limit));
+
+        let mut requested: Vec<(String, Option<u64>)> = batches
+            .into_iter()
+            .flatten()
+            .map(|req| (req.object_id.unwrap_or_default(), req.version))
+            .collect();
+        requested.sort();
+        let mut expected: Vec<(String, Option<u64>)> = keys
+            .iter()
+            .map(|key| {
+                let address: Address = key.0.into();
+                (address.to_string(), Some(key.1))
+            })
+            .collect();
+        expected.sort();
+        assert_eq!(requested, expected);
+
+        server.abort();
+    }
+}