T09 · Insecure Skill Coding Practices
Error
- Location
- src/TaskQueue.ts:41
- Finding
- Non-Atomic Task Claiming Allows Duplicate Distributed Execution<![CDATA[ ## Vulnerability Details **File Location**: `src/TaskQueue.ts:41-56`; affected persistence operations also appear in `src/storage/RedisStorage.ts:41-49` and `src/storage/SQLiteStorage.ts:45-52` **Vulnerability Type**: Race condition / non-atomic distributed task claim **Risk Level**: High ### Vulnerable Code ```ts async claimNextReady(): Promise<TaskRecord | undefined> { const tasks = await this.storage.listTasks(); const now = Date.now(); const ready = tasks .filter((task) => ["queued", "retry_scheduled"].includes(task.status)) .filter((task) => new Date(task.runAt).getTime() <= now) .sort(compareReadyTasks); const next = ready[0]; if (!next) { return undefined; } next.status = "running"; next.attempts += 1; next.startedAt = new Date().toISOString(); next.updatedAt = next.startedAt; await this.storage.updateTask(next); await this.log(next.id, "info", "Task claimed for execution", { attempt: next.attempts }); return next; } ``` The Redis backend performs independent reads and writes: ```ts async updateTask(task: TaskRecord): Promise<void> { await this.saveTask(task); } async listTasks(): Promise<TaskRecord[]> { await this.connect(); const entries = await this.client.hVals(this.key("tasks")); return entries.map((entry) => JSON.parse(entry) as TaskRecord); } ``` The SQLite backend likewise separates listing from updating: ```ts async updateTask(task: TaskRecord): Promise<void> { await this.saveTask(task); } async listTasks(): Promise<TaskRecord[]> { const rows = this.db.prepare("SELECT id, data FROM tasks").all() as Record<string, unknown>[]; return rows.map(parseTask); } ``` ### Technical Analysis Task claiming is implemented as a read-select-write sequence rather than an atomic storage operation. Each scheduler first obtains all task records, selects the highest-priority ready task locally, and only afterward writes the updated `running` state. When two schedulers poll the same Redis or ...[truncated 2089 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions 1. Move claiming into the `QueueStorage` interface as a dedicated atomic operation, such as: ```ts claimNextReady(now: Date, workerId: string): Promise<TaskRecord | undefined>; ``` 2. For Redis, implement the selection and state transition in one Lua script or equivalent atomic transaction. The script should: - Select only an eligible queued task. - Atomically change its state to `running`. - Assign an unpredictable claim or lease token. - Record the worker identity and lease expiration. - Return the task only when the claim succeeds. 3. For SQLite, use a transaction and conditional update. The update must include an eligibility predicate and verify that exactly one row changed, for example: ```sql UPDATE tasks SET data = ? WHERE id = ? AND current_status IN ('queued', 'retry_scheduled'); ``` Consider storing status and scheduling fields in dedicated columns rather than only inside JSON so they can be selected and updated safely in SQL. 4. Require the claim token or matching worker ownership for `complete()` and `fail()` operations. This prevents a stale worker from overwriting the result after its lease expires. 5. Add lease renewal and recovery for workers that crash while holding a task. 6. Add concurrency tests using multiple queue and scheduler instances against the same Redis and SQLite storage, asserting that a task handler executes exactly once. 7. Document whether handlers must be idempotent. Even with atomic claiming, idempotency keys should be used for sensitive external side effects. ]]>