feat(ethexe): delayed messaging (#4290)

common/src/auxiliary/task_pool.rs

@@ -20,28 +20,28 @@
 
 use super::{AuxiliaryDoubleStorageWrap, BlockNumber, DoubleBTreeMap};
 use crate::scheduler::TaskPoolImpl;
-use gear_core::{ids::ProgramId, tasks::ScheduledTask};
+use gear_core::{ids::ProgramId, tasks::VaraScheduledTask};
 use std::cell::RefCell;
 
 /// Task pool implementation that can be used in a native, non-wasm runtimes.
 pub type AuxiliaryTaskpool<TaskPoolCallbacks> = TaskPoolImpl<
     TaskPoolStorageWrap,
-    ScheduledTask<ProgramId>,
+    VaraScheduledTask<ProgramId>,
     TaskPoolErrorImpl,
     TaskPoolErrorImpl,
     TaskPoolCallbacks,
 >;
 
 std::thread_local! {
-    pub(crate) static TASKPOOL_STORAGE: RefCell<DoubleBTreeMap<BlockNumber, ScheduledTask<ProgramId>, ()>> = const { RefCell::new(DoubleBTreeMap::new()) };
+    pub(crate) static TASKPOOL_STORAGE: RefCell<DoubleBTreeMap<BlockNumber, VaraScheduledTask<ProgramId>, ()>> = const { RefCell::new(DoubleBTreeMap::new()) };
 }
 
 /// `TaskPool` double storage map manager
 pub struct TaskPoolStorageWrap;
 
 impl AuxiliaryDoubleStorageWrap for TaskPoolStorageWrap {
     type Key1 = BlockNumber;
-    type Key2 = ScheduledTask<ProgramId>;
+    type Key2 = VaraScheduledTask<ProgramId>;
     type Value = ();
 
     fn with_storage<F, R>(f: F) -> R

core/src/tasks.rs

@@ -23,13 +23,16 @@ use gsys::Gas;
 use parity_scale_codec::{Decode, Encode, MaxEncodedLen};
 use scale_info::TypeInfo;
 
+/// Alias for ScheduledTask used in vara-runtime, generic across AccountId used.
+pub type VaraScheduledTask<AccountId> = ScheduledTask<AccountId, MessageId, bool>;
+
 /// Scheduled task sense and required data for processing action.
 ///
-/// CAUTION: NEVER ALLOW `ScheduledTask<AccountId>` BE A BIG DATA.
+/// CAUTION: NEVER ALLOW `ScheduledTask` BE A BIG DATA.
 /// To avoid redundant migrations only append new variant(s) to the enum
 /// with an explicit corresponding scale codec index.
 #[derive(Clone, Debug, Eq, PartialEq, PartialOrd, Ord, Encode, Decode, TypeInfo, MaxEncodedLen)]
-pub enum ScheduledTask<AccountId> {
+pub enum ScheduledTask<RFM, SD, SUM> {
     // Rent charging section.
     // -----
     /// Pause program as out of rent one.
@@ -42,7 +45,7 @@ pub enum ScheduledTask<AccountId> {
 
     /// Remove message from mailbox as out of rent one.
     #[codec(index = 2)]
-    RemoveFromMailbox(AccountId, MessageId),
+    RemoveFromMailbox(RFM, MessageId),
 
     /// Remove message from waitlist as out of rent one.
     #[codec(index = 3)]
@@ -62,7 +65,7 @@ pub enum ScheduledTask<AccountId> {
     ///
     /// The message itself stored in DispatchStash.
     #[codec(index = 6)]
-    SendDispatch(MessageId),
+    SendDispatch(SD),
 
     /// Delayed message to user sending.
     ///
@@ -72,7 +75,7 @@ pub enum ScheduledTask<AccountId> {
         /// What message to send.
         message_id: MessageId,
         /// Should it be inserted into users mailbox.
-        to_mailbox: bool,
+        to_mailbox: SUM,
     },
 
     /// Remove gas reservation.
@@ -85,9 +88,9 @@ pub enum ScheduledTask<AccountId> {
     RemoveResumeSession(u32),
 }
 
-impl<AccountId> ScheduledTask<AccountId> {
+impl<RFM, SD, SUM> ScheduledTask<RFM, SD, SUM> {
     /// Processing function of current task with given handler.
-    pub fn process_with(self, handler: &mut impl TaskHandler<AccountId>) -> Gas {
+    pub fn process_with(self, handler: &mut impl TaskHandler<RFM, SD, SUM>) -> Gas {
         use ScheduledTask::*;
 
         match self {
@@ -116,15 +119,15 @@ impl<AccountId> ScheduledTask<AccountId> {
 }
 
 /// Task handler trait for dealing with required tasks.
-pub trait TaskHandler<AccountId> {
+pub trait TaskHandler<RFM, SD, SUM> {
     // Rent charging section.
     // -----
     /// Pause program action.
     fn pause_program(&mut self, program_id: ProgramId) -> Gas;
     /// Remove code action.
     fn remove_code(&mut self, code_id: CodeId) -> Gas;
     /// Remove from mailbox action.
-    fn remove_from_mailbox(&mut self, user_id: AccountId, message_id: MessageId) -> Gas;
+    fn remove_from_mailbox(&mut self, user_id: RFM, message_id: MessageId) -> Gas;
     /// Remove from waitlist action.
     fn remove_from_waitlist(&mut self, program_id: ProgramId, message_id: MessageId) -> Gas;
     /// Remove paused program action.
@@ -136,10 +139,10 @@ pub trait TaskHandler<AccountId> {
     fn wake_message(&mut self, program_id: ProgramId, message_id: MessageId) -> Gas;
 
     /// Send delayed message to program action.
-    fn send_dispatch(&mut self, stashed_message_id: MessageId) -> Gas;
+    fn send_dispatch(&mut self, stashed_message_id: SD) -> Gas;
 
     /// Send delayed message to user action.
-    fn send_user_message(&mut self, stashed_message_id: MessageId, to_mailbox: bool) -> Gas;
+    fn send_user_message(&mut self, stashed_message_id: MessageId, to_mailbox: SUM) -> Gas;
 
     /// Remove gas reservation action.
     fn remove_gas_reservation(
@@ -158,7 +161,5 @@ fn task_encoded_size() {
     const MAX_SIZE: usize = 256;
 
     // For example we will take `AccountId` = `ProgramId` from `gear_core`.
-    type AccountId = ProgramId;
-
-    assert!(ScheduledTask::<AccountId>::max_encoded_len() <= MAX_SIZE);
+    assert!(VaraScheduledTask::<ProgramId>::max_encoded_len() <= MAX_SIZE);
 }

ethexe/common/src/db.rs

@@ -27,11 +27,20 @@ use gear_core::{
     code::InstrumentedCode,
     ids::{ActorId, CodeId, ProgramId},
 };
-use gprimitives::H256;
+use gprimitives::{MessageId, H256};
 use parity_scale_codec::{Decode, Encode};
 
-/// NOTE: key for actor id is (program_id, user_id). only used for mailbox.
-pub type ScheduledTask = gear_core::tasks::ScheduledTask<(ProgramId, ActorId)>;
+/// RemoveFromMailbox key; (msgs sources program (mailbox and queue provider), destination user id)
+pub type Rfm = (ProgramId, ActorId);
+
+/// SendDispatch key; (msgs destinations program (stash and queue provider), message id)
+pub type Sd = (ProgramId, MessageId);
+
+/// SendUserMessage key; (msgs sources program (mailbox and stash provider))
+pub type Sum = ProgramId;
+
+/// NOTE: generic keys differs to Vara and have been chosen dependent on storage organization of ethexe.
+pub type ScheduledTask = gear_core::tasks::ScheduledTask<Rfm, Sd, Sum>;
 
 #[derive(Debug, Clone, Default, Encode, Decode, serde::Serialize)]
 pub struct BlockHeader {

ethexe/db/src/database.rs

@@ -28,7 +28,7 @@ use ethexe_common::{
     BlockRequestEvent,
 };
 use ethexe_runtime_common::state::{
-    Allocations, Mailbox, MemoryPages, MessageQueue, ProgramState, Storage, Waitlist,
+    Allocations, DispatchStash, Mailbox, MemoryPages, MessageQueue, ProgramState, Storage, Waitlist,
 };
 use gear_core::{
     code::InstrumentedCode,
@@ -504,6 +504,17 @@ impl Storage for Database {
         self.cas.write(&waitlist.encode())
     }
 
+    fn read_stash(&self, hash: H256) -> Option<DispatchStash> {
+        self.cas.read(&hash).map(|data| {
+            DispatchStash::decode(&mut data.as_slice())
+                .expect("Failed to decode data into `DispatchStash`")
+        })
+    }
+
+    fn write_stash(&self, stash: DispatchStash) -> H256 {
+        self.cas.write(&stash.encode())
+    }
+
     fn read_mailbox(&self, hash: H256) -> Option<Mailbox> {
         self.cas.read(&hash).map(|data| {
             Mailbox::decode(&mut data.as_slice()).expect("Failed to decode data into `Mailbox`")

ethexe/processor/src/lib.rs

@@ -85,13 +85,16 @@ impl Processor {
         let states = self
             .db
             .block_start_program_states(block_hash)
-            .unwrap_or_default(); // TODO (breathx): shouldn't it be a panic?
+            .ok_or_else(|| {
+                anyhow!("failed to get block start program states for under-processing block")
+            })?;
 
-        let schedule = self.db.block_start_schedule(block_hash).unwrap_or_default(); // TODO (breathx): shouldn't it be a panic?
+        let schedule = self.db.block_start_schedule(block_hash).ok_or_else(|| {
+            anyhow!("failed to get block start schedule for under-processing block")
+        })?;
 
         let mut in_block_transitions = InBlockTransitions::new(header, states, schedule);
 
-        // TODO (breathx): handle resulting addresses that were changed (e.g. top up balance wont be dumped as outcome).
         for event in events {
             match event {
                 BlockRequestEvent::Router(event) => {

ethexe/runtime/common/src/journal.rs

@@ -1,7 +1,7 @@
 use crate::{
     state::{
         self, ActiveProgram, ComplexStorage, Dispatch, HashAndLen, MaybeHash, Program,
-        ProgramState, Storage,
+        ProgramState, Storage, MAILBOX_VALIDITY,
     },
     InBlockTransitions,
 };
@@ -163,49 +163,71 @@ impl<S: Storage> JournalHandler for Handler<'_, S> {
             unreachable!("deprecated: {dispatch:?}");
         }
 
-        if delay != 0 {
-            todo!("delayed sending isn't supported yet");
-        }
-
         if self
             .in_block_transitions
             .state_of(&dispatch.destination())
             .is_none()
         {
-            if !dispatch.is_reply() {
-                let expiry = self.in_block_transitions.schedule_task(
-                    state::MAILBOX_VALIDITY.try_into().expect("infallible"),
-                    ScheduledTask::RemoveFromMailbox(
-                        (dispatch.source(), dispatch.destination()),
-                        dispatch.id(),
-                    ),
-                );
-
-                self.update_state_with_storage(dispatch.source(), |storage, state| {
-                    state.mailbox_hash =
-                        storage.modify_mailbox(state.mailbox_hash.clone(), |mailbox| {
-                            mailbox
-                                .entry(dispatch.destination())
-                                .or_default()
-                                .insert(dispatch.id(), (dispatch.value(), expiry));
-                        })?;
+            let user_id = dispatch.destination();
 
-                    Ok(())
-                });
+            if !dispatch.is_reply() {
+                if let Ok(non_zero_delay) = delay.try_into() {
+                    let expiry = self.in_block_transitions.schedule_task(
+                        non_zero_delay,
+                        ScheduledTask::SendUserMessage {
+                            message_id: dispatch.id(),
+                            to_mailbox: dispatch.source(),
+                        },
+                    );
+
+                    self.update_state_with_storage(dispatch.source(), |storage, state| {
+                        let dispatch = Dispatch::from_stored(storage, dispatch.into_stored());
+
+                        state.stash_hash =
+                            storage.modify_stash(state.stash_hash.clone(), |stash| {
+                                let r =
+                                    stash.insert(dispatch.id, ((dispatch, Some(user_id)), expiry));
+                                debug_assert!(r.is_none());
+                            })?;
+
+                        Ok(())
+                    });
+
+                    return;
+                } else {
+                    let expiry = self.in_block_transitions.schedule_task(
+                        MAILBOX_VALIDITY.try_into().expect("infallible"),
+                        ScheduledTask::RemoveFromMailbox(
+                            (dispatch.source(), user_id),
+                            dispatch.id(),
+                        ),
+                    );
+
+                    self.update_state_with_storage(dispatch.source(), |storage, state| {
+                        state.mailbox_hash =
+                            storage.modify_mailbox(state.mailbox_hash.clone(), |mailbox| {
+                                mailbox
+                                    .entry(user_id)
+                                    .or_default()
+                                    .insert(dispatch.id(), (dispatch.value(), expiry));
+                            })?;
+
+                        Ok(())
+                    });
+                }
             }
 
-            // TODO (breathx): send here to in_block_transitions.
             let source = dispatch.source();
             let message = dispatch.into_parts().1;
 
-            let source_state_hash = self
+            let state_hash = self
                 .in_block_transitions
                 .state_of(&source)
                 .expect("must exist");
 
             self.in_block_transitions.modify_state_with(
                 source,
-                source_state_hash,
+                state_hash,
                 0,
                 vec![],
                 vec![OutgoingMessage::from(message)],
@@ -214,27 +236,31 @@ impl<S: Storage> JournalHandler for Handler<'_, S> {
             return;
         }
 
-        let (kind, message) = dispatch.into_parts();
-        let (id, source, destination, payload, gas_limit, value, details) = message.into_parts();
+        let destination = dispatch.destination();
+        let dispatch = Dispatch::from_stored(self.storage, dispatch.into_stored());
 
-        let payload_hash = self.storage.write_payload(payload).into();
+        if let Ok(non_zero_delay) = delay.try_into() {
+            let expiry = self.in_block_transitions.schedule_task(
+                non_zero_delay,
+                ScheduledTask::SendDispatch((destination, dispatch.id)),
+            );
 
-        let dispatch = Dispatch {
-            id,
-            kind,
-            source,
-            payload_hash,
-            value,
-            details,
-            context: None,
-        };
+            self.update_state_with_storage(destination, |storage, state| {
+                state.stash_hash = storage.modify_stash(state.stash_hash.clone(), |stash| {
+                    let r = stash.insert(dispatch.id, ((dispatch, None), expiry));
+                    debug_assert!(r.is_none());
+                })?;
 
-        self.update_state_with_storage(destination, |storage, state| {
-            state.queue_hash = storage.modify_queue(state.queue_hash.clone(), |queue| {
-                queue.push_back(dispatch);
-            })?;
-            Ok(())
-        });
+                Ok(())
+            });
+        } else {
+            self.update_state_with_storage(destination, |storage, state| {
+                state.queue_hash = storage.modify_queue(state.queue_hash.clone(), |queue| {
+                    queue.push_back(dispatch);
+                })?;
+                Ok(())
+            });
+        }
     }
 
     fn wait_dispatch(