chore: merge feat/tasks-cancel-and-redis into main

Author: exec-unit

src/modules/executor/executor.module.ts

@@ -1,16 +1,18 @@
 import { Module } from '@nestjs/common'
 import { TasksRepository } from '../tasks/tasks.repository.js'
-import { EventsRepository } from '../delivery/events.repository.js'
 import { CallbackAuthGuard } from './guards/callback-auth.guard.js'
 import { ExecutorService } from './executor.service.js'
 import { ExecutorController } from './executor.controller.js'
+import { DeliveryModule } from '../delivery/delivery.module.js'
 
 /**
- * API-specific module for the Executor domain.
- * Contains only HTTP controllers and their dependencies.
+ * Imports DeliveryModule to share the EventsRepository singleton
+ * already registered there — avoids a duplicate provider instance
+ * that would break NestJS DI scope semantics.
  */
 @Module({
+  imports: [DeliveryModule],
   controllers: [ExecutorController],
-  providers: [TasksRepository, EventsRepository, CallbackAuthGuard, ExecutorService],
+  providers: [TasksRepository, CallbackAuthGuard, ExecutorService],
 })
 export class ExecutorModule {}

src/modules/tasks/schemas/tasks.sql.ts

@@ -110,7 +110,7 @@ export const tasks = pgTable(
     // Hot partial index for the timeout sweeper.
     index('tasks_sweeper_idx')
       .on(t.expiresAt)
-      .where(sql`${t.status} = 'PROCESSING'`),
+      .where(sql`${t.status} IN ('PENDING', 'PROCESSING')`),
 
     uniqueIndex('tasks_idempotency_key_idx')
       .on(t.idempotencyKey)

src/modules/tasks/tasks.module.ts

@@ -2,10 +2,11 @@ import { Module } from '@nestjs/common'
 import { TasksController } from './tasks.controller.js'
 import { TasksService } from './tasks.service.js'
 import { TasksRepository } from './tasks.repository.js'
+import { EventsRepository } from '../delivery/events.repository.js'
 
 @Module({
   controllers: [TasksController],
-  providers: [TasksService, TasksRepository],
+  providers: [TasksService, TasksRepository, EventsRepository],
   exports: [],
 })
 export class TasksModule {}

src/modules/tasks/tasks.repository.ts

@@ -1,7 +1,9 @@
 import { Injectable } from '@nestjs/common'
-import { and, eq, inArray } from 'drizzle-orm'
+import { and, eq, inArray, sql } from 'drizzle-orm'
 import { tasks } from './schemas/tasks.sql.js'
 import type { TaskInsert, TaskSelect, TaskStatus } from './schemas/tasks.sql.js'
+import { taskEvents } from '../delivery/schemas/events.sql.js'
+import type { TaskEventSelect } from '../delivery/schemas/events.sql.js'
 import type { DrizzleDb } from '#src/database/drizzle/drizzle.provider.js'
 import { InjectDb } from '#src/database/drizzle/drizzle.provider.js'
 
@@ -55,4 +57,45 @@ export class TasksRepository {
 
     return result.length > 0 ? (result[0] ?? null) : null
   }
+
+  /**
+   * Atomically transitions a task to CANCELLED and inserts a 'cancelled' event.
+   * Eliminates the risk of a race condition or partial failure leaving the state
+   * machine and event log out of sync.
+   */
+  async cancelTask(
+    id: string,
+  ): Promise<{ task: TaskSelect; event: TaskEventSelect } | null> {
+    return this.db.transaction(async (tx) => {
+      const updatedTasks = await tx
+        .update(tasks)
+        .set({ status: 'CANCELLED', completedAt: new Date() })
+        .where(and(eq(tasks.id, id), inArray(tasks.status, ['PENDING', 'PROCESSING'])))
+        .returning()
+
+      if (updatedTasks.length === 0) return null
+      const task = updatedTasks[0]
+      if (!task) return null
+
+      await tx.execute(
+        sql`SELECT pg_advisory_xact_lock(hashtext('task_events'), hashtext(${id}))`,
+      )
+      const [seqRow] = await tx
+        .select({ nextSeq: sql<number>`COALESCE(MAX(${taskEvents.seq}), 0) + 1` })
+        .from(taskEvents)
+        .where(eq(taskEvents.taskId, id))
+
+      const [event] = await tx
+        .insert(taskEvents)
+        .values({
+          taskId: id,
+          seq: seqRow?.nextSeq ?? 1,
+          eventType: 'cancelled',
+          payload: {},
+        })
+        .returning()
+
+      return { task, event: event as TaskEventSelect }
+    })
+  }
 }

src/modules/tasks/tasks.service.ts

