From 8c518df029eb2043b50a48f2e6dce286f578bb8d Mon Sep 17 00:00:00 2001 From: goshawk-3 Date: Thu, 22 Feb 2024 16:53:30 +0200 Subject: [PATCH] node: Instantiate Acceptor in ChainSrv::Initialize --- node/src/chain.rs | 50 ++++++++++++++++++++++++-------------- node/src/chain/acceptor.rs | 3 +-- 2 files changed, 33 insertions(+), 20 deletions(-) diff --git a/node/src/chain.rs b/node/src/chain.rs index 3932eb846a..f38365b358 100644 --- a/node/src/chain.rs +++ b/node/src/chain.rs @@ -44,38 +44,29 @@ const TOPICS: &[u8] = &[ const ACCEPT_BLOCK_TIMEOUT_SEC: Duration = Duration::from_secs(20); const HEARTBEAT_SEC: Duration = Duration::from_secs(1); -pub struct ChainSrv { +pub struct ChainSrv { /// Inbound wire messages queue inbound: AsyncQueue, keys_path: String, + acceptor: Option>>>, } #[async_trait] impl - LongLivedService for ChainSrv + LongLivedService for ChainSrv { - async fn execute( + async fn initialize( &mut self, network: Arc>, db: Arc>, vm: Arc>, - ) -> anyhow::Result { - // Register routes - LongLivedService::::add_routes( - self, - TOPICS, - self.inbound.clone(), - &network, - ) - .await?; - - // Restore/Load most recent block + ) -> anyhow::Result<()> { let mrb = Self::load_most_recent_block(db.clone(), vm.clone()).await?; let state_hash = mrb.inner().header().state_hash; let provisioners_list = vm.read().await.get_provisioners(state_hash)?; - // Initialize Acceptor and trigger consensus task + // Initialize Acceptor let acc = Acceptor::init_consensus( &self.keys_path, mrb, @@ -85,7 +76,29 @@ impl vm.clone(), ) .await?; - let acc = Arc::new(RwLock::new(acc)); + + self.acceptor = Some(Arc::new(RwLock::new(acc))); + + Ok(()) + } + + async fn execute( + &mut self, + network: Arc>, + _db: Arc>, + _vm: Arc>, + ) -> anyhow::Result { + // Register routes + LongLivedService::::add_routes( + self, + TOPICS, + self.inbound.clone(), + &network, + ) + .await?; + + let acc = self.acceptor.as_mut().unwrap(); + acc.write().await.spawn_task().await; // Start-up FSM instance let mut fsm = SimpleFSM::new(acc.clone(), network.clone()); @@ -200,11 +213,12 @@ impl } } -impl ChainSrv { +impl ChainSrv { pub fn new(keys_path: String) -> Self { Self { inbound: Default::default(), keys_path, + acceptor: None, } } @@ -213,7 +227,7 @@ impl ChainSrv { /// Panics /// /// If register entry is read but block is not found. - async fn load_most_recent_block( + async fn load_most_recent_block( db: Arc>, vm: Arc>, ) -> Result { diff --git a/node/src/chain/acceptor.rs b/node/src/chain/acceptor.rs index 59b6dbfad8..801497af29 100644 --- a/node/src/chain/acceptor.rs +++ b/node/src/chain/acceptor.rs @@ -161,11 +161,10 @@ impl Acceptor { acc.try_revert(RevertTarget::LastFinalizedState).await?; } - acc.spawn_task().await; Ok(acc) } - async fn spawn_task(&self) { + pub async fn spawn_task(&self) { let provisioners_list = self.provisioners_list.read().await.clone(); let base_timeouts = self.adjust_round_base_timeouts().await;