Transmission 026 ended with one open item in the status table: the PostgreSQL startup race. The worker crashes on fresh docker compose up because it tries to connect before Postgres finishes initialising. The restart policy absorbs it, but it takes up to 60 seconds and seven restarts to stabilise. That is not a clean startup.

That was the first thing in this session. It was not the last.

The constraint

026 left three gaps:

1. The worker crashes on init_pool if PostgreSQL is still starting — no retry logic, just a cold exception and a Docker restart.
2. GET /tasks/{task_id} returns task_id, name, status, result — none of the retry columns added in 026. A caller looking at a failed task cannot see how many times it was attempted or what the error was without querying Postgres directly.
3. The retry back-off is flat. Every attempt waits 1 second before re-queuing. A downstream service that is overwhelmed gets hammered at a fixed rate regardless of how many retries have already failed.

A fourth gap came up during planning: if a worker process dies mid-task — after XREADGROUP delivers the message but before it can XACK or re-queue — that message stays in the Redis Pending Entries List indefinitely. No other consumer receives it. It is stuck until a human runs XCLAIM manually or the process comes back up.

Then a fifth: the gateway accepts duplicate submissions for the same logical operation. Nothing prevents a client from submitting the same task twice if a network timeout causes it to retry the POST.

Five things. Three batches.

The proof

Batch 1: PostgreSQL startup retry loop and retry telemetry

worker/db.py — init_pool with back-off

The fix was a loop, not a longer startup delay. asyncpg.create_pool is now wrapped in a for block that retries up to 10 times with a 2-second wait between attempts:

_PG_MAX_RETRIES = 10
_PG_RETRY_DELAY = 2  # seconds between each attempt

async def init_pool(min_size: int = 1, max_size: int = 5) -> None:
    global _pool

    last_exc: Exception | None = None
    for attempt in range(1, _PG_MAX_RETRIES + 1):
        try:
            _pool = await asyncpg.create_pool(dsn=_DSN, min_size=min_size, max_size=max_size)
            log.info("PostgreSQL pool ready  host=%s db=%s  min=%d max=%d",
                     _PG_HOST, _PG_DB, min_size, max_size)
            return
        except (
            asyncpg.PostgresError,
            asyncpg.exceptions.CannotConnectNowError,
            ConnectionRefusedError,
            OSError,
        ) as exc:
            last_exc = exc
            log.warning("PostgreSQL not ready (attempt %d/%d): %s — retrying in %ds …",
                        attempt, _PG_MAX_RETRIES, exc, _PG_RETRY_DELAY)
            await asyncio.sleep(_PG_RETRY_DELAY)

    raise RuntimeError(
        f"Could not connect to PostgreSQL after {_PG_MAX_RETRIES} attempts."
    ) from last_exc

OSError is in the catch tuple alongside asyncpg.PostgresError because the low-level TCP refusal arrives before the Postgres handshake begins — asyncpg raises OSError in that case, not a Postgres-specific exception. The loop catches both. After 10 failed attempts it re-raises as RuntimeError, chained to the last exception so the traceback survives into Docker logs.

The alternative was a healthcheck dependency in docker-compose.yml with condition: service_healthy on the postgres service. That works at the orchestrator level but does not help in Kubernetes, where the worker and postgres pods start independently with no compose-style dependency graph. A retry loop inside the process is portable across both environments. Compose healthcheck solves today’s problem. The retry loop solves it everywhere.

gateway/db.py — fetch_task expanded

The SELECT in fetch_task previously stopped at updated_at. Three columns added:

SELECT task_id, name, status, payload, result,
       retry_count, max_retries, error_message,
       created_at, updated_at
  FROM tasks
 WHERE task_id = $1

The returned dict includes all three. gateway/main.py’s TaskStatusResponse model picks them up directly — no extra mapping step.

gateway/main.py — TaskStatusResponse

class TaskStatusResponse(BaseModel):
    task_id: str
    name: str
    status: str
    payload: dict | None = None
    result: dict | None = None
    retry_count: int = Field(0, description="Number of times this task has been retried.")
    max_retries: int = Field(3, description="Maximum retry attempts allowed for this task.")
    error_message: str | None = Field(None, description="Last error recorded for this task, if any.")
    created_at: str
    updated_at: str

