Transmission 025 ended with one confirmed gap. The stack was running. The schema was there. The state machine was wired. What had not been verified was whether a task submitted through the gateway actually made it through the worker and into the database.

That was one curl and one psql query. Both ran at the start of this session.

The constraint

025 left the worker in a simple state: on success, write completed. On any exception, write failed and leave the message in the Redis PEL — unacknowledged, waiting for reclaim. That is not a retry mechanism. That is a message stuck in the PEL until a human intervenes or a timeout triggers XCLAIM.

The goal for this session was to replace that with a proper retry policy:

• Track how many times a task has been attempted.
• On a transient failure with attempts remaining, reset the task to pending, increment the counter, and re-queue it.
• On permanent failure — attempts exhausted — mark it failed and acknowledge the message so it stops blocking the stream.
• Add a simulated failure condition so the entire path could be tested without needing a real downstream service to misbehave.

That meant touching three places: the database schema, the database helper module, and the worker’s main loop.

The proof

First: closing the gap from 025

The first thing that ran in this session was the task that 025 had not confirmed.

$ curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"name": "generate_report", "payload": {"user_id": 42, "target": "q3_metrics"}}'
{"task_id":"26fd9233-55a2-453b-8825-3b5ea4e479d2","name":"generate_report","status":"accepted","accepted_at":"2026-08-27T06:34:57.439656+00:00"}

worker-1  | 2026-08-27T06:35:09Z [INFO] nexusflow.worker — Processing task_id=26fd9233-55a2-453b-8825-3b5ea4e479d2  name=generate_report
worker-1  | 2026-08-27T06:35:09Z [INFO] nexusflow.worker — Task completed  task_id=26fd9233-55a2-453b-8825-3b5ea4e479d2

               task_id                |      name       |  status   |          created_at           |          updated_at
--------------------------------------+-----------------+-----------+-------------------------------+-------------------------------
 26fd9233-55a2-453b-8825-3b5ea4e479d2 | generate_report | completed | 2026-08-27 06:34:57.439656+00 | 2026-08-27 06:35:19.662195+00

Task submitted, worker consumed it, Postgres row updated to completed. The gap from 025 is closed.

The worker also shut down cleanly when signalled:

worker-1  | 2026-08-27T07:05:46Z [INFO] nexusflow.worker — Received signal 15 — finishing current task then shutting down.
worker-1  | 2026-08-27T07:05:47Z [INFO] nexusflow.db — PostgreSQL pool closed.
worker-1  | 2026-08-27T07:05:47Z [INFO] nexusflow.worker — Shutdown flag set. Worker 'worker-1' exiting cleanly.
worker-1 exited with code 0

Signal 15 is SIGTERM — the same signal Kubernetes sends before killing a pod. The loop finished its current task, closed the pool, and exited with code 0.

Schema update

Three columns added to postgres/init.sql:

CREATE TABLE IF NOT EXISTS tasks (
    task_id       VARCHAR(64)  PRIMARY KEY,
    name          VARCHAR(128) NOT NULL,
    status        VARCHAR(32)  NOT NULL DEFAULT 'pending',
    payload       JSONB,
    result        JSONB,
    retry_count   INTEGER      NOT NULL DEFAULT 0,
    max_retries   INTEGER      NOT NULL DEFAULT 3,
    error_message TEXT,
    created_at    TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    updated_at    TIMESTAMPTZ  NOT NULL DEFAULT NOW()
);

retry_count tracks attempts made so far. max_retries sets the ceiling — defaulting to 3 at the schema level means every task gets retry behaviour without the gateway needing to specify it. error_message is nullable TEXT, written only on failure. max_retries lives per row rather than as a global constant so individual tasks can be given a different ceiling at insert time if needed.

New database helpers

worker/db.py got two new functions alongside the existing update_task_status.

fetch_task_retry_info runs a single SELECT before each attempt and returns the current retry_count and max_retries for the row:

async def fetch_task_retry_info(task_id: str) -> dict[str, Any]:
    async with _pool_or_raise().acquire() as conn:
        row = await conn.fetchrow(
            "SELECT retry_count, max_retries FROM tasks WHERE task_id = $1",
            task_id,
        )
    if row is None:
        raise LookupError(f"Task not found in database: task_id={task_id!r}")
    return {"retry_count": row["retry_count"], "max_retries": row["max_retries"]}

update_task_for_retry does one atomic UPDATE — increment retry_count, reset status to pending, and write the exception string to error_message. One round-trip, no race window:

