diff --git a/benchmarks/bench_data.json b/benchmarks/bench_data.json index 897508add..26ad6022b 100644 --- a/benchmarks/bench_data.json +++ b/benchmarks/bench_data.json @@ -1,33 +1,33 @@ { "compute": { - "median": 465525077223 + "median": 465525097695 }, "alloc": { - "0": 599843943, - "12": 603255248, - "143": 763104169, - "986": 865787429, - "10945": 2216328240, - "46367": 6545787850, - "121392": 17172248387, - "317810": 43909564540 + "0": 599863562, + "12": 603274867, + "143": 763123788, + "986": 865807048, + "10945": 2216347859, + "46367": 6545807469, + "121392": 17172268006, + "317810": 43909584159 }, "counter": { - "async_call": 865267022, - "sync_call": 695201799 + "async_call": 817171163, + "sync_call": 653756647 }, "cross_program": { - "median": 2391505600 + "median": 2198136314 }, "redirect": { - "median": 3464158516 + "median": 3170024744 }, "message_stack": { - "0": 792932822, - "1": 3752847200, - "5": 15633706718, - "10": 29340040097, - "20": 65942391269 + "0": 696547455, + "1": 3357401971, + "5": 14035799870, + "10": 26225131218, + "20": 51911611286 }, "noop_baseline": { "median": 363446797 diff --git a/docs/syscall-mapping-spec.md b/docs/syscall-mapping-spec.md index 48cfd0c27..6066ea3d0 100644 --- a/docs/syscall-mapping-spec.md +++ b/docs/syscall-mapping-spec.md @@ -46,6 +46,7 @@ See `rs/src/types.rs`. | `Syscall::exit(inheritor_id)` | `!` | `gcore::exec::exit(inheritor_id)` | | | `Syscall::panic(data)` | `!` | `gcore::ext::panic(data)` | | | `Syscall::read_bytes()` | `Result, gcore::errors::Error>` | allocate `vec![0u8; gcore::msg::size()]`, then `gcore::msg::read(result.as_mut())` | | +| `Syscall::wake(message_id)` | `Result<(), gcore::errors::Error>` | `gcore::exec::wake(message_id)` | | | `Syscall::system_reserve_gas(amount)` | `Result<(), gcore::errors::Error>` | `gcore::exec::system_reserve_gas(amount)` | only when `ethexe` is disabled | ## Behavioral Notes @@ -58,7 +59,7 @@ The `message_*`, `reply_*`, and `signal_*` methods are wrappers over `gcore::msg ### Execution Context -The `program_id`, `block_height`, `block_timestamp`, `value_available`, `gas_available`, `env_vars`, `exit`, and `system_reserve_gas` methods are wrappers over `gcore::exec::*` accessors and control flow. +The `program_id`, `block_height`, `block_timestamp`, `value_available`, `gas_available`, `env_vars`, `exit`, `wake`, and `system_reserve_gas` methods are wrappers over `gcore::exec::*` accessors and control flow. ### Panic Surface @@ -75,6 +76,7 @@ On non-`wasm32` targets with feature `std`: - `env_vars()` returns a constructed `gcore::EnvVars` value rather than delegating to `gcore::exec::env_vars()`. - `exit()` and `panic()` call Rust `panic!` with diagnostic text instead of delegating to runtime syscalls. - `read_bytes()` reads from thread-local mock state. +- `wake()` is a no-op that returns `Ok(())` (it takes a `MessageId` argument, so it is a manual mock rather than a generated getter). - `system_reserve_gas()` returns `Ok(())` when present, that is, when `ethexe` is disabled. These behaviors are intentionally test-oriented and are not part of the `gcore` mapping defined above. diff --git a/examples/demo/app/src/chaos/mod.rs b/examples/demo/app/src/chaos/mod.rs index d95303c50..f22e1ebc1 100644 --- a/examples/demo/app/src/chaos/mod.rs +++ b/examples/demo/app/src/chaos/mod.rs @@ -1,7 +1,8 @@ -use sails_rs::gstd::debug; +use sails_rs::gstd::{Lock, debug}; use sails_rs::{gstd, prelude::*}; static mut REPLY_HOOK_COUNTER: u32 = 0; +static mut CRITICAL_HOOK_COUNTER: u32 = 0; pub struct ChaosService; @@ -10,9 +11,7 @@ impl ChaosService { #[export] pub async fn panic_after_wait(&self) { let source = Syscall::message_source(); - let _ = gstd::msg::send_for_reply::<()>(source, (), 0, 0) - .unwrap() - .await; + let _ = gstd::send_for_reply::<()>(source, (), 0).unwrap().await; debug!("Message received, now panicking!"); panic!("Simulated panic after wait"); } @@ -22,16 +21,19 @@ impl ChaosService { let source = Syscall::message_source(); debug!("before handle_reply"); - let fut = gstd::msg::send_for_reply::<()>(source, (), 0, 10_000_000_000).unwrap(); - let fut = fut - .handle_reply(|| { + let fut = gstd::send_bytes_for_reply( + source, + &[], + 0, + Lock::up_to(1), + None, + Some(10_000_000_000), + Some(Box::new(|| { unsafe { REPLY_HOOK_COUNTER += 1 }; debug!("handle_reply triggered"); - }) - .unwrap() - .up_to(Some(1)) - .unwrap(); - + })), + ) + .unwrap(); let _ = fut.await; debug!("after handle_reply"); } @@ -40,4 +42,56 @@ impl ChaosService { pub fn reply_hook_counter(&self) -> u32 { unsafe { REPLY_HOOK_COUNTER } } + + /// Suspends the message for `blocks` blocks via `sleep_for`, then returns + /// the actual block delta. Used to verify the `MessageSleepFuture` lifecycle: + /// register sleep lock -> `exec::wait_for` -> resume at deadline -> complete. + #[export] + pub async fn sleep_then_return(&self, blocks: u32) -> u32 { + let start = Syscall::block_height(); + gstd::sleep_for(blocks).await; + Syscall::block_height().saturating_sub(start) + } + + /// Sends two messages for reply concurrently and resolves with the first to + /// reply via `futures::select`. Exercises `next_lock` picking the earliest + /// of several armed locks and `forget_future` reclaiming the losing branch + /// (its `MessageFuture` is dropped unresolved). Returns `0` if the first + /// send (`b"A"`) wins, `1` if the second (`b"B"`) does. + #[export] + pub async fn select_first_reply(&self) -> u8 { + use sails_rs::futures::future::{Either, select}; + + let source = Syscall::message_source(); + let a = gstd::send_bytes_for_reply(source, b"A", 0, Lock::up_to(100), None, None, None) + .unwrap(); + let b = gstd::send_bytes_for_reply(source, b"B", 0, Lock::up_to(100), None, None, None) + .unwrap(); + + match select(a, b).await { + Either::Left(_) => 0, + Either::Right(_) => 1, + } + } + + /// Registers a critical hook, awaits a reply, then panics after resuming. + /// The trap on a message that reserved system gas makes the runtime invoke + /// `handle_signal`, which runs the stored critical hook. Used to verify the + /// `set_critical_hook` -> `handle_signal` path (distinct from the userspace + /// error reply). Observe the effect via [`Self::critical_hook_counter`]. + #[export] + pub async fn critical_hook_on_signal(&self) { + gstd::set_critical_hook(|_msg_id| { + unsafe { CRITICAL_HOOK_COUNTER += 1 }; + debug!("critical hook fired in handle_signal"); + }); + let source = Syscall::message_source(); + let _ = gstd::send_for_reply::<()>(source, (), 0).unwrap().await; + panic!("panic after wait to trigger a signal"); + } + + #[export] + pub fn critical_hook_counter(&self) -> u32 { + unsafe { CRITICAL_HOOK_COUNTER } + } } diff --git a/examples/demo/app/tests/gtest.rs b/examples/demo/app/tests/gtest.rs index 8a9a0ecab..a774df4e5 100644 --- a/examples/demo/app/tests/gtest.rs +++ b/examples/demo/app/tests/gtest.rs @@ -651,6 +651,147 @@ fn chaos_service_timeout_wait() { ); } +#[test] +fn chaos_service_sleep_for_works() { + use demo_client::{chaos::io::SleepThenReturn, io::Default}; + use sails_rs::gtest::{Program, System}; + + let system = System::new(); + system.init_logger_with_default_filter("gwasm=debug,gtest=info,sails_rs=debug"); + system.mint_to(ACTOR_ID, 1_000_000_000_000_000); + let program = Program::from_file(&system, DEMO_WASM_PATH); + program.send_bytes(ACTOR_ID, Default::encode_call(0)); + system.run_next_block(); + + const SLEEP_BLOCKS: u32 = 3; + let msg_id = program.send_bytes( + ACTOR_ID, + SleepThenReturn::encode_call(DemoClientProgram::ROUTE_ID_CHAOS, SLEEP_BLOCKS), + ); + + // First block sends the message and suspends via sleep_for(SLEEP_BLOCKS), + // then the runtime must wake the task at the deadline. Scan a few blocks + // beyond the deadline; the reply lands in whichever block the wake occurs. + let mut reply_payload: Option> = None; + for _ in 0..(SLEEP_BLOCKS + 3) { + let run = system.run_next_block(); + if let Some(payload) = run + .log() + .iter() + .find(|log| log.reply_to() == Some(msg_id)) + .map(|log| log.payload().to_vec()) + { + reply_payload = Some(payload); + break; + } + } + let payload = reply_payload.expect("sleep_then_return reply not produced"); + let elapsed = + SleepThenReturn::decode_reply(DemoClientProgram::ROUTE_ID_CHAOS, payload).unwrap(); + assert!( + elapsed >= SLEEP_BLOCKS, + "sleep_for should suspend at least {SLEEP_BLOCKS} blocks, got {elapsed}" + ); +} + +#[test] +fn chaos_service_select_first_reply() { + use demo_client::{chaos::io::SelectFirstReply, io::Default}; + use sails_rs::gtest::{Log, Program, System}; + + let system = System::new(); + system.init_logger_with_default_filter("gwasm=debug,gtest=info,sails_rs=debug"); + system.mint_to(ACTOR_ID, 1_000_000_000_000_000); + let program = Program::from_file(&system, DEMO_WASM_PATH); + program.send_bytes(ACTOR_ID, Default::encode_call(0)); + system.run_next_block(); + + let msg_id = program.send_bytes( + ACTOR_ID, + SelectFirstReply::encode_call(DemoClientProgram::ROUTE_ID_CHAOS), + ); + // First block: the method sends b"A" and b"B" concurrently, then suspends + // with two armed locks. + system.run_next_block(); + + // Reply only to the first send (b"A"); the second future is dropped + // unresolved when `select` resolves Left, exercising `forget_future`. + let log = Log::builder().source(program.id()).dest(ACTOR_ID); + system + .get_mailbox(ACTOR_ID) + .reply_bytes(log.payload_bytes(b"A"), vec![], 0) + .unwrap(); + let run = system.run_next_block(); + + let payload = run + .log() + .iter() + .find(|log| log.reply_to() == Some(msg_id)) + .map(|log| log.payload().to_vec()) + .expect("select_first_reply did not produce a reply"); + let winner = + SelectFirstReply::decode_reply(DemoClientProgram::ROUTE_ID_CHAOS, payload).unwrap(); + assert_eq!(winner, 0, "the first send (b\"A\") must win the select"); +} + +#[test] +fn chaos_service_critical_hook_on_signal() { + use demo_client::{ + chaos::io::{CriticalHookCounter, CriticalHookOnSignal}, + io::Default, + }; + use sails_rs::gtest::{Log, Program, System}; + + let system = System::new(); + system.init_logger_with_default_filter("gwasm=debug,gtest=info,sails_rs=debug"); + system.mint_to(ACTOR_ID, 1_000_000_000_000_000); + let program = Program::from_file(&system, DEMO_WASM_PATH); + program.send_bytes(ACTOR_ID, Default::encode_call(0)); + system.run_next_block(); + + let read_counter = |system: &System, program: &Program| { + let msg_id = program.send_bytes( + ACTOR_ID, + CriticalHookCounter::encode_call(DemoClientProgram::ROUTE_ID_CHAOS), + ); + let run = system.run_next_block(); + let payload = run + .log() + .iter() + .find(|log| log.reply_to() == Some(msg_id)) + .map(|log| log.payload().to_vec()) + .expect("counter reply not found"); + CriticalHookCounter::decode_reply(DemoClientProgram::ROUTE_ID_CHAOS, payload).unwrap() + }; + + assert_eq!(read_counter(&system, &program), 0, "hook must not fire yet"); + + program.send_bytes( + ACTOR_ID, + CriticalHookOnSignal::encode_call(DemoClientProgram::ROUTE_ID_CHAOS), + ); + // Sends the inner message (reserving system gas) and suspends. + system.run_next_block(); + + // Reply so the method resumes and panics; the trap on a message that + // reserved system gas drives the runtime into `handle_signal`, which runs + // the stored critical hook. + let log = Log::builder().source(program.id()).dest(ACTOR_ID); + system + .get_mailbox(ACTOR_ID) + .reply_bytes(log.payload_bytes(().encode()), vec![], 0) + .unwrap(); + system.run_next_block(); + // Signal processing may land in the following block. + system.run_next_block(); + + assert_eq!( + read_counter(&system, &program), + 1, + "critical hook must fire exactly once via handle_signal" + ); +} + #[tokio::test] async fn chaos_panic_does_not_affect_other_services() { use demo_client::chaos::Chaos as _; diff --git a/examples/demo/client/demo_client.idl b/examples/demo/client/demo_client.idl index 1b5440a37..8d32f91aa 100644 --- a/examples/demo/client/demo_client.idl +++ b/examples/demo/client/demo_client.idl @@ -148,12 +148,33 @@ service Validator@0x4e78bafffdb4fb1c { } } -service Chaos@0xf0c8c80dfabf72d5 { +service Chaos@0x6412b1f7e47f892d { functions { + @query + CriticalHookCounter() -> u32; + /// Registers a critical hook, awaits a reply, then panics after resuming. + /// The trap on a message that reserved system gas makes the runtime invoke + /// `handle_signal`, which runs the stored critical hook. Used to verify the + /// `set_critical_hook` -> `handle_signal` path (distinct from the userspace + /// error reply). Observe the effect via [`Self::critical_hook_counter`]. + @query + CriticalHookOnSignal(); @query PanicAfterWait(); @query ReplyHookCounter() -> u32; + /// Sends two messages for reply concurrently and resolves with the first to + /// reply via `futures::select`. Exercises `next_lock` picking the earliest + /// of several armed locks and `forget_future` reclaiming the losing branch + /// (its `MessageFuture` is dropped unresolved). Returns `0` if the first + /// send (`b"A"`) wins, `1` if the second (`b"B"`) does. + @query + SelectFirstReply() -> u8; + /// Suspends the message for `blocks` blocks via `sleep_for`, then returns + /// the actual block delta. Used to verify the `MessageSleepFuture` lifecycle: + /// register sleep lock -> `exec::wait_for` -> resume at deadline -> complete. + @query + SleepThenReturn(blocks: u32) -> u32; @query TimeoutWait(); } @@ -198,7 +219,7 @@ program DemoClient { ThisThat@0x381e13fdd02d675f, ValueFee@0x61261a86528bf9d5, Validator@0x4e78bafffdb4fb1c, - Chaos@0xf0c8c80dfabf72d5, + Chaos@0x6412b1f7e47f892d, Chain@0x01fcbe183e2199b0, OverrideGenerics@0xa33febc87c18925b, } diff --git a/examples/demo/client/src/demo_client.rs b/examples/demo/client/src/demo_client.rs index 7c7e47e96..65620f45f 100644 --- a/examples/demo/client/src/demo_client.rs +++ b/examples/demo/client/src/demo_client.rs @@ -895,10 +895,36 @@ pub mod chaos { pub trait Chaos { type Env: sails_rs::client::GearEnv; + fn critical_hook_counter( + &self, + ) -> sails_rs::client::PendingCall; + /// Registers a critical hook, awaits a reply, then panics after resuming. + /// The trap on a message that reserved system gas makes the runtime invoke + /// `handle_signal`, which runs the stored critical hook. Used to verify the + /// `set_critical_hook` -> `handle_signal` path (distinct from the userspace + /// error reply). Observe the effect via [`Self::critical_hook_counter`]. + fn critical_hook_on_signal( + &self, + ) -> sails_rs::client::PendingCall; fn panic_after_wait(&self) -> sails_rs::client::PendingCall; fn reply_hook_counter( &self, ) -> sails_rs::client::PendingCall; + /// Sends two messages for reply concurrently and resolves with the first to + /// reply via `futures::select`. Exercises `next_lock` picking the earliest + /// of several armed locks and `forget_future` reclaiming the losing branch + /// (its `MessageFuture` is dropped unresolved). Returns `0` if the first + /// send (`b"A"`) wins, `1` if the second (`b"B"`) does. + fn select_first_reply( + &self, + ) -> sails_rs::client::PendingCall; + /// Suspends the message for `blocks` blocks via `sleep_for`, then returns + /// the actual block delta. Used to verify the `MessageSleepFuture` lifecycle: + /// register sleep lock -> `exec::wait_for` -> resume at deadline -> complete. + fn sleep_then_return( + &self, + blocks: u32, + ) -> sails_rs::client::PendingCall; fn timeout_wait(&self) -> sails_rs::client::PendingCall; } @@ -906,11 +932,21 @@ pub mod chaos { impl sails_rs::client::Identifiable for ChaosImpl { const INTERFACE_ID: sails_rs::InterfaceId = - sails_rs::InterfaceId::from_bytes_8([240, 200, 200, 13, 250, 191, 114, 213]); + sails_rs::InterfaceId::from_bytes_8([100, 18, 177, 247, 228, 127, 137, 45]); } impl Chaos for sails_rs::client::Service { type Env = E; + fn critical_hook_counter( + &self, + ) -> sails_rs::client::PendingCall { + self.pending_call(()) + } + fn critical_hook_on_signal( + &self, + ) -> sails_rs::client::PendingCall { + self.pending_call(()) + } fn panic_after_wait(&self) -> sails_rs::client::PendingCall { self.pending_call(()) } @@ -919,6 +955,17 @@ pub mod chaos { ) -> sails_rs::client::PendingCall { self.pending_call(()) } + fn select_first_reply( + &self, + ) -> sails_rs::client::PendingCall { + self.pending_call(()) + } + fn sleep_then_return( + &self, + blocks: u32, + ) -> sails_rs::client::PendingCall { + self.pending_call((blocks,)) + } fn timeout_wait(&self) -> sails_rs::client::PendingCall { self.pending_call(()) } @@ -926,9 +973,13 @@ pub mod chaos { pub mod io { use super::*; - sails_rs::io_struct_impl!(PanicAfterWait () -> (), 0, ::INTERFACE_ID); - sails_rs::io_struct_impl!(ReplyHookCounter () -> u32, 1, ::INTERFACE_ID); - sails_rs::io_struct_impl!(TimeoutWait () -> (), 2, ::INTERFACE_ID); + sails_rs::io_struct_impl!(CriticalHookCounter () -> u32, 0, ::INTERFACE_ID); + sails_rs::io_struct_impl!(CriticalHookOnSignal () -> (), 1, ::INTERFACE_ID); + sails_rs::io_struct_impl!(PanicAfterWait () -> (), 2, ::INTERFACE_ID); + sails_rs::io_struct_impl!(ReplyHookCounter () -> u32, 3, ::INTERFACE_ID); + sails_rs::io_struct_impl!(SelectFirstReply () -> u8, 4, ::INTERFACE_ID); + sails_rs::io_struct_impl!(SleepThenReturn (blocks: u32) -> u32, 5, ::INTERFACE_ID); + sails_rs::io_struct_impl!(TimeoutWait () -> (), 6, ::INTERFACE_ID); } #[cfg(feature = "with_mocks")] @@ -943,7 +994,7 @@ pub mod chaos { #[allow(clippy::type_complexity)] impl chaos::Chaos for Chaos { type Env = sails_rs::client::GstdEnv; - fn panic_after_wait (&self, ) -> sails_rs::client::PendingCall;fn reply_hook_counter (&self, ) -> sails_rs::client::PendingCall;fn timeout_wait (&self, ) -> sails_rs::client::PendingCall; + fn critical_hook_counter (&self, ) -> sails_rs::client::PendingCall;fn critical_hook_on_signal (&self, ) -> sails_rs::client::PendingCall;fn panic_after_wait (&self, ) -> sails_rs::client::PendingCall;fn reply_hook_counter (&self, ) -> sails_rs::client::PendingCall;fn select_first_reply (&self, ) -> sails_rs::client::PendingCall;fn sleep_then_return (&self, blocks: u32) -> sails_rs::client::PendingCall;fn timeout_wait (&self, ) -> sails_rs::client::PendingCall; } } } diff --git a/examples/event-routes/app/src/lib.rs b/examples/event-routes/app/src/lib.rs index 162a7ff4a..b85b57ecd 100644 --- a/examples/event-routes/app/src/lib.rs +++ b/examples/event-routes/app/src/lib.rs @@ -28,9 +28,7 @@ impl Service { pub async fn foo(&mut self) { let source = Syscall::message_source(); self.emit_event(Events::Start).unwrap(); - let _res = gstd::msg::send_for_reply(source, self.0, 0, 0) - .unwrap() - .await; + let _res = gstd::send_for_reply(source, self.0, 0).unwrap().await; self.emit_event(Events::End).unwrap(); } } diff --git a/rs/Cargo.toml b/rs/Cargo.toml index c9f5d2104..5da772267 100644 --- a/rs/Cargo.toml +++ b/rs/Cargo.toml @@ -49,7 +49,7 @@ log = { workspace = true, optional = true } tokio = { workspace = true, features = ["rt", "macros"] } [features] -default = ["gstd"] +default = ["gstd", "async-runtime"] build = ["client-builder", "wasm-builder"] debug = ["gstd?/debug"] ethexe = [ @@ -73,3 +73,4 @@ idl-embed = ["dep:sails-idl-embed"] mockall = ["std", "dep:mockall"] std = ["futures/std", "sails-idl-gen?/std", "gear-core?/std"] wasm-builder = ["dep:gwasm-builder"] +async-runtime = ["gstd"] diff --git a/rs/src/client/gstd_env.rs b/rs/src/client/gstd_env.rs index ee70d64f9..7dedf7fb3 100644 --- a/rs/src/client/gstd_env.rs +++ b/rs/src/client/gstd_env.rs @@ -1,32 +1,39 @@ use super::*; -use ::gstd::{ - errors::Error, - msg::{CreateProgramFuture, MessageFuture}, -}; +use crate::gstd::{CreateProgramFuture, Lock, MessageFuture}; +use ::gstd::errors::Error; #[derive(Default)] pub struct GstdParams { + pub value: Option, + pub wait: Option, + pub redirect_on_exit: bool, #[cfg(not(feature = "ethexe"))] pub gas_limit: Option, - pub value: Option, - pub wait_up_to: Option, #[cfg(not(feature = "ethexe"))] pub reply_deposit: Option, #[cfg(not(feature = "ethexe"))] - pub reply_hook: Option>, - pub redirect_on_exit: bool, + pub reply_hook: Option>, } crate::params_for_pending_impl!(GstdEnv, GstdParams { #[cfg(not(feature = "ethexe"))] pub gas_limit: GasUnit, pub value: ValueUnit, - pub wait_up_to: BlockCount, + pub wait: Lock, #[cfg(not(feature = "ethexe"))] pub reply_deposit: GasUnit, }); impl GstdParams { + /// Wait *up to* `block_count` blocks for a reply. + /// + /// Convenience over [`Self::with_wait`]; equivalent to + /// `with_wait(Lock::up_to(block_count))`. The internal [`Lock`] stores an + /// absolute deadline, so the timeout survives a redirect unchanged. + pub fn with_wait_up_to(self, block_count: BlockCount) -> Self { + self.with_wait(Lock::up_to(block_count)) + } + pub fn with_redirect_on_exit(self, redirect_on_exit: bool) -> Self { Self { redirect_on_exit, @@ -35,7 +42,7 @@ impl GstdParams { } #[cfg(not(feature = "ethexe"))] - pub fn with_reply_hook(self, f: F) -> Self { + pub fn with_reply_hook(self, f: F) -> Self { Self { reply_hook: Some(Box::new(f)), ..self @@ -44,6 +51,11 @@ impl GstdParams { } impl PendingCall { + /// Wait *up to* `block_count` blocks for a reply. See [`GstdParams::with_wait_up_to`]. + pub fn with_wait_up_to(self, block_count: BlockCount) -> Self { + self.with_params(|params| params.with_wait_up_to(block_count)) + } + /// Set `redirect_on_exit` flag to `true`` /// /// This flag is used to redirect a message to a new program when the target program exits @@ -58,11 +70,18 @@ impl PendingCall { } #[cfg(not(feature = "ethexe"))] - pub fn with_reply_hook(self, f: F) -> Self { + pub fn with_reply_hook(self, f: F) -> Self { self.with_params(|params| params.with_reply_hook(f)) } } +impl PendingCtor { + /// Wait *up to* `block_count` blocks for a reply. See [`GstdParams::with_wait_up_to`]. + pub fn with_wait_up_to(self, block_count: BlockCount) -> Self { + self.with_params(|params| params.with_wait_up_to(block_count)) + } +} + #[derive(Debug, Default, Clone)] pub struct GstdEnv; @@ -92,30 +111,26 @@ impl ReplyError for Error { } impl GstdEnv { + #[cfg_attr(feature = "ethexe", allow(unused_mut))] pub fn send_one_way( &self, destination: ActorId, payload: impl AsRef<[u8]>, - params: GstdParams, + mut params: GstdParams, ) -> Result { - let value = params.value.unwrap_or_default(); - let payload_bytes = payload.as_ref(); - - #[cfg(not(feature = "ethexe"))] - let waiting_reply_to = if let Some(gas_limit) = params.gas_limit { - ::gcore::msg::send_with_gas(destination, payload_bytes, gas_limit, value)? - } else { - ::gcore::msg::send(destination, payload_bytes, value)? - }; - #[cfg(feature = "ethexe")] - let waiting_reply_to = ::gcore::msg::send(destination, payload_bytes, value)?; - - #[cfg(not(feature = "ethexe"))] - if let Some(reply_deposit) = params.reply_deposit { - ::gcore::exec::reply_deposit(waiting_reply_to, reply_deposit)?; - } - - Ok(waiting_reply_to) + let reply_to = crate::ok!(crate::gstd::send_one_way( + destination, + payload.as_ref(), + params.value.unwrap_or_default(), + #[cfg(not(feature = "ethexe"))] + params.gas_limit, + #[cfg(not(feature = "ethexe"))] + params.reply_deposit, + #[cfg(not(feature = "ethexe"))] + params.reply_hook.take(), + )); + + Ok(reply_to) } } @@ -129,52 +144,7 @@ impl PendingCall { #[cfg(target_arch = "wasm32")] const _: () = { use core::task::ready; - - #[cfg(not(feature = "ethexe"))] - #[inline] - fn send_for_reply_future( - destination: ActorId, - payload: &[u8], - params: &mut GstdParams, - ) -> Result { - let value = params.value.unwrap_or_default(); - let reply_deposit = params.reply_deposit.unwrap_or_default(); - // here can be a redirect target - let mut message_future = if let Some(gas_limit) = params.gas_limit { - ::gstd::msg::send_bytes_with_gas_for_reply( - destination, - payload, - gas_limit, - value, - reply_deposit, - )? - } else { - ::gstd::msg::send_bytes_for_reply(destination, payload, value, reply_deposit)? - }; - - message_future = message_future.up_to(params.wait_up_to)?; - - if let Some(reply_hook) = params.reply_hook.take() { - message_future = message_future.handle_reply(reply_hook)?; - } - Ok(message_future) - } - - #[cfg(feature = "ethexe")] - #[inline] - fn send_for_reply_future( - destination: ActorId, - payload: &[u8], - params: &mut GstdParams, - ) -> Result { - let value = params.value.unwrap_or_default(); - // here can be a redirect target - let mut message_future = ::gstd::msg::send_bytes_for_reply(destination, payload, value)?; - - message_future = message_future.up_to(params.wait_up_to)?; - - Ok(message_future) - } + use futures::future::FusedFuture; #[inline] fn send_for_reply( @@ -182,12 +152,20 @@ const _: () = { payload: Vec, params: &mut GstdParams, ) -> Result { - // send message - let future = send_for_reply_future(destination, payload.as_ref(), params)?; + let future = crate::ok!(crate::gstd::send_bytes_for_reply( + destination, + payload.as_ref(), + params.value.unwrap_or_default(), + params.wait.unwrap_or_default(), + #[cfg(not(feature = "ethexe"))] + params.gas_limit, + #[cfg(not(feature = "ethexe"))] + params.reply_deposit, + #[cfg(not(feature = "ethexe"))] + params.reply_hook.take(), + )); if params.redirect_on_exit { - let created_block = params.wait_up_to.map(|_| gstd::exec::block_height()); Ok(GstdFuture::MessageWithRedirect { - created_block, future, destination, payload, @@ -197,6 +175,28 @@ const _: () = { } } + fn create_program( + code_id: CodeId, + salt: impl AsRef<[u8]>, + payload: impl AsRef<[u8]>, + params: &mut GstdParams, + ) -> Result<(GstdFuture, ActorId), Error> { + let (future, program_id) = crate::ok!(crate::gstd::create_program_for_reply( + code_id, + salt.as_ref(), + payload.as_ref(), + params.value.unwrap_or_default(), + params.wait.unwrap_or_default(), + #[cfg(not(feature = "ethexe"))] + params.gas_limit, + #[cfg(not(feature = "ethexe"))] + params.reply_deposit, + #[cfg(not(feature = "ethexe"))] + params.reply_hook.take(), + )); + Ok((GstdFuture::CreateProgram { future }, program_id)) + } + impl PendingCall { /// Sends the message and returns the `PendingCall` for subsequent `poll`/`await`. pub fn send_for_reply(mut self) -> Result { @@ -219,6 +219,7 @@ const _: () = { impl Future for PendingCall { type Output = Result::Error>; + #[inline(always)] fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { if self.state.is_none() { let args = self @@ -252,7 +253,6 @@ const _: () = { let params = this.params.get_or_insert_default(); if let Replace::MessageWithRedirect { destination: _destination, - created_block, payload, .. } = state.as_mut().project_replace(GstdFuture::Dummy) @@ -261,16 +261,6 @@ const _: () = { { gstd::debug!("Redirecting message from {_destination} to {new_target}"); - // Calculate updated `wait_up_to` if provided - // wait_up_to = wait_up_to - (current_block - created_block) - params.wait_up_to = params.wait_up_to.and_then(|wait_up_to| { - created_block.map(|created_block| { - let current_block = gstd::exec::block_height(); - wait_up_to - .saturating_sub(current_block.saturating_sub(created_block)) - }) - }); - // send message to new target let future = send_for_reply(new_target, payload, params)?; // Replace the future with a new one @@ -283,7 +273,8 @@ const _: () = { ErrorReplyReason::UnavailableActor( SimpleUnavailableActorError::ProgramExited, ), - ))) + ) + .into())) } } output => Poll::Ready(decode_reply_or_throw::(this.route, output)), @@ -291,6 +282,20 @@ const _: () = { } } + impl FusedFuture for PendingCall { + fn is_terminated(&self) -> bool { + self.state + .as_ref() + .map(|future| match future { + GstdFuture::CreateProgram { future } => future.is_terminated(), + GstdFuture::Message { future } => future.is_terminated(), + GstdFuture::MessageWithRedirect { future, .. } => future.is_terminated(), + GstdFuture::Dummy => false, + }) + .unwrap_or_default() + } + } + impl Future for PendingCtor where T: ServiceCall, @@ -303,8 +308,7 @@ const _: () = { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { if self.state.is_none() { - let params = self.params.take().unwrap_or_default(); - let value = params.value.unwrap_or_default(); + let mut params = self.params.take().unwrap_or_default(); let salt = self.salt.take().unwrap(); let args = self @@ -313,35 +317,14 @@ const _: () = { .unwrap_or_else(|| panic!("{PENDING_CALL_INVALID_STATE}")); let payload = T::encode_call(&self.route, args); // Send message - #[cfg(not(feature = "ethexe"))] - let future = if let Some(gas_limit) = params.gas_limit { - ::gstd::prog::create_program_bytes_with_gas_for_reply( - self.code_id, - salt, - payload, - gas_limit, - value, - params.reply_deposit.unwrap_or_default(), - )? - } else { - ::gstd::prog::create_program_bytes_for_reply( - self.code_id, - salt, - payload, - value, - params.reply_deposit.unwrap_or_default(), - )? - }; - #[cfg(feature = "ethexe")] - let future = ::gstd::prog::create_program_bytes_for_reply( - self.code_id, - salt, - payload, - value, - )?; - - // self.program_id = Some(program_future.program_id); - self.state = Some(GstdFuture::CreateProgram { future }); + let (future, program_id) = + match create_program(self.code_id, salt, payload, &mut params) { + Ok(res) => res, + Err(err) => return Poll::Ready(Err(err)), + }; + + self.program_id = Some(program_id); + self.state = Some(future); // No need to poll the future return Poll::Pending; } @@ -349,10 +332,8 @@ const _: () = { // SAFETY: checked in the code above. let state = unsafe { this.state.as_pin_mut().unwrap_unchecked() }; if let Projection::CreateProgram { future } = state.project() { - let (reply, program_id) = match ready!(future.poll(cx)) { - Ok((program_id, payload)) => (Ok(payload), program_id), - Err(err) => (Err(err), ActorId::zero()), - }; + let reply = ready!(future.poll(cx)); + let program_id = unsafe { this.program_id.unwrap_unchecked() }; match decode_reply_or_throw::(this.route, reply) { Ok(output) => Poll::Ready(Ok(output.map_result(this.env.clone(), program_id))), Err(err) => Poll::Ready(Err(err)), @@ -374,7 +355,6 @@ pin_project_lite::pin_project! { #[pin] future: MessageFuture, destination: ActorId, - created_block: Option, payload: Vec, // reuse encoded payload when redirecting }, Dummy, diff --git a/rs/src/gstd/async_runtime.rs b/rs/src/gstd/async_runtime.rs new file mode 100644 index 000000000..0cdd4220f --- /dev/null +++ b/rs/src/gstd/async_runtime.rs @@ -0,0 +1,1807 @@ +use super::*; +use crate::collections::{BinaryHeap, HashMap}; +use core::{ + cmp::Reverse, + hash::{BuildHasherDefault, Hasher}, + pin::Pin, + task::{Context, Poll}, +}; +use futures::future::{FusedFuture, FutureExt as _, LocalBoxFuture}; +use gstd::{BlockNumber, errors::Error}; + +/// `gstd::debug!` only emits in the wasm runtime; on host (e.g. unit tests) it +/// is a no-op so the debug syscall is never reached. +#[cfg(target_arch = "wasm32")] +macro_rules! debug { + ($($arg:tt)*) => { ::gstd::debug!($($arg)*) }; +} +#[cfg(not(target_arch = "wasm32"))] +macro_rules! debug { + ($($arg:tt)*) => {}; +} + +/// Identity-hasher for `MessageId`. `MessageId` is itself a 32-byte +/// cryptographic hash, so the first 8 bytes are already uniformly +/// distributed — running them through a general-purpose hash function is +/// wasted gas. Saves the 32-byte mixing on every HashMap op. +#[derive(Default)] +struct IdHasher(u64); + +impl Hasher for IdHasher { + #[inline] + fn write(&mut self, bytes: &[u8]) { + // MessageId encode_to writes 32 bytes; we read the first 8. + let n = bytes.len().min(8); + let mut buf = [0u8; 8]; + buf[..n].copy_from_slice(&bytes[..n]); + self.0 = u64::from_ne_bytes(buf); + } + + #[inline] + fn finish(&self) -> u64 { + self.0 + } +} + +type IdMap = HashMap>; + +fn tasks() -> &'static mut IdMap { + static mut MAP: Option> = None; + unsafe { &mut *core::ptr::addr_of_mut!(MAP) }.get_or_insert_with(IdMap::default) +} + +fn signals() -> &'static mut WakeSignals { + static mut MAP: Option = None; + unsafe { &mut *core::ptr::addr_of_mut!(MAP) }.get_or_insert_with(WakeSignals::new) +} + +/// Runs `f` with a short-lived `&mut Task` for `message_id`, returning `None` +/// when no task exists. +/// +/// The borrow is confined to `f`, so it never overlaps another `&mut Task` to +/// the same entry. This is what keeps the reentrant accesses performed from +/// inside a polled future (`register_signal`, `set_critical_hook`, `sleep_for`) +/// sound: [`message_loop`] never holds a live `&mut Task` across the poll, so +/// each `with_task_mut` reborrow of the `static mut` task map is disjoint in +/// time from every other. +#[inline] +fn with_task_mut(message_id: &MessageId, f: impl FnOnce(&mut Task) -> R) -> Option { + tasks().get_mut(message_id).map(f) +} + +/// Matches a task to a some message in order to avoid duplicate execution +/// of code that was running before the program was interrupted by `wait`. +/// +/// The [`Task`] lifecycle matches to the single message processing in the `handle()` entry-point +/// and ends when all internal futures are resolved or `handle_signal()` received for this `message_id`. +pub struct Task { + /// Wrapped in `Option` so [`message_loop`] can move the future *out* for the + /// duration of a poll. While it is taken, the slot is `None` and the + /// reentrant `with_task_mut` accesses from inside the future cannot alias the + /// future allocation through this `Task` (no `Box`-noalias hazard). + future: Option>, + locks: BinaryHeap<(Reverse, Option)>, + #[cfg(not(feature = "ethexe"))] + critical_hook: Option>, +} + +impl Task { + fn new(future: F) -> Self + where + F: Future + 'static, + { + Self { + future: Some(future.boxed_local()), + locks: Default::default(), + #[cfg(not(feature = "ethexe"))] + critical_hook: None, + } + } + + /// Stores the lock associated with an outbound reply, keeping it ordered by deadline. + /// + /// - pushes `(lock, Some(reply_to))` into the binary heap so the task can efficiently retrieve the + /// earliest lock when deciding how long to sleep. + /// + /// # Context + /// Called from [`WakeSignals::register_signal`] when the [`message_loop`] schedules a reply wait. + #[inline] + fn insert_lock(&mut self, reply_to: MessageId, lock: Lock) { + self.locks.push((Reverse(lock), Some(reply_to))); + } + + /// Tracks a sleep-specific lock (without a reply identifier) for the inbound message. + /// + /// # Context + /// Used when the task needs to suspend itself via `exec::wait_*` without tying the wait to a + /// particular reply id. + fn insert_sleep(&mut self, lock: Lock) { + self.locks.push((Reverse(lock), None)); + } + + /// Returns the earliest lock still awaiting completion, removing stale or cleared entries. + /// + /// # Context + /// Called from [`message_loop`] whenever the user future remains pending after polling. + #[inline] + fn next_lock(&mut self, now: BlockNumber) -> Option { + let signals_map = signals(); + while let Some((Reverse(lock), reply_to)) = self.locks.peek() { + // 1. skip and remove expired + if now >= lock.deadline() { + // If the user future never polled the matching MessageFuture, + // its signal entry would sit in Pending forever. Push it + // through the same timeout transition `WakeSignals::poll` + // would apply, so the global signals map stays bounded. + if let Some(reply_to) = reply_to { + signals_map.record_timeout(*reply_to, now); + } + self.locks.pop(); + continue; + } + // 2. skip and remove if not waits for reply_to + if let Some(reply_to) = reply_to + && !signals_map.waits_for(reply_to) + { + self.locks.pop(); + continue; + } + // 3. keep lock in `self.locks` for `WakeSignal::Pending` in case of `clear_signals` + return Some(*lock); + } + None + } + + /// Removes all outstanding reply locks from the signal registry without waiting on them. + /// + /// - iterates every stored `(_, reply_to)` pair and asks [`WakeSignals`] to drop the wake entry; + /// - used as part of task teardown to avoid keeping stale replies alive. + /// + /// # Context + /// Called from [`handle_signal`]. + #[cfg(not(feature = "ethexe"))] + #[inline] + fn clear_signals(&self) { + let now = Syscall::block_height(); + let signals_map = signals(); + self.locks.iter().for_each(|(_, reply_to)| { + if let Some(reply_to) = reply_to { + // set the `WakeSignal::Expired` for further processing in `handle_reply` + signals_map.record_timeout(*reply_to, now); + } + }); + } +} + +/// Sets a critical hook. +/// +/// # Context +/// If called in the `handle_reply` or `handle_signal` entrypoints. +/// +/// # SAFETY +/// Ensure that sufficient `gstd::Config::SYSTEM_RESERVE` is set in your +/// program, as this gas is locked during each async call to provide resources +/// for hook execution in case it is triggered. +#[cfg(not(feature = "ethexe"))] +pub fn set_critical_hook(f: F) { + if Syscall::reply_code().is_ok() { + panic!( + "`sails_rs::gstd::set_critical_hook()` must not be called in `handle_reply` entrypoint" + ) + } + + if Syscall::signal_code().is_ok() { + panic!( + "`sails_rs::gstd::set_critical_hook()` must not be called in `handle_signal` entrypoint" + ) + } + let message_id = Syscall::message_id(); + + with_task_mut(&message_id, |task| { + task.critical_hook = Some(Box::new(f)); + }) + .unwrap(); +} + +/// Drives asynchronous handling for the currently executing inbound message. +/// +/// - locates or creates the `Task` holding the user future for the current message id; +/// - polls the future once and, if it completes, tears down the bookkeeping; +/// - when the future stays pending, arms the shortest wait lock so the runtime suspends until a wake. +/// +/// # Context +/// Called from the contract's `handle` entry point while [`message_loop`] runs single-threaded inside the +/// actor. It must be invoked exactly once per incoming message to advance the async state machine. +/// +/// # Panics +/// Panics propagated from the user future bubble up, and the function will panic if no wait lock is +/// registered when a pending future requests suspension, signalling a contract logic bug. +#[inline] +pub fn message_loop(future: F) +where + F: Future + 'static, +{ + let msg_id = Syscall::message_id(); + + // Locate-or-create the task and move its future *out* in a single map + // lookup: `or_insert_with` hands back the `&mut Task`, and that borrow ends + // with `.take()` — before the poll. While we own the future as an + // independent local, the task entry's `future` slot is `None`, so the + // reentrant `with_task_mut` accesses from inside the future + // (`register_signal` / `set_critical_hook` / `sleep_for`) borrow a `Task` + // that no longer owns this future allocation. There is therefore no overlap + // between the polled future and the `&mut Task` — sound under both Stacked + // and Tree Borrows, with no `Box`-noalias hazard and no `unsafe`. + let mut fut = tasks() + .entry(msg_id) + .or_insert_with(|| { + #[cfg(not(feature = "ethexe"))] + { + Syscall::system_reserve_gas(gstd::Config::system_reserve()).unwrap(); + } + Task::new(future) + }) + .future + .take() + .unwrap(); + + let completed = { + let mut cx = Context::from_waker(task::Waker::noop()); + debug!("message_loop: polling future for {msg_id}"); + fut.as_mut().poll(&mut cx).is_ready() + }; + + if completed { + tasks().remove(&msg_id); + } else { + // Still pending: return the future to its `Task` and arm the wait. The + // returned lock is awaited *outside* the borrow so `wait` never runs + // while a `&mut Task` is live. + let now = Syscall::block_height(); + with_task_mut(&msg_id, |task| { + task.future = Some(fut); + task.next_lock(now) + }) + .flatten() + .unwrap() + .wait(now); + } +} + +pub type Payload = Vec; + +/// The [`WakeSignal`] lifecycle corresponds to waiting for a reply to a sent message +/// and ends when `handle_reply()` is received. +/// +/// May outlive parent [`Task`] in [`WakeSignal::Expired`] state. +/// +/// Can be created in [`WakeSignal::Expired`] state if there is no [`Task`] to await. +enum WakeSignal { + /// Reply is still pending; tracks origin message, deadline, and optional hook to run on completion or timeout. + Pending { + message_id: MessageId, + deadline: BlockNumber, + reply_hook: Option>, + }, + /// Reply handled; captures payload and reply code so the waiting future can resolve. + Ready { + payload: Payload, + reply_code: ReplyCode, + }, + /// Reply missed its deadline; retains timing data and hook so late arrivals can still be acknowledged. + Expired { + expected: BlockNumber, + now: BlockNumber, + reply_hook: Option>, + }, +} + +impl WakeSignal { + /// Transition a [`WakeSignal::Pending`] signal in place to + /// [`WakeSignal::Expired`], preserving the optional reply hook. No-op for + /// already-`Expired`/`Ready` signals. + /// + /// Centralizes the `Pending -> Expired` transition shared by + /// [`WakeSignals::record_timeout`] (natural timeout), [`WakeSignals::poll`] + /// (deferred timeout) and [`WakeSignals::forget_future`] (cancellation), so + /// a late reply can still fire the hook and deferred polls observe + /// `Err(Timeout)` rather than a missing entry. + #[inline] + fn expire(&mut self, now: BlockNumber) { + if let WakeSignal::Pending { + deadline, + reply_hook, + .. + } = self + { + *self = WakeSignal::Expired { + expected: *deadline, + now, + reply_hook: reply_hook.take(), + }; + } + } +} + +struct WakeSignals { + signals: IdMap, +} + +impl WakeSignals { + pub fn new() -> Self { + Self { + signals: IdMap::default(), + } + } + + /// Registers a reply wait for `waiting_reply_to` while the current message is being processed. + /// + /// - stores [`WakeSignal::Pending`] together with an optional hook so `poll`/`record_reply` can resolve it later; + /// - records the lock deadline for timeout detection and attaches the lock to the owning [`Task`] for + /// consistent wake bookkeeping. + /// + /// # Context + /// Called from helpers such as `send_bytes_for_reply` / `create_program_for_reply` while the message + /// handler executes inside [`message_loop`] in `handle()` entry point, see [Gear Protocol](https://wiki.vara.network/docs/build/introduction). + /// The current `message_id` is read from the runtime and used to fetch the associated `Task` entry. + /// + /// # Panics + /// Panics if the `Task` for the current `message_id` cannot be found, which indicates the function + /// was invoked outside the [`message_loop`] context (programmer error). + pub fn register_signal( + &mut self, + waiting_reply_to: MessageId, + lock: locks::Lock, + reply_hook: Option>, + ) { + let message_id = Syscall::message_id(); + let deadline = lock.deadline(); + + self.signals.insert( + waiting_reply_to, + WakeSignal::Pending { + message_id, + deadline, + reply_hook, + }, + ); + + // ::gstd::debug!( + // "register_signal: add lock for reply_to {waiting_reply_to} in message {message_id}" + // ); + // The task for the current `message_id` is inserted by `message_loop` + // before the user future driving this call is polled, so it is always + // present here. `with_task_mut` confines the `&mut Task` to the closure, + // keeping this reentrant access disjoint from `message_loop`'s borrow. + with_task_mut(&message_id, |task| task.insert_lock(waiting_reply_to, lock)).unwrap(); + } + + /// Registers a reply hook for `waiting_reply_to` without creating a tracked wait. + /// + /// - stores a [`WakeSignal::Expired`] entry so `record_reply` will still execute the hook if a reply + /// arrives later; + /// - intended for one-way sends that want to observe replies from outside [`message_loop`]. + /// + /// # Context + /// Called from [`send_one_way`] and other synchronous helpers; may be invoked outside [`message_loop`]. + #[cfg(not(feature = "ethexe"))] + #[inline] + pub fn register_hook( + &mut self, + waiting_reply_to: MessageId, + reply_hook: Option>, + ) { + if let Some(reply_hook) = reply_hook { + let now = Syscall::block_height(); + self.signals.insert( + waiting_reply_to, + WakeSignal::Expired { + expected: now, + now, + reply_hook: Some(reply_hook), + }, + ); + } + } + + /// Processes an incoming reply for `reply_to` and transitions the stored wake state. + /// + /// - upgrades the [`WakeSignal::Pending`] entry to [`WakeSignal::Ready`], capturing payload and reply code; + /// - executes the optional reply hook once the reply becomes available. + /// - for the [`WakeSignal::Expired`] entry executes the optional reply hook and remove entry; + /// + /// # Context + /// Invoked by [`handle_reply_with_hook`] when a reply arrives during `handle_reply()` execution. The + /// runtime supplies `reply_to`. + /// + /// # Panics + /// Panics if it encounters an already finalised entry [`WakeSignal::Ready`] or the associated task is + /// missing. Both scenarios indicate logic bugs or duplicate delivery. + pub fn record_reply(&mut self, reply_to: &MessageId) { + if let hashbrown::hash_map::EntryRef::Occupied(mut entry) = self.signals.entry_ref(reply_to) + { + match entry.get_mut() { + WakeSignal::Pending { + message_id, + deadline: _, + reply_hook, + } => { + let message_id = *message_id; + let reply_hook = reply_hook.take(); + // replace entry with `WakeSignal::Ready` + _ = entry.insert(WakeSignal::Ready { + payload: Syscall::read_bytes().unwrap(), + // SAFETY: `record_reply` runs only in the `handle_reply` + // entrypoint, where the reply context always exists. + reply_code: unsafe { Syscall::reply_code().unwrap_unchecked() }, + }); + debug!( + "record_reply: remove lock for reply_to {reply_to} in message {message_id}" + ); + // wake message loop after receiving reply + Syscall::wake(message_id).unwrap(); + + // execute reply hook + if let Some(f) = reply_hook { + f() + } + } + WakeSignal::Expired { reply_hook, .. } => { + let reply_hook = reply_hook.take(); + _ = entry.remove(); + // execute reply hook and remove entry + if let Some(f) = reply_hook { + f() + } + } + WakeSignal::Ready { .. } => panic!("A reply has already received"), + }; + } else { + debug!( + "A message has received a reply though it wasn't to receive one, or a processed message has received a reply" + ); + } + } + + /// Marks a pending reply as timed out and preserves context for later handling. + /// + /// - upgrades a [`WakeSignal::Pending`] entry to [`WakeSignal::Expired`], capturing when the reply was expected + /// and when the timeout was detected; + /// - retains the optional reply hook so it can still be executed if a late reply arrives and reuses the + /// stored state when `record_reply` is called afterwards. + /// + /// # Context + /// Triggered from [`Task::clear_signals`]. + pub fn record_timeout(&mut self, reply_to: MessageId, now: BlockNumber) { + // Always transition to Expired (with or without hook). The entry must + // stay consumable by a deferred `MessageFuture::poll` — that poll has + // to observe `Err(Timeout)` instead of panicking on a missing entry. + // Final cleanup happens in `MessageFuture::drop` when the future is no + // longer reachable. + if let Some(signal @ WakeSignal::Pending { .. }) = self.signals.get_mut(&reply_to) { + signal.expire(now); + } else { + debug!("A message has timed out after reply"); + } + } + + /// Release the entry tied to a `MessageFuture` that is being dropped. + /// + /// Called from `MessageFuture::drop`. Reclaims entries that no one will + /// observe via `poll` anymore; preserves entries that still hold a + /// reply hook so a late reply can fire it through `record_reply`. + /// + /// Cancellation must be symmetric with timeout: + /// - `Pending { reply_hook: Some, .. }` -> `Expired { reply_hook: Some, .. }` + /// (same transition `record_timeout` would do, so the hook survives). + /// - `Pending { reply_hook: None, .. }` -> removed. + /// - `Ready` -> removed (payload will never be consumed; the hook, if any, + /// already ran inside `record_reply`). + /// - `Expired { reply_hook: Some, .. }` -> kept. + /// - `Expired { reply_hook: None, .. }` -> removed. + pub fn forget_future(&mut self, reply_to: &MessageId) { + if let hashbrown::hash_map::EntryRef::Occupied(mut entry) = self.signals.entry_ref(reply_to) + { + match entry.get_mut() { + // Cancellation before timeout with a live hook: transition to + // Expired (same as a natural timeout) so a late reply can still + // fire the hook and a deferred poll observes `Err(Timeout)`. + signal @ WakeSignal::Pending { + reply_hook: Some(_), + .. + } => signal.expire(Syscall::block_height()), + // Expired entry still holding a hook: keep it for a late reply. + WakeSignal::Expired { + reply_hook: Some(_), + .. + } => {} + // Nothing left to observe (Pending/Expired without a hook, or a + // Ready payload no one will consume): drop the entry. + _ => { + entry.remove(); + } + } + } + } + + pub fn waits_for(&self, reply_to: &MessageId) -> bool { + self.signals + .get(reply_to) + .is_some_and(|signal| !matches!(signal, WakeSignal::Expired { .. })) + } + + /// Polls the stored wake signal for `reply_to`, returning the appropriate future state. + /// + /// - inspects the current `WakeSignal` variant, promoting pending entries whose deadline has passed to + /// [`WakeSignal::Expired`]; + /// - returns `Pending`, a `Ready` payload, or propagates a timeout error; when `Ready`, the entry is + /// removed so subsequent polls observe completion. + /// + /// # Context + /// Called by [`MessageFuture::poll`] (and any wrappers) while a consumer awaits a reply produced by + /// [`message_loop`]. It runs on the same execution thread and must be non-blocking. + /// + /// # Panics + /// Panics if the signal was never registered for `reply_to`, which indicates misuse of the async API + /// (polling without having called one of the [`send_bytes_for_reply`]/[`create_program_for_reply`] methods first). + pub fn poll( + &mut self, + reply_to: &MessageId, + _cx: &mut Context<'_>, + ) -> Poll, Error>> { + let hashbrown::hash_map::EntryRef::Occupied(mut entry) = self.signals.entry_ref(reply_to) + else { + panic!("Poll not registered feature") + }; + + match entry.get_mut() { + WakeSignal::Pending { deadline, .. } => { + let now = Syscall::block_height(); + let expected = *deadline; + if now >= expected { + // Transition to Expired and keep the entry. The future + // may be polled again (idempotent timeout) or dropped; + // `MessageFuture::drop` does the final cleanup. + entry.get_mut().expire(now); + Poll::Ready(Err(Error::Timeout(expected, now))) + } else { + Poll::Pending + } + } + WakeSignal::Expired { expected, now, .. } => { + // Entry persists either for a late reply (if there's a hook) + // or until the owning `MessageFuture` is dropped. + Poll::Ready(Err(Error::Timeout(*expected, *now))) + } + WakeSignal::Ready { .. } => { + // remove entry if `WakeSignal::Ready` + let WakeSignal::Ready { + payload, + reply_code, + } = entry.remove() + else { + // SAFETY: checked in the code above. + unsafe { hint::unreachable_unchecked() } + }; + match reply_code { + ReplyCode::Success(_) => Poll::Ready(Ok(payload)), + ReplyCode::Error(reason) => { + Poll::Ready(Err(Error::ErrorReply(payload.into(), reason))) + } + ReplyCode::Unsupported => Poll::Ready(Err(Error::UnsupportedReply(payload))), + } + } + } + } +} + +pub struct MessageFuture { + /// A message identifier for an expected reply. + /// + /// This identifier is generated by the corresponding send function (e.g. + /// [`gcore::msg::send`](::gcore::msg::send)). + pub waiting_reply_to: MessageId, +} + +impl Unpin for MessageFuture {} + +impl Future for MessageFuture { + type Output = Result, Error>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + poll(&self.waiting_reply_to, cx) + } +} + +impl Drop for MessageFuture { + /// Reclaim the `WakeSignals` entry tied to this future. + /// + /// `WakeSignals::poll` and `record_timeout` deliberately keep entries + /// alive so a deferred poll can still observe `Err(Timeout)`. That means + /// the only safe place to delete an entry that no one will ever poll + /// again is here, when the future itself is being dropped. + /// + /// Entries that retain a reply hook (`WakeSignal::Expired` with + /// `reply_hook.is_some()`) stay in the map so a late reply can still + /// fire the hook through `record_reply`. + fn drop(&mut self) { + signals().forget_future(&self.waiting_reply_to); + } +} + +impl FusedFuture for MessageFuture { + fn is_terminated(&self) -> bool { + is_terminated(&self.waiting_reply_to) + } +} + +#[inline] +pub fn send_for_reply( + destination: ActorId, + payload: E, + value: ValueUnit, +) -> Result { + let size = Encode::encoded_size(&payload); + stack_buffer::with_byte_buffer(size, |buffer: &mut [mem::MaybeUninit]| { + let mut buffer_writer = MaybeUninitBufferWriter::new(buffer); + Encode::encode_to(&payload, &mut buffer_writer); + buffer_writer.with_buffer(|buffer| { + send_bytes_for_reply( + destination, + buffer, + value, + Default::default(), + #[cfg(not(feature = "ethexe"))] + None, + #[cfg(not(feature = "ethexe"))] + None, + #[cfg(not(feature = "ethexe"))] + None, + ) + }) + }) +} + +#[inline] +pub fn send_one_way( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + #[cfg(not(feature = "ethexe"))] gas_limit: Option, + #[cfg(not(feature = "ethexe"))] reply_deposit: Option, + #[cfg(not(feature = "ethexe"))] reply_hook: Option>, +) -> Result { + let waiting_reply_to = crate::ok!(send_bytes( + destination, + payload, + value, + #[cfg(not(feature = "ethexe"))] + gas_limit, + #[cfg(not(feature = "ethexe"))] + reply_deposit + )); + + #[cfg(not(feature = "ethexe"))] + signals().register_hook(waiting_reply_to, reply_hook); + + Ok(waiting_reply_to) +} + +#[cfg(not(feature = "ethexe"))] +#[inline] +pub fn send_bytes_for_reply( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + wait: Lock, + gas_limit: Option, + reply_deposit: Option, + reply_hook: Option>, +) -> Result { + let waiting_reply_to = crate::ok!(send_bytes( + destination, + payload, + value, + gas_limit, + reply_deposit + )); + + // Register the wait lock and reply hook so the runtime wakes this future + // when the reply arrives. + signals().register_signal(waiting_reply_to, wait, reply_hook); + + Ok(MessageFuture { waiting_reply_to }) +} + +#[cfg(feature = "ethexe")] +#[inline] +pub fn send_bytes_for_reply( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + wait: Lock, +) -> Result { + let waiting_reply_to = crate::ok!(send_bytes(destination, payload, value)); + + signals().register_signal(waiting_reply_to, wait, None); + + Ok(MessageFuture { waiting_reply_to }) +} + +#[cfg(not(feature = "ethexe"))] +#[allow(clippy::too_many_arguments)] +#[inline] +pub fn create_program_for_reply( + code_id: CodeId, + salt: &[u8], + payload: &[u8], + value: ValueUnit, + wait: Lock, + gas_limit: Option, + reply_deposit: Option, + reply_hook: Option>, +) -> Result<(MessageFuture, ActorId), ::gstd::errors::Error> { + let (waiting_reply_to, program_id) = if let Some(gas_limit) = gas_limit { + crate::ok!(::gcore::prog::create_program_with_gas( + code_id, salt, payload, gas_limit, value + )) + } else { + crate::ok!(::gcore::prog::create_program(code_id, salt, payload, value)) + }; + + if let Some(reply_deposit) = reply_deposit { + // Reserve gas for handling the reply. The error is propagated, not + // ignored, matching gstd's `#[wait_for_reply]`. On the awaited path the + // caller panics, trapping before the program creation commits. + crate::ok!(::gcore::exec::reply_deposit( + waiting_reply_to, + reply_deposit + )); + } + + // Register the wait lock and reply hook so the runtime wakes this future + // when the reply arrives. + signals().register_signal(waiting_reply_to, wait, reply_hook); + + Ok((MessageFuture { waiting_reply_to }, program_id)) +} + +#[cfg(feature = "ethexe")] +#[inline] +pub fn create_program_for_reply( + code_id: CodeId, + salt: &[u8], + payload: &[u8], + value: ValueUnit, + wait: Lock, +) -> Result<(MessageFuture, ActorId), ::gstd::errors::Error> { + let (waiting_reply_to, program_id) = + crate::ok!(::gcore::prog::create_program(code_id, salt, payload, value)); + + signals().register_signal(waiting_reply_to, wait, None); + + Ok((MessageFuture { waiting_reply_to }, program_id)) +} + +/// Default reply handler. +#[inline] +pub fn handle_reply_with_hook() { + // SAFETY: only called from the `handle_reply` entrypoint, where the reply + // context always exists. + let reply_to = unsafe { Syscall::reply_to().unwrap_unchecked() }; + + signals().record_reply(&reply_to); +} + +/// Default signal handler. +#[cfg(not(feature = "ethexe"))] +#[inline] +pub fn handle_signal() { + // SAFETY: only called from the `handle_signal` entrypoint, where the signal + // context always exists. + let msg_id = unsafe { Syscall::signal_from().unwrap_unchecked() }; + // Remove Task and all associated signals, execute critical hook + if let Some(mut task) = tasks().remove(&msg_id) { + if let Some(critical_hook) = task.critical_hook.take() { + critical_hook(msg_id); + } + task.clear_signals(); + } +} + +pub fn poll(message_id: &MessageId, cx: &mut Context<'_>) -> Poll, Error>> { + signals().poll(message_id, cx) +} + +pub fn is_terminated(message_id: &MessageId) -> bool { + !signals().waits_for(message_id) +} + +struct MessageSleepFuture { + deadline: BlockNumber, +} + +impl MessageSleepFuture { + fn new(deadline: BlockNumber) -> Self { + Self { deadline } + } +} + +impl Unpin for MessageSleepFuture {} + +impl Future for MessageSleepFuture { + type Output = (); + + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll { + let now = Syscall::block_height(); + + if now >= self.deadline { + Poll::Ready(()) + } else { + Poll::Pending + } + } +} + +/// Delays message execution in asynchronous way for the specified number of blocks. +/// +/// It works pretty much like the [`gcore::exec::wait_for`] function, but +/// allows to continue execution after the delay in the same handler. It is +/// worth mentioning that the program state gets persisted inside the call, and +/// the execution resumes with potentially different state. +pub fn sleep_for(block_count: BlockCount) -> impl Future { + let message_id = Syscall::message_id(); + let lock = Lock::exactly(block_count); + // `sleep_for` runs inside the user future, which `message_loop` polls only + // after inserting the task for the current `message_id`. `with_task_mut` + // confines the `&mut Task` to the closure so this reentrant access never + // aliases `message_loop`'s borrow. + with_task_mut(&message_id, |task| task.insert_sleep(lock)).unwrap(); + MessageSleepFuture::new(lock.deadline()) +} + +#[cfg(feature = "std")] +#[cfg(test)] +mod tests { + use super::*; + use crate::gstd::syscalls::Syscall; + use core::{sync::atomic::AtomicU64, task, task::Context}; + use std::sync::Mutex; + + // `tasks()` / `signals()` are `static mut` — correct for the single-threaded + // WASM runtime, but cargo test runs this module on a thread pool. Serialize + // every test in here through one mutex so the parallel runner never races + // on the shared HashMaps. Recover from poisoning so a panicking test doesn't + // cascade into PoisonError for every subsequent one. + static SERIAL: Mutex<()> = Mutex::new(()); + + fn serial(f: impl FnOnce() -> R) -> R { + let _guard = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + f() + } + + static MSG_ID: AtomicU64 = AtomicU64::new(1); + + fn msg_id() -> MessageId { + MessageId::from(MSG_ID.fetch_add(1, core::sync::atomic::Ordering::SeqCst)) + } + + fn set_context(message_id: MessageId, block_height: u32) { + Syscall::with_message_id(message_id); + Syscall::with_block_height(block_height); + } + + #[test] + fn task_insert_lock_adds_entry() { + serial(|| { + set_context(msg_id(), 10); + + let mut task = Task::new(async {}); + let reply_to = msg_id(); + let lock = Lock::up_to(3); + + task.insert_lock(reply_to, lock); + task.insert_lock(msg_id(), Lock::exactly(5)); + + let Some((Reverse(next_lock), next_reply_to)) = task.locks.peek() else { + unreachable!() + }; + + assert_eq!(task.locks.len(), 2); + assert_eq!(Some(&reply_to), next_reply_to.as_ref()); + assert_eq!(next_lock, &lock); + }); + } + + #[test] + fn signals_poll_converts_pending_into_timeout() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 20); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(5); + let deadline = lock.deadline(); + + signals().register_signal(reply_to, lock, None); + + Syscall::with_block_height(deadline - 1); + let mut cx = Context::from_waker(task::Waker::noop()); + assert!(matches!(signals().poll(&reply_to, &mut cx), Poll::Pending)); + + Syscall::with_block_height(deadline); + let mut cx = Context::from_waker(task::Waker::noop()); + match signals().poll(&reply_to, &mut cx) { + Poll::Ready(Err(Error::Timeout(expected, now))) => { + assert_eq!(expected, deadline); + assert_eq!(now, deadline); + } + other => panic!("expected timeout, got {other:?}"), + } + + signals().signals.remove(&reply_to); + tasks().remove(&message_id); + }); + } + + #[test] + fn task_remove_signal_skip_not_waited_lock() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 30); + let reply_to = msg_id(); + let lock = Lock::up_to(5); + + tasks().insert(message_id, Task::new(async {})); + signals().register_signal(reply_to, lock, None); + + // Re-borrow `tasks()` for each read instead of holding a `&mut Task` + // across `register_signal` (which reborrows the `static mut` task + // map): two live `&mut Task` to the same entry is UB under Stacked + // Borrows. Short, disjoint borrows keep the test sound. + assert_eq!(1, tasks().get_mut(&message_id).unwrap().locks.len()); + + signals().signals.remove(&reply_to); + + assert_eq!(1, tasks().get_mut(&message_id).unwrap().locks.len()); + assert_eq!(None, tasks().get_mut(&message_id).unwrap().next_lock(31)); + tasks().remove(&message_id); + }); + } + + #[test] + fn task_insert_sleep_adds_entry_without_reply() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 40); + + let mut task = Task::new(async {}); + let lock = Lock::exactly(4); + + task.insert_sleep(lock); + assert_eq!(Some(lock), task.next_lock(42)); + assert_eq!(None, task.next_lock(lock.deadline())); + }); + } + + /// After a timed-out poll, the entry must stay alive as `Expired` so a + /// deferred re-poll (or a late reply with a hook) can still observe it. + /// Final cleanup is `MessageFuture::drop`'s job, not `poll`'s. + #[test] + fn signals_poll_keeps_expired_for_deferred_repoll() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 50); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(2); + let deadline = lock.deadline(); + signals().register_signal(reply_to, lock, None); + + Syscall::with_block_height(deadline + 1); + let mut cx = Context::from_waker(task::Waker::noop()); + + // First poll: Pending -> Expired, returns Timeout. + assert!(matches!( + signals().poll(&reply_to, &mut cx), + Poll::Ready(Err(Error::Timeout(..))) + )); + assert!( + matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Expired { .. }) + ), + "entry must remain so a re-poll can observe Timeout" + ); + + // Second poll on the same reply_to: same answer, no panic. + assert!(matches!( + signals().poll(&reply_to, &mut cx), + Poll::Ready(Err(Error::Timeout(..))) + )); + + signals().signals.remove(&reply_to); + tasks().remove(&message_id); + }); + } + + /// `Task::next_lock` must lift the signal out of `Pending` when reaping + /// the matching expired lock — otherwise the future that polls later + /// would still see `Pending` past its deadline. The entry stays as + /// `Expired` so a deferred poll can observe `Err(Timeout)`. + #[test] + fn next_lock_expiring_lock_transitions_signal_to_expired() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 100); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(2); + signals().register_signal(reply_to, lock, None); + + // Advance past the deadline WITHOUT polling the MessageFuture + // (simulates select!/join! where another branch finished first). + let now = lock.deadline() + 5; + Syscall::with_block_height(now); + + let task = tasks().get_mut(&message_id).unwrap(); + assert_eq!(task.next_lock(now), None, "expired lock must be popped"); + + assert!( + matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Expired { .. }) + ), + "next_lock must transition Pending -> Expired (not leave Pending, not delete)" + ); + + // Deferred poll: the user future finally awaits this MessageFuture. + // It must produce Err(Timeout), not panic on a missing entry. + let mut cx = Context::from_waker(task::Waker::noop()); + assert!(matches!( + signals().poll(&reply_to, &mut cx), + Poll::Ready(Err(Error::Timeout(..))) + )); + + signals().signals.remove(&reply_to); + tasks().remove(&message_id); + }); + } + + /// `MessageFuture::drop` must remove a `Pending` entry — the user gave + /// up before polling, no one will ever observe a reply or timeout. + #[test] + fn message_future_drop_removes_pending_entry() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 200); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + signals().register_signal(reply_to, Lock::up_to(5), None); + assert!(signals().signals.contains_key(&reply_to)); + + drop(MessageFuture { + waiting_reply_to: reply_to, + }); + + assert!( + !signals().signals.contains_key(&reply_to), + "Pending entry must be removed when MessageFuture is dropped" + ); + + tasks().remove(&message_id); + }); + } + + /// `MessageFuture::drop` must remove an `Expired` entry that holds no + /// reply hook — keeping it serves no late-delivery purpose. + #[test] + fn message_future_drop_removes_expired_entry_when_no_hook() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 300); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(2); + let deadline = lock.deadline(); + signals().register_signal(reply_to, lock, None); + + // Drive Pending -> Expired (no hook). + Syscall::with_block_height(deadline + 1); + let mut cx = Context::from_waker(task::Waker::noop()); + assert!(matches!( + signals().poll(&reply_to, &mut cx), + Poll::Ready(Err(Error::Timeout(..))) + )); + assert!(matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Expired { .. }) + )); + + drop(MessageFuture { + waiting_reply_to: reply_to, + }); + + assert!( + !signals().signals.contains_key(&reply_to), + "Expired-no-hook entry must be removed when MessageFuture is dropped" + ); + + tasks().remove(&message_id); + }); + } + + /// Cancellation symmetry: dropping a `MessageFuture` while still + /// `Pending` with a reply hook must keep the hook alive (transition to + /// Expired-with-hook). Otherwise, whether `with_reply_hook` fires on a + /// late reply would depend on whether the future was dropped before or + /// after the timeout — a footgun. + #[test] + fn message_future_drop_preserves_pending_entry_with_hook() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 500); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(5); + let deadline = lock.deadline(); + signals().register_signal(reply_to, lock, Some(Box::new(|| {}))); + + // Drop while still well within the deadline — natural timeout has + // not fired. The hook must survive. + assert!(Syscall::block_height() < deadline); + drop(MessageFuture { + waiting_reply_to: reply_to, + }); + + assert!( + matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Expired { + reply_hook: Some(_), + .. + }) + ), + "Pending-with-hook drop must transition to Expired-with-hook \ + so a late reply can still fire the hook" + ); + + signals().signals.remove(&reply_to); + tasks().remove(&message_id); + }); + } + + /// Conversely, `MessageFuture::drop` MUST preserve an `Expired` entry + /// that still holds a reply hook — a late reply via `record_reply` is + /// the only thing that will ever fire that hook. + #[test] + fn message_future_drop_preserves_expired_entry_with_hook() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 400); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(2); + let deadline = lock.deadline(); + signals().register_signal(reply_to, lock, Some(Box::new(|| {}))); + + Syscall::with_block_height(deadline + 1); + let mut cx = Context::from_waker(task::Waker::noop()); + assert!(matches!( + signals().poll(&reply_to, &mut cx), + Poll::Ready(Err(Error::Timeout(..))) + )); + + drop(MessageFuture { + waiting_reply_to: reply_to, + }); + + assert!( + matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Expired { + reply_hook: Some(_), + .. + }) + ), + "Expired entry with a hook must survive future drop so a late \ + reply can still fire the hook" + ); + + signals().signals.remove(&reply_to); + tasks().remove(&message_id); + }); + } + + /// Reset the (thread-local) reply/signal context to "no context" so a test + /// that ran earlier on the same worker thread can't leak an `Ok` reply/signal + /// code into a test that expects to be outside those entrypoints. + fn clear_reply_signal_ctx() { + use gear_core_errors::{ExecutionError, ExtError}; + Syscall::with_reply_code(Err( + ExtError::Execution(ExecutionError::NoReplyContext).into() + )); + Syscall::with_signal_code(Err( + ExtError::Execution(ExecutionError::NoSignalContext).into() + )); + } + + fn poll_once(f: &mut F) -> Poll { + let mut cx = Context::from_waker(task::Waker::noop()); + core::pin::Pin::new(f).poll(&mut cx) + } + + /// A `Ready` signal carrying a successful reply resolves the poll to + /// `Ok(payload)` and removes the entry so it can't be double-consumed. + #[test] + fn poll_ready_success_returns_ok_and_removes_entry() { + serial(|| { + let reply_to = msg_id(); + signals().signals.insert( + reply_to, + WakeSignal::Ready { + payload: b"OK".to_vec(), + reply_code: ReplyCode::Success(SuccessReplyReason::Manual), + }, + ); + let mut cx = Context::from_waker(task::Waker::noop()); + match signals().poll(&reply_to, &mut cx) { + Poll::Ready(Ok(p)) => assert_eq!(p, b"OK".to_vec()), + other => panic!("expected Ok, got {other:?}"), + } + assert!( + !signals().signals.contains_key(&reply_to), + "Ready entry must be removed once consumed" + ); + }); + } + + #[test] + fn poll_ready_error_and_unsupported_map_to_errors() { + serial(|| { + let err_reply = msg_id(); + signals().signals.insert( + err_reply, + WakeSignal::Ready { + payload: b"e".to_vec(), + reply_code: ReplyCode::Error(ErrorReplyReason::Execution( + SimpleExecutionError::UserspacePanic, + )), + }, + ); + let mut cx = Context::from_waker(task::Waker::noop()); + assert!(matches!( + signals().poll(&err_reply, &mut cx), + Poll::Ready(Err(Error::ErrorReply(..))) + )); + + let uns = msg_id(); + signals().signals.insert( + uns, + WakeSignal::Ready { + payload: b"u".to_vec(), + reply_code: ReplyCode::Unsupported, + }, + ); + assert!(matches!( + signals().poll(&uns, &mut cx), + Poll::Ready(Err(Error::UnsupportedReply(_))) + )); + }); + } + + #[test] + #[should_panic(expected = "Poll not registered")] + fn poll_unregistered_panics() { + serial(|| { + let reply_to = msg_id(); + let mut cx = Context::from_waker(task::Waker::noop()); + let _ = signals().poll(&reply_to, &mut cx); + }); + } + + /// A reply for a `Pending` signal captures the payload and reply code into + /// `Ready`, fires the reply hook, and wakes the loop; the awaiting poll then + /// yields the payload. + #[test] + fn record_reply_pending_fires_hook_and_resolves_to_ready() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 10); + clear_reply_signal_ctx(); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let fired = std::rc::Rc::new(core::cell::Cell::new(false)); + let f = fired.clone(); + signals().register_signal( + reply_to, + Lock::up_to(10), + Some(Box::new(move || f.set(true))), + ); + + // Simulate the `handle_reply` context the runtime sees. + Syscall::with_read_bytes(Ok(b"PONG".to_vec())); + Syscall::with_reply_code(Ok(ReplyCode::Success(SuccessReplyReason::Manual))); + + signals().record_reply(&reply_to); + + assert!(fired.get(), "reply hook must fire on a successful reply"); + assert!(matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Ready { .. }) + )); + + let mut cx = Context::from_waker(task::Waker::noop()); + match signals().poll(&reply_to, &mut cx) { + Poll::Ready(Ok(p)) => assert_eq!(p, b"PONG".to_vec()), + other => panic!("expected Ok(PONG), got {other:?}"), + } + assert!(!signals().signals.contains_key(&reply_to)); + + tasks().remove(&message_id); + clear_reply_signal_ctx(); + }); + } + + #[test] + fn record_reply_on_expired_with_hook_fires_and_removes() { + serial(|| { + let reply_to = msg_id(); + let fired = std::rc::Rc::new(core::cell::Cell::new(false)); + let f = fired.clone(); + signals().signals.insert( + reply_to, + WakeSignal::Expired { + expected: 1, + now: 2, + reply_hook: Some(Box::new(move || f.set(true))), + }, + ); + // Expired branch takes no wake path. + signals().record_reply(&reply_to); + assert!(fired.get(), "late reply must fire the preserved hook"); + assert!(!signals().signals.contains_key(&reply_to)); + }); + } + + #[test] + #[should_panic(expected = "already received")] + fn record_reply_on_ready_panics() { + serial(|| { + let reply_to = msg_id(); + signals().signals.insert( + reply_to, + WakeSignal::Ready { + payload: vec![], + reply_code: ReplyCode::Success(SuccessReplyReason::Manual), + }, + ); + signals().record_reply(&reply_to); + }); + } + + #[test] + fn record_reply_unknown_is_noop() { + serial(|| { + let reply_to = msg_id(); + signals().record_reply(&reply_to); + assert!(!signals().signals.contains_key(&reply_to)); + }); + } + + /// `record_timeout` only transitions `Pending` entries; `Ready`, `Expired`, + /// and missing entries are left untouched (captured timing is never + /// overwritten). + #[test] + fn record_timeout_is_noop_on_non_pending() { + serial(|| { + let ready = msg_id(); + signals().signals.insert( + ready, + WakeSignal::Ready { + payload: b"x".to_vec(), + reply_code: ReplyCode::Unsupported, + }, + ); + signals().record_timeout(ready, 5); + assert!( + matches!( + signals().signals.get(&ready), + Some(WakeSignal::Ready { .. }) + ), + "Ready must stay Ready" + ); + + let expired = msg_id(); + signals().signals.insert( + expired, + WakeSignal::Expired { + expected: 3, + now: 4, + reply_hook: None, + }, + ); + signals().record_timeout(expired, 99); + assert!( + matches!( + signals().signals.get(&expired), + Some(WakeSignal::Expired { + expected: 3, + now: 4, + .. + }) + ), + "Expired timing must not be overwritten" + ); + + let missing = msg_id(); + signals().record_timeout(missing, 1); + assert!(!signals().signals.contains_key(&missing)); + + signals().signals.remove(&ready); + signals().signals.remove(&expired); + }); + } + + /// The `(expected, now)` of a timeout is captured on the first observation + /// and must not drift when the future is polled again at a later block. + #[test] + fn poll_expired_timing_is_captured_once() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 700); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let lock = Lock::up_to(2); + let deadline = lock.deadline(); + signals().register_signal(reply_to, lock, None); + + Syscall::with_block_height(deadline + 1); + let mut cx = Context::from_waker(task::Waker::noop()); + let first = deadline + 1; + assert!(matches!( + signals().poll(&reply_to, &mut cx), + Poll::Ready(Err(Error::Timeout(d, n))) if d == deadline && n == first + )); + + Syscall::with_block_height(deadline + 50); + match signals().poll(&reply_to, &mut cx) { + Poll::Ready(Err(Error::Timeout(expected, now))) => { + assert_eq!(expected, deadline); + assert_eq!(now, first, "now must stay the first-observed timeout block"); + } + other => panic!("expected timeout, got {other:?}"), + } + + signals().signals.remove(&reply_to); + tasks().remove(&message_id); + }); + } + + /// Dropping the future after its reply already arrived (`Ready`) discards + /// the unconsumed payload — nothing will ever read it. + #[test] + fn message_future_drop_removes_ready_entry() { + serial(|| { + let reply_to = msg_id(); + signals().signals.insert( + reply_to, + WakeSignal::Ready { + payload: b"z".to_vec(), + reply_code: ReplyCode::Success(SuccessReplyReason::Manual), + }, + ); + drop(MessageFuture { + waiting_reply_to: reply_to, + }); + assert!( + !signals().signals.contains_key(&reply_to), + "a Ready payload no one will consume must be dropped" + ); + }); + } + + /// `waits_for` is true for `Pending`/`Ready` and false for `Expired`/missing + /// entries; `is_terminated` is its inverse. + #[test] + fn waits_for_and_is_terminated_truth_table() { + serial(|| { + let pending = msg_id(); + let ready = msg_id(); + let expired = msg_id(); + let missing = msg_id(); + signals().signals.insert( + pending, + WakeSignal::Pending { + message_id: pending, + deadline: 1, + reply_hook: None, + }, + ); + signals().signals.insert( + ready, + WakeSignal::Ready { + payload: vec![], + reply_code: ReplyCode::Success(SuccessReplyReason::Manual), + }, + ); + signals().signals.insert( + expired, + WakeSignal::Expired { + expected: 1, + now: 2, + reply_hook: None, + }, + ); + + assert!(signals().waits_for(&pending)); + assert!( + signals().waits_for(&ready), + "Ready still counts as waited-for" + ); + assert!(!signals().waits_for(&expired)); + assert!(!signals().waits_for(&missing)); + + assert!(!is_terminated(&pending)); + assert!(!is_terminated(&ready)); + assert!(is_terminated(&expired)); + assert!(is_terminated(&missing)); + + signals().signals.remove(&pending); + signals().signals.remove(&ready); + signals().signals.remove(&expired); + }); + } + + /// `next_lock` returns the earliest-deadline lock, breaking deadline ties in + /// favour of `UpTo` over `Exactly`. + #[test] + fn next_lock_prefers_earliest_then_up_to_on_tie() { + serial(|| { + set_context(msg_id(), 1000); + let mut task = Task::new(async {}); + let up = Lock::up_to(5); // deadline 1005, UpTo + let exact = Lock::exactly(5); // deadline 1005, Exactly + let later = Lock::up_to(10); // deadline 1010 + task.insert_sleep(later); + task.insert_sleep(exact); + task.insert_sleep(up); + + let next = task.next_lock(1000).unwrap(); + assert_eq!(next.deadline(), up.deadline()); + assert_eq!( + next.wait_type(), + WaitType::UpTo, + "UpTo must win the same-deadline tie-break" + ); + }); + } + + #[test] + fn next_lock_skips_dropped_signal_and_returns_later_pending() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 2000); + tasks().insert(message_id, Task::new(async {})); + + let gone = msg_id(); + let pending = msg_id(); + signals().register_signal(gone, Lock::up_to(3), None); // deadline 2003 + signals().register_signal(pending, Lock::up_to(7), None); // deadline 2007 + // The earlier lock's signal disappears (e.g. consumed/cleared). + signals().signals.remove(&gone); + + let task = tasks().get_mut(&message_id).unwrap(); + let next = task.next_lock(2000).unwrap(); + assert_eq!( + next.deadline(), + 2007, + "the dropped earlier lock must be skipped, returning the live one" + ); + + signals().signals.remove(&pending); + tasks().remove(&message_id); + }); + } + + /// `register_hook` stores an `Expired`-with-hook entry only when a hook is + /// supplied (a later reply fires it); a `None` hook stores nothing. + #[cfg(not(feature = "ethexe"))] + #[test] + fn register_hook_some_stores_expired_with_hook_none_stores_nothing() { + serial(|| { + set_context(msg_id(), 10); + + let with = msg_id(); + let fired = std::rc::Rc::new(core::cell::Cell::new(false)); + let f = fired.clone(); + signals().register_hook(with, Some(Box::new(move || f.set(true)))); + assert!(matches!( + signals().signals.get(&with), + Some(WakeSignal::Expired { + reply_hook: Some(_), + .. + }) + )); + + let without = msg_id(); + signals().register_hook(without, None); + assert!( + !signals().signals.contains_key(&without), + "a None hook must store nothing" + ); + + // A later reply fires the stored hook (Expired branch, no wake). + signals().record_reply(&with); + assert!(fired.get()); + assert!(!signals().signals.contains_key(&with)); + }); + } + + /// `MessageSleepFuture` stays `Pending` until the block height reaches its + /// deadline, then resolves. + #[test] + fn message_sleep_future_pending_then_ready_at_deadline() { + serial(|| { + Syscall::with_block_height(800); + let mut fut = MessageSleepFuture::new(803); + assert!(matches!(poll_once(&mut fut), Poll::Pending)); + Syscall::with_block_height(803); + assert!(matches!(poll_once(&mut fut), Poll::Ready(()))); + }); + } + + #[test] + fn sleep_for_inserts_unkeyed_lock_and_zero_is_immediately_ready() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 900); + tasks().insert(message_id, Task::new(async {})); + + let mut fut = sleep_for(3); // deadline 903 + assert!( + tasks() + .get_mut(&message_id) + .unwrap() + .locks + .iter() + .any(|(_, reply_to)| reply_to.is_none()), + "sleep_for must push a lock with no reply id" + ); + assert!(matches!(poll_once(&mut fut), Poll::Pending)); + + let mut zero = sleep_for(0); // deadline == now + assert!(matches!(poll_once(&mut zero), Poll::Ready(()))); + + tasks().remove(&message_id); + }); + } + + /// Drives `message_loop` with a future that reentrantly mutates its own + /// `Task` (`sleep_for` -> `with_task_mut(insert_sleep)`) *while being + /// polled*. This is the exact aliasing shape the runtime must keep sound: + /// `message_loop` moves the future out of its `Task` for the poll, so the + /// reentrant `with_task_mut` borrow cannot overlap the polled future (no + /// `&mut future` / `&mut Task` overlap, no `Box`-noalias hazard). Miri + /// (Stacked Borrows) validates the absence of UB on this path. + #[test] + fn message_loop_reentrant_with_task_mut_during_poll_is_sound() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 7); + + // `sleep_for(0)` resolves immediately (deadline == now), so the + // future completes in a single poll without reaching `Lock::wait` + // (the suspend syscall is not available off-chain) — but it still + // performs the reentrant `with_task_mut(insert_sleep)` mid-poll. + message_loop(async { + sleep_for(0).await; + }); + + assert!( + !tasks().contains_key(&message_id), + "a completed future must tear its task down" + ); + }); + } + + /// `handle_signal` drops the task and turns its pending reply locks into + /// `Expired`-with-hook entries, so a late reply can still fire the hook. + #[cfg(not(feature = "ethexe"))] + #[test] + fn handle_signal_drops_task_and_expires_signals_preserving_hook() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 60); + clear_reply_signal_ctx(); + tasks().insert(message_id, Task::new(async {})); + + let reply_to = msg_id(); + let fired = std::rc::Rc::new(core::cell::Cell::new(false)); + let f = fired.clone(); + signals().register_signal( + reply_to, + Lock::up_to(5), + Some(Box::new(move || f.set(true))), + ); + + Syscall::with_signal_from(Ok(message_id)); + handle_signal(); + + assert!(!tasks().contains_key(&message_id), "task must be removed"); + assert!( + matches!( + signals().signals.get(&reply_to), + Some(WakeSignal::Expired { + reply_hook: Some(_), + .. + }) + ), + "pending reply locks become Expired-with-hook on signal teardown" + ); + + // A late reply after the signal still fires the hook. + signals().record_reply(&reply_to); + assert!(fired.get()); + assert!(!signals().signals.contains_key(&reply_to)); + }); + } + + /// `set_critical_hook` must reject being called inside the `handle_reply` + /// entrypoint (a reply context is present). + #[cfg(not(feature = "ethexe"))] + #[test] + #[should_panic(expected = "handle_reply")] + fn set_critical_hook_panics_in_reply_context() { + serial(|| { + Syscall::with_reply_code(Ok(ReplyCode::Success(SuccessReplyReason::Manual))); + set_critical_hook(|_| {}); + }); + } + + #[cfg(not(feature = "ethexe"))] + #[test] + fn set_critical_hook_stores_hook_on_current_task() { + serial(|| { + let message_id = msg_id(); + set_context(message_id, 1); + clear_reply_signal_ctx(); + tasks().insert(message_id, Task::new(async {})); + + set_critical_hook(|_| {}); + assert!( + tasks().get(&message_id).unwrap().critical_hook.is_some(), + "hook must be stored on the current message's task" + ); + + tasks().remove(&message_id); + }); + } + + /// `IdHasher` derives its 64-bit hash from the first 8 bytes of the id, so + /// distinct ids hash distinctly. + #[test] + fn id_hasher_reads_first_eight_le_bytes() { + use core::hash::Hasher; + let mut h = IdHasher::default(); + let mut bytes = [0u8; 32]; + bytes[..8].copy_from_slice(&7u64.to_ne_bytes()); + h.write(&bytes); + assert_eq!(h.finish(), 7u64); + + let mut h2 = IdHasher::default(); + let mut other = [0u8; 32]; + other[..8].copy_from_slice(&8u64.to_ne_bytes()); + h2.write(&other); + assert_ne!(h.finish(), h2.finish(), "distinct ids must hash distinctly"); + } +} diff --git a/rs/src/gstd/locks.rs b/rs/src/gstd/locks.rs new file mode 100644 index 000000000..1eebc91ec --- /dev/null +++ b/rs/src/gstd/locks.rs @@ -0,0 +1,90 @@ +use crate::prelude::*; +use core::cmp::Ordering; +use gstd::{BlockCount, BlockNumber, Config, exec}; + +/// Type of wait locks. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Lock { + deadline: BlockNumber, + ty: WaitType, +} + +/// Wait types. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum WaitType { + Exactly, + #[default] + UpTo, +} + +impl Lock { + /// Wait for + pub fn exactly(b: BlockCount) -> Self { + let current = Syscall::block_height(); + Self { + deadline: current.saturating_add(b), + ty: WaitType::Exactly, + } + } + + /// Wait up to + pub fn up_to(b: BlockCount) -> Self { + let current = Syscall::block_height(); + Self { + deadline: current.saturating_add(b), + ty: WaitType::UpTo, + } + } + + /// Gets the deadline of the current lock. + pub fn deadline(&self) -> BlockNumber { + self.deadline + } + + /// Gets the duration from current [`Syscall::block_height()`]. + pub fn duration(&self) -> Option { + let current = Syscall::block_height(); + self.deadline.checked_sub(current) + } + + pub fn wait_type(&self) -> WaitType { + self.ty + } + + /// Call wait functions by the lock type. + pub fn wait(&self, now: BlockNumber) { + // SAFETY: `message_loop` only calls `wait(now)` with a lock returned by + // `Task::next_lock(now)`, which guarantees `lock.deadline() > now`, so + // the subtraction never underflows. + let duration = unsafe { self.deadline.unchecked_sub(now) }; + match self.ty { + WaitType::Exactly => exec::wait_for(duration), + WaitType::UpTo => exec::wait_up_to(duration), + } + } +} + +impl PartialOrd for Lock { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for Lock { + fn cmp(&self, other: &Self) -> Ordering { + let mut ord = self.deadline().cmp(&other.deadline()); + if ord == Ordering::Equal { + ord = match self.wait_type() { + WaitType::Exactly => Ordering::Greater, + WaitType::UpTo => Ordering::Less, + } + } + ord + } +} + +impl Default for Lock { + fn default() -> Self { + Lock::up_to(Config::wait_up_to()) + } +} diff --git a/rs/src/gstd/macros.rs b/rs/src/gstd/macros.rs index be14df271..c1a756442 100644 --- a/rs/src/gstd/macros.rs +++ b/rs/src/gstd/macros.rs @@ -282,16 +282,14 @@ macro_rules! service_route_dispatch { if is_async { $crate::gstd::message_loop(async move { $svc.try_handle_async($interface_id, $entry_id, $input, |encoded_result, value| { - $crate::gstd::msg::reply_bytes(encoded_result, value) - .expect("Failed to send output"); + $crate::gstd::msg::reply_bytes(encoded_result, value).unwrap(); }) .await .unwrap_or_else(|| $crate::gstd::unknown_input_panic("Unknown request", &[])); }); } else { $svc.try_handle($interface_id, $entry_id, $input, |encoded_result, value| { - $crate::gstd::msg::reply_bytes(encoded_result, value) - .expect("Failed to send output"); + $crate::gstd::msg::reply_bytes(encoded_result, value).unwrap(); }) .unwrap_or_else(|| $crate::gstd::unknown_input_panic("Unknown request", &[])); } diff --git a/rs/src/gstd/mod.rs b/rs/src/gstd/mod.rs index 00d935f47..a04126714 100644 --- a/rs/src/gstd/mod.rs +++ b/rs/src/gstd/mod.rs @@ -1,14 +1,31 @@ +#[cfg(feature = "async-runtime")] +pub use async_runtime::{ + MessageFuture, create_program_for_reply, handle_reply_with_hook, message_loop, + send_bytes_for_reply, send_for_reply, send_one_way, sleep_for, +}; +#[cfg(feature = "async-runtime")] +pub type CreateProgramFuture = MessageFuture; +#[cfg(feature = "async-runtime")] +#[cfg(not(feature = "ethexe"))] +#[doc(hidden)] +pub use async_runtime::{handle_signal, set_critical_hook}; #[doc(hidden)] #[cfg(feature = "ethexe")] pub use ethexe::{EthEvent, EthEventExpo}; #[doc(hidden)] pub use events::{EventEmitter, SailsEvent}; +#[cfg(not(feature = "async-runtime"))] #[cfg(not(feature = "ethexe"))] #[doc(hidden)] pub use gstd::handle_signal; +#[cfg(not(feature = "async-runtime"))] #[doc(hidden)] -pub use gstd::{async_init, async_main, handle_reply_with_hook, message_loop}; +pub use gstd::msg::{CreateProgramFuture, MessageFuture}; pub use gstd::{debug, exec, msg}; +#[doc(hidden)] +#[cfg(not(feature = "async-runtime"))] +pub use gstd::{handle_reply_with_hook, message_loop}; +pub use locks::{Lock, WaitType}; use sails_idl_meta::{InterfaceId, MethodMeta}; #[doc(hidden)] pub use sails_macros::{event, export, program, service}; @@ -27,9 +44,12 @@ use crate::{ }; use gcore::stack_buffer; +#[cfg(feature = "async-runtime")] +mod async_runtime; #[cfg(feature = "ethexe")] mod ethexe; mod events; +mod locks; mod macros; pub mod services; mod syscalls; @@ -178,3 +198,195 @@ pub fn with_optimized_encode( pub fn is_empty_tuple() -> bool { TypeId::of::() == TypeId::of::<()>() } + +#[doc(hidden)] +#[macro_export] +macro_rules! ok { + ($e:expr) => { + match $e { + Ok(t) => t, + Err(err) => { + return Err(err.into()); + } + } + }; +} + +#[cfg(not(feature = "ethexe"))] +#[inline] +fn send_bytes( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + gas_limit: Option, + reply_deposit: Option, +) -> Result { + let waiting_reply_to = if let Some(gas_limit) = gas_limit { + crate::ok!(::gcore::msg::send_with_gas( + destination, + payload, + gas_limit, + value + )) + } else { + crate::ok!(::gcore::msg::send(destination, payload, value)) + }; + + if let Some(reply_deposit) = reply_deposit { + // Reserve gas for handling the reply. The error is propagated, not + // ignored, matching gstd's `#[wait_for_reply]`. On the awaited path the + // caller panics, trapping before the staged message commits. + crate::ok!(::gcore::exec::reply_deposit( + waiting_reply_to, + reply_deposit + )); + } + Ok(waiting_reply_to) +} + +#[cfg(feature = "ethexe")] +#[inline] +fn send_bytes( + destination: ActorId, + payload: &[u8], + value: ValueUnit, +) -> Result { + ::gcore::msg::send(destination, payload, value).map_err(::gstd::errors::Error::Core) +} + +#[cfg(not(feature = "async-runtime"))] +#[inline] +pub fn send_one_way( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + #[cfg(not(feature = "ethexe"))] gas_limit: Option, + #[cfg(not(feature = "ethexe"))] reply_deposit: Option, + #[cfg(not(feature = "ethexe"))] reply_hook: Option>, +) -> Result { + // The legacy gstd fallback cannot deliver a reply hook on a fire-and-forget + // send (its only hook API is bound to an awaited future). Catch misuse in + // debug builds instead of silently dropping the hook. + #[cfg(not(feature = "ethexe"))] + debug_assert!( + reply_hook.is_none(), + "reply hooks on one-way sends require the `async-runtime` feature" + ); + let waiting_reply_to = crate::ok!(send_bytes( + destination, + payload, + value, + #[cfg(not(feature = "ethexe"))] + gas_limit, + #[cfg(not(feature = "ethexe"))] + reply_deposit + )); + + Ok(waiting_reply_to) +} + +#[cfg(not(feature = "async-runtime"))] +#[cfg(not(feature = "ethexe"))] +#[inline] +pub fn send_bytes_for_reply( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + wait: Lock, + gas_limit: Option, + reply_deposit: Option, + reply_hook: Option>, +) -> Result { + let reply_deposit = reply_deposit.unwrap_or_default(); + // here can be a redirect target + let mut message_future = if let Some(gas_limit) = gas_limit { + ::gstd::msg::send_bytes_with_gas_for_reply( + destination, + payload, + gas_limit, + value, + reply_deposit, + )? + } else { + ::gstd::msg::send_bytes_for_reply(destination, payload, value, reply_deposit)? + }; + + message_future = match wait.wait_type() { + WaitType::Exactly => message_future.exactly(wait.duration())?, + WaitType::UpTo => message_future.up_to(wait.duration())?, + }; + + if let Some(reply_hook) = reply_hook { + message_future = message_future.handle_reply(reply_hook)?; + } + Ok(message_future) +} + +#[cfg(not(feature = "async-runtime"))] +#[cfg(feature = "ethexe")] +#[inline] +pub fn send_bytes_for_reply( + destination: ActorId, + payload: &[u8], + value: ValueUnit, + wait: Lock, +) -> Result { + // here can be a redirect target + let mut message_future = ::gstd::msg::send_bytes_for_reply(destination, payload, value)?; + + message_future = match wait.wait_type() { + WaitType::Exactly => message_future.exactly(wait.duration())?, + WaitType::UpTo => message_future.up_to(wait.duration())?, + }; + + Ok(message_future) +} + +#[cfg(not(feature = "async-runtime"))] +#[allow(clippy::too_many_arguments)] +#[inline] +pub fn create_program_for_reply( + code_id: CodeId, + salt: &[u8], + payload: &[u8], + value: ValueUnit, + wait: Lock, + #[cfg(not(feature = "ethexe"))] gas_limit: Option, + #[cfg(not(feature = "ethexe"))] reply_deposit: Option, + #[cfg(not(feature = "ethexe"))] reply_hook: Option>, +) -> Result<(CreateProgramFuture, ActorId), ::gstd::errors::Error> { + #[cfg(not(feature = "ethexe"))] + let mut future = if let Some(gas_limit) = gas_limit { + ::gstd::prog::create_program_bytes_with_gas_for_reply( + code_id, + salt, + payload, + gas_limit, + value, + reply_deposit.unwrap_or_default(), + )? + } else { + ::gstd::prog::create_program_bytes_for_reply( + code_id, + salt, + payload, + value, + reply_deposit.unwrap_or_default(), + )? + }; + #[cfg(feature = "ethexe")] + let mut future = ::gstd::prog::create_program_bytes_for_reply(code_id, salt, payload, value)?; + let program_id = future.program_id; + + future = match wait.wait_type() { + WaitType::Exactly => future.exactly(wait.duration())?, + WaitType::UpTo => future.up_to(wait.duration())?, + }; + + #[cfg(not(feature = "ethexe"))] + if let Some(reply_hook) = reply_hook { + future = future.handle_reply(reply_hook)?; + } + + Ok((future, program_id)) +} diff --git a/rs/src/gstd/syscalls.rs b/rs/src/gstd/syscalls.rs index 6c50a676c..092bb8325 100644 --- a/rs/src/gstd/syscalls.rs +++ b/rs/src/gstd/syscalls.rs @@ -91,6 +91,10 @@ impl Syscall { Ok(result) } + pub fn wake(message_id: MessageId) -> Result<(), ::gcore::errors::Error> { + ::gcore::exec::wake(message_id) + } + #[cfg(not(feature = "ethexe"))] pub fn system_reserve_gas(amount: GasUnit) -> Result<(), ::gcore::errors::Error> { ::gcore::exec::system_reserve_gas(amount) @@ -135,6 +139,7 @@ syscall_unimplemented!( exit(_inheritor_id: ActorId) -> !, panic(_data: &[u8]) -> !, read_bytes() -> Result, gcore::errors::Error>, + wake(_message_id: MessageId) -> Result<(), gcore::errors::Error>, system_reserve_gas(_amount: GasUnit) -> Result<(), ::gcore::errors::Error>, ); @@ -242,6 +247,10 @@ const _: () = { panic!("{:?}", data); } + pub fn wake(_message_id: MessageId) -> Result<(), ::gcore::errors::Error> { + Ok(()) + } + #[cfg(not(feature = "ethexe"))] pub fn system_reserve_gas(_amount: GasUnit) -> Result<(), ::gcore::errors::Error> { Ok(())