A caller hitting GET /tasks/{task_id} on a failed task now sees exactly how many attempts were made and what the last exception was. No database access required.

Batch 2: Exponential back-off and PEL reclaim

worker/worker.py — exponential back-off

REQUEUE_DELAY_S is gone. Two constants replace it:

REQUEUE_BASE_DELAY_S = int(os.getenv("REQUEUE_BASE_DELAY_MS", "1000")) / 1000.0
REQUEUE_MAX_DELAY_S  = int(os.getenv("REQUEUE_MAX_DELAY_MS", "30000")) / 1000.0

The back-off calculation runs on every retry:

backoff_s = min(
    REQUEUE_BASE_DELAY_S * (2 ** max(retry_count - 1, 0)),
    REQUEUE_MAX_DELAY_S,
)
log.info("task_id=%s back-off %.1fs before retry %d/%d.",
         task_id, backoff_s, retry_count + 1, max_retries)
await asyncio.sleep(backoff_s)

retry_count here is the pre-increment value — what was in the database before update_task_for_retry ran. So the progression is:

max(retry_count - 1, 0) at retry_count = 0 gives 2^0 = 1, so the first failure never waits zero seconds.

worker/worker.py — PEL reclaim background task

reclaim_idle_pel runs as asyncio.create_task("pel-reclaim") at the start of run(). It sleeps for PEL_SCAN_INTERVAL_S (default 30 seconds), then calls XPENDING_RANGE to list all pending entries in the group:

pending = client.xpending_range(
    name=STREAM_NAME,
    groupname=CONSUMER_GROUP,
    min="-",
    max="+",
    count=100,
)

For each entry with time_since_delivered >= PEL_IDLE_THRESHOLD_MS (default 60 seconds), it calls XCLAIM to take ownership, re-publishes the message fields with XADD, and removes the stale PEL entry with XACK:

claimed = client.xclaim(
    name=STREAM_NAME,
    groupname=CONSUMER_GROUP,
    consumername=CONSUMER_NAME,
    min_idle_time=PEL_IDLE_THRESHOLD_MS,
    message_ids=[msg_id],
)

for _claimed_id, fields in claimed:
    task_id = fields.get("task_id", msg_id)
    log.warning("Reclaiming orphaned PEL message  msg_id=%s  task_id=%s  "
                "idle_ms=%d  delivery_count=%d",
                msg_id, task_id, idle_ms, entry["times_delivered"])
    client.xadd(STREAM_NAME, fields)
    client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)

The loop catches asyncio.CancelledError separately from everything else. On cancel it logs and returns immediately — no silent suppression. In the finally block of run(), the task is cancelled and awaited before the Postgres pool closes:

finally:
    pel_task.cancel()
    try:
        await pel_task
    except asyncio.CancelledError:
        pass
    await db.close_pool()

Batch 3: Idempotency

Schema — postgres/init.sql

One new column and one new index:

