sui-indexer-alt-framework: fix indexer startup during streaming hang (#26198)

## Description 

**Commit 1:**

Adds back #26096 and #26116 which were reverted in #26188. 

**Commit 2:**

Specify `timeout` along with `connect_timeout` when creating the gRPC
client.
* `connect_timeout` only bounds the TCP/TLS handshake - the low-level
socket connect. Once the TCP connection is established,
`connect_timeout` has no further effect. The `subscribe_checkpoints` RPC
is a separate HTTP/2 request sent over the already-established
connection, so it's not covered by that setting. If a server accepts
TCP, but stalls before sending response headers, the client hangs the
indefinitely.
* `timeout` is the counterpart that bounds individual RPC calls on the
channel.

**Commit 3:**

Include changes from #26182.


## Test plan 

Added new unit test to prove indexer does not hang during startup when
streaming hangs.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] Indexing Framework:

Author: evan-wall-mysten

Cargo.lock

@@ -665,9 +665,9 @@ anemo-cli = { git = "https://github.com/mystenlabs/anemo.git", rev = "9248f8d395
 anemo-tower = { git = "https://github.com/mystenlabs/anemo.git", rev = "9248f8d3951df750360308df22618e5978ad6dc0" }
 
 # core-types from the new sdk for use in gRPC
-sui-sdk-types = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "e5c53de2e9570229b63bd0508b43f27c11cb5f76", features = [ "hash", "serde" ] }
-sui-crypto = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "e5c53de2e9570229b63bd0508b43f27c11cb5f76", features = [ "ed25519", "secp256r1", "secp256k1", "passkey", "zklogin" ] }
-sui-rpc = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "e5c53de2e9570229b63bd0508b43f27c11cb5f76" }
+sui-sdk-types = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "e494a36a76a0aab8c5d66d5557995faee5c1fb09", features = [ "hash", "serde" ] }
+sui-crypto = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "e494a36a76a0aab8c5d66d5557995faee5c1fb09", features = [ "ed25519", "secp256r1", "secp256k1", "passkey", "zklogin" ] }
+sui-rpc = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "e494a36a76a0aab8c5d66d5557995faee5c1fb09" }
 
 ### Workspace Members ###
 anemo-benchmark = { path = "crates/anemo-benchmark" }

Cargo.toml

@@ -95,7 +95,7 @@ impl Handler for TokenTransferDataHandler {
     async fn commit<'a>(
         values: &[Self::Value],
         conn: &mut Connection<'a>,
-    ) -> sui_indexer_alt_framework::Result<usize> {
+    ) -> anyhow::Result<usize> {
         Ok(diesel::insert_into(token_transfer_data::table)
             .values(values)
             .on_conflict_do_nothing()

crates/sui-bridge-indexer-alt/src/handlers/token_transfer_data_handler.rs

@@ -146,14 +146,15 @@ impl FormalSnapshot {
         info!(epoch, "Connected to valid formal snapshot");
 
         // Use the remote store to associate the epoch with a watermark pointing to its last
-        // transaction.
+        // transaction. We only use `end_of_epoch_checkpoints` here, so no byte counter is needed.
         let client = StoreIngestionClient::new(
             HttpBuilder::new()
                 .with_url(snapshot_args.remote_store_url.to_string())
                 .with_client_options(ClientOptions::new().with_allow_http(true))
                 .build()
                 .map(Arc::new)
                 .context("Failed to connect to remote checkpoint store")?,
+            None,
         );
 
         let end_of_epoch_checkpoints: Vec<_> = client

crates/sui-indexer-alt-consistent-store/src/restore/formal_snapshot.rs

@@ -19,8 +19,9 @@ diesel = { workspace = true, features = ["chrono"] }
 diesel-async = { workspace = true, features = ["bb8", "postgres", "async-connection-wrapper"] }
 diesel_migrations.workspace = true
 futures.workspace = true
+http.workspace = true
+mysten-network.workspace = true
 num_cpus.workspace = true
-pin-project-lite.workspace = true
 prometheus.workspace = true
 object_store.workspace = true
 serde_json.workspace = true