@@ -5,6 +5,7 @@ import { IdempotencyService } from '#libs/idempotency/index.js'
 import { QUEUE_ADAPTER } from '#libs/queue/index.js'
 import type { IQueueAdapter } from '#libs/queue/index.js'
 import { Inject } from '@nestjs/common'
+import { RedisService } from '#libs/redis/index.js'
 import type { CreateTaskDto, TaskCreatedResponseDto } from './dto/tasks.dto.js'
 import type { TaskSelect } from './schemas/tasks.sql.js'
 
@@ -26,13 +27,13 @@ export class TasksService {
   constructor(
     private readonly tasksRepo: TasksRepository,
     private readonly idempotency: IdempotencyService,
+    private readonly redis: RedisService,
     @Inject(QUEUE_ADAPTER) private readonly queue: IQueueAdapter,
   ) {}
 
   async create(dto: CreateTaskDto, baseUrl: string): Promise<TaskCreatedResponseDto> {
     const { idempotencyKey } = dto
 
-    // --- Idempotency check (fast path) ---
     if (idempotencyKey) {
       const slot = await this.idempotency.occupySlot(
         idempotencyKey,
@@ -63,22 +64,20 @@ export class TasksService {
         cancelUrl: dto.cancelUrl,
         webhookUrl: dto.webhookUrl,
         timeoutSeconds: dto.timeoutSeconds,
+        expiresAt: new Date(Date.now() + dto.timeoutSeconds * 1000),
         idempotencyKey: dto.idempotencyKey,
-        // Generated once and stored; the worker embeds it in the Authorization header.
-        // The executor must present it back when pushing results - never logged or
-        // returned to clients.
         callbackToken: randomUUID(),
       })
     } catch (error) {
       // Release the slot so the next retry can attempt creation cleanly.
       if (idempotencyKey) {
-        await this.idempotency.releaseSlot(idempotencyKey)
+        await this.idempotency.releaseSlot(idempotencyKey).catch((e: unknown) => {
+          this.logger.error(`Failed to release idempotency slot`, e)
+        })
       }
       throw error
     }
 
-    // attempts=1: prevents BullMQ from re-dispatching a job that has already
-    // transitioned the task to PROCESSING (double-execution guard).
     try {
       await this.queue.enqueue(
         DISPATCH_QUEUE,
@@ -93,9 +92,19 @@ export class TasksService {
         `Failed to enqueue dispatch for task ${task.id}, rolling back`,
         error instanceof Error ? error.stack : String(error),
       )
-      await this.tasksRepo.deleteById(task.id)
-      if (idempotencyKey) {
-        await this.idempotency.releaseSlot(idempotencyKey)
+      try {
+        await this.tasksRepo.deleteById(task.id)
+      } catch (rollbackErr) {
+        this.logger.error(
+          `Rollback failed for task ${task.id}`,
+          rollbackErr instanceof Error ? rollbackErr.stack : String(rollbackErr),
+        )
+      } finally {
+        if (idempotencyKey) {
+          await this.idempotency.releaseSlot(idempotencyKey).catch((e: unknown) => {
+            this.logger.error(`Failed to release idempotency slot for task ${task.id}`, e)
+          })
+        }
       }
       throw error
     }
@@ -114,16 +123,11 @@ export class TasksService {
     return task
   }
 
-  /** Cancels a PENDING task directly. Worker notifies executors for PROCESSING tasks via cancelUrl. */
+  /** Cancels a PENDING or PROCESSING task. Publishes 'cancelled' event for SSE consumers. */
   async cancel(id: string): Promise<void> {
-    const updatedTask = await this.tasksRepo.updateStatus(
-      id,
-      ['PENDING', 'PROCESSING'],
-      'CANCELLED',
-      { completedAt: new Date() },
-    )
-
-    if (!updatedTask) {
+    const result = await this.tasksRepo.cancelTask(id)
+
+    if (!result) {
       // Task was not PENDING/PROCESSING or doesn't exist. Fetch it to give an accurate error.
       const currentTask = await this.tasksRepo.findById(id)
       if (!currentTask) throw new NotFoundException(`Task ${id} not found`)
@@ -133,9 +137,12 @@ export class TasksService {
       )
     }
 
-    // cancelUrl notification is best-effort: executor receives it async
-    // via cancel.processor.
-    if (updatedTask.cancelUrl) {
+    const { task, event } = result
+
+    // Notify live SSE clients.
+    await this.redis.client.publish(`task:${id}`, JSON.stringify(event))
+
+    if (task.cancelUrl) {
       await this.queue.enqueue('cancel', { name: 'cancel', data: { taskId: id } })
     }