feat: add retry logic

migrations/20260128113327_add_retry_logic.ts

@@ -0,0 +1,23 @@
+import type { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  // Add retry columns
+  await knex.raw(
+    `ALTER TABLE tasks ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0;`,
+  );
+  await knex.raw(
+    `ALTER TABLE tasks ADD COLUMN max_attempts INTEGER NOT NULL DEFAULT 3;`,
+  );
+  await knex.raw(`ALTER TABLE tasks ADD COLUMN error_message TEXT;`);
+
+  await knex.raw(`ALTER TYPE t_status ADD VALUE IF NOT EXISTS 'failed';`);
+}
+
+export async function down(knex: Knex): Promise<void> {
+  await knex.raw(`ALTER TABLE tasks DROP COLUMN retry_count;`);
+  await knex.raw(`ALTER TABLE tasks DROP COLUMN max_attempts;`);
+  await knex.raw(`ALTER TABLE tasks DROP COLUMN error_message;`);
+
+  // Note: PostgreSQL does not support simply removing values from enums.
+  // The 'failed' value will remain in t_status; only the added columns are rolled back.
+}

src/consumer.ts

@@ -4,6 +4,10 @@ import type { Task } from './types.ts';
 
 const executeTask = async (task: Task): Promise<void> => {
   // simulate task execution, just sleep for the specified time
+  // randomly fail 30% of the time to test retry logic
+  if (Math.random() < 0.3) {
+    throw new Error('Random failure for testing');
+  }
   await new Promise((resolve) => setTimeout(resolve, task.payload.sleep));
 };
 
@@ -16,10 +20,45 @@ while (true) {
     continue;
   }
 
-  await executeTask(task);
+  try {
+    await executeTask(task);
+
+    await db('tasks')
+      .update({ status: 'done', executed_at: db.fn.now() })
+      .where({ id: task.id });
+    console.log('✅ Task done:', task.id);
+  } catch (error) {
+    const newRetryCount = task.retry_count + 1;
+
+    if (newRetryCount >= task.max_attempts) {
+      // Max attempts exceeded, mark as failed
+      await db('tasks')
+        .update({
+          status: 'failed',
+          retry_count: newRetryCount,
+          error_message: error.message,
+          executed_at: db.fn.now(),
+        })
+        .where({ id: task.id });
 
-  await db('tasks')
-    .update({ status: 'done', executed_at: db.fn.now() })
-    .where({ id: task.id });
-  console.log('Task done:', task.id);
+      console.log(
+        `❌ Task failed permanently (${newRetryCount} failures):`,
+        task.id,
+      );
+    } else {
+      await db('tasks')
+        .update({
+          status: 'pending',
+          retry_count: newRetryCount,
+          error_message: error.message,
+          picked_at: null,
+        })
+        .where({ id: task.id });
+
+      console.log(
+        `🔄 Task failed (failure ${newRetryCount}/${task.max_attempts}):`,
+        task.id,
+      );
+    }
+  }
 }

src/types.ts

@@ -3,9 +3,13 @@ export type Payload = {
 };
 
 export type Task = {
-  id: number;
+  id: string;
   payload: Payload;
-  status: 'pending' | 'in_progress' | 'done';
+  status: 'pending' | 'in_progress' | 'done' | 'failed';
+  retry_count: number;
+  max_attempts: number;
+  error_message: string | null;
   created_at: string;
-  updated_at: string;
+  picked_at: string | null;
+  executed_at: string | null;
 };