enable vm2 contract event notification

Author: simonjiao

Cargo.lock

@@ -5,6 +5,8 @@ starcoin-logger = { workspace = true }
 starcoin-service-registry = { workspace = true }
 starcoin-storage = { workspace = true }
 starcoin-types = { workspace = true }
+starcoin-vm2-storage = { workspace = true }
+starcoin-vm2-vm-types = { workspace = true }
 
 [package]
 authors = { workspace = true }

chain/chain-notify/Cargo.toml

@@ -3,25 +3,27 @@
 
 pub mod message;
 
-use crate::message::{ContractEventNotification, Event, Notification, ThinBlock};
+use crate::message::{ContractEventNotification, Event, Event2, Notification, ThinBlock};
 use anyhow::{format_err, Result};
 use starcoin_logger::prelude::*;
 use starcoin_service_registry::{ActorService, EventHandler, ServiceContext, ServiceFactory};
 use starcoin_storage::{Storage, Store};
 use starcoin_types::block::Block;
 use starcoin_types::system_events::NewHeadBlock;
+use starcoin_vm2_storage::{Storage as Storage2, Store as Store2};
 use std::sync::Arc;
 
 /// ChainNotify watch `NewHeadBlock` message from bus,
 /// and then reproduce `Notification<ThinBlock>` and `Notification<Arc<[Event]>>` message to bus.
 /// User can subscribe the two notification to watch onchain events.
 pub struct ChainNotifyHandlerService {
     store: Arc<dyn Store>,
+    store2: Arc<dyn Store2>,
 }
 
 impl ChainNotifyHandlerService {
-    pub fn new(store: Arc<dyn Store>) -> Self {
-        Self { store }
+    pub fn new(store: Arc<dyn Store>, store2: Arc<dyn Store2>) -> Self {
+        Self { store, store2 }
     }
 }
 
@@ -30,7 +32,8 @@ impl ServiceFactory<Self> for ChainNotifyHandlerService {
         ctx: &mut ServiceContext<ChainNotifyHandlerService>,
     ) -> Result<ChainNotifyHandlerService> {
         let storage = ctx.get_shared::<Arc<Storage>>()?;
-        Ok(Self::new(storage))
+        let storage2 = ctx.get_shared::<Arc<Storage2>>()?;
+        Ok(Self::new(storage, storage2))
     }
 }
 
