wip: detect and sleep during 403 slots

Author: dylanlott

Cargo.toml

@@ -27,11 +27,17 @@ integration = []
 [dependencies]
 init4-bin-base = "0.3"
 
-signet-constants = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
-signet-sim = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
-signet-tx-cache = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
-signet-types = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
-signet-zenith = { git = "https://github.com/init4tech/signet-sdk", branch = "main" }
+# signet-constants = { path = "../signet-sdk/crates/constants" }
+# signet-sim = { path = "../signet-sdk/crates/sim" }
+# signet-tx-cache = { path = "../signet-sdk/crates/tx-cache" }
+# signet-types = { path = "../signet-sdk/crates/types" }
+# signet-zenith = { path = "../signet-sdk/crates/zenith" }
+
+signet-constants = { git = "https://github.com/init4tech/signet-sdk", branch = "dylan/block-number" }
+signet-sim = { git = "https://github.com/init4tech/signet-sdk", branch = "dylan/block-number" }
+signet-tx-cache = { git = "https://github.com/init4tech/signet-sdk", branch = "dylan/block-number" }
+signet-types = { git = "https://github.com/init4tech/signet-sdk", branch = "dylan/block-number" }
+signet-zenith = { git = "https://github.com/init4tech/signet-sdk", branch = "dylan/block-number" }
 
 trevm = { version = "0.20.10", features = ["concurrent-db", "test-utils"] }
 

src/tasks/block/sim.rs

@@ -87,17 +87,18 @@ impl Simulator {
             db,
             constants,
             self.config.cfg_env(),
-            block,
+            block.clone(),
             finish_by,
             self.config.concurrency_limit,
             sim_items,
             self.config.rollup_block_gas_limit,
         );
 
-        let block = block_build.build().await;
-        debug!(block = ?block, "finished block simulation");
+        let mut built_block = block_build.build().await;
+        built_block.set_block_number(block.number);
+        debug!(block = ?built_block, "finished block simulation");
 
-        Ok(block)
+        Ok(built_block)
     }
 
     /// Spawns the simulator task, which handles the setup and sets the deadline
@@ -152,11 +153,10 @@ impl Simulator {
                 error!("block_env channel closed");
                 return;
             }
-
+            
             // If no env, skip this run
             let Some(block_env) = self.block_env.borrow_and_update().clone() else { return };
-
-            debug!(block_env = ?block_env, "building on block");
+            debug!(block_env = ?block_env, "building on block env");
 
             match self.handle_build(constants, sim_cache, finish_by, block_env).await {
                 Ok(block) => {

src/tasks/submit.rs

@@ -111,6 +111,7 @@ impl SubmitTask {
     async fn next_host_block_height(&self) -> eyre::Result<u64> {
         let result = self.provider().get_block_number().await?;
         let next = result.checked_add(1).ok_or_else(|| eyre!("next host block height overflow"))?;
+        debug!(next, "next host block height");
         Ok(next)
     }
 
@@ -257,7 +258,6 @@ impl SubmitTask {
             nonce = ?tx.nonce,
             "sending transaction to network"
         );
-        
 
         // assign the nonce and fill the rest of the values
         let SendableTx::Envelope(tx) = self.provider().fill(tx).await? else {
@@ -285,12 +285,15 @@ impl SubmitTask {
             return Ok(ControlFlow::Skip);
         }
 
+        // Okay so the code gets all the way to this log
+        // but we don't see the tx hash in the logs or in the explorer, 
+        // not even as a failed TX, just not at all.
         info!(
             tx_hash = %tx.tx_hash(),
             ru_chain_id = %resp.req.ru_chain_id,
             gas_limit = %resp.req.gas_limit,
             "dispatched to network"
-        );
+        ); 
 
         Ok(ControlFlow::Done)
     }
@@ -336,22 +339,31 @@ impl SubmitTask {
             let span = debug_span!("SubmitTask::retrying_handle_inbound", retries);
             debug!(retries, "number of retries");
 
-            let inbound_result = match self.handle_inbound(retries, block).instrument(span.clone()).await {
+            let inbound_result = match self
+                .handle_inbound(retries, block)
+                .instrument(span.clone())
+                .await
+            {
                 Ok(control_flow) => {
                     debug!(?control_flow, retries, "successfully handled inbound block");
                     control_flow
                 }
                 Err(err) => {
+                    // Log the retry attempt
                     retries += 1;
                     error!(error = %err, "error handling inbound block");
 
+                    // Delay until next slot if we get a 403 error
                     if err.to_string().contains("403") {
-                        debug!("403 error - skipping block");
-                        let (slot_number, start, end) = self.calculate_slot_window()?;
-                        debug!(slot_number, start, end, "403 sleep until skipping block");
-                        // TODO: Sleep until the end of the next slot and return retry
-                        return Ok(ControlFlow::Done);
+                        let (slot_number, _, end) = self.calculate_slot_window()?;
+                        let now = self.now();
+                        if end > now {
+                            let sleep_duration = std::time::Duration::from_secs(end - now);
+                            debug!(sleep_duration = ?sleep_duration, slot_number, "403 detected - sleeping until end of slot");
+                            tokio::time::sleep(sleep_duration).await;
+                        }
                     }
+
                     ControlFlow::Retry
                 }
             };
@@ -413,19 +425,36 @@ impl SubmitTask {
     }
 
     /// Task future for the submit task
+    /// NB: This task assumes that the simulator will only send it blocks for
+    /// slots that it's assigned.
     async fn task_future(self, mut inbound: mpsc::UnboundedReceiver<BuiltBlock>) {
+        // Holds a reference to the last block we attempted to submit
+        let mut last_block_attempted: u64 = 0;
+
         loop {
+            // Wait to receive a new block
             let Some(block) = inbound.recv().await else {
                 debug!("upstream task gone");
                 break;
             };
-            debug!(?block, "submit channel received block");
+            debug!(block_number = block.block_number(), ?block, "submit channel received block");
+
+            // Check if a block number was set and skip if not
+            if block.block_number() == 0 {
+                debug!("block number is 0 - skipping");
+                continue;
+            }
+
+            // Only attempt each block number once
+            if block.block_number() == last_block_attempted {
+                debug!("block number is unchanged from last attempt - skipping");
+                continue;
+            }
 
-            // TODO: Pass a BlockEnv to this function to give retrying handle inbound access to the block
-            // env and thus the block number so that we can be sure that we try for only our assigned slots.
+            // This means we have encountered a new block, so reset the last block attempted
+            last_block_attempted = block.block_number();
+            debug!(last_block_attempted, "resetting last block attempted");
 
-            // Instead this needs to fire off a task that attempts to land the block for the given slot
-            // Once that slot is up, it's invalid for the next anyway, so this job can be ephemeral.
             if self.retrying_handle_inbound(&block, 3).await.is_err() {
                 debug!("error handling inbound block");
                 continue;