CREATE TABLE IF NOT EXISTS tasks (
    ...
    idempotency_key VARCHAR(255),
    ...
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_tasks_idempotency_key
    ON tasks (idempotency_key)
    WHERE idempotency_key IS NOT NULL;

The index is partial — WHERE idempotency_key IS NOT NULL. Standard submissions with no key write NULL to that column. NULL values are not subject to the uniqueness constraint in PostgreSQL, so every keyless submission still creates a new row. Only rows with an actual key value are compared. A second INSERT with the same non-null key would violate the index, but the application checks for duplicates before inserting so that case never reaches the database.

gateway/db.py — two changes

create_task gains a new parameter, idempotency_key: str | None = None, written as $4 in the INSERT:

INSERT INTO tasks (task_id, name, status, payload, idempotency_key,
                   created_at, updated_at)
VALUES ($1, $2, 'pending', $3::jsonb, $4, $5, $5)

A new helper, fetch_task_by_idempotency_key, runs a SELECT on the column and returns the slim dict shape that maps directly onto TaskResponse:

async def fetch_task_by_idempotency_key(key: str) -> dict[str, Any] | None:
    async with _pool_or_raise().acquire() as conn:
        row = await conn.fetchrow(
            """
            SELECT task_id, name, status, created_at
              FROM tasks
             WHERE idempotency_key = $1
            """,
            key,
        )
    if row is None:
        return None
    return {
        "task_id":     row["task_id"],
        "name":        row["name"],
        "status":      row["status"],
        "accepted_at": row["created_at"].isoformat(),
    }

gateway/main.py — the endpoint

TaskRequest gains an optional idempotency_key body field. TaskResponse echoes it back. The handler accepts an Idempotency-Key HTTP header too, with the header taking precedence over the body field:

async def create_task(
    task: TaskRequest,
    idempotency_key_header: Optional[str] = Header(
        None,
        alias="Idempotency-Key",
    ),
) -> TaskResponse:
    effective_key: str | None = idempotency_key_header or task.idempotency_key

    if effective_key is not None:
        existing = await db.fetch_task_by_idempotency_key(effective_key)
        if existing is not None:
            return JSONResponse(
                status_code=status.HTTP_200_OK,
                content=TaskResponse(
                    task_id=existing["task_id"],
                    name=existing["name"],
                    status=existing["status"],
                    accepted_at=existing["accepted_at"],
                    idempotency_key=effective_key,
                ).model_dump(),
            )

    # new task — normal path
    ...

A hit returns 200 OK via JSONResponse directly, overriding the endpoint’s default 202 Accepted. A miss falls through to the normal INSERT + XADD path and returns 202 Accepted. Both status codes are declared in the responses dict on the decorator so they appear in the OpenAPI docs.

The header takes precedence over the body field so HTTP-level infrastructure — load balancers, API gateways — can inject or strip the key independently of the request body.

Testing

The first docker compose up --build -d produced two images in 31 seconds — all layers cached except the new source files.

[+] up 9/9
 ✔ Image nexusflow-worker          Built   31.0s
 ✔ Image nexusflow-gateway         Built   31.0s
 ✔ Network nexusflow_nexusflow_net Created
 ✔ Volume nexusflow_redis_data     Created
 ... 5 more

The first idempotency test used Idempotency-Key: demo-key-001. Both requests returned 202 Accepted with different task_id values:

$ curl -i -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: demo-key-001" \
  -d '{"name": "process_video", "payload": {"file": "clip.mp4"}}'
HTTP/1.1 202 Accepted
{"task_id":"126cd0d0-0bc1-4a40-a4fb-dc0e7e3d7194","name":"process_video","status":"accepted","accepted_at":"2026-08-28T06:24:15.009716+00:00"}

$ curl -i -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: demo-key-001" \
  -d '{"name": "process_video", "payload": {"file": "clip.mp4"}}'
HTTP/1.1 202 Accepted
{"task_id":"97ca9c8b-c4bb-4f03-9c55-013dfaf4806c","name":"process_video","status":"accepted","accepted_at":"2026-08-28T06:24:57.422885+00:00"}

Two different task_id values for the same key. The deduplication was not working. The idempotency_key column existed in the schema but the volume from the previous session still had the old schema — the column was never there. docker compose down without -v left the Postgres data volume intact with the old table definition.

docker compose down -v wiped the volumes. The next docker compose up --build -d provisioned a fresh Postgres instance and ran the updated init.sql. The same test:

$ curl -i -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: demo-key-001" \
  -d '{"name": "process_video", "payload": {"file": "clip.mp4"}}'
HTTP/1.1 202 Accepted
{"task_id":"4644ca04-d1af-4f74-8223-ffbf5f9d7880","name":"process_video","status":"accepted","accepted_at":"2026-08-28T06:43:22.946902+00:00","idempotency_key":"demo-key-001"}

$ curl -i -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: demo-key-001" \
  -d '{"name": "process_video", "payload": {"file": "clip.mp4"}}'
HTTP/1.1 200 OK
{"task_id":"4644ca04-d1af-4f74-8223-ffbf5f9d7880","name":"process_video","status":"completed","accepted_at":"2026-08-28T06:43:22.946902+00:00","idempotency_key":"demo-key-001"}

Same task_id on the second request. 200 OK instead of 202 Accepted. status is completed — the worker had already processed it in the 24 seconds between the two submissions. The original row came back unchanged.

The stale volume was the only blocker in the session. Not a code issue.

Where this sits

The application layer is hardened. The startup race is gone. The API surfaces failure detail. Retries back off progressively. Orphaned PEL messages get reclaimed. Duplicate submissions get deduplicated. The next layer is Kubernetes.