@@ -58,7 +61,7 @@ impl EventHandler<Self, NewHeadBlock> for ChainNotifyHandlerService {
         self.notify_new_block(block, ctx);
 
         // notify events
-        if let Err(e) = self.notify_events(block, self.store.clone(), ctx) {
+        if let Err(e) = self.notify_events(block, self.store.clone(), self.store2.clone(), ctx) {
             error!(target: "pubsub", "fail to notify events to client, err: {}", &e);
         }
     }
@@ -77,18 +80,23 @@ impl ChainNotifyHandlerService {
         &self,
         block: &Block,
         store: Arc<dyn Store>,
+        store2: Arc<dyn Store2>,
         ctx: &mut ServiceContext<Self>,
     ) -> Result<()> {
         let block_number = block.header().number();
         let block_id = block.id();
         let txn_info_ids = store.get_block_txn_info_ids(block_id)?;
         let mut all_events: Vec<Event> = vec![];
+        let mut all_events2 = vec![];
         for txn_info_id in txn_info_ids.into_iter().rev() {
             let txn_info = store
                 .get_transaction_info(txn_info_id)?
                 .ok_or_else(|| format_err!("cannot find txn info by it's id {}", &txn_info_id))?;
             // get events directly by txn_info_id
-            let events = store.get_contract_events(txn_info_id)?.unwrap_or_default();
+            let (in_vm1, events) = store
+                .get_contract_events(txn_info_id)?
+                .map(|e| (true, e))
+                .unwrap_or((false, vec![]));
             all_events.extend(events.into_iter().enumerate().map(|(idx, evt)| {
                 Event::new(
                     block_id,
@@ -100,9 +108,26 @@ impl ChainNotifyHandlerService {
                     evt,
                 )
             }));
+            if !in_vm1 {
+                let events = store2.get_contract_events(txn_info_id)?.unwrap_or_default();
+                all_events2.extend(events.into_iter().enumerate().map(|(idx, evt)| {
+                    Event2::new(
+                        block_id,
+                        block_number,
+                        txn_info.transaction_hash(),
+                        Some(txn_info.transaction_index),
+                        Some(txn_info.transaction_global_index),
+                        Some(idx as u32),
+                        evt,
+                    )
+                }));
+            }
         }
-        let events_notification: ContractEventNotification =
-            Notification((block.header.state_root(), all_events.into()));
+        let events_notification: ContractEventNotification = Notification((
+            block.header.state_root(),
+            all_events.into(),
+            all_events2.into(),
+        ));
         ctx.broadcast(events_notification);
         Ok(())
     }

chain/chain-notify/src/lib.rs

@@ -4,12 +4,13 @@
 use starcoin_crypto::HashValue;
 use starcoin_types::block::BlockHeader;
 use starcoin_types::{block::BlockNumber, contract_event::ContractEvent};
+use starcoin_vm2_vm_types::contract_event::ContractEvent as ContractEvent2;
 use std::sync::Arc;
 
 #[derive(Debug, Clone)]
 pub struct Notification<T>(pub T);
 
-pub type ContractEventNotification = Notification<(HashValue, Arc<[Event]>)>;
+pub type ContractEventNotification = Notification<(HashValue, Arc<[Event]>, Arc<[Event2]>)>;
 pub type NewHeadEventNotification = Notification<ThinBlock>;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -46,6 +47,40 @@ impl Event {
     }
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Event2 {
+    pub block_hash: HashValue,
+    pub block_number: BlockNumber,
+    pub transaction_hash: HashValue,
+    // txn index in block
+    pub transaction_index: Option<u32>,
+    pub transaction_global_index: Option<u64>,
+    pub event_index: Option<u32>,
+    pub contract_event: ContractEvent2,
+}
+
+impl Event2 {
+    pub fn new(
+        block_hash: HashValue,
+        block_number: BlockNumber,
+        transaction_hash: HashValue,
+        transaction_index: Option<u32>,
+        transaction_global_index: Option<u64>,
+        event_index: Option<u32>,
+        contract_event: ContractEvent2,
+    ) -> Self {
+        Self {
+            block_hash,
+            block_number,
+            transaction_hash,
+            transaction_index,
+            transaction_global_index,
+            event_index,
+            contract_event,
+        }
+    }
+}
+
 /// Block with only txn hashes.
 #[derive(Debug, Clone, Hash, PartialEq, Eq)]
 pub struct ThinBlock {

chain/chain-notify/src/message.rs

@@ -31,6 +31,7 @@ starcoin-types = { workspace = true }
 starcoin-vm2-storage = { workspace = true }
 starcoin-vm2-types = { workspace = true }
 starcoin-vm2-vm-types = { workspace = true }
+starcoin-vm2-account-service = { workspace = true }
 
 [dev-dependencies]
 starcoin-network-rpc = { workspace = true }

miner/Cargo.toml

@@ -70,6 +70,8 @@ impl ServiceFactory<Self> for BlockBuilderService {
             .expect("Startup info should exist when service start.");
         //TODO support get service ref by AsyncAPI;
         let account_service = ctx.service_ref::<AccountService>()?;
+        // todo: use vm2 account as default?
+        //let _account_service = ctx.service_ref::<AccountService2>()?;
         let miner_account = block_on(async { account_service.get_default_account().await })?
             .ok_or_else(|| {
                 format_err!("Default account should exist when BlockBuilderService start.")

miner/src/create_block_template/mod.rs

@@ -29,6 +29,7 @@ starcoin-storage = { workspace = true }
 starcoin-types = { workspace = true }
 starcoin-state-tree = { workspace = true }
 starcoin-vm-types = { workspace = true }
+starcoin-vm2-state-service = { workspace = true }
 
 [dev-dependencies]
 starcoin-account-api = { workspace = true }

network-rpc/Cargo.toml

@@ -22,6 +22,7 @@ use starcoin_service_registry::{
 use starcoin_state_service::ChainStateService;
 use starcoin_storage::{Storage, Store};
 use starcoin_txpool::TxPoolService;
+use starcoin_vm2_state_service::ChainStateService as ChainStateService2;
 use std::sync::Arc;
 
 mod rpc;
@@ -84,6 +85,8 @@ impl ServiceFactory<Self> for NetworkRpcService {
         let chain_service = ctx.service_ref::<ChainReaderService>()?.clone();
         let txpool_service = ctx.get_shared::<TxPoolService>()?;
         let state_service = ctx.service_ref::<ChainStateService>()?.clone();
+        // todo: register state_service2
+        let _state_service2 = ctx.service_ref::<ChainStateService2>()?.clone();
         let node_config = ctx.get_shared::<Arc<NodeConfig>>()?;
         let quotas = node_config.network.network_rpc_quotas.clone();
         Ok(Self::new(

network-rpc/src/lib.rs

@@ -23,6 +23,7 @@ use starcoin_state_service::ChainStateService;
 use starcoin_storage::Storage;
 use starcoin_sync::sync::SyncService;
 use starcoin_txpool::TxPoolService;
+use starcoin_vm2_state_service::ChainStateService as ChainStateService2;
 use starcoin_vm2_storage::Storage as Storage2;
 use std::sync::Arc;
 
@@ -63,6 +64,7 @@ impl ServiceFactory<RpcService> for RpcServiceFactory {
             .service_ref_opt::<ChainStateService>()?
             .map(|service_ref| StateRpcImpl::new(service_ref.clone(), storage.clone()));
         let chain_state_service = ctx.service_ref::<ChainStateService>()?.clone();
+        let _chain_state_service = ctx.service_ref_opt::<ChainStateService2>()?.cloned();
         let account_service = ctx.service_ref_opt::<AccountService>()?.cloned();
         let account_api = account_service.clone().map(|service_ref| {
             AccountRpcImpl::new(

node/src/rpc_service_factory.rs

@@ -522,7 +522,7 @@ pub struct ContractEventHandler {
 
 impl EventHandler<ContractEventNotification> for ContractEventHandler {
     fn handle(&self, msg: ContractEventNotification) -> Vec<jsonrpc_core::Result<pubsub::Result>> {
-        let Notification((state_root, events)) = msg;
+        let Notification((state_root, events, _events2)) = msg;
         let filtered = events
             .as_ref()
             .iter()

rpc/server/src/module/pubsub.rs

@@ -2,6 +2,7 @@
 anyhow = { workspace = true }
 starcoin-vm2-account = { workspace = true }
 starcoin-vm2-account-api = { features = ["mock"], workspace = true }
+starcoin-chain-notify = { workspace = true }
 starcoin-config = { workspace = true }
 starcoin-vm2-crypto = { workspace = true }
 starcoin-logger = { workspace = true }

vm2/service/account/Cargo.toml

@@ -1,8 +1,8 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::message::ContractEventNotification;
 use anyhow::{Error, Result};
+use starcoin_chain_notify::message::ContractEventNotification;
 use starcoin_logger::prelude::*;
 use starcoin_service_registry::{ActorService, EventHandler, ServiceContext, ServiceFactory};
 use starcoin_vm2_account::account_storage::AccountStorage;
@@ -56,7 +56,7 @@ impl EventHandler<Self, ContractEventNotification> for AccountEventService {
             return;
         }
 
-        for i in item.0 .1.as_ref() {
+        for i in item.0 .2.as_ref() {
             let contract_event = &i.contract_event;
             if watched_keys.contains(&contract_event.event_key()) {
                 if let Err(e) = self.handle_contract_event(contract_event) {

vm2/service/account/src/account_events.rs

@@ -2,10 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 mod account_events;
-mod message;
 mod service;
 
 pub use account_events::AccountEventService;
-pub use message::{ContractEventNotification, Event};
 pub use service::AccountService;
 pub use starcoin_vm2_account::account_storage::AccountStorage;