feat(ethexe): add storage key for upcoming scheduler and manage it

ethexe/common/Cargo.toml

@@ -12,6 +12,7 @@ repository.workspace = true
 [dependencies]
 gear-core.workspace = true
 gprimitives = { workspace = true, features = ["serde"] }
+common.workspace = true
 
 parity-scale-codec.workspace = true
 derive_more.workspace = true

ethexe/common/src/db.rs

@@ -30,6 +30,8 @@ use gear_core::{
 use gprimitives::H256;
 use parity_scale_codec::{Decode, Encode};
 
+pub type ScheduledTask = common::scheduler::ScheduledTask<ActorId>;
+
 #[derive(Debug, Clone, Default, Encode, Decode, serde::Serialize)]
 pub struct BlockHeader {
     pub height: u32,
@@ -73,6 +75,12 @@ pub trait BlockMetaStorage: Send + Sync {
 
     fn latest_valid_block(&self) -> Option<(H256, BlockHeader)>;
     fn set_latest_valid_block(&self, block_hash: H256, header: BlockHeader);
+
+    fn block_start_schedule(&self, block_hash: H256) -> Option<BTreeMap<u32, Vec<ScheduledTask>>>;
+    fn set_block_start_schedule(&self, block_hash: H256, map: BTreeMap<u32, Vec<ScheduledTask>>);
+
+    fn block_end_schedule(&self, block_hash: H256) -> Option<BTreeMap<u32, Vec<ScheduledTask>>>;
+    fn set_block_end_schedule(&self, block_hash: H256, map: BTreeMap<u32, Vec<ScheduledTask>>);
 }
 
 pub trait CodesStorage: Send + Sync {

ethexe/db/src/database.rs

@@ -25,7 +25,7 @@ use crate::{
     CASDatabase, KVDatabase,
 };
 use ethexe_common::{
-    db::{BlockHeader, BlockMetaStorage, CodesStorage},
+    db::{BlockHeader, BlockMetaStorage, CodesStorage, ScheduledTask},
     router::StateTransition,
     BlockRequestEvent,
 };
@@ -56,6 +56,8 @@ enum KeyPrefix {
     LatestValidBlock = 8,
     BlockHeader = 9,
     CodeValid = 10,
+    BlockStartSchedule = 11,
+    BlockEndSchedule = 12,
 }
 
 impl KeyPrefix {
@@ -271,6 +273,40 @@ impl BlockMetaStorage for Database {
             (block_hash, header).encode(),
         );
     }
+
+    fn block_start_schedule(&self, block_hash: H256) -> Option<BTreeMap<u32, Vec<ScheduledTask>>> {
+        self.kv
+            .get(&KeyPrefix::BlockStartSchedule.two(self.router_address, block_hash))
+            .map(|data| {
+                BTreeMap::decode(&mut data.as_slice())
+                    .expect("Failed to decode data into `BTreeMap`")
+            })
+    }
+
+    fn set_block_start_schedule(&self, block_hash: H256, map: BTreeMap<u32, Vec<ScheduledTask>>) {
+        log::trace!(target: LOG_TARGET, "For block {block_hash} set block start schedule: {map:?}");
+        self.kv.put(
+            &KeyPrefix::BlockStartSchedule.two(self.router_address, block_hash),
+            map.encode(),
+        );
+    }
+
+    fn block_end_schedule(&self, block_hash: H256) -> Option<BTreeMap<u32, Vec<ScheduledTask>>> {
+        self.kv
+            .get(&KeyPrefix::BlockEndSchedule.two(self.router_address, block_hash))
+            .map(|data| {
+                BTreeMap::decode(&mut data.as_slice())
+                    .expect("Failed to decode data into `BTreeMap`")
+            })
+    }
+
+    fn set_block_end_schedule(&self, block_hash: H256, map: BTreeMap<u32, Vec<ScheduledTask>>) {
+        log::trace!(target: LOG_TARGET, "For block {block_hash} set block end schedule: {map:?}");
+        self.kv.put(
+            &KeyPrefix::BlockEndSchedule.two(self.router_address, block_hash),
+            map.encode(),
+        );
+    }
 }
 
 impl CodesStorage for Database {

ethexe/observer/src/query.rs

@@ -70,6 +70,8 @@ impl Query {
         self.database.set_block_is_empty(hash, true);
         self.database
             .set_block_end_program_states(hash, Default::default());
+        self.database
+            .set_block_end_schedule(hash, Default::default());
 
         // set latest valid if empty.
         if self.database.latest_valid_block().is_none() {
@@ -275,6 +277,13 @@ impl Query {
         self.database
             .set_block_start_program_states(block_hash, program_state_hashes);
 
+        // Propagate scheduled tasks
+        let schedule = self
+            .database
+            .block_end_schedule(parent)
+            .ok_or_else(|| anyhow!("parent block schedule not found"))?;
+        self.database.set_block_start_schedule(block_hash, schedule);
+
         // Propagate `wait for commitment` blocks queue
         let queue = self
             .database

ethexe/processor/Cargo.toml

@@ -18,6 +18,7 @@ gprimitives.workspace = true
 gear-runtime-interface = { workspace = true, features = ["std"] }
 gear-lazy-pages.workspace = true
 core-processor.workspace = true
+common = { workspace = true, features = ["std"] }
 
 anyhow = { workspace = true, features = ["std"] }
 wasmtime.workspace = true

ethexe/processor/src/handling/events.rs

@@ -1,3 +1,21 @@
+// This file is part of Gear.
+//
+// Copyright (C) 2024 Gear Technologies Inc.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
 use anyhow::Result;
 use ethexe_common::{
     mirror::RequestEvent as MirrorEvent,

ethexe/processor/src/handling/mod.rs

@@ -1,3 +1,21 @@
+// This file is part of Gear.
+//
+// Copyright (C) 2024 Gear Technologies Inc.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
 use crate::Processor;
 use anyhow::Result;
 use ethexe_db::CodesStorage;
@@ -6,6 +24,7 @@ use gprimitives::{CodeId, H256};
 
 pub(crate) mod events;
 pub(crate) mod run;
+pub(crate) mod tasks;
 
 impl Processor {
     pub(crate) fn handle_message_queueing(

ethexe/processor/src/handling/tasks.rs

@@ -0,0 +1,99 @@
+// This file is part of Gear.
+//
+// Copyright (C) 2024 Gear Technologies Inc.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+use crate::{LocalOutcome, Processor};
+use anyhow::Result;
+use common::{
+    scheduler::{ScheduledTask, TaskHandler},
+    CodeId, Gas, MessageId, ProgramId, ReservationId,
+};
+use ethexe_db::BlockMetaStorage;
+use gear_core::message::Message;
+use gprimitives::{ActorId, H256};
+use std::collections::BTreeMap;
+
+impl Processor {
+    pub fn run_tasks(
+        &mut self,
+        block_hash: H256,
+        states: &mut BTreeMap<ProgramId, H256>,
+        tasks: &mut BTreeMap<u32, Vec<ScheduledTask<ActorId>>>,
+    ) -> Result<Vec<LocalOutcome>> {
+        let mut handler = TasksHandler {
+            states,
+            results: Default::default(),
+            to_users_messages: Default::default(),
+        };
+
+        let block_meta = self
+            .db
+            .block_header(block_hash)
+            .ok_or_else(|| anyhow::anyhow!("block header for chain head wasn't found"))?;
+
+        let tasks = tasks.remove(&block_meta.height).unwrap_or_default();
+
+        for task in tasks {
+            let _gas = task.process_with(&mut handler);
+        }
+
+        Ok(vec![])
+    }
+}
+
+#[allow(unused)]
+pub struct TasksHandler<'a> {
+    pub states: &'a mut BTreeMap<ProgramId, H256>,
+    pub results: BTreeMap<ActorId, H256>,
+    pub to_users_messages: Vec<Message>,
+}
+
+impl<'a> TaskHandler<ActorId> for TasksHandler<'a> {
+    fn remove_from_mailbox(&mut self, _user_id: ActorId, _message_id: MessageId) -> Gas {
+        unimplemented!("TODO (breathx)")
+    }
+    fn remove_from_waitlist(&mut self, _program_id: ProgramId, _message_id: MessageId) -> Gas {
+        unimplemented!("TODO (breathx)")
+    }
+    fn send_dispatch(&mut self, _stashed_message_id: MessageId) -> Gas {
+        unimplemented!("TODO (breathx)")
+    }
+    fn send_user_message(&mut self, _stashed_message_id: MessageId, _to_mailbox: bool) -> Gas {
+        unimplemented!("TODO (breathx)")
+    }
+    fn wake_message(&mut self, _program_id: ProgramId, _message_id: MessageId) -> Gas {
+        // TODO (breathx): consider deprecation of delayed wakes + non-concrete waits.
+        unimplemented!("TODO (breathx)")
+    }
+
+    /* Deprecated APIs */
+    fn pause_program(&mut self, _: ProgramId) -> Gas {
+        unreachable!("deprecated")
+    }
+    fn remove_code(&mut self, _: CodeId) -> Gas {
+        unreachable!("deprecated")
+    }
+    fn remove_gas_reservation(&mut self, _: ProgramId, _: ReservationId) -> Gas {
+        unreachable!("deprecated")
+    }
+    fn remove_paused_program(&mut self, _: ProgramId) -> Gas {
+        unreachable!("deprecated")
+    }
+    fn remove_resume_session(&mut self, _: u32) -> Gas {
+        unreachable!("deprecated")
+    }
+}

ethexe/processor/src/lib.rs

@@ -86,7 +86,9 @@ impl Processor {
         let mut states = self
             .db
             .block_start_program_states(block_hash)
-            .unwrap_or_default();
+            .unwrap_or_default(); // TODO (breathx): shouldn't it be a panic?
+
+        let mut schedule = self.db.block_start_schedule(block_hash).unwrap_or_default(); // TODO (breathx): shouldn't it be a panic?
 
         let mut all_value_claims = Default::default();
 
@@ -104,6 +106,9 @@ impl Processor {
             }
         }
 
+        // TODO (breathx): handle outcomes.
+        let mut _outcomes = self.run_tasks(block_hash, &mut states, &mut schedule)?;
+
         let mut outcomes = self.run(block_hash, &mut states)?;
 
         for outcome in &mut outcomes {
@@ -118,6 +123,7 @@ impl Processor {
         }
 
         self.db.set_block_end_program_states(block_hash, states);
+        self.db.set_block_end_schedule(block_hash, schedule);
 
         Ok(outcomes)
     }

ethexe/processor/src/tests.rs

@@ -36,6 +36,7 @@ fn init_new_block(processor: &mut Processor, meta: BlockHeader) -> H256 {
     chain_head
 }
 
+#[track_caller]
 fn init_new_block_from_parent(processor: &mut Processor, parent_hash: H256) -> H256 {
     let parent_block_header = processor.db.block_header(parent_hash).unwrap_or_default();
     let height = parent_block_header.height + 1;
@@ -48,13 +49,35 @@ fn init_new_block_from_parent(processor: &mut Processor, parent_hash: H256) -> H
             parent_hash,
         },
     );
+
     let parent_out_program_hashes = processor
         .db
         .block_end_program_states(parent_hash)
-        .unwrap_or_default();
+        .unwrap_or_else(|| {
+            if parent_hash.is_zero() {
+                Default::default()
+            } else {
+                panic!("process block events before new block; start states not found")
+            }
+        });
     processor
         .db
         .set_block_start_program_states(chain_head, parent_out_program_hashes);
+
+    let parent_out_schedule = processor
+        .db
+        .block_end_schedule(parent_hash)
+        .unwrap_or_else(|| {
+            if parent_hash.is_zero() {
+                Default::default()
+            } else {
+                panic!("process block events before new block; start schedule not found")
+            }
+        });
+    processor
+        .db
+        .set_block_start_schedule(chain_head, parent_out_schedule);
+
     chain_head
 }
 
@@ -83,6 +106,7 @@ fn process_observer_event() {
         }]
     );
 
+    let _ = processor.process_block_events(ch0, vec![]).unwrap();
     let ch1 = init_new_block_from_parent(&mut processor, ch0);
 
     let actor_id = ActorId::from(42);