---
name: taskengine-workerprocess-pending-running-race
description: |
  Fix two co-occurring classes of bug in BuyerKiosk TaskEngine's
  WorkerProcess.processJobInner that let two workers process the same
  task_executions row, OR silently strand a row in 'pending' forever after
  a DB exception. Use when: (1) writing or reviewing the worker pickup
  path, (2) a job's side-effects appear to fire twice in production,
  (3) executions sit in 'pending' state with no queue item pointing to
  them, (4) building new Job classes whose handle() is NOT idempotent
  and that need a single-worker guarantee per execution attempt, (5)
  building any system that uses a check-then-act sequence of load +
  in-memory-mutate + plain-WHERE-id-update for state transitions.
author: Claude Code
version: 1.0.0
date: 2026-06-03
---

# WorkerProcess pending→running race + strand-on-throw

## Problem

Two bugs that co-occur in BuyerKiosk TaskEngine's worker pickup
path. Both pre-date the spec-050 work; T20 (commit `3ff0db2c6`) closes
both.

### Bug 1 — Non-atomic pending→running transition

`WorkerProcess::processJobInner` historically did:

```php
$execution = $this->executionRepo->find($executionId);       // 1: load
if (!$execution || (!$execution->isPending() && !$execution->isRunning())) {
    return true;
}
// ... jobDef lookup ...
$execution->markStarted($this->worker?->getId() ?? 'unknown'); // 2: in-memory mutate
$this->executionRepo->update($execution);                       // 3: UPDATE WHERE id = ?
```

Step 3 has no `AND status = 'pending'` guard. If two workers picked up
the same `task_executions.id` (a single duplicate Redis delivery is
enough), both passed step 1 with the row still in `pending`, both
mutated in memory at step 2, and both updated at step 3. Both then
proceeded into `$job->handle()`, causing duplicate side effects.

Single-shot unit tests with one mocked execution never catch this
— it only manifests with concurrent invocation. The TaskEngine's own
test suite assumes a single worker.

### Bug 2 — Strand-on-throw (pre-existing in the same code path)

`processJob` has an outer `try { processJobInner } finally { ack }`
wrapper that ACKs the Redis reservation on any path that exits without
explicitly setting `$outerAcked = true`. The intent was "never orphan
a reservation in Redis." But if `processJobInner` exits early because
the DB transition write threw (Redis down, MySQL gone away, etc.):

```php
try {
    $execution->markStarted(...);
    $this->executionRepo->update($execution);
} catch (\Throwable $e) {
    error_log(...);
    return false;       // outer finally ACKs
}
```

…the outer finally ACKs the reservation. The `task_executions` row is
still in `pending` state (never transitioned), but the Redis processing
list no longer contains the message. **There is NO scheduler-side
recovery for stranded `pending` task_executions rows.** Only
`RedisQueueAdapter::getStaleProcessingItems` + `requeueItem` is
implemented, and it only finds rows still in the processing list. The
row sits in `pending` forever.

This bug is invisible until you grep the repo for
"stale recovery will re-queue" comments and realize there's no
corresponding scheduler implementation.

## Context / Trigger Conditions

- Writing or reviewing `WorkerProcess::processJobInner` or any class
  that drives `ExecutionRepository::update()` with a status transition.
- A new Job has non-idempotent side effects and you want a
  single-worker-per-attempt guarantee.
- Production symptom: a job's audit/emit/write side effects fire twice
  for the same execution id (often surfaced via duplicate audit rows
  or a downstream API rejecting the second call).
- Production symptom: `task_executions` row stuck in `pending` with
  `queued_at` long ago, no `started_at`, no Redis processing-list item.
- Code review: any DB-exception catch in WorkerProcess that exits
  without setting `$outerAcked = true`.

## Solution

### Fix 1 — Atomic pending→running

Add a single guarded UPDATE to `ExecutionRepository`:

```php
public function markStartedAtomic(int $executionId, string $workerId): bool
{
    $this->reconnectIfNeeded();
    $now = (new DateTime('now', new DateTimeZone('UTC')))->format('Y-m-d H:i:s');
    $stmt = $this->db->prepare(
        "UPDATE task_executions
            SET status     = 'running',
                workerId   = :workerId,
                started_at = :startedAt
          WHERE id     = :id
            AND status = 'pending'"
    );
    $stmt->bindValue(':workerId',  $workerId,    PDO::PARAM_STR);
    $stmt->bindValue(':startedAt', $now,         PDO::PARAM_STR);
    $stmt->bindValue(':id',        $executionId, PDO::PARAM_INT);
    $stmt->execute();
    return $stmt->rowCount() === 1;
}
```

In `WorkerProcess::processJobInner`, replace the load-then-update with:

