Implement stale tasks

Cargo.toml

@@ -58,8 +58,8 @@ spl-token = "4.0.0"
 itertools = "0.13"
 tokio-graceful-shutdown = "0.15"
 solana-transaction-utils = { version = "0.1.0", path = "./solana-transaction-utils" }
-tuktuk-sdk = { version = "0.1.0", path = "./tuktuk-sdk" }
-tuktuk-program = { version = "0.1.0", path = "./tuktuk-program" }
+tuktuk-sdk = { version = "0.2.0", path = "./tuktuk-sdk" }
+tuktuk-program = { version = "0.2.0", path = "./tuktuk-program" }
 # solana-transaction-utils = { version = "0.1.0" }
 # tuktuk-sdk = { version = "0.1.0" }
 # tuktuk-program = { version = "0.1.0" }

solana-programs/programs/tuktuk/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tuktuk"
-version = "0.1.3"
+version = "0.2.0"
 description = "Created with Anchor"
 edition = "2021"
 

solana-programs/programs/tuktuk/src/instructions/initialize_task_queue_v0.rs

@@ -14,6 +14,7 @@ pub struct InitializeTaskQueueArgsV0 {
     pub name: String,
     pub capacity: u16,
     pub lookup_tables: Vec<Pubkey>,
+    pub stale_task_age: u32,
 }
 
 pub fn hash_name(name: &str) -> [u8; 32] {
@@ -81,6 +82,7 @@ pub fn handler(ctx: Context<InitializeTaskQueueV0>, args: InitializeTaskQueueArg
         created_at: Clock::get()?.unix_timestamp,
         updated_at: Clock::get()?.unix_timestamp,
         num_queue_authorities: 0,
+        stale_task_age: args.stale_task_age,
     });
     ctx.accounts
         .task_queue_name_mapping

solana-programs/programs/tuktuk/src/instructions/run_task_v0.rs

@@ -353,14 +353,19 @@ pub fn handler<'info>(
     ctx: Context<'_, '_, '_, 'info, RunTaskV0<'info>>,
     args: RunTaskArgsV0,
 ) -> Result<()> {
+    let now = Clock::get()?.unix_timestamp;
+    let task_time = match ctx.accounts.task.trigger {
+        TriggerV0::Now => now,
+        TriggerV0::Timestamp(timestamp) => timestamp,
+    };
+    ctx.accounts.task_queue.updated_at = now;
     for id in args.free_task_ids.clone() {
         require_gt!(
             ctx.accounts.task_queue.capacity,
             id,
             ErrorCode::InvalidTaskId
         );
     }
-    ctx.accounts.task_queue.updated_at = Clock::get()?.unix_timestamp;
     let remaining_accounts = ctx.remaining_accounts;
 
     let transaction = match ctx.accounts.task.transaction.clone() {
@@ -447,11 +452,15 @@ pub fn handler<'info>(
         );
     }
 
