Merge branch 'master' into revert-4457-rmasl-fix-windows-ub-corosense…

…i-stack-precommit

Cargo.toml

@@ -309,10 +309,10 @@ ethexe-runtime-common = { path = "ethexe/runtime/common", default-features = fal
 ethexe-prometheus = { path = "ethexe/prometheus", default-features = false }
 ethexe-validator = { path = "ethexe/validator", default-features = false }
 ethexe-rpc = { path = "ethexe/rpc", default-features = false }
-ethexe-common = { path = "ethexe/common" }
+ethexe-common = { path = "ethexe/common", default-features = false }
 
 # Common executor between `sandbox-host` and `lazy-pages-fuzzer`
-wasmi = { package = "wasmi", version = "0.38"}
+wasmi = { package = "wasmi", version = "0.38" }
 
 # Substrate deps
 binary-merkle-tree = { version = "15.0.1", git = "https://github.com/gear-tech/polkadot-sdk.git", branch = "gear-polkadot-stable2409", default-features = false }

ethexe/common/Cargo.toml

@@ -12,9 +12,11 @@ repository.workspace = true
 [dependencies]
 gear-core.workspace = true
 gprimitives = { workspace = true, features = ["serde"] }
-
 parity-scale-codec.workspace = true
 derive_more.workspace = true
 hex.workspace = true
 anyhow.workspace = true
 serde.workspace = true
+
+[features]
+std = ["gear-core/std", "gprimitives/serde"]

ethexe/common/src/db.rs

@@ -42,7 +42,8 @@ pub type Sum = ProgramId;
 /// NOTE: generic keys differs to Vara and have been chosen dependent on storage organization of ethexe.
 pub type ScheduledTask = gear_core::tasks::ScheduledTask<Rfm, Sd, Sum>;
 
-#[derive(Debug, Clone, Default, Encode, Decode, serde::Serialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub struct BlockHeader {
     pub height: u32,
     pub timestamp: u64,

ethexe/common/src/events/mirror.rs

@@ -20,7 +20,6 @@ use alloc::vec::Vec;
 use gear_core::message::ReplyCode;
 use gprimitives::{ActorId, MessageId, H256};
 use parity_scale_codec::{Decode, Encode};
-use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Encode, Decode, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Event {
@@ -103,7 +102,8 @@ impl Event {
     }
 }
 
-#[derive(Clone, Debug, Encode, Decode, Serialize, Deserialize)]
+#[derive(Clone, Debug, Encode, Decode)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub enum RequestEvent {
     ExecutableBalanceTopUpRequested {
         value: u128,

ethexe/common/src/events/mod.rs

@@ -18,7 +18,6 @@
 
 use gprimitives::ActorId;
 use parity_scale_codec::{Decode, Encode};
-use serde::{Deserialize, Serialize};
 
 mod mirror;
 mod router;
@@ -73,7 +72,8 @@ impl From<WVaraEvent> for BlockEvent {
     }
 }
 
-#[derive(Clone, Debug, Encode, Decode, Serialize, Deserialize)]
+#[derive(Clone, Debug, Encode, Decode)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub enum BlockRequestEvent {
     Router(RouterRequestEvent),
     Mirror {

ethexe/common/src/events/router.rs

@@ -18,7 +18,6 @@
 
 use gprimitives::{ActorId, CodeId, H256};
 use parity_scale_codec::{Decode, Encode};
-use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Encode, Decode, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Event {
@@ -72,7 +71,8 @@ impl Event {
     }
 }
 
-#[derive(Clone, Debug, Encode, Decode, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub enum RequestEvent {
     CodeValidationRequested {
         code_id: CodeId,

ethexe/common/src/events/wvara.rs

@@ -18,7 +18,6 @@
 
 use gprimitives::{ActorId, U256};
 use parity_scale_codec::{Decode, Encode};
-use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Encode, Decode, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Event {
@@ -43,7 +42,8 @@ impl Event {
     }
 }
 
-#[derive(Clone, Debug, Encode, Decode, Serialize, Deserialize)]
+#[derive(Clone, Debug, Encode, Decode)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub enum RequestEvent {
     Transfer {
         /// Never router, wvara or zero address.

ethexe/common/src/gear.rs

@@ -100,6 +100,7 @@ pub struct ComputationSettings {
 }
 
 #[derive(Clone, Debug, Default, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub struct Message {
     pub id: MessageId,
     pub destination: ActorId,
@@ -130,6 +131,7 @@ pub struct ProtocolData {
 }
 
 #[derive(Clone, Debug, Default, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub struct StateTransition {
     pub actor_id: ActorId,
     pub new_state_hash: H256,
@@ -147,6 +149,7 @@ pub struct ValidationSettings {
 }
 
 #[derive(Clone, Debug, Default, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
 pub struct ValueClaim {
     pub message_id: MessageId,
     pub destination: ActorId,

ethexe/db/src/database.rs

@@ -340,7 +340,7 @@ impl CodesStorage for Database {
 
         self.kv
             .iter_prefix(&key_prefix)
-            .map(|#[allow(unused_variables)] (key, code_id)| {
+            .map(|(key, code_id)| {
                 let (split_key_prefix, program_id) = key.split_at(key_prefix.len());
                 debug_assert_eq!(split_key_prefix, key_prefix);
                 let program_id =

ethexe/network/src/lib.rs

@@ -28,7 +28,7 @@ pub mod export {
 use anyhow::{anyhow, Context};
 use ethexe_db::Database;
 use ethexe_signer::{PublicKey, Signer};
-use futures::{future::Either, stream::FusedStream, Stream};
+use futures::{future::Either, ready, stream::FusedStream, Stream};
 use libp2p::{
     connection_limits,
     core::{muxing::StreamMuxerBox, upgrade},
@@ -121,6 +121,31 @@ pub struct NetworkService {
     swarm: Swarm<Behaviour>,
 }
 
+impl Stream for NetworkService {
+    type Item = NetworkEvent;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        loop {
+            let Some(event) = ready!(self.swarm.poll_next_unpin(cx)) else {
+                return Poll::Ready(None);
+            };
+
+            if let Some(event) = self.handle_swarm_event(event) {
+                return Poll::Ready(Some(event));
+            }
+        }
+    }
+}
+
+impl FusedStream for NetworkService {
+    fn is_terminated(&self) -> bool {
+        self.swarm.is_terminated()
+    }
+}
+
 impl NetworkService {
     pub fn new(
         config: NetworkConfig,
@@ -407,31 +432,6 @@ impl NetworkService {
     }
 }
 
-impl Stream for NetworkService {
-    type Item = NetworkEvent;
-
-    fn poll_next(
-        mut self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
-        if let Poll::Ready(Some(event)) = self.swarm.poll_next_unpin(cx) {
-            if let Some(event) = self.get_mut().handle_swarm_event(event) {
-                return Poll::Ready(Some(event));
-            } else {
-                cx.waker().wake_by_ref();
-            }
-        }
-
-        Poll::Pending
-    }
-}
-
-impl FusedStream for NetworkService {
-    fn is_terminated(&self) -> bool {
-        self.swarm.is_terminated()
-    }
-}
-
 #[cfg(test)]
 impl NetworkService {
     async fn connect(&mut self, service: &mut Self) {

ethexe/observer/src/lib.rs

@@ -27,12 +27,20 @@ use alloy::{
 use anyhow::{Context as _, Result};
 use ethexe_common::events::{BlockEvent, BlockRequestEvent, RouterEvent};
 use ethexe_db::BlockHeader;
-use ethexe_service_utils::AsyncFnStream;
 use ethexe_signer::Address;
-use futures::{future::BoxFuture, stream::FuturesUnordered, Stream, StreamExt};
+use futures::{
+    future::BoxFuture,
+    stream::{FusedStream, FuturesUnordered},
+    Stream, StreamExt,
+};
 use gprimitives::{CodeId, H256};
 use parity_scale_codec::{Decode, Encode};
-use std::{pin::Pin, sync::Arc, time::Duration};
+use std::{
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+    time::Duration,
+};
 
 pub(crate) type Provider = RootProvider<BoxTransport>;
 
@@ -77,29 +85,31 @@ pub struct ObserverService {
     codes_futures: FuturesUnordered<BlobDownloadFuture>,
 }
 
-impl AsyncFnStream for ObserverService {
+impl Stream for ObserverService {
     type Item = Result<ObserverEvent>;
 
-    async fn like_next(&mut self) -> Option<Self::Item> {
-        Some(self.next().await)
-    }
-}
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if let Poll::Ready(Some((hash, header, events))) = self.stream.poll_next_unpin(cx) {
+            let event = Ok(self.handle_stream_next(hash, header, events));
 
-// TODO: fix it by some wrapper. It's not possible to implement Stream for SequencerService like this.
-// impl Stream for ObserverService {
-//     type Item = Result<ObserverEvent>;
+            return Poll::Ready(Some(event));
+        };
 
-//     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-//         let e = ready!(pin!(self.next_event()).poll(cx));
-//         Poll::Ready(Some(e))
-//     }
-// }
+        if let Poll::Ready(Some(res)) = self.codes_futures.poll_next_unpin(cx) {
+            let event = res.map(|(code_id, code)| ObserverEvent::Blob { code_id, code });
 
-// impl FusedStream for ObserverService {
-//     fn is_terminated(&self) -> bool {
-//         false
-//     }
-// }
+            return Poll::Ready(Some(event));
+        }
+
+        Poll::Pending
+    }
+}
+
+impl FusedStream for ObserverService {
+    fn is_terminated(&self) -> bool {
+        false
+    }
+}
 
 impl ObserverService {
     pub async fn new(config: &EthereumConfig) -> Result<Self> {
@@ -171,7 +181,7 @@ impl ObserverService {
         router: Address,
     ) -> impl Stream<Item = (H256, BlockHeader, Vec<BlockEvent>)> {
         async_stream::stream! {
-            while let Some(header) = stream.like_next().await {
+            while let Some(header) = stream.next().await {
                 let hash = (*header.hash).into();
                 let parent_hash = (*header.parent_hash).into();
                 let block_number = header.number as u32;
@@ -190,31 +200,31 @@ impl ObserverService {
         }
     }
 
-    pub async fn next(&mut self) -> Result<ObserverEvent> {
-        tokio::select! {
-            Some((hash, header, events)) = self.stream.next() => {
-                // TODO (breathx): set in db?
-                log::trace!("Received block: {hash:?}");
-
-                self.last_block_number = header.height;
-
-                // TODO: replace me with proper processing of all events, including commitments.
-                for event in &events {
-                    if let BlockEvent::Router(RouterEvent::CodeValidationRequested { code_id, tx_hash }) = event {
-                        self.lookup_code(*code_id, *tx_hash);
-                    }
-                }
-
-                Ok(ObserverEvent::Block(BlockData {
-                    hash,
-                    header,
-                    events,
-                }))
-            },
-            Some(res) = self.codes_futures.next() => {
-                res.map(|(code_id, code)| ObserverEvent::Blob { code_id, code })
+    fn handle_stream_next(
+        &mut self,
+        hash: H256,
+        header: BlockHeader,
+        events: Vec<BlockEvent>,
+    ) -> ObserverEvent {
+        // TODO (breathx): set in db?
+        log::trace!("Received block: {hash:?}");
+
+        self.last_block_number = header.height;
+
+        // TODO: replace me with proper processing of all events, including commitments.
+        for event in &events {
+            if let BlockEvent::Router(RouterEvent::CodeValidationRequested { code_id, tx_hash }) =
+                event
+            {
+                self.lookup_code(*code_id, *tx_hash);
             }
         }
+
+        ObserverEvent::Block(BlockData {
+            hash,
+            header,
+            events,
+        })
     }
 }
 

ethexe/observer/src/tests.rs

@@ -92,14 +92,16 @@ async fn test_deployment() -> Result<()> {
     let event = observer
         .next()
         .await
-        .expect("observer did not receive event");
+        .expect("observer did not receive event")
+        .expect("received error instead of event");
 
     assert!(matches!(event, ObserverEvent::Block(..)));
 
     let event = observer
         .next()
         .await
-        .expect("observer did not receive event");
+        .expect("observer did not receive event")
+        .expect("received error instead of event");
 
     assert_eq!(
         event,