fix(l1): only rebuild payload when mempool receives new transactions (Notify)

crates/blockchain/Cargo.toml

@@ -22,7 +22,7 @@ tracing.workspace = true
 bytes.workspace = true
 hex.workspace = true
 rustc-hash.workspace = true
-tokio = { workspace = true, features = ["time", "rt"] }
+tokio = { workspace = true, features = ["time", "rt", "sync", "macros"] }
 tokio-util.workspace = true
 
 [dev-dependencies]

crates/blockchain/mempool.rs

@@ -78,15 +78,23 @@ impl MempoolInner {
 #[derive(Debug, Default)]
 pub struct Mempool {
     inner: RwLock<MempoolInner>,
+    /// Signaled on transaction and blobs bundle insertions so payload
+    /// builders can await new work instead of busy-looping.
+    tx_added: tokio::sync::Notify,
 }
 
 impl Mempool {
     pub fn new(max_mempool_size: usize) -> Self {
         Mempool {
             inner: RwLock::new(MempoolInner::new(max_mempool_size)),
+            tx_added: tokio::sync::Notify::new(),
         }
     }
 
+    pub(crate) fn tx_added(&self) -> &tokio::sync::Notify {
+        &self.tx_added
+    }
+
     fn write(&self) -> Result<std::sync::RwLockWriteGuard<'_, MempoolInner>, StoreError> {
         self.inner
             .write()
@@ -123,6 +131,9 @@ impl Mempool {
             .insert((sender, transaction.nonce()), hash);
         inner.transaction_pool.insert(hash, transaction);
         inner.broadcast_pool.insert(hash);
+        // Drop the write lock before notifying to avoid holding it while waking waiters
+        drop(inner);
+        self.tx_added.notify_waiters();
 
         Ok(())
     }
@@ -157,6 +168,9 @@ impl Mempool {
         self.write()?
             .blobs_bundle_pool
             .insert(tx_hash, blobs_bundle);
+        // Notify after the bundle is inserted so builders that woke up from
+        // add_transaction see the complete blob data.
+        self.tx_added.notify_waiters();
         Ok(())
     }
 

crates/blockchain/payload.rs

@@ -390,6 +390,14 @@ impl Blockchain {
         // TODO(#4997): start with an empty block
         let mut res = self.build_payload(payload.clone())?;
         while start.elapsed() < SECONDS_PER_SLOT && !cancel_token.is_cancelled() {
+            // Wait for new transactions, cancellation, or slot deadline before rebuilding
+            let remaining = SECONDS_PER_SLOT.saturating_sub(start.elapsed());
+            let notified = self.mempool.tx_added().notified();
+            tokio::select! {
+                _ = notified => {}
+                _ = cancel_token.cancelled() => break,
+                _ = tokio::time::sleep(remaining) => break,
+            }
             let payload = payload.clone();
             let self_clone = self.clone();
             let building_task =