-    let mut processor = TaskProcessor::new(ctx, &transaction, args.free_task_ids)?;
+    if now.saturating_sub(task_time) <= ctx.accounts.task_queue.stale_task_age as i64 {
+        let mut processor = TaskProcessor::new(ctx, &transaction, args.free_task_ids)?;
 
-    // Process each instruction
-    for ix in &transaction.instructions {
-        processor.process_instruction(ix, remaining_accounts)?;
+        // Process each instruction
+        for ix in &transaction.instructions {
+            processor.process_instruction(ix, remaining_accounts)?;
+        }
+    } else {
+        msg!("Task is stale, closing task");
     }
 
     msg!(

solana-programs/programs/tuktuk/src/instructions/update_task_queue_v0.rs

@@ -10,6 +10,7 @@ pub struct UpdateTaskQueueArgsV0 {
     pub capacity: Option<u16>,
     pub lookup_tables: Option<Vec<Pubkey>>,
     pub update_authority: Option<Pubkey>,
+    pub stale_task_age: Option<u32>,
 }
 
 #[derive(Accounts)]
@@ -52,6 +53,9 @@ pub fn handler(ctx: Context<UpdateTaskQueueV0>, args: UpdateTaskQueueArgsV0) ->
     if let Some(update_authority) = args.update_authority {
         ctx.accounts.task_queue.update_authority = update_authority;
     }
+    if let Some(stale_task_age) = args.stale_task_age {
+        ctx.accounts.task_queue.stale_task_age = stale_task_age;
+    }
 
     resize_to_fit(
         &ctx.accounts.payer.to_account_info(),

solana-programs/programs/tuktuk/src/state.rs

@@ -39,6 +39,11 @@ pub struct TaskQueueV0 {
     pub name: String,
     pub lookup_tables: Vec<Pubkey>,
     pub num_queue_authorities: u16,
+    // Age before a task is considered stale and can be run/deleted without running the instructions.
+    // The longer this value, the more likely you have stale tasks clogging up your queue, which can cause
+    // the queue to be full and prevent new tasks from being added.
+    // The shorter this value, the more difficult it will be to debug, as failed tasks dissappear.
+    pub stale_task_age: u32,
 }
 
 #[macro_export]

tuktuk-cli/src/cmd/task_queue.rs

@@ -43,6 +43,11 @@ pub enum Cmd {
         min_crank_reward: u64,
         #[arg(long, help = "Lookup tables to create")]
         lookup_tables: Option<Vec<Pubkey>>,
+        #[arg(
+            long,
+            help = "Age before a task is considered stale and can be deleted without running the instructions. This is effectively the retention rate for debugging purposes."
+        )]
+        stale_task_age: u32,
     },
     Update {
         #[command(flatten)]
@@ -55,6 +60,11 @@ pub enum Cmd {
         update_authority: Option<Pubkey>,
         #[arg(long)]
         capacity: Option<u16>,
+        #[arg(
+            long,
+            help = "Age before a task is considered stale and can be deleted without running the instructions. This is effectively the retention rate for debugging purposes."
+        )]
+        stale_task_age: Option<u32>,
     },
     Get {
         #[command(flatten)]
@@ -137,6 +147,7 @@ impl TaskQueueCmd {
                 min_crank_reward,
                 funding_amount,
                 lookup_tables,
+                stale_task_age,
             } => {
                 let client = opts.client().await?;
 
@@ -148,6 +159,7 @@ impl TaskQueueCmd {
                         min_crank_reward: *min_crank_reward,
                         name: name.clone(),
                         lookup_tables: lookup_tables.clone().unwrap_or_default(),
+                        stale_task_age: *stale_task_age,
                     },
                     *update_authority,
                 )
@@ -204,6 +216,7 @@ impl TaskQueueCmd {
                 lookup_tables,
                 update_authority,
                 capacity,
+                stale_task_age,
             } => {
                 let client = opts.client().await?;
                 let task_queue_key = task_queue.get_pubkey(&client).await?.ok_or_else(|| {
@@ -220,6 +233,7 @@ impl TaskQueueCmd {
                         min_crank_reward: *min_crank_reward,
                         lookup_tables: lookup_tables.clone(),
                         update_authority: *update_authority,
+                        stale_task_age: *stale_task_age,
                     },
                 )
                 .await?;

tuktuk-crank-turner/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tuktuk-crank-turner"
-version = "0.1.17"
+version = "0.1.18"
 authors.workspace = true
 edition.workspace = true
 license.workspace = true

tuktuk-crank-turner/src/task_processor.rs

@@ -303,8 +303,8 @@ impl TimedTask {
                         ctx.task_queue
                             .add_task(TimedTask {
                                 task: self.task.clone(),
-                                total_retries: self.total_retries + 1,
-                                // Try again in 30 seconds with exponential backoff
+                                total_retries: 0,
+                                // Try again when task is stale
                                 task_time: now + retry_delay,
                                 task_key: self.task_key,
                                 task_queue_key: self.task_queue_key,
@@ -318,6 +318,20 @@ impl TimedTask {
                             "task {:?} failed after {} retries",
                             self.task_key, self.max_retries
                         );
+                        let task_queue = self.get_task_queue(ctx.clone()).await;
+                        ctx.task_queue
+                            .add_task(TimedTask {
+                                task: self.task.clone(),
+                                total_retries: self.total_retries + 1,
+                                // Try again in 30 seconds with exponential backoff
+                                task_time: self.task_time + task_queue.stale_task_age as u64,
+                                task_key: self.task_key,
+                                task_queue_key: self.task_queue_key,
+                                task_queue_name: self.task_queue_name.clone(),
+                                max_retries: self.max_retries,
+                                in_flight_task_ids: vec![],
+                            })
+                            .await?;
                         TASKS_FAILED
                             .with_label_values(&[self.task_queue_name.as_str(), "RetriesExceeded"])
                             .inc();

tuktuk-program/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tuktuk-program"
-version = "0.1.0"
+version = "0.2.0"
 description = "Raw rust sdk for interacting with the tuktuk program"
 
 authors.workspace = true

tuktuk-program/idls/tuktuk.json

@@ -2,7 +2,7 @@
   "address": "tuktukUrfhXT6ZT77QTU8RQtvgL967uRuVagWF57zVA",
   "metadata": {
     "name": "tuktuk",
-    "version": "0.1.2",
+    "version": "0.1.3",
     "spec": "0.1.0",
     "description": "Created with Anchor"
   },
@@ -652,6 +652,10 @@
             "type": {
               "vec": "pubkey"
             }
+          },
+          {
+            "name": "stale_task_age",
+            "type": "u32"
           }
         ]
       }
@@ -848,6 +852,10 @@
           {
             "name": "num_queue_authorities",
             "type": "u16"
+          },
+          {
+            "name": "stale_task_age",
+            "type": "u32"
           }
         ]
       }
@@ -1016,6 +1024,12 @@
             "type": {
               "option": "pubkey"
             }
+          },
+          {
+            "name": "stale_task_age",
+            "type": {
+              "option": "u32"
+            }
           }
         ]
       }

tuktuk-sdk/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tuktuk-sdk"
-version = "0.1.0"
+version = "0.2.0"
 authors.workspace = true
 edition.workspace = true
 license.workspace = true