feat(indexer): batch state sync

crates/discovery-core/Cargo.toml

@@ -11,7 +11,7 @@ starknet-core = "0.16"
 starknet-crypto = "0.8"
 starknet-types-core = { version = "0.2", features = ["curve", "serde"] }
 starknet-providers = "0.16"
-tokio = { version = "1", features = ["rt"] }
+futures = "0.3"
 tracing = "0.1"
 serde = { version = "1.0", features = ["derive"] }
 url = "2"

crates/discovery-core/src/discovery/cursor.rs

@@ -65,9 +65,11 @@ pub struct SubchannelCursor {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub last_note_index: Option<u64>,
 
-    /// First index where no note exists (sentinel).
-    /// - Incoming: not used.
-    /// - Outgoing: upper bound (hi) for exponential search; `Some` = bisection phase.
+    /// - Incoming: last index confirmed to exist by exponential probe. Linear
+    ///   scan reads notes up to this index. Kept after scan — used to bound the
+    ///   next exponential probe range (`max_note_index * 2`). Re-probe triggers
+    ///   when `last_note_index == max_note_index`.
+    /// - Outgoing: first index confirmed empty (hi); `Some` = bisection phase.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_note_index: Option<u64>,
 }

crates/discovery-core/src/discovery/incoming_channels.rs

@@ -104,44 +104,40 @@ pub async fn discover_incoming_channels<PrivacyPool: IViews>(
         });
     }
 
-    // Discover and decrypt each channel.
-    // Cap pre-allocation: total_n_channels may come from an untrusted cursor,
-    // so a malicious value must not cause OOM via Vec::with_capacity.
-    const MAX_CAPACITY: usize = 1024;
-    let capacity = usize::try_from(total_n_channels.saturating_sub(start_index))
-        .unwrap_or(0)
-        .min(MAX_CAPACITY);
-    let mut channels = Vec::with_capacity(capacity);
-    let mut index = start_index;
-    let mut out_of_budget = false;
-
-    loop {
-        // Check if we've processed all channels
-        if index >= total_n_channels {
-            break;
-        }
+    // Batch-read as many channels as budget allows in a single RPC call.
+    let remaining = usize::try_from(total_n_channels.saturating_sub(start_index)).unwrap_or(0);
+    let batch_size = budget.consume_up_to(remaining, COST_CHANNEL_INFO);
+    if batch_size == 0 {
+        return Ok(ChannelDiscoveryResult {
+            channels: vec![],
+            last_index: None,
+            has_more: true,
+        });
+    }
 
-        // Consume budget for get_channel_info
-        if !budget.consume(COST_CHANNEL_INFO) {
-            out_of_budget = true;
-            break;
-        }
+    let encrypted_batch = privacy_pool
+        .get_channel_info_batch(recipient_addr, start_index, batch_size)
+        .await?;
 
-        let encrypted = privacy_pool.get_channel_info(recipient_addr, index).await?;
+    let mut channels = Vec::with_capacity(batch_size);
+    for (i, encrypted) in encrypted_batch.into_iter().enumerate() {
+        let index = start_index
+            + u64::try_from(i)
+                .map_err(|_| DiscoveryError::InvalidCursor("channel index overflow".into()))?;
 
         let info = decrypt_channel_info(&encrypted, private_key)
             .map_err(|source| DiscoveryError::Decryption { index, source })?;
 
         channels.push(IncomingChannel { index, info });
-        index += 1;
     }
 
     let last_index = channels.last().map(|c| c.index);
+    let has_more = batch_size < remaining;
 
     Ok(ChannelDiscoveryResult {
         channels,
         last_index,
-        has_more: out_of_budget,
+        has_more,
     })
 }