feat: schedule tasks (#493)

Closes #492

Enables scheduling static periodic tasks (repeats the exact same
instruction a fixed number of times) in the ephemeral rollup. Refer to
the
[MIMD](#479)
for detailed explanations.

# Changes

## Task scheduler

A new component that starts along the validator. It listens for changes
to a newly introduced special `TaskContext` account. A new
`TaskSchedulerConfig` is introduced to set its parameters.

## Magic program

Introduces two new Magic program instructions:
- `ScheduleTask` creates a new task that will be executed periodically.
This instruction can only be called via CPI. When scheduling the task,
all accounts used by the task must be signers of the transaction or
owned by the program that is calling it. This is to ensure that the
scheduled instructions are valid, as we do not verify the signatures of
scheduled tasks.
- `CancelTask` cancels an existing task.

Author: Dodecahedr0x

Cargo.lock

@@ -32,6 +32,7 @@ members = [
   "magicblock-rpc",
   "magicblock-rpc-client",
   "magicblock-table-mania",
+  "magicblock-task-scheduler",
   "magicblock-tokens",
   "magicblock-transaction-status",
   "magicblock-validator",
@@ -69,6 +70,7 @@ bs58 = "0.4.0"
 byteorder = "1.5.0"
 cargo-expand = "1"
 cargo-lock = "10.0.0"
+chrono = "0.4"
 clap = "4.5.40"
 conjunto-transwise = { git = "https://github.com/magicblock-labs/conjunto.git", rev = "bf82b45" }
 console-subscriber = "0.2.0"
@@ -122,7 +124,9 @@ magicblock-config = { path = "./magicblock-config" }
 magicblock-config-helpers = { path = "./magicblock-config-helpers" }
 magicblock-config-macro = { path = "./magicblock-config-macro" }
 magicblock-core = { path = "./magicblock-core" }
-magicblock-delegation-program = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "5fb8d20", features = ["no-entrypoint"] }
+magicblock-delegation-program = { git = "https://github.com/magicblock-labs/delegation-program.git", rev = "5fb8d20", features = [
+  "no-entrypoint",
+] }
 magicblock-geyser-plugin = { path = "./magicblock-geyser-plugin" }
 magicblock-ledger = { path = "./magicblock-ledger" }
 magicblock-metrics = { path = "./magicblock-metrics" }
@@ -135,6 +139,7 @@ magicblock-pubsub = { path = "./magicblock-pubsub" }
 magicblock-rpc = { path = "./magicblock-rpc" }
 magicblock-rpc-client = { path = "./magicblock-rpc-client" }
 magicblock-table-mania = { path = "./magicblock-table-mania" }
+magicblock-task-scheduler = { path = "./magicblock-task-scheduler" }
 magicblock-tokens = { path = "./magicblock-tokens" }
 magicblock-transaction-status = { path = "./magicblock-transaction-status" }
 magicblock-validator-admin = { path = "./magicblock-validator-admin" }
@@ -152,7 +157,7 @@ protobuf-src = "1.1"
 quote = "1.0"
 rand = "0.8.5"
 rayon = "1.10.0"
-rusqlite = { version = "0.34.0", features = ["bundled"] } # bundled sqlite 3.44
+rusqlite = { version = "0.37.0", features = ["bundled"] } # bundled sqlite 3.44
 rustc_version = "0.4"
 semver = "1.0.22"
 serde = "1.0.217"

Cargo.toml

@@ -0,0 +1,41 @@
+# Task Scheduler API
+
+## Architecture
+
+### Components
+
+1. **TaskContext Account**: Stores tasks and cancellation requests on-chain
+2. **Task Scheduler Service**: Runs alongside the validator to execute scheduled tasks
+3. **Database**: SQLite database for efficient task storage and retrieval
+4. **Geyser Integration**: Monitors TaskContext account changes
+
+### Data Flow
+
+1. User schedules task via program instruction
+2. Task is stored in TaskContext account
+3. Task Scheduler Service monitors TaskContext periodically
+4. Service adds task to local database
+5. Service executes tasks at scheduled intervals
+6. Service updates task state after execution
+
+## Configuration
+
+The task scheduler can be configured via the validator configuration:
+
+```toml
+[task_scheduler]
+reset = false
+millis_per_tick = 100
+```
+
+## Security Considerations
+
+- Only task authorities can cancel their own tasks
+- Database is protected by file system permissions
+
+## Performance Considerations
+
+- Database is indexed for efficient task retrieval
+- Tasks are executed in batches to minimize overhead
+- Failed task executions are logged but don't block other tasks
+- Database operations are optimized for high-frequency access 

docs/task-scheduler.md

@@ -213,6 +213,7 @@ pub fn standard_blacklisted_accounts(
     blacklisted_accounts.insert(NATIVE_SOL_ID);
     blacklisted_accounts.insert(magicblock_program::ID);
     blacklisted_accounts.insert(magicblock_program::MAGIC_CONTEXT_PUBKEY);
+    blacklisted_accounts.insert(magicblock_program::TASK_CONTEXT_PUBKEY);
     blacklisted_accounts.insert(*validator_id);
     blacklisted_accounts.insert(*faucet_id);
     blacklisted_accounts

magicblock-account-cloner/src/account_cloner.rs

@@ -141,7 +141,7 @@ impl AccountsStorage {
         // https://github.com/magicblock-labs/magicblock-validator/issues/334
         assert!(
             head.load(Relaxed) < self.meta.total_blocks as u64,
-            "database is full"
+            "database is full",
         );
 
         // SAFETY:

magicblock-accounts-db/src/storage.rs

@@ -29,7 +29,9 @@ use magicblock_committor_service::{
     transactions::MAX_PROCESS_PER_TX,
     types::{ScheduledBaseIntentWrapper, TriggerType},
 };
-use magicblock_magic_program_api::{self, MAGIC_CONTEXT_PUBKEY};
+use magicblock_magic_program_api::{
+    self, MAGIC_CONTEXT_PUBKEY, TASK_CONTEXT_PUBKEY,
+};
 use magicblock_program::{
     magic_scheduled_base_intent::{
         CommitType, CommittedAccount, MagicBaseIntent, ScheduledBaseIntent,
@@ -445,7 +447,7 @@ where
 }
 
 fn should_clone_account(pubkey: &Pubkey) -> bool {
-    pubkey != &MAGIC_CONTEXT_PUBKEY
+    pubkey != &MAGIC_CONTEXT_PUBKEY && pubkey != &TASK_CONTEXT_PUBKEY
 }
 
 /// Creates deterministic hashes from account lamports, owner and data

magicblock-accounts/src/external_accounts_manager.rs

@@ -38,6 +38,7 @@ magicblock-pubsub = { workspace = true }
 magicblock-rpc = { workspace = true }
 magicblock-transaction-status = { workspace = true }
 magicblock-validator-admin = { workspace = true }
+magicblock-task-scheduler = { workspace = true }
 magic-domain-program = { workspace = true }
 solana-geyser-plugin-interface = { workspace = true }
 solana-rpc-client = { workspace = true }

magicblock-api/Cargo.toml

@@ -90,4 +90,12 @@ pub enum ApiError {
     #[error("Accounts Database couldn't be initialized"
     )]
     AccountsDbError(#[from] AccountsDbError),
+
+    #[error("TaskSchedulerServiceError")]
+    TaskSchedulerServiceError(
+        #[from] magicblock_task_scheduler::TaskSchedulerError,
+    ),
+
+    #[error("Failed to sanitize transaction: {0}")]
+    FailedToSanitizeTransaction(#[from] solana_sdk::transaction::TransactionError),
 }

magicblock-api/src/errors.rs

@@ -2,8 +2,9 @@ use std::path::Path;
 
 use magicblock_bank::bank::Bank;
 use magicblock_magic_program_api::{
-    self, MAGIC_CONTEXT_PUBKEY, MAGIC_CONTEXT_SIZE,
+    self, MAGIC_CONTEXT_PUBKEY, TASK_CONTEXT_PUBKEY,
 };
+use magicblock_program::{MagicContext, TaskContext};
 use solana_sdk::{
     account::{Account, WritableAccount},
     clock::Epoch,
@@ -77,6 +78,15 @@ pub(crate) fn fund_magic_context(bank: &Bank) {
         bank,
         &MAGIC_CONTEXT_PUBKEY,
         u64::MAX,
-        vec![0; MAGIC_CONTEXT_SIZE],
+        MagicContext::ZERO.to_vec(),
+    );
+}
+
+pub(crate) fn fund_task_context(bank: &Bank) {
+    fund_account_with_data(
+        bank,
+        &TASK_CONTEXT_PUBKEY,
+        u64::MAX,
+        TaskContext::ZERO.to_vec(),
     );
 }

magicblock-api/src/fund_account.rs

@@ -56,7 +56,8 @@ use magicblock_metrics::MetricsService;
 use magicblock_perf_service::SamplePerformanceService;
 use magicblock_processor::execute_transaction::TRANSACTION_INDEX_LOCK;
 use magicblock_program::{
-    init_persister, validator, validator::validator_authority,
+    init_persister,
+    validator::{self, validator_authority},
     TransactionScheduler,
 };
 use magicblock_pubsub::pubsub_service::{
@@ -65,6 +66,7 @@ use magicblock_pubsub::pubsub_service::{
 use magicblock_rpc::{
     json_rpc_request_processor::JsonRpcConfig, json_rpc_service::JsonRpcService,
 };
+use magicblock_task_scheduler::{SchedulerDatabase, TaskSchedulerService};
 use magicblock_transaction_status::{
     TransactionStatusMessage, TransactionStatusSender,
 };
@@ -96,7 +98,8 @@ use crate::{
     errors::{ApiError, ApiResult},
     external_config::{cluster_from_remote, try_convert_accounts_config},
     fund_account::{
-        fund_magic_context, fund_validator_identity, funded_faucet,
+        fund_magic_context, fund_task_context, fund_validator_identity,
+        funded_faucet,
     },
     geyser_transaction_notify_listener::GeyserTransactionNotifyListener,
     init_geyser_service::{init_geyser_service, InitGeyserServiceConfig},
@@ -175,6 +178,7 @@ pub struct MagicValidator {
     pubsub_config: PubsubConfig,
     pub transaction_status_sender: TransactionStatusSender,
     claim_fees_task: ClaimFeesTask,
+    task_scheduler_handle: Option<tokio::task::JoinHandle<()>>,
 }
 
 impl MagicValidator {
@@ -247,6 +251,7 @@ impl MagicValidator {
 
         fund_validator_identity(&bank, &validator_pubkey);
         fund_magic_context(&bank);
+        fund_task_context(&bank);
         let faucet_keypair =
             funded_faucet(&bank, ledger.ledger_path().as_path())?;
 
@@ -444,6 +449,7 @@ impl MagicValidator {
             transaction_listener,
             transaction_status_sender,
             claim_fees_task: ClaimFeesTask::new(),
+            task_scheduler_handle: None,
         })
     }
 
@@ -788,6 +794,42 @@ impl MagicValidator {
         self.pubsub_handle.write().unwrap().replace(pubsub_handle);
         self.pubsub_close_handle = pubsub_close_handle;
 
+        let task_scheduler_db_path =
+            SchedulerDatabase::path(self.ledger.ledger_path().parent().expect(
+                "ledger_path didn't have a parent, should never happen",
+            ));
+        debug!(
+            "Task scheduler persists to: {}",
+            task_scheduler_db_path.display()
+        );
+        let task_scheduler_handle = TaskSchedulerService::start(
+            &task_scheduler_db_path,
+            &self.config.task_scheduler,
+            self.bank.clone(),
+            self.token.clone(),
+        )?;
+        // TODO: we should shutdown gracefully.
+        // This is discussed in this comment:
+        // https://github.com/magicblock-labs/magicblock-validator/pull/493#discussion_r2324560798
+        // However there is no proper solution for this right now.
+        // An issue to create a shutdown system is open here:
+        // https://github.com/magicblock-labs/magicblock-validator/issues/524
+        self.task_scheduler_handle = Some(tokio::spawn(async move {
+            match task_scheduler_handle.await {
+                Ok(Ok(())) => {}
+                Ok(Err(err)) => {
+                    error!("An error occurred while running the task scheduler: {:?}", err);
+                    error!("Exiting process...");
+                    std::process::exit(1);
+                }
+                Err(err) => {
+                    error!("Failed to start task scheduler: {:?}", err);
+                    error!("Exiting process...");
+                    std::process::exit(1);
+                }
+            }
+        }));
+
         self.sample_performance_service
             .replace(SamplePerformanceService::new(
                 &self.bank,