Start scraper message_dispatch sync from high watermark (#2377)

### Description

This PR addresses an issue in which BSC RPC providers have a very hard
time providing very old transactions.

This caused BSC scraping to fail, since the scraper had never actually
indexed very early BSC messages (e.g. nonce 0).

By starting at the latest scraped nonce - 1, the scraper will bypass
this issue.

### Drive-by changes

None

rust/agents/scraper/src/agent.rs

@@ -246,11 +246,42 @@ macro_rules! spawn_sync_task {
     }
 }
 impl Scraper {
-    spawn_sync_task!(
-        build_message_indexer,
-        forward_message_sync_cursor,
-        "message_dispatch"
-    );
+    async fn build_message_indexer(
+        &self,
+        domain: HyperlaneDomain,
+        metrics: Arc<CoreMetrics>,
+        contract_sync_metrics: Arc<ContractSyncMetrics>,
+        db: HyperlaneSqlDb,
+        index_settings: IndexSettings,
+    ) -> Instrumented<JoinHandle<eyre::Result<()>>> {
+        let sync = self
+            .as_ref()
+            .settings
+            .build_message_indexer(
+                &domain,
+                &metrics.clone(),
+                &contract_sync_metrics.clone(),
+                Arc::new(db.clone()),
+            )
+            .await
+            .unwrap();
+        let latest_nonce = self
+            .scrapers
+            .get(&domain.id())
+            .unwrap()
+            .db
+            .last_message_nonce()
+            .await
+            .unwrap_or(None)
+            .unwrap_or(0);
+        let cursor = sync
+            .forward_message_sync_cursor(index_settings.clone(), latest_nonce.saturating_sub(1))
+            .await;
+        tokio::spawn(async move { sync.sync("message_dispatch", cursor).await }).instrument(
+            info_span!("ChainContractSync", chain=%domain.name(), event="message_dispatch"),
+        )
+    }
+
     spawn_sync_task!(
         build_delivery_indexer,
         rate_limited_cursor,

rust/agents/scraper/src/chain_scraper/mod.rs

@@ -65,6 +65,12 @@ impl HyperlaneSqlDb {
         &self.domain
     }
 
+    pub async fn last_message_nonce(&self) -> Result<Option<u32>> {
+        self.db
+            .last_message_nonce(self.domain.id(), &self.mailbox_address)
+            .await
+    }
+
     /// Takes a list of txn and block hashes and ensure they are all in the
     /// database. If any are not it will fetch the data and insert them.
     ///

rust/hyperlane-base/src/contract_sync/mod.rs

@@ -123,14 +123,15 @@ impl MessageContractSync {
     pub async fn forward_message_sync_cursor(
         &self,
         index_settings: IndexSettings,
+        next_nonce: u32,
     ) -> Box<dyn ContractSyncCursor<HyperlaneMessage>> {
         let forward_data = MessageSyncCursor::new(
             self.indexer.clone(),
             self.db.clone(),
             index_settings.chunk_size,
             index_settings.from,
             index_settings.from,
-            0,
+            next_nonce,
         );
         Box::new(ForwardMessageSyncCursor::new(forward_data))
     }