```php
$workerId = $this->worker?->getId() ?? 'unknown';
try {
    $claimed = $this->executionRepo->markStartedAtomic($execution->getId(), $workerId);
} catch (\Throwable $e) {
    error_log("[TaskEngine Worker] Failed atomic pending→running on #{$executionId}: {$e->getMessage()} — leaving Redis reservation in flight for stale-recovery re-delivery");
    $outerAcked = true;  // see Fix 2 below
    return false;
}

if (!$claimed) {
    error_log("[TaskEngine Worker] Execution #{$executionId} no longer pending (claimed by another worker or status already advanced); skipping");
    return true; // outer finally ACKs — winner owns the work
}

// Sync the in-memory model so downstream code sees running/workerId/started_at.
$execution->markStarted($workerId);
$this->markBusy($execution->getId());
```

InnoDB row-locking on the `WHERE id = :id AND status = 'pending'`
predicate is mutually exclusive across concurrent invocations:
exactly one UPDATE returns `rowCount = 1`; the rest get 0.

### Fix 2 — Don't ACK on claim-write throw

On the DB-exception path inside processJobInner, set
`$outerAcked = true` **before** returning. This suppresses the outer
finally's ACK. The Redis reservation stays in the processing list and
`RedisQueueAdapter::getStaleProcessingItems` + `requeueItem` will
re-deliver it on the next stale-recovery tick.

This is the corollary of "never orphan a reservation." The outer ACK
is designed for the case where the worker FINISHED with the message
(success, failure, cancel — all terminal states). On a DB-write throw
we did NOT finish — we never transitioned the row. ACKing here strands
both Redis (we won't try again) AND the DB (the row is still pending).

## Verification

Three test cases close the regression door:

1. **Race-loss returns true + ACKs** — stub `markStartedAtomic` to
   return false, assert `registry->resolve` is never called,
   `update` is never called, `queue->ack` is called once,
   `processJob` returns true.

2. **Race-loss before any side effect** — same setup, additionally
   assert no audit/profile/log writes happen.

3. **Throw does NOT ACK** — stub `markStartedAtomic` to throw,
   assert `queue->ack` is **never** called, `processJob` returns false.

Plus a repository-layer regression guard that the SQL contains both
`AND status = 'pending'` (the race-closing predicate) and
`WHERE id = :id`.

Example test (from spec-050 T20):

```php
public function test_bails_without_ack_when_markStartedAtomic_throws(): void
{
    $execution = $this->createMockExecution();
    $jobDef    = $this->createMockJobDefinition();
    $this->executionRepo->method('find')->willReturn($execution);
    $this->jobDefRepo->method('find')->willReturn($jobDef);
    $this->executionRepo->expects($this->once())
        ->method('markStartedAtomic')
        ->willThrowException(new \RuntimeException('DB unreachable'));

    $this->registry->expects($this->never())->method('resolve');
    $this->executionRepo->expects($this->never())->method('update');
    $this->queue->expects($this->never())->method('ack');  // CRITICAL

    $result = $this->workerProcess->processJob('default', ['executionId' => 999]);
    $this->assertFalse($result);
}
```

## Example: spec-050 T20 (commit 3ff0db2c6)

`userfrosting/src/BuyerKiosk/TaskEngine/Infrastructure/Persistence/ExecutionRepository.php`
adds `markStartedAtomic`. `userfrosting/src/BuyerKiosk/TaskEngine/Domain/Worker/WorkerProcess.php`
applies both fixes. The work was done as the TaskEngine-layer
companion to spec-050 T19's payroll per-event atomic claim — together
they provide layered defence against the webhook double-process race
(TaskEngine: one worker per task_executions.id; Payroll: one execution
per payrollWebhookEvents.id).

## Notes

- PHPUnit gotcha: re-stubbing `->method('foo')->willReturn(X)` then
  `->method('foo')->willReturn(Y)` does NOT reliably override. Use
  `->expects(...)->method('foo')->willReturn(Y)` (the `expects` form
  overrides) OR drive the value via a property and a willReturnCallback
  in setUp.
- The pre-flight gate at the top of processJobInner that accepts
  `isPending() || isRunning()` is now largely dead code — the atomic
  UPDATE is the authoritative gate. You can tighten to pending-only,
  but it's not required for correctness.
- This pattern generalizes: any "single-handler-per-row" semantic that
  uses a non-atomic check + plain-WHERE-id update is exposed to the
  same race. Look for it whenever you see `$model->setX(...); $repo->update($model)`
  on a status column that's also used as a worker-claim gate.
- The companion payroll-layer skill is
  `taskengine-recovery-key-stale-execution-trap` — together they
  describe the layered defence model for webhook reliability in this
  codebase.

## References

- `userfrosting/src/BuyerKiosk/TaskEngine/Infrastructure/Persistence/ExecutionRepository.php`
  — `markStartedAtomic` implementation.
- `userfrosting/src/BuyerKiosk/TaskEngine/Domain/Worker/WorkerProcess.php`
  — `processJobInner` showing both fixes.
- `userfrosting/src/BuyerKiosk/TaskEngine/Infrastructure/Queue/RedisQueueAdapter.php`
  — `getStaleProcessingItems` + `requeueItem`, the stale-reservation
  recovery this fix depends on.
- spec-050 commit `3ff0db2c6` for the full diff.