ntx-builder: replace dataless Notify with per-account watch (#2301)

Author: SantiagoPittella

bin/ntx-builder/src/actor/mod.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use miden_protocol::account::{AccountId, AccountUpdateDetails};
 use miden_protocol::block::{BlockHeader, SignedBlock};
 use miden_protocol::note::Nullifier;
@@ -90,4 +92,14 @@ impl CommittedBlockEffects {
             .then_some(*account_id)
         })
     }
+
+    /// The latest transaction committed against each account in this block.
+    ///
+    /// `account_transactions` is in block order, so collecting into a map keeps the last
+    /// transaction per account. Both `apply_committed_block` (to persist `accounts.last_tx_id`) and
+    /// the coordinator (to populate each [`AccountView`](crate::coordinator)'s `last_committed_tx`)
+    /// derive landing state from this single definition, so the two never disagree.
+    pub fn latest_tx_per_account(&self) -> HashMap<AccountId, TransactionId> {
+        self.account_transactions.iter().copied().collect()
+    }
 }

bin/ntx-builder/src/committed_block.rs

@@ -5,50 +5,53 @@ use anyhow::Context;
 use miden_node_utils::shutdown::CancellationToken;
 use miden_node_utils::tracing::miden_instrument;
 use miden_protocol::account::AccountId;
+use miden_protocol::block::BlockNumber;
+use miden_protocol::transaction::TransactionId;
 use miden_standards::note::AccountTargetNetworkNote;
-use tokio::sync::{Notify, Semaphore};
+use tokio::sync::{Semaphore, watch};
 use tokio::task::JoinSet;
 
 use crate::LOG_TARGET;
 use crate::actor::{AccountActor, AccountActorContext};
 use crate::committed_block::CommittedBlockEffects;
 
+// ACCOUNT VIEW
+// ================================================================================================
+
+/// Per-account state the coordinator pushes to an actor on every committed block.
+///
+/// Every field is cumulative, so an actor that wakes after several blocks reads the latest view and
+/// answers entirely in memory: "did my submission land" (`last_committed_tx`), "has it expired"
+/// (`chain_tip` vs the submission block), and "is there new work" (`notes_seen` vs a local cursor).
+/// The view is intentionally bounded: a single latest-tx slot and a monotone counter, never a
+/// growing list.
+#[derive(Clone, Debug)]
+pub(crate) struct AccountView {
+    /// Chain tip as of the latest committed block. Advances every block, so the view always changes
+    /// and every actor wakes (cheaply, in memory) once per block.
+    pub chain_tip: BlockNumber,
+    /// Latest transaction committed against this account, mirroring `accounts.last_tx_id`. An actor
+    /// waiting on its submission compares this against its own transaction id to confirm landing.
+    pub last_committed_tx: Option<TransactionId>,
+    /// Monotone count of network notes seen targeting this account since the actor was spawned. A
+    /// local cursor on the actor side answers "is there new work" without a DB query.
+    pub notes_seen: u64,
+}
+
 // ACTOR HANDLE
 // ================================================================================================
 
 /// Handle to an account actor spawned by the coordinator.
-#[derive(Clone)]
 struct ActorHandle {
-    /// [`Notify`] shared with the actor. The coordinator calls [`Notify::notify_one`] when DB state
-    /// relevant to the actor may have changed, the actor awaits [`Notify::notified`] and
-    /// re-evaluates its state on wake-up.
-    notify: Arc<Notify>,
+    /// Sender half of the per-account [`AccountView`] watch channel. The coordinator updates the
+    /// view on every committed block; the actor awaits changes on its receiver and re-evaluates its
+    /// state from the pushed data rather than querying the DB.
+    view_tx: watch::Sender<AccountView>,
 }
 
 impl ActorHandle {
-    fn new(notify: Arc<Notify>) -> Self {
-        Self { notify }
-    }
-
-    /// Signals the actor that DB state may have changed. Notifications coalesce when one is already
-    /// pending.
-    fn notify(&self) {
-        self.notify.notify_one();
-    }
-
-    /// Returns `true` if a notification is queued but not yet consumed by the actor.
-    ///
-    /// Used after an actor has shut down to detect the race where a notification arrived just
-    /// as the actor timed out. If so, the coordinator should respawn the actor.
-    fn has_pending_notification(&self) -> bool {
-        use futures::FutureExt;
-        if self.notify.notified().now_or_never().is_some() {
-            // Restore the permit so the respawned actor still sees the notification.
-            self.notify.notify_one();
-            true
-        } else {
-            false
-        }
+    fn new(view_tx: watch::Sender<AccountView>) -> Self {
+        Self { view_tx }
     }
 }
 
@@ -64,18 +67,18 @@ impl ActorHandle {
 /// 1. At the catch-up boundary, to spawn one actor per account returned by
 ///    `Db::accounts_with_pending_notes()`.
 /// 2. On every committed block in steady state, via [`Coordinator::handle_committed_block`], which
-///    spawns missing actors for accounts that just received new network notes and wakes every
-///    active actor so it can re-evaluate its state from the DB.
+///    spawns missing actors for accounts that just received new network notes and pushes a fresh
+///    [`AccountView`] to every active actor so it can re-evaluate its state in memory.
 ///
 /// Actors only operate on committed account state, so spawning is restricted to accounts whose
 /// creation has been committed: a spawn requested for a not-yet-committed account is deferred
 /// until the block carrying the account's creation arrives.
 ///
-/// Notifications are coalesced through [`Notify`]: multiple wakes while an actor is busy
-/// collapse into one. Actors that crash repeatedly are deactivated after `max_account_crashes`
-/// failures.
+/// State changes are pushed through a per-account [`watch`] channel: intermediate views are
+/// coalesced (an actor busy for several blocks only ever sees the latest). Actors that crash
+/// repeatedly are deactivated after `max_account_crashes` failures.
 pub struct Coordinator {
-    /// Mapping of network account IDs to their notification handles.
+    /// Mapping of network account IDs to their view-channel handles.
     actor_registry: HashMap<AccountId, ActorHandle>,
 
     /// Join set tracking each spawned actor task; used to detect intentional shutdowns vs. crashes.
@@ -159,14 +162,20 @@ impl Coordinator {
             return;
         }
 
-        let notify = Arc::new(Notify::new());
-        let actor = AccountActor::new(account_id, &self.actor_context, notify.clone());
-        let handle = ActorHandle::new(notify);
+        let initial_view = AccountView {
+            chain_tip: self.actor_context.state.chain.chain_tip_block_number(),
+            last_committed_tx: None,
+            notes_seen: 0,
+        };
+        let (view_tx, view_rx) = watch::channel(initial_view);
+        let actor = AccountActor::new(account_id, &self.actor_context);
+        let handle = ActorHandle::new(view_tx);
 
         let semaphore = self.semaphore.clone();
         let shutdown = self.shutdown.clone();
-        self.actor_join_set
-            .spawn(Box::pin(async move { (account_id, actor.run(semaphore, shutdown).await) }));
+        self.actor_join_set.spawn(Box::pin(async move {
+            (account_id, actor.run(semaphore, view_rx, shutdown).await)
+        }));
 
         self.actor_registry.insert(account_id, handle);
         tracing::debug!(
@@ -212,8 +221,8 @@ impl Coordinator {
 
     /// Reacts to a committed block: spawns actors for any newly-targeted network accounts whose
     /// committed state exists (deferring the rest until their creation commits), releases deferred
-    /// spawns for accounts created by this block, and wakes every active actor so it can
-    /// re-evaluate its state.
+    /// spawns for accounts created by this block, and pushes a fresh [`AccountView`] to every
+    /// active actor so it can re-evaluate its state in memory.
     pub async fn handle_committed_block(
         &mut self,
         effects: &CommittedBlockEffects,
@@ -230,33 +239,51 @@ impl Coordinator {
             .iter()
             .map(AccountTargetNetworkNote::target_account_id)
             .collect();
-        for account_id in targeted {
-            self.spawn_actor_when_committed(account_id).await?;
+        for account_id in &targeted {
+            self.spawn_actor_when_committed(*account_id).await?;
         }
 
-        for handle in self.actor_registry.values() {
-            handle.notify();
+        // Push the block's effects to every active actor. The latest transaction per account is the
+        // same map `apply_committed_block` uses for `accounts.last_tx_id`, so the pushed
+        // `last_committed_tx` agrees with the persisted state; the per-account note counts feed the
+        // `notes_seen` work counter.
+        let chain_tip = effects.header.block_num();
+        let latest_tx = effects.latest_tx_per_account();
+        let mut new_notes: HashMap<AccountId, u64> = HashMap::new();
+        for note in &effects.network_notes {
+            *new_notes.entry(note.target_account_id()).or_default() += 1;
+        }
+
+        for (account_id, handle) in &self.actor_registry {
+            let committed_tx = latest_tx.get(account_id).copied();
+            let notes = new_notes.get(account_id).copied().unwrap_or(0);
+            handle.view_tx.send_modify(|view| {
+                view.chain_tip = chain_tip;
+                if let Some(tx) = committed_tx {
+                    view.last_committed_tx = Some(tx);
+                }
+                view.notes_seen += notes;
+            });
         }
         Ok(())
     }
 
     /// Waits for the next actor to complete and handles the outcome.
     ///
-    /// Returns `Some(account_id)` if an actor should be respawned (because a notification arrived
-    /// just as it shut down on idle timeout), or `None` otherwise. If no actors are currently
-    /// running, this method waits indefinitely until new actors are spawned.
+    /// Returns `Some(account_id)` if an actor should be respawned (because work reappeared for the
+    /// account between its last view observation and its idle shutdown), or `None` otherwise. If no
+    /// actors are currently running, this method waits indefinitely until new actors are spawned.
     pub async fn next(&mut self) -> anyhow::Result<Option<AccountId>> {
         let actor_result = self.actor_join_set.join_next().await;
         match actor_result {
             Some(Ok((account_id, Ok(())))) => {
-                // Actor shut down intentionally (idle timeout or account removed). Remove from
-                // registry and check if a notification arrived just as it shut down. If so, the
-                // caller should respawn it.
-                let should_respawn = self
-                    .actor_registry
-                    .remove(&account_id)
-                    .is_some_and(|handle| handle.has_pending_notification());
-
+                // Actor shut down intentionally on idle timeout, which only happens when it had no
+                // pending notes. Reap it, then respawn if a block committed between its last
+                // observation and its exit added (or re-armed) work for the account: that view
+                // update went to a now-dropped receiver and would otherwise wait for the next
+                // block.
+                self.actor_registry.remove(&account_id);
+                let should_respawn = self.account_has_pending_notes(account_id).await?;
                 Ok(should_respawn.then_some(account_id))
             },
             Some(Ok((account_id, Err(err)))) => {
@@ -284,6 +311,17 @@ impl Coordinator {
         }
     }
 
+    /// Returns `true` if the account has any pending notes: eligible now, or awaiting a backoff or
+    /// execution-hint window. Used to decide whether to respawn an actor that just idle-timed-out.
+    async fn account_has_pending_notes(&self, account_id: AccountId) -> anyhow::Result<bool> {
+        self.actor_context
+            .state
+            .db
+            .account_has_pending_notes(account_id, self.actor_context.config.max_note_attempts)
+            .await
+            .context("failed to check pending notes when reaping an idle actor")
+    }
+
     /// Waits for all currently running actors to exit after cancellation.
     pub async fn shutdown(&mut self) -> anyhow::Result<()> {
         while let Some(result) = self.actor_join_set.join_next().await {
@@ -322,15 +360,22 @@ impl Coordinator {
 
 #[cfg(test)]
 mod tests {
-    use futures::FutureExt;
-
     use super::*;
     use crate::test_utils::*;
 
-    /// Registers a dummy actor handle (no real actor task) in the coordinator's registry.
-    fn register_dummy_actor(coordinator: &mut Coordinator, account_id: AccountId) {
-        let notify = Arc::new(Notify::new());
-        coordinator.actor_registry.insert(account_id, ActorHandle::new(notify));
+    /// Registers a dummy actor handle (no real actor task) in the coordinator's registry and
+    /// returns the view receiver so the test can observe what the coordinator pushes.
+    fn register_dummy_actor(
+        coordinator: &mut Coordinator,
+        account_id: AccountId,
+    ) -> watch::Receiver<AccountView> {
+        let (view_tx, view_rx) = watch::channel(AccountView {
+            chain_tip: BlockNumber::GENESIS,
+            last_committed_tx: None,
+            notes_seen: 0,
+        });
+        coordinator.actor_registry.insert(account_id, ActorHandle::new(view_tx));
+        view_rx
     }
 
     /// Seeds a committed row for `account_id` so the coordinator's spawn check sees the account.
@@ -474,12 +519,13 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn handle_committed_block_notifies_existing_actors() {
+    async fn handle_committed_block_pushes_view_to_existing_actors() {
         let (mut coordinator, _dir, _rx) = Coordinator::test().await;
 
         let bystander = mock_network_account_id();
-        register_dummy_actor(&mut coordinator, bystander);
-        let bystander_notify = coordinator.actor_registry.get(&bystander).unwrap().notify.clone();
+        let mut bystander_rx = register_dummy_actor(&mut coordinator, bystander);
+        // Mark the initial view as seen so the post-block update is observable as a change.
+        let _ = bystander_rx.borrow_and_update();
 
         let target = mock_network_account_id_seeded(42);
         seed_committed_account(&coordinator, target).await;
@@ -495,13 +541,51 @@ mod tests {
         coordinator.handle_committed_block(&effects).await.unwrap();
 
         assert!(
-            bystander_notify.notified().now_or_never().is_some(),
-            "every registered actor should be notified on a committed block",
+            bystander_rx.has_changed().unwrap(),
+            "every registered actor should receive a view update on a committed block",
         );
+        let view = bystander_rx.borrow_and_update();
+        assert_eq!(view.chain_tip, 1_u32.into(), "the view carries the new chain tip");
+        assert_eq!(view.notes_seen, 0, "a bystander targeted by no note sees no new work");
+        drop(view);
 
         assert!(
             coordinator.actor_registry.contains_key(&target),
             "freshly-targeted account should get an actor",
         );
     }
+
+    /// The pushed view carries the account's latest committed transaction (for landing detection)
+    /// and a bumped note counter (for the work signal).
+    #[tokio::test]
+    async fn handle_committed_block_view_carries_landing_and_new_notes() {
+        let (mut coordinator, _dir, _rx) = Coordinator::test().await;
+
+        let account_id = mock_network_account_id();
+        // A dummy handle for the targeted account so the coordinator updates it in place instead of
+        // spawning a real actor (which would own the receiver and hide it from the test).
+        let mut rx = register_dummy_actor(&mut coordinator, account_id);
+        let _ = rx.borrow_and_update();
+
+        let tx_id = mock_transaction_id(5);
+        let note = mock_single_target_note(account_id, 10);
+        let effects = CommittedBlockEffects {
+            header: mock_block_header(3_u32.into()),
+            network_notes: vec![note],
+            nullifiers: vec![],
+            network_account_updates: vec![],
+            account_transactions: vec![(account_id, tx_id)],
+        };
+
+        coordinator.handle_committed_block(&effects).await.unwrap();
+
+        let view = rx.borrow_and_update();
+        assert_eq!(view.chain_tip, 3_u32.into());
+        assert_eq!(
+            view.last_committed_tx,
+            Some(tx_id),
+            "the account's latest committed tx is pushed for in-memory landing detection",
+        );
+        assert_eq!(view.notes_seen, 1, "one note targeting the account bumps the work counter");
+    }
 }