Add function to requeue cron tasks, store schedule task on cron job (#45)

Author: ChewingGlass

Cargo.lock

@@ -58,8 +58,8 @@ spl-token = "4.0.0"
 itertools = "0.13"
 tokio-graceful-shutdown = "0.15"
 solana-transaction-utils = { version = "0.3.4", path = "./solana-transaction-utils" }
-tuktuk-sdk = { version = "0.3.3", path = "./tuktuk-sdk" }
-tuktuk-program = { version = "0.3.1", path = "./tuktuk-program" }
+tuktuk-sdk = { version = "0.3.4", path = "./tuktuk-sdk" }
+tuktuk-program = { version = "0.3.2", path = "./tuktuk-program" }
 solana-account-decoder = { version = "2.2.3" }
 solana-clock = { version = "2.2.1" }
 solana-transaction-status = "2.2.3"

Cargo.toml

@@ -251,10 +251,12 @@ export async function runTask({
   task,
   crankTurner,
   fetcher = defaultFetcher,
+  nextAvailableTaskIds: argsNextAvailableTaskIds,
 }: {
   program: Program<Tuktuk>;
   task: PublicKey;
   crankTurner: PublicKey;
+  nextAvailableTaskIds?: number[],
   fetcher?: ({
     task,
     taskQueuedAt,
@@ -287,7 +289,7 @@ export async function runTask({
       };
     });
 
-    const nextAvailable = nextAvailableTaskIds(
+    const nextAvailable = argsNextAvailableTaskIds || nextAvailableTaskIds(
       taskQueueAcc.taskBitmap,
       freeTasks
     );

solana-programs/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "cron"
-version = "0.2.1"
+version = "0.2.3"
 description = "Created with Anchor"
 edition = "2021"
 

solana-programs/packages/tuktuk-sdk/src/transaction.ts

@@ -152,6 +152,7 @@ pub fn handler(ctx: Context<InitializeCronJobV0>, args: InitializeCronJobArgsV0)
         bump_seed: ctx.bumps.cron_job,
         removed_from_queue: false,
         num_transactions: 0,
+        next_schedule_task: ctx.accounts.task.key(),
     });
     ctx.accounts.user_cron_jobs.next_cron_job_id += 1;
     ctx.accounts

solana-programs/programs/cron/Cargo.toml

@@ -5,12 +5,14 @@ pub mod close_cron_job_v0;
 pub mod initialize_cron_job_v0;
 pub mod queue_cron_tasks_v0;
 pub mod remove_cron_transaction_v0;
+pub mod requeue_cron_task_v0;
 
 pub use add_cron_transaction_v0::*;
 pub use close_cron_job_v0::*;
 pub use initialize_cron_job_v0::*;
 pub use queue_cron_tasks_v0::*;
 pub use remove_cron_transaction_v0::*;
+pub use requeue_cron_task_v0::*;
 
 pub fn hash_name(name: &str) -> [u8; 32] {
     hash(name.as_bytes()).to_bytes()

solana-programs/programs/cron/src/instructions/initialize_cron_job_v0.rs

@@ -138,31 +138,34 @@ pub fn handler(ctx: Context<QueueCronTasksV0>) -> Result<RunTaskReturnV0> {
         .chars()
         .take(32)
         .collect::<String>();
-    let tasks = (0..num_tasks_to_queue as usize)
-        .filter_map(|i| {
-            let transaction = ctx.remaining_accounts[i].clone();
-            if transaction.data_is_empty() {
-                return None;
-            }
+    let tasks = std::iter::once(TaskReturnV0 {
+        trigger: TriggerV0::Timestamp(ctx.accounts.cron_job.current_exec_ts - QUEUE_TASK_DELAY),
+        transaction: TransactionSourceV0::CompiledV0(queue_tx),
+        crank_reward: None,
+        free_tasks: ctx.accounts.cron_job.num_tasks_per_queue_call + 1,
+        description: format!("queue {}", trunc_name),
+    })
+    .chain((0..num_tasks_to_queue as usize).filter_map(|i| {
+        let transaction = ctx.remaining_accounts[i].clone();
+        if transaction.data_is_empty() {
+            return None;
+        }
 
-            let parsed_transaction: CronJobTransactionV0 =
-                AccountDeserialize::try_deserialize(&mut &transaction.data.borrow()[..]).ok()?;
+        let parsed_transaction: CronJobTransactionV0 =
+            AccountDeserialize::try_deserialize(&mut &transaction.data.borrow()[..]).ok()?;
 
-            Some(TaskReturnV0 {
-                trigger,
-                transaction: parsed_transaction.transaction,
-                crank_reward: None,
-                free_tasks: free_tasks_per_transaction,
-                description: format!("{} {}", trunc_name, parsed_transaction.id),
-            })
-        })
-        .chain(std::iter::once(TaskReturnV0 {
-            trigger: TriggerV0::Timestamp(ctx.accounts.cron_job.current_exec_ts - QUEUE_TASK_DELAY),
-            transaction: TransactionSourceV0::CompiledV0(queue_tx),
+        Some(TaskReturnV0 {
+            trigger,
+            transaction: parsed_transaction.transaction,
             crank_reward: None,
-            free_tasks: ctx.accounts.cron_job.num_tasks_per_queue_call + 1,
-            description: format!("queue {}", trunc_name),
-        }));
+            free_tasks: free_tasks_per_transaction,
+            description: format!("{} {}", trunc_name, parsed_transaction.id),
+        })
+    }));
+
+    // Past all the CronJobTransaction are the free tasks
+    ctx.accounts.cron_job.next_schedule_task =
+        ctx.remaining_accounts[num_tasks_to_queue as usize].key();
 
     let res = write_return_tasks(WriteReturnTasksArgs {
         program_id: crate::ID,
@@ -206,6 +209,7 @@ pub fn handler(ctx: Context<QueueCronTasksV0>) -> Result<RunTaskReturnV0> {
                     cron_job_info.lamports()
                 );
                 ctx.accounts.cron_job.removed_from_queue = true;
+                ctx.accounts.cron_job.next_schedule_task = Pubkey::default();
                 Ok(RunTaskReturnV0 {
                     tasks: vec![],
                     accounts: vec![],

solana-programs/programs/cron/src/instructions/mod.rs

@@ -0,0 +1,151 @@
+use std::str::FromStr;
+
+use anchor_lang::{prelude::*, solana_program::instruction::Instruction, InstructionData};
+use chrono::{DateTime, Utc};
+use clockwork_cron::Schedule;
+use tuktuk_program::{
+    compile_transaction,
+    tuktuk::{
+        cpi::{accounts::QueueTaskV0, queue_task_v0},
+        program::Tuktuk,
+    },
+    types::QueueTaskArgsV0,
+    TaskQueueAuthorityV0, TaskQueueV0, TransactionSourceV0, TriggerV0,
+};
+
+use super::QUEUE_TASK_DELAY;
+use crate::{error::ErrorCode, state::CronJobV0};
+
+#[derive(AnchorSerialize, AnchorDeserialize, Clone, Default)]
+pub struct RequeueCronTaskArgsV0 {
+    pub task_id: u16,
+}
+
+#[derive(Accounts)]
+#[instruction(args: RequeueCronTaskArgsV0)]
+pub struct RequeueCronTaskV0<'info> {
+    #[account(mut)]
+    pub payer: Signer<'info>,
+    pub authority: Signer<'info>,
+    pub queue_authority: Signer<'info>,
+    #[account(
+        seeds = [b"task_queue_authority", task_queue.key().as_ref(), queue_authority.key().as_ref()],
+        bump = task_queue_authority.bump_seed,
+        seeds::program = tuktuk_program.key(),
+    )]
+    pub task_queue_authority: Box<Account<'info, TaskQueueAuthorityV0>>,
+    #[account(
+        mut,
+        has_one = authority,
+        constraint = cron_job.removed_from_queue || cron_job.next_schedule_task == Pubkey::default()
+    )]
+    pub cron_job: Box<Account<'info, CronJobV0>>,
+    #[account(mut)]
+    pub task_queue: Box<Account<'info, TaskQueueV0>>,
+    /// CHECK: Initialized in CPI
+    #[account(mut)]
+    pub task: AccountInfo<'info>,
+    /// CHECK: Used to write return data
+    #[account(
+        mut,
+        seeds = [b"task_return_account_1", cron_job.key().as_ref()],
+        bump
+    )]
+    pub task_return_account_1: AccountInfo<'info>,
+    /// CHECK: Used to write return data
+    #[account(
+        mut,
+        seeds = [b"task_return_account_2", cron_job.key().as_ref()],
+        bump
+    )]
+    pub task_return_account_2: AccountInfo<'info>,
+    pub system_program: Program<'info, System>,
+    pub tuktuk_program: Program<'info, Tuktuk>,
+}
+
+pub fn handler(ctx: Context<RequeueCronTaskV0>, args: RequeueCronTaskArgsV0) -> Result<()> {
+    let schedule = Schedule::from_str(&ctx.accounts.cron_job.schedule);
+    if let Err(e) = schedule {
+        msg!("Invalid schedule: {}", e);
+        return Err(error!(ErrorCode::InvalidSchedule));
+    }
+
+    let ts = Clock::get().unwrap().unix_timestamp;
+    let now = &DateTime::<Utc>::from_naive_utc_and_offset(
+        DateTime::from_timestamp(ts, 0).unwrap().naive_utc(),
+        Utc,
+    );
+
+    ctx.accounts.cron_job.next_schedule_task = ctx.accounts.task.key();
+    ctx.accounts.cron_job.removed_from_queue = false;
+    ctx.accounts.cron_job.current_exec_ts = schedule.unwrap().next_after(now).unwrap().timestamp();
+
+    let remaining_accounts = (ctx.accounts.cron_job.current_transaction_id
+        ..ctx.accounts.cron_job.current_transaction_id
+            + ctx.accounts.cron_job.num_tasks_per_queue_call as u32)
+        .map(|i| {
+            Pubkey::find_program_address(
+                &[
+                    b"cron_job_transaction",
+                    ctx.accounts.cron_job.key().as_ref(),
+                    &i.to_le_bytes(),
+                ],
+                &crate::ID,
+            )
+            .0
+        })
+        .collect::<Vec<Pubkey>>();
+    let (queue_tx, _) = compile_transaction(
+        vec![Instruction {
+            program_id: crate::ID,
+            accounts: [
+                crate::__cpi_client_accounts_queue_cron_tasks_v0::QueueCronTasksV0 {
+                    cron_job: ctx.accounts.cron_job.to_account_info(),
+                    task_queue: ctx.accounts.task_queue.to_account_info(),
+                    task_return_account_1: ctx.accounts.task_return_account_1.to_account_info(),
+                    task_return_account_2: ctx.accounts.task_return_account_2.to_account_info(),
+                    system_program: ctx.accounts.system_program.to_account_info(),
+                }
+                .to_account_metas(None),
+                remaining_accounts
+                    .iter()
+                    .map(|pubkey| AccountMeta::new_readonly(*pubkey, false))
+                    .collect::<Vec<AccountMeta>>(),
+            ]
+            .concat(),
+            data: crate::instruction::QueueCronTasksV0.data(),
+        }],
+        vec![],
+    )?;
+
+    let trunc_name = ctx
+        .accounts
+        .cron_job
+        .name
+        .chars()
+        .take(32)
+        .collect::<String>();
+    queue_task_v0(
+        CpiContext::new(
+            ctx.accounts.tuktuk_program.to_account_info(),
+            QueueTaskV0 {
+                payer: ctx.accounts.payer.to_account_info(),
+                queue_authority: ctx.accounts.queue_authority.to_account_info(),
+                task_queue_authority: ctx.accounts.task_queue_authority.to_account_info(),
+                task_queue: ctx.accounts.task_queue.to_account_info(),
+                task: ctx.accounts.task.to_account_info(),
+                system_program: ctx.accounts.system_program.to_account_info(),
+            },
+        ),
+        QueueTaskArgsV0 {
+            trigger: TriggerV0::Timestamp(ctx.accounts.cron_job.current_exec_ts - QUEUE_TASK_DELAY),
+            transaction: TransactionSourceV0::CompiledV0(queue_tx),
+            crank_reward: None,
+            free_tasks: ctx.accounts.cron_job.num_tasks_per_queue_call + 1,
+            id: args.task_id,
+            description: format!("queue {}", trunc_name),
+        },
+    )?;
+
+    Ok(())
+}

solana-programs/programs/cron/src/instructions/queue_cron_tasks_v0.rs

@@ -42,4 +42,11 @@ pub mod cron {
     pub fn close_cron_job_v0(ctx: Context<CloseCronJobV0>) -> Result<()> {
         close_cron_job_v0::handler(ctx)
     }
+
+    pub fn requeue_cron_task_v0(
+        ctx: Context<RequeueCronTaskV0>,
+        args: RequeueCronTaskArgsV0,
+    ) -> Result<()> {
+        requeue_cron_task_v0::handler(ctx, args)
+    }
 }