async def update_task_for_retry(task_id: str, error_message: str) -> None:
    async with _pool_or_raise().acquire() as conn:
        await conn.execute(
            """
            UPDATE tasks
               SET status        = 'pending',
                   retry_count   = retry_count + 1,
                   error_message = $2,
                   updated_at    = $3
             WHERE task_id = $1
            """,
            task_id, error_message, datetime.now(tz=timezone.utc),
        )

update_task_as_failed sets status = 'failed' and writes the final error message. No counter increment — the count already reflects how many attempts were made.

The retry loop

The run() function in worker/worker.py was rewritten. The shape of the exception handler is the part that changed:

# Before each attempt — reads current state
retry_info = await db.fetch_task_retry_info(task_id)
retry_count = retry_info["retry_count"]
max_retries = retry_info["max_retries"]

await db.update_task_status(task_id, "processing")

try:
    result = await process_task(task_id, fields)

    # Success
    client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)
    await db.update_task_status(task_id, "completed", result=result)

except Exception as exc:
    error_str = str(exc)

    if retry_count < max_retries:
        # Retries remaining — reset, back off, re-queue, ACK original
        await db.update_task_for_retry(task_id, error_str)
        await asyncio.sleep(REQUEUE_DELAY_S)
        client.xadd(STREAM_NAME, fields)   # new stream entry
        client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)  # clear original from PEL

    else:
        # Retries exhausted — mark failed, ACK to clear PEL
        await db.update_task_as_failed(task_id, error_str)
        client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)

The re-queue strategy uses XADD rather than leaving the original message in the PEL and waiting for XCLAIM to reclaim it. XCLAIM would work — it is the canonical Redis Streams approach to at-least-once re-delivery — but it requires a separate idle-time scan and introduces a reclaim delay that is separate from the intentional back-off. XADD gives the same re-delivery guarantee with a simpler implementation: acknowledge the original to clean the PEL, publish a fresh copy with identical fields. Any consumer in the group picks it up on the next read. The back-off (REQUEUE_DELAY_S, defaulting to 1 second) is configurable via REQUEUE_DELAY_MS in the environment.

The simulated failure is a single condition in process_task:

if task_name == "fail_task":
    raise ValueError("Simulated failure for testing")

The rebuild

docker compose down -v wiped the volumes, which wiped the old schema. docker compose up --build -d rebuilt both images and started all four containers. The first attempt exited with a race condition — PostgreSQL was still in crash recovery from the unclean shutdown and the worker tried to connect before it was ready:

asyncpg.exceptions.CannotConnectNowError: the database system is starting up

This is the same failure mode from 025. Docker’s restart policy brought the worker back up. It cycled through seven attempts over about 60 seconds before PostgreSQL finished its fsync walk and accepted connections:

worker-1  | 2026-08-27T06:19:18Z [INFO] nexusflow.worker — Connected to Redis at redis:6379
worker-1  | 2026-08-27T06:19:18Z [INFO] nexusflow.db — PostgreSQL pool ready  host=postgres db=nexusflow  min=1 max=5

The second docker compose up -d skipped the build and brought the remaining containers up cleanly.

NAME                   IMAGE                STATUS                        PORTS
nexusflow-gateway-1    nexusflow-gateway    Up About a minute (healthy)   0.0.0.0:8000->8000/tcp
nexusflow-postgres-1   postgres:16-alpine   Up 4 minutes (healthy)        5432/tcp
nexusflow-redis-1      redis:7-alpine       Up 4 minutes (healthy)        6379/tcp
nexusflow-worker-1     nexusflow-worker     Up About a minute

Proving the retry path

$ curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"name": "fail_task", "payload": {"test": "retry_logic"}}'
{"task_id":"e56d7cf4-1957-4b2d-bf3d-53cb8dadb065","name":"fail_task","status":"accepted","accepted_at":"2026-08-27T07:15:07.650275+00:00"}

The task entered the stream. The worker picked it up, raised ValueError("Simulated failure for testing") on every attempt, re-queued it twice, and on the third attempt exhausted max_retries. Then:

               task_id                |   name    | status | retry_count | max_retries |         error_message
--------------------------------------+-----------+--------+-------------+-------------+-------------------------------
 e56d7cf4-1957-4b2d-bf3d-53cb8dadb065 | fail_task | failed |           3 |           3 | Simulated failure for testing

retry_count = 3, max_retries = 3, status = failed. The machine counted to the ceiling, wrote the error, and stopped.

Where this sits

The retry path works end to end. The open item is the PostgreSQL startup race — the worker crashes multiple times before the database is ready on a fresh docker compose up. The restart policy absorbs it for now, but it is not a clean startup. That gets a proper fix next: a retry loop with exponential back-off inside init_pool so the worker waits for Postgres rather than crashing and relying on Docker to restart it.