Actor proposal

account_actor.rs

enum AccountActorMsg {
    MempoolEvent(MempoolEvent),
    TxStarted(TransactionId),
    TxCompleted(TransactionId),
    Shutdown,
}

pub struct AccountActor {
    mode: ActorMode,
    origin: AccountOrigin,
    active: HashMap<TransactionId, MempoolEvent>,
    state: NetworkAccountState,
}

impl AccountActor {
    pub fn on_message(&mut self, msg: AccountActorMsg, ctx: ActorContext) {
        match (msg, self.mode) {
            AccountActorMsg::MempoolEvent(MempoolEvent::TransactionAdded { id, .. }) => {
                self.active.insert(id, event.clone());
                ctx.send::<ChainActor>(ChainActorMsg::ProcessTx {
                    account: self.origin.clone(),
                    id,
                });
            },

            AccountActorMsg::MempoolEvent(event) => {
                self.state.mempool_update(event);
            },

            AccountActorMsg::TxStarted(id) => {
                // TODO Log event if already processing a transaction
                self.mode = ActorMode::ProcessingTx(id);
            },

            AccountActorMsg::TxCompleted(id) => {
                if let Some(event) = self.active.get(&id) {
                    if let Some(reason) = self.state.mempool_update(event) {
                        // Failure occurred
                        ctx.stop();
                    } else {
                        self.mode = ActorMode::Idle;
                    }
                }
            },

            AccountActorMsg::Shutdown => {
                // Send message to ChainActor to cancel current transaction
            },
        }
    }
}

chain_actor.rs

pub enum ChainActorMsg {
    ProcessTx {
        account: AccountOrigin,
        id: TransactionId,
    },
    BlockNumberChanged(BlockNumber),
}

pub struct ChainActor {
    store: Arc<StoreClient>,
    chain_state: ChainState,
    block_num: BlockNumber,
    rt: tokio::runtime::Runtime,
}

impl ChainActor {
    pub fn on_message(&mut self, msg: ChainActorMsg, ctx: ActorContext) {
        match msg {
            ChainActorMsg::BlockNumberChanged(block_number) => {
                self.block_num = block_number;
            },

            ChainActorMsg::ProcessTx { account } => {
                let store = Arc::clone(&self.store);
                let block_num = self.block_num;
                let mut chain_state = self.chain_state.clone();

                // TODO Network accounts could be cached within the actor
                let account_data = self.rt.block_on(async move || match account {
                    AccountOrigin::Store(prefix) => store
                        .get_network_account(prefix)
                        .await
                        .expect("actor should be able to load account")
                        .expect("actor account should exist"),
                    AccountOrigin::Transaction(ref acc) => acc.clone(),
                });

                let store_ref = Arc::clone(&self.store);
                let origin_prefix = account_data.prefix();
                let mut state = self.rt.block_on(async move || {
                    NetworkAccountState::load(account_data, origin_prefix, &store_ref, block_num)
                        .expect("actor should be able to load account state")
                        .await
                });

                if let Some(tx_candidate) =
                    state.select_candidate(crate::MAX_NOTES_PER_TX, chain_state)
                {
                    self.rt.block_on(async move || {
                        let tx_id = state.execute_transactions(tx_candidate).await;
                        ctx.select::<AccountActor>(tx_candidate.account)
                            .send(AccountActorMsg::TxCompleted(tx_id));
                    });
                }
            },
        }
    }
}