SHAAKER

Personal Transmissions

Technical log ACTIVE

NexusFlow · transmissions 024–027 · 2026-08-28

NexusFlow: A complete study guide

Every concept in transmissions 024–027 explained from first principles. Microservices, Redis Streams, asyncpg, retry state machines, exponential back-off, PEL reclaim, idempotency, Docker multi-stage builds — with decision rationale, alternatives, and references.

Every concept that was built, why it was built that way, what the alternatives were, and how every piece connects to every other piece. Written for a developer who is curious but starting from zero on this stack. Covers transmissions 024, 025, 026, and 027.

Table of contents

Foundations

The message broker

The worker

Containers

The database layer

Error handling and retries

Hardening (transmission 027)

Putting it all together

1. The big picture

NexusFlow is a microservices platform. It is being built as a learning vehicle to understand Docker, Kubernetes, and AWS EKS by actually building something real — not by reading about it.

The word “platform” here means: a system where you can drop in a task (like “send an email” or “generate a report”), and the system processes it in the background without making the person who submitted it wait for it to finish.

Think of it like a restaurant. You place an order and get a receipt with a number. You do not stand at the kitchen counter watching the chef cook. You sit down. The kitchen processes your order independently. You pick up your food when it is ready.

NexusFlow works the same way. The gateway is the front counter — you submit a task and immediately get a receipt (HTTP 202 + a task ID). The worker is the kitchen — it processes the task in the background. Redis is the ticket rail between them. PostgreSQL is the record of every order and its current state.

The system is not complicated. The value of building it is in the specifics: what happens when the worker crashes mid-task? What if it fails three times in a row? What if PostgreSQL is still starting when the worker tries to connect? These are the problems that matter in production, and they are the problems this project forces you to solve.

2. Microservices vs a monolith

A monolith is one application. The web server, the database logic, and the background job processing all live in the same codebase, running as one process. Simple to start, and right for small systems. Gets complicated fast as the system grows, because changing one part risks breaking another.

Microservices split that one application into multiple smaller services, each responsible for one thing. In NexusFlow:

  • The gateway handles HTTP traffic. It does not process tasks.
  • The worker processes tasks. It has no HTTP server.
  • Redis moves messages between them.
  • PostgreSQL stores the state.

Each service can be deployed, scaled, and updated independently. If the worker is slow, you run more workers without touching the gateway. If the gateway is getting too many requests, you scale it independently.

The trade-off: microservices add complexity. You now have network calls between services that can fail. You have to think about what happens when one service is down while another is up. For a learning project, that complexity is exactly the point — it forces you to solve the problems that large systems encounter.

3. The gateway pattern

The gateway is the only public-facing service in NexusFlow. Nothing in the outside world speaks directly to the worker or the database. Everything goes through the gateway.

This is called the API Gateway pattern. The gateway accepts incoming HTTP requests, validates them, forwards the work to the appropriate internal service, and responds to the caller.

In NexusFlow, the gateway does this through Redis. A client sends a POST request. The gateway validates it, puts a message on a Redis Stream, and tells the client: “Accepted.” The client does not know a worker exists. It does not know Redis is involved. It gets a task ID and a status.

If you want to add authentication, rate limiting, or logging later, you add it in one place: the gateway. Not in every service.

4. FastAPI

FastAPI is the Python web framework the gateway is built on.

FrameworkWhat it isWhy not chosen
FlaskMinimal, very popularNo built-in async, no built-in validation, no automatic API docs
DjangoFull-featured, batteries-includedDesigned for server-rendered web apps, too heavy for a small API
FastAPIModern, async-native, built on PydanticChosen

FastAPI gives you three things automatically:

1. Type-based validation. You declare what the request body should look like using Python type hints, and FastAPI refuses any request that does not match — before your code runs.

2. Automatic API documentation. Because you declare the types, FastAPI generates an interactive Swagger UI at /docs with zero extra work. Useful for testing endpoints without writing curl commands.

3. Async support. FastAPI is built on Starlette, which is asynchronous. One worker process can handle many requests concurrently without threads. This becomes important when the gateway starts awaiting database calls.

Uvicorn is the server that runs FastAPI. FastAPI defines how to handle requests. Uvicorn is the process that listens on a port and passes connections to FastAPI. Uvicorn is the building. FastAPI is the rooms inside it.

uvicorn[standard] in requirements.txt means Uvicorn plus optional extras: websocket support and a faster event loop implementation called uvloop.

5. Pydantic

Pydantic is the data validation library. In the gateway, the request body is defined as:

class TaskRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=128)
    payload: dict = Field(default_factory=dict)

When a POST request arrives, FastAPI reads the JSON body and feeds it into TaskRequest. Pydantic checks: is name present? Is it a string? Between 1 and 128 characters? Is payload a dictionary?

If any check fails, FastAPI returns HTTP 422 Unprocessable Entity automatically with a detailed error showing exactly what was wrong. Your endpoint function is never called.

Without validation at the door, invalid data gets deeper into your system — into the queue, into the database, into the worker, where it causes confusing errors far from the source. Pydantic stops bad data at the boundary, where it is cheapest to stop.

6. HTTP 202 vs 200

When a task is submitted, the gateway returns HTTP 202 Accepted, not 200 OK.

CodeMeaningWhen to use
200 OKHere is the result of your requestThe work is done right now
202 AcceptedI received your request and will process itThe work happens later, asynchronously

Returning 200 would be a lie. The task has not been processed when the gateway responds. It has only been put on a queue. The worker has not touched it yet. 202 is the honest answer.

A 202 sets the right expectation. Do not assume the work is done. Use the task_id to check the status later. The GET /tasks/{task_id} endpoint exists because the 202 implies one must.

7. UUID

When a task arrives, the gateway generates:

task_id = str(uuid.uuid4())

UUID stands for Universally Unique Identifier. uuid4() generates a random one: 571a0512-8dde-48e0-9e0f-32354cf34552.

Why not a simple incrementing number? A database auto-incrementing ID is generated by the database — after a write. At this point in the flow, no database write has happened yet. The task just landed in Redis. We need an ID before the database is involved.

UUID is also safe to generate in multiple places simultaneously. If you run five gateway instances, each generates UUIDs without coordinating. They never collide. An incrementing counter would require all five instances to coordinate around a shared counter — which means a network call, which means a potential failure point, for something as simple as generating an ID.

8. Redis

Redis is an in-memory data store. Data lives in RAM, not on disk. This makes it extremely fast: a PostgreSQL query takes milliseconds, a Redis operation takes microseconds.

Why in-memory is fine for the stream: the stream is not the source of truth. It is a transport layer. Messages flow through it quickly. Once the worker processes a task and writes the result to PostgreSQL, Redis is no longer the record. PostgreSQL is. If Redis lost all its data, no permanent business data would be lost — the worker writes outcomes to PostgreSQL.

What about the Redis volume in docker-compose? It is there for development continuity. The compose file uses AOF (append-only file) persistence with appendfsync everysec: at most one second of data loss on a hard crash. Pending tasks survive a Redis container restart during development, so you are not constantly losing work in progress.

Port 6379 is Redis’s default port — to Redis what 5432 is to PostgreSQL and 8000 is to the gateway.

9. Why Redis Streams, not something else

Four alternatives existed.

Option A: Redis BLPOP (list-based queue) BLPOP reads from a Redis List. It blocks until an item arrives, then returns it and removes it from the list. Problem: once you pop a message, it is gone. If the worker crashes before finishing, the message is lost forever. You have to build your own re-delivery system. There is no history replay, no consumer group concept.

Option B: Redis Pub/Sub Pub/Sub is a broadcast system. All subscribers receive every message simultaneously. Problem: if no subscriber is connected when a message is published, the message is gone. There is no storage. Pub/Sub is fire-and-forget. A task queue where tasks must not be lost needs storage.

Option C: RabbitMQ A full message broker with advanced routing, dead-letter queues, and fine-grained acknowledgement. It solves all the above problems. Problem: it is a separate system to learn, run, and operate. Redis was already in the stack. Redis Streams covers 90% of what RabbitMQ does for this use case, with no new dependency.

Option D: Apache Kafka A distributed log system used at enormous scale. Kafka requires its own cluster management (ZooKeeper, or KRaft in newer versions). It is designed for millions of events per second at Netflix and LinkedIn scale. Using Kafka here is using a freight train to deliver a letter.

Why Redis Streams won: Redis was already in the stack. Streams add — without a new dependency — persistent messages, consumer groups, acknowledgement (XACK), the Pending Entries List for re-delivery of unacknowledged messages, and message replay. For the scale NexusFlow targets, Streams is the right tool.

10. How a Redis Stream works

A Redis Stream is an append-only log. You can only add rows at the bottom. You cannot delete or edit an entry (without explicit trim commands).

XADD: writing to the stream

redis_client.xadd(
    "tasks",
    {
        "task_id":     task_id,
        "name":        task.name,
        "payload":     json.dumps(task.payload),
        "accepted_at": accepted_at,
    },
)

Redis appends a new entry and assigns an ID based on the current timestamp. The ID 1787556301266-0 is a millisecond timestamp (1787556301266) plus a sequence number (0) in case two messages arrive in the same millisecond.

Why json.dumps(task.payload)? Redis Stream field values must be scalars: a string, a number, or bytes. A Python dict is not a scalar. Storing a dict directly raises a DataError. json.dumps() converts the dict to a JSON string, which is a scalar. The worker reverses this with json.loads().

XREADGROUP: reading from the stream

response = client.xreadgroup(
    groupname=CONSUMER_GROUP,     # "workers"
    consumername=CONSUMER_NAME,   # "worker-1"
    streams={STREAM_NAME: ">"},   # ">" means: give me only new messages
    count=1,
    block=BLOCK_MS,               # wait up to 2000ms if empty
)

The ">" means: give me messages no other consumer in my group has received yet. This prevents two workers from getting the same message.

block=BLOCK_MS means: if the stream is empty, park this call for up to 2000ms before returning empty. Without blocking, the worker would spin in a tight loop consuming 100% CPU doing nothing. Blocking is the efficient way to wait.

XACK: acknowledging a message

client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)

XACK tells Redis: consumer group “workers” has successfully processed this message. Remove it from the Pending Entries List. XACK is called only after success. If processing fails, XACK is withheld — which is the signal that something went wrong.

11. Consumer groups

A consumer group is a named group of workers that collectively read from a stream. Redis tracks, per group, which messages have been delivered and which have been acknowledged.

Stream: tasks
────────────────────────────────────────────────────
 msg-001  |  msg-002  |  msg-003  |  msg-004
────────────────────────────────────────────────────
    |                      |
 worker-1              worker-2
(processing)          (processing)

Worker-1 gets msg-001. Worker-2 gets msg-003. Neither gets a message the other is processing. Each message goes to exactly one worker — the key difference from Pub/Sub, where every subscriber gets every message.

MKSTREAM in the worker startup code means: if the stream tasks does not exist yet when the worker starts, create it automatically. This prevents a race condition where the worker starts before any task has ever been published.

When the stack restarts, the worker prints: Consumer group 'workers' already exists. Joining. This is the BUSYGROUP error being caught gracefully. The group was created in the first session and persisted in the Redis volume. The worker detected it and joined.

12. At-least-once delivery

The Pending Entries List (PEL) tracks messages that have been delivered to a worker but not yet acknowledged.

When a message is delivered:

  1. It appears in the PEL with the worker’s name and a delivery timestamp
  2. The worker processes it
  3. If XACK is called → message removed from PEL (success)
  4. If XACK is never called (crash, exception) → message stays in PEL

On restart, a worker can reclaim PEL messages from a crashed worker using XAUTOCLAIM. That is the definition of at-least-once delivery: a message will be delivered at least once. It might be delivered more than once if a worker dies mid-processing and the message is re-delivered.

Exactly-once delivery — guaranteed to process exactly once, never more — is very hard in distributed systems. It requires the consumer to be idempotent: able to handle duplicate messages without producing duplicate effects. For example, if “send email” is processed twice, the user gets two emails. Making that safe requires tracking which tasks have already been completed and skipping duplicates. That is a later problem for NexusFlow.

13. The worker

The worker is a plain Python script. No web server. No HTTP. It connects to Redis and loops forever.

while not _shutdown:
    messages = XREADGROUP(block=2000)   <- waits up to 2s for a message
    if no messages:
        continue                         <- check _shutdown flag, loop again
    for each message:
        process_task()                   <- do the actual work
        XACK()                           <- tell Redis we are done

The worker has no HTTP server because it does not receive requests from the outside world. It reads from Redis. Adding Uvicorn and FastAPI to the worker would be unnecessary complexity. worker/requirements.txt contains only redis, asyncpg, and pydantic — not fastapi or uvicorn.

The process_task() function is the plug-in point. The surrounding infrastructure — connection management, graceful shutdown, retry logic, database writes — is complete. The actual business logic (calling an email API, resizing an image, running a report) is what gets dropped in when the platform is used for a real purpose.

14. Graceful shutdown

The worker registers signal handlers:

signal.signal(signal.SIGTERM, _handle_signal)
signal.signal(signal.SIGINT, _handle_signal)

A signal is a notification the operating system sends to a process. SIGTERM means “please stop soon.” SIGINT means “the user pressed Ctrl+C.” Both are polite requests — as opposed to SIGKILL, which means “stop immediately, no argument.”

When Kubernetes wants to stop a pod, it sends SIGTERM and waits 30 seconds. If the process is still running after that, it sends SIGKILL. Without the signal handler, SIGTERM would terminate the Python process immediately, mid-task, mid-database-write. The message would not be ACKed. The task would be re-delivered and potentially processed twice.

With the handler, _shutdown is set to True. The loop finishes its current task, calls XACK, then exits cleanly. The in-flight message is handled correctly.

The logs from transmission 026 show this working exactly as designed:

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

Signal 15 is SIGTERM. The worker caught it, finished its work, closed the database pool, and exited with code 0 — the OS code for “success.”

15. Exponential backoff

The worker retries the Redis connection up to 10 times:

wait = 2 ** attempt   # 2, 4, 8, 16, 32... seconds

Why not retry immediately? If Redis is not ready and 10 workers all retry every 100 milliseconds, they hammer a server that is already struggling to start. They can prevent Redis from starting at all.

Why not retry at fixed intervals? If all 10 workers retry every 5 seconds, they all arrive at the same moment: a thundering herd. Exponential backoff with slightly different starting times (because containers start at slightly different moments) spreads the load naturally.

Exponential backoff appears in many places in distributed systems: HTTP clients retrying failed requests, database connection pools waiting for a server to come back, message queue consumers waiting for their broker. The pattern is always the same: wait longer after each failure, up to a maximum.

16. Docker

Docker packages software into containers. A container is an isolated process with its own filesystem, its own network, and its own installed software — but sharing the host machine’s operating system kernel.

Why not just run Python directly? Running python main.py on your laptop depends on which Python version is installed, which packages are installed, your OS, your environment variables. On another machine, any of these might differ. “Works on my machine” is the classic problem Docker solves.

A Docker container packages the application and its exact runtime environment together. You ship the container to any machine and it runs identically.

Image vs container:

  • An image is the blueprint: a read-only snapshot of the filesystem and configuration. Like a class definition.
  • A container is a running instance of an image. Like an object created from that class.

docker build creates an image. docker run or docker compose up creates containers from images.

17. The Dockerfile

A Dockerfile is a recipe for building an image. Here is the gateway Dockerfile explained instruction by instruction.

FROM python:3.12-slim AS builder

Start from the official Python 3.12 image, slim variant. The full python:3.12 image is ~350MB and includes compilers and documentation. slim strips those to ~45MB. This stage is named builder.

WORKDIR /install
COPY requirements.txt .

Set the working directory. Copy only requirements.txt first, not the whole project. This is the layer caching trick (section 19).

RUN pip install --upgrade pip --no-cache-dir \
 && pip install --no-cache-dir --prefix=/install/packages -r requirements.txt

Install packages into /install/packages. --no-cache-dir skips pip’s download cache, which is useless in a container (the container is thrown away after the build). --prefix puts packages in a specific directory so they can be cleanly copied into the second stage.

FROM python:3.12-slim AS runtime

Start a brand new stage: runtime. A fresh python:3.12-slim with nothing from the builder except what we explicitly copy.

RUN addgroup --system appgroup && adduser --system --ingroup appgroup appuser
WORKDIR /app
COPY --from=builder /install/packages /usr/local
COPY . .
USER appuser
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Create a non-root user. Set working directory. Copy packages from builder. Copy application source. Drop to non-root. Declare the port. Set the startup command.

--host 0.0.0.0 is mandatory inside a container. By default, Uvicorn binds to 127.0.0.1 (localhost), accessible only from within the container itself. 0.0.0.0 means “accept connections from any network interface” — including the Docker network.

18. Multi-stage builds

Both Dockerfiles use a two-stage build: the single most important Dockerfile practice for production images.

Stage 1 (builder): has pip, potentially build tools, installs all dependencies. Produces: the installed packages in /install/packages.

Stage 2 (runtime): starts fresh from python:3.12-slim. Copies only installed packages from stage 1. Has no pip, no build tools, no download cache, no wheel files.

Why this matters: the final image is what gets deployed. Smaller images:

  • Deploy faster (less data to pull from a registry)
  • Have a smaller attack surface (fewer binaries an attacker could use)
  • Use less disk space on nodes

Without multi-stage builds, the final image contains pip, build tools, and all intermediate build artifacts. With multi-stage, the runtime image contains exactly what is needed to run the application — nothing else.

19. Layer caching

A Docker image is made of layers. Each Dockerfile instruction adds a layer. Docker caches layers. If a layer has not changed since the last build, Docker reuses the cached version and skips re-running that instruction.

The order of COPY instructions is deliberate:

COPY requirements.txt .          <- layer A
RUN pip install ...              <- layer B (depends on A)
COPY . .                         <- layer C (independent of B's content)

When you edit main.py, Docker detects layer C changed. It rebuilds from C. Layers A and B are unaffected. Pip does not re-run. The 90-second dependency install is skipped.

If the order were reversed — source code before requirements — editing any source file would invalidate the pip install layer. 90 seconds every time for a one-line code change.

The second docker compose up --build in transmission 024 took 17 seconds instead of 188 — the expensive pip install layer was cached. Only the application source layer was rebuilt.

20. Non-root user

By default, processes in a Docker container run as root. Root inside a container can read any file, write anywhere, and install software.

Why this is dangerous: if the application is ever exploited through a dependency vulnerability, and the process runs as root, the attacker has root access inside the container. Depending on the container runtime configuration, this escapes the container and reaches the host machine.

Running as appuser limits the blast radius. An attacker who exploits the application gets the permissions of appuser, which are minimal by design.

Kubernetes Pod Security Standards (restricted) enforces non-root containers across a cluster. If your image runs as root, the policy refuses to schedule the pod. Since NexusFlow is heading to EKS, the images were built correctly from the start — not retrofitted later.

21. Environment variables

Two Python-specific environment variables appear in both Dockerfiles:

PYTHONDONTWRITEBYTECODE=1 Python normally compiles .py files to .pyc bytecode and caches them for faster subsequent imports. In a container where the filesystem is ephemeral and the app starts fresh every time, these cache files are never reused between restarts. They waste disk space. This variable prevents Python from writing them.

PYTHONUNBUFFERED=1 Python buffers stdout and stderr by default — it collects output in a buffer and writes in chunks. In a container, you view logs with docker logs or kubectl logs. If Python is buffering, a log message might sit in the buffer for seconds before appearing — or never appear at all if the container crashes before the buffer is flushed.

Setting PYTHONUNBUFFERED=1 forces Python to flush every write immediately. The worker logs appear in real time as tasks are processed.

22. Docker Compose

Docker Compose defines and runs multi-container applications locally. Instead of four separate docker run commands with many flags, you describe the entire system in docker-compose.yml and start everything with one command: docker compose up.

The compose file for NexusFlow declares 4 services, 2 named volumes, and 1 network.

docker compose up --build does, in order:

  1. Builds images for services with a build: block (gateway, worker)
  2. Pulls images for pre-built services (postgres, redis)
  3. Creates the network
  4. Creates the volumes
  5. Starts containers in dependency order, respecting health checks

docker compose down stops containers and removes the network but leaves volumes intact. docker compose down -v also removes volumes, giving a clean slate — which is how transmission 026 started, to force a fresh schema with the new columns.

23. The bridge network

All four services share a network called nexusflow_net:

networks:
  nexusflow_net:
    driver: bridge

Bridge means Docker creates a virtual network switch. Each container gets a virtual network interface connected to this switch. Containers on the same bridge network can reach each other by their service name via Docker’s built-in DNS.

The gateway connects to Redis with:

redis.Redis(host="redis", port=6379)

Not with an IP address. Docker’s internal DNS resolves "redis" to the IP of the Redis container automatically. The same applies to PostgreSQL: host="postgres" in the connection string.

If you used localhost inside the gateway container, it would resolve to the gateway container’s own loopback interface — not Redis. This is the most common mistake when moving from local development to containers.

The network is also isolated from other Docker Compose projects on the same machine. A different project’s containers cannot accidentally reach NexusFlow’s services.

24. Named volumes

By default, a container’s filesystem is ephemeral. When a container is removed, all data written inside it is gone. For a database, this is obviously not acceptable.

Named volumes persist data across container removal and recreation. When you run docker compose up again, the volume is mounted back and the database finds its data exactly where it left off.

The worker prints “Consumer group ‘workers’ already exists. Joining.” on restart — the Redis volume still held the stream and consumer group data from the previous session. No data was lost across restarts.

Named volumes are managed by Docker in Docker’s own storage area on the host. They are more portable than bind mounts (which map a specific host directory into the container) and more appropriate for production-like data.

25. Health checks and service_healthy

Without health checks, docker compose up starts containers based on dependency declarations but does not wait for the container to be ready — only running. A container is “running” within one second. A database is “ready to accept connections” after 10–40 seconds of initialization.

Health checks define a test command that Docker runs periodically:

test: ["CMD-SHELL", "pg_isready -U nexus -d nexusflow"]
interval: 10s
timeout: 5s
retries: 5
start_period: 10s

Docker runs pg_isready every 10 seconds. If it exits with code 0, the container is marked healthy. After 5 consecutive failures, it is unhealthy.

depends_on:
  postgres:
    condition: service_healthy
  redis:
    condition: service_healthy

condition: service_healthy means: do not start this service until the dependency is healthy, not just running. condition: service_started — the default — only waits for the container process to start, not for the service inside to be ready. The default is almost always wrong for databases.

26. PostgreSQL

PostgreSQL is a full relational database. In transmission 024 it was in the compose stack but no code wrote to it. From transmission 025 onward, it is the source of truth for every task.

Why PostgreSQL and not something else:

AlternativeWhy not used
SQLiteNo network access, single-writer only, cannot handle concurrent writes from multiple containers
MySQL/MariaDBValid choice, very similar. PostgreSQL preferred in modern Python stacks for JSONB column type, richer query planner, and better async driver support
MongoDBDocument store, appropriate for schema-free nested data. Task records are structured and relational — fixed fields, fixed transitions

PostgreSQL stores every task as a row with known columns: task_id, name, status, payload, result, retry_count, max_retries, error_message, created_at, updated_at. The schema enforces structure. A document store would let the worker write whatever shape it wanted, making queries and state tracking harder.

27. asyncio — why the worker needed it

When the database layer was added in transmission 025, the worker had to make network calls to PostgreSQL — calls that take time. In a synchronous Python program, a network call blocks the entire thread: while waiting for PostgreSQL to respond, nothing else runs.

asyncio is Python’s built-in concurrency model for I/O-bound work. Instead of blocking and waiting, an async function yields control while it waits for I/O, and the event loop runs other things in the meantime.

# Synchronous — blocks the thread
result = conn.execute(query)  # thread does nothing for 5ms

# Asynchronous — yields control while waiting
result = await conn.execute(query)  # event loop can do other things during those 5ms

The await keyword means: “start this operation, give up control until it is done, then resume here.” The async def keyword marks a function as one that can contain await calls.

Why does the worker care? The worker makes at least two PostgreSQL calls per task: one to update status to processing, one to update it to completed or failed. With retries, it makes more. If these were synchronous, each call would block the Redis read loop. Making them async keeps the worker responsive while database calls are in flight.

The worker entry point changed from:

# Before — synchronous
if __name__ == "__main__":
    run(r)

# After — async event loop
if __name__ == "__main__":
    asyncio.run(run(r))

28. asyncpg — the async PostgreSQL driver

asyncpg is a Python library for talking to PostgreSQL asynchronously. It is written in Cython (a mix of Python and C) and communicates directly with PostgreSQL using the binary protocol rather than the text protocol used by most drivers.

What the binary protocol means: PostgreSQL has two wire formats for sending data back and forth. Text format converts everything to strings. Binary format sends native types — integers as 4 or 8 bytes, timestamps as a 64-bit integer. asyncpg uses binary, which gives faster serialization and deserialization with no type conversion overhead.

A typical query:

async with pool.acquire() as conn:
    row = await conn.fetchrow(
        "SELECT retry_count, max_retries FROM tasks WHERE task_id = $1",
        task_id,
    )

pool.acquire() checks out a connection from the pool. fetchrow() runs the query and returns one row as a dict-like object. When the async with block exits, the connection is returned to the pool automatically.

Note the parameterized query: $1 not a Python f-string. Parameterized queries prevent SQL injection — the database driver handles the value safely rather than interpolating it into the SQL string.

29. Why asyncpg, not psycopg3 or SQLAlchemy

Three async PostgreSQL options exist for Python:

OptionWhat it isWhy not chosen
psycopg3The new version of the most popular PostgreSQL driverValid alternative. asyncpg has a cleaner pool API and was more mature for this pattern at the time
SQLAlchemy (async)ORM + query builder with async supportAn ORM abstracts SQL into Python objects. For NexusFlow’s small, explicit query set, the abstraction adds complexity without benefit. Writing SQL directly is clearer
asyncpgBare async driver, no ORMChosen: clean pool API, binary protocol, explicit SQL

asyncpg was chosen because create_pool + async with pool.acquire() as conn is exactly the pattern needed. One call to create the pool, one context manager per query, no connection juggling. The worker and gateway each have a module-level pool and a few functions that use it. That is the entire database layer.

30. Connection pools

A connection pool is a collection of pre-opened database connections that are reused across requests.

Why not open a new connection for every query? Opening a PostgreSQL connection involves a TCP handshake, authentication, SSL negotiation if configured, and session setup. This takes 20–100ms. For a gateway handling many requests per second, opening a new connection for every query would add 20–100ms to every single response.

A pool opens connections once at startup and keeps them ready. When a request needs a database call, it borrows a connection from the pool, uses it, and returns it. The pool has a minimum and maximum size:

_pool = await asyncpg.create_pool(
    dsn=_DSN,
    min_size=1,
    max_size=5,
)

min_size=1 means at least one connection is always open. max_size=5 means at most five concurrent connections. If six database calls arrive simultaneously, the sixth waits until one of the five returns to the pool.

The worker uses min_size=1, max_size=5 because it processes tasks sequentially — it is never making six database calls at once. A small pool is sufficient.

31. The lifespan context manager in FastAPI

The gateway needs to open the database connection pool when it starts and close it when it stops. FastAPI provides the lifespan context manager for exactly this:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    await db.init_pool()      # runs at startup
    yield                     # the application runs here
    await db.close_pool()     # runs at shutdown

app = FastAPI(lifespan=lifespan)

Everything before yield runs when the application starts. Everything after yield runs when the application shuts down. The pool is available for the entire lifetime of the application.

The older approach used @app.on_event("startup") and @app.on_event("shutdown") decorators. FastAPI deprecated those in favour of lifespan because the context manager makes the startup/shutdown pair explicit and co-located. You can see both in the same function, rather than in two separate decorated functions scattered across the file.

32. init.sql — how PostgreSQL initializes in Docker

The official PostgreSQL Docker image has a built-in feature: any .sql or .sh files placed in /docker-entrypoint-initdb.d/ inside the container are executed once, on first startup, before the database accepts connections.

In docker-compose.yml:

postgres:
  image: postgres:16-alpine
  volumes:
    - ./postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
    - postgres_data:/var/lib/postgresql/data

The bind mount makes postgres/init.sql from the project appear at /docker-entrypoint-initdb.d/init.sql inside the container. On first startup, PostgreSQL finds it and runs it — creating the tasks table and its index.

“First startup” means: when the data directory is empty. Once the data directory exists (on a named volume), PostgreSQL skips the init scripts. This is correct behaviour — you do not want to re-run CREATE TABLE every time the database restarts.

The consequence: if you change the schema (adding columns as in transmission 026), you cannot simply restart the container. PostgreSQL will not re-run the init script because the data directory already exists. You either run a migration (ALTER TABLE) against the running database, or destroy the volume with docker compose down -v and restart with the new schema.

In transmission 026, docker compose down -v was used because this is a development environment with no data worth preserving. In production, you would write a migration.

33. The db.py pattern — two modules, two contracts

Both the gateway and the worker have a db.py module. They are not shared. They have different jobs.

gateway/db.py has three functions:

  • create_task() — inserts a new row as pending
  • fetch_task() — reads a row by task_id for the status endpoint
  • update_task_status() — changes the status column

worker/db.py has five functions (after transmission 026):

  • fetch_task_retry_info() — reads retry_count and max_retries
  • update_task_status() — changes the status column
  • update_task_for_retry() — increments retry_count, resets status to pending, writes error_message
  • update_task_as_failed() — sets status to failed, writes error_message

The worker never inserts. The gateway never reads retry counters. Each module contains only what its service needs. If you put all of this in one shared module, you would have a database layer full of functions that half the codebase never uses — and a coupling between two services that should be independent.

34. The state machine — pending, processing, completed, failed

A state machine is a model where a system can be in exactly one state at a time, and transitions between states follow defined rules. The tasks table’s status column is a state machine.

pending
  |
  | worker picks up the message
  v
processing
  |                  |
  | success          | failure
  v                  v
completed         failed  (if retry_count >= max_retries)
                     |
                     | (if retry_count < max_retries)
                     v
                   pending  <- back to the start, retry_count incremented

Each transition is deliberate:

  • pending → processing: the worker fetched the task from Redis. Status is updated before processing starts.
  • processing → completed: process_task() succeeded, XACK was called.
  • processing → pending: process_task() raised an exception, retries remain. retry_count is incremented.
  • processing → failed: process_task() raised an exception, retries are exhausted.

The state machine makes the system observable. At any point, you can query the database and know exactly where every task is. pending means waiting to be picked up. processing means in flight right now. completed means done. failed means gave up after all allowed attempts.

35. COPY . . and the .dockerignore fix

In transmission 024, both Dockerfiles copied files explicitly:

# gateway/Dockerfile
COPY main.py .

# worker/Dockerfile
COPY worker.py .

When db.py was added in transmission 025, neither Dockerfile knew about it. The images were built, the containers started, and both crashed immediately with ModuleNotFoundError: No module named 'db'. The file existed on disk. It was not in any image.

The fix was switching to COPY . . — copy the entire build context. With one instruction, any file added to the service directory is automatically included in the next build. No Dockerfile update required.

The consequence of COPY . . without control: it copies everything, including things you do not want in the image: __pycache__/, .env files with secrets, development tools.

The solution is .dockerignore — a file in each service directory that lists what to exclude from the build context:

__pycache__/
*.pyc
.env
*.env.*
.git/
.venv/

Docker reads .dockerignore before copying. Excluded paths never enter the build context at all. COPY . . then copies the cleaned context, not the raw directory.

36. The curl problem in slim images

The gateway’s health check in docker-compose.yml was originally:

test: ["CMD-SHELL", "curl -f http://localhost:8000/health || exit 1"]

python:3.12-slim does not ship curl. It is a minimal image with only what Python needs. The probe ran curl, got command not found, exited non-zero, and Docker marked the container unhealthy on every cycle. The gateway had been silently “unhealthy” since day one — it just had not mattered until other services started using condition: service_healthy.

The fix was replacing curl with Python’s standard library:

test: ["CMD-SHELL", "python -c \"import urllib.request, sys; urllib.request.urlopen('http://localhost:8000/health', timeout=4); sys.exit(0)\""]

urllib.request.urlopen() raises an exception on non-2xx responses or connection failures. An unhandled exception causes Python to exit with code 1. Docker reads that as unhealthy. A successful response causes sys.exit(0) to run, which Docker reads as healthy.

No extra package, no shell dependency. The same Python interpreter that runs the application runs the health check.

The general lesson: slim images trade size for completeness. Many standard Unix tools — curl, wget, netcat — are not present. Health checks and diagnostic commands need to use what is actually in the image.

37. PostgreSQL crash recovery and fsync

In transmission 026, docker compose down -v was used to wipe the old schema. That is a hard stop — not a graceful shutdown. PostgreSQL did not finish writing its checkpoint before the container was killed.

On the next docker compose up --build, PostgreSQL started and immediately began crash recovery:

postgres-1 | database system was interrupted; last known up at 2026-08-25 06:32:17 UTC
postgres-1 | syncing data directory (fsync), elapsed time: 70.75 s

What PostgreSQL is doing: it keeps a Write-Ahead Log (WAL). Before writing data to the actual data files, it writes the intended change to the WAL. On crash recovery, it reads the WAL and replays any changes that were logged but not yet written to the data files. The fsync step walks the data directory and forces all pending writes to disk, ensuring nothing is half-written.

Recovery took 70 seconds. Docker’s health check had a start_period of 10 seconds and 5 retries at 10-second intervals — 60 seconds total. PostgreSQL was still recovering at second 61. The health check declared it unhealthy. The gateway and worker refused to start because their dependency was unhealthy.

docker compose restart postgres let it finish recovery. After that, docker compose up -d started everything cleanly.

38. Why the PEL alone is not a retry mechanism

After transmission 025, the error handling in the worker was:

except Exception as exc:
    # Do NOT xack — message stays in PEL
    await db.update_task_status(task_id, "failed", result={"error": str(exc)})

This looks like retry behaviour but it is not. What actually happens:

  1. Task fails. Status → failed. Message stays in PEL.
  2. No re-delivery happens automatically. The PEL holds the message, but no one reclaims it.
  3. The message sits in the PEL indefinitely, marked as delivered-but-unacknowledged.
  4. On restart, XAUTOCLAIM could reclaim it — but that requires a separate idle-time scan that was not implemented.

In practice, that message was dead. A developer would have to manually intervene to reclaim it. That is not a retry mechanism. That is a message graveyard.

A real retry mechanism requires: tracking how many attempts have been made, defining a maximum, re-queuing on failure while attempts remain, and definitively acknowledging on permanent failure.

39. Schema design for retry state

Three columns were added to the tasks table:

retry_count   INTEGER NOT NULL DEFAULT 0,
max_retries   INTEGER NOT NULL DEFAULT 3,
error_message TEXT,

retry_count tracks how many attempts have been made. It starts at 0 and is incremented by 1 each time the task fails and is re-queued. When retry_count equals max_retries, no more retries occur.

max_retries sets the ceiling. It defaults to 3, meaning a task gets 4 total attempts (the first attempt, then 3 retries). The condition retry_count < max_retries is true for attempts 1, 2, and 3, and false on attempt 4, which triggers the permanent failure path.

error_message stores the last exception string. It is TEXT (unlimited length) rather than VARCHAR(n) because exception messages are unpredictable in length. It is nullable because a task that completes successfully has no error to record.

40. Per-row max_retries — why it lives in the schema

max_retries is a column in the database, not a constant in the worker code.

If max_retries were a constant:

MAX_RETRIES = 3  # in worker.py

Every task would get exactly 3 retries, forever. No exceptions.

With max_retries in the schema, the gateway can set different values per task at insert time:

  • A critical payment task might get max_retries = 10
  • A low-priority report task might get max_retries = 1
  • A test task might get max_retries = 0 to fail immediately and predictably

The worker does not hard-code a policy. It reads the policy from the database and applies it. The DEFAULT 3 means: if nothing is specified at insert time, 3 retries is the default. The gateway does not need to pass a value for ordinary tasks.

41. Reading state before acting

Before each attempt, the worker reads the current retry state from the database:

retry_info = await db.fetch_task_retry_info(task_id)
retry_count = retry_info["retry_count"]
max_retries = retry_info["max_retries"]

Why read from the database rather than tracking in memory?

If the worker maintained retry counts in memory, they would be lost on restart. A task that failed twice, with retry_count = 2 in the worker’s memory, would start over at retry_count = 0 after the worker restarts. It would retry indefinitely.

With the count in the database, it survives restarts. The worker picks up the task, reads retry_count = 2, knows it has already failed twice, and behaves accordingly. The database is the source of truth for all state.

Multiple workers can handle the same task across restarts without losing count. Worker-1 processes attempt 1 and crashes. Worker-2 picks up the re-queued message, reads retry_count = 1 from the database, and knows it is on attempt 2.

42. Atomic database operations

When a task fails and needs to be retried, three things must happen: increment retry_count, reset status to pending, and write the error_message. These could be done in three separate queries — but three separate queries create a window for inconsistency. If the worker crashes between queries 1 and 2, retry_count is incremented but status is still processing.

The solution is one UPDATE statement that does all three:

UPDATE tasks
   SET status        = 'pending',
       retry_count   = retry_count + 1,
       error_message = $2,
       updated_at    = $3
 WHERE task_id = $1

retry_count = retry_count + 1 — the increment happens in the database, not in Python. The database reads the current value, adds 1, and writes the result in a single atomic operation. If the worker crashes after this query runs, the database is in a consistent state: status pending, retry_count correctly incremented.

43. The retry state machine

The full exception handler after transmission 026:

except Exception as exc:
    error_str = str(exc)

    if retry_count < max_retries:
        # Retries remaining
        await db.update_task_for_retry(task_id, error_str)  # status → pending, count + 1
        await asyncio.sleep(backoff_s)                      # exponential back-off (section 45)
        client.xadd(STREAM_NAME, fields)                    # new stream entry
        client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)   # clear original from PEL

    else:
        # Retries exhausted
        await db.update_task_as_failed(task_id, error_str)  # status → failed
        client.xack(STREAM_NAME, CONSUMER_GROUP, msg_id)   # clear from PEL permanently

The four-step retry sequence:

Step 1 — update the database: write the incremented retry_count, reset status to pending, record error_message. This happens first. If the worker crashes after this, the task is pending in the database. The new entry from step 3 will be processed independently.

Step 2 — back-off: wait before re-queuing. This gives a struggling downstream service time to recover.

Step 3 — XADD: publish a new message to the stream with the same fields (including the original task_id). Any consumer in the group picks this up on the next read.

Step 4 — XACK: acknowledge the original message, removing it from the PEL. The new entry from step 3 is now the authoritative one.

44. XADD vs XCLAIM — the re-queue decision

Two valid strategies exist for re-delivering a failed message in Redis Streams.

XCLAIM / XAUTOCLAIM Take ownership of a message that has been sitting in the PEL for too long. The message is still the same entry in the stream — the consumer changes, the content does not. XCLAIM is the canonical Redis Streams approach to at-least-once re-delivery on worker failure. It requires a separate idle-time scan: a process that periodically checks the PEL, finds messages pending for more than N milliseconds, and reclaims them. The re-delivery time is the PEL idle threshold, not the intentional back-off delay.

XADD (new entry) Acknowledge the original message to clear the PEL. Publish a brand-new entry with identical fields. The new entry has a new stream ID. Any consumer picks it up normally on the next XREADGROUP.

XADD was chosen for NexusFlow because it is simpler. No idle-time scan. No XCLAIM logic. The back-off is an explicit asyncio.sleep() before the XADD. The PEL stays clean after every failure. The trade-off: duplicate entries appear in the stream (the original, now ACKed, and the new one). That trade-off is acceptable because the stream is a transport, not a history, and the database is the source of truth.

Transmission 027 keeps XADD for the normal retry re-queue but adds XCLAIM for a different job — a background scan that reclaims messages orphaned when a worker dies mid-task (section 49).

45. Exponential back-off for retries

After a failure, the worker waits before it re-queues the task. The wait grows with every attempt.

Two environment-tunable constants replace the flat REQUEUE_DELAY_S from transmission 026:

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 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 — the count read from the database before update_task_for_retry ran. max(retry_count - 1, 0) at retry_count = 0 gives 2^0 = 1, so the first failure never waits zero seconds. The min(..., REQUEUE_MAX_DELAY_S) caps the wait at 30 seconds.

Attemptretry_count (pre-increment)MultiplierBack-off
1st failure02⁰ = 11 s
2nd failure12¹ = 22 s
3rd failure22² = 44 s
4th failure32³ = 88 s
………capped at 30 s

The alternative was the flat 1-second delay. A flat delay re-queues at the same rate whether the downstream service has failed once or ten times. Exponential back-off gives a struggling service more room to recover before the next attempt lands.

46. Simulated failures for testing distributed systems

Testing distributed system behaviour requires triggering failures on demand. In the real world, failures come from network outages, service crashes, or data errors that are hard to reproduce. For local testing, you need a controlled, deterministic failure.

NexusFlow uses a named trigger:

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

The trigger is data (the task name), not a code flag or a config variable. It is easy to test — curl -d '{"name": "fail_task"}' — and easy to find and remove before production.

From transmission 026, the result after submitting a fail_task:

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

Every step of the retry state machine executed correctly and was recorded in the database.

47. The PostgreSQL startup retry loop

On a fresh docker compose up --build, PostgreSQL may still be initializing its data directory or recovering from a crash when the worker connects. asyncpg.create_pool() raises CannotConnectNowError, and the worker from transmission 026 crashed and relied on Docker’s restart policy to try again. It took seven restart cycles to stabilize.

Transmission 027 wraps the pool creation in a retry loop. Ten attempts, two seconds apart:

_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

The catch tuple spans both layers of failure: CannotConnectNowError for the server’s “database system is starting up” message, and ConnectionRefusedError plus OSError for the raw socket refusal that arrives before any Postgres handshake begins. After ten failed attempts the loop re-raises as RuntimeError, chained to the last exception so the traceback survives into the logs.

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

48. Telemetry exposure in the task status endpoint

Before transmission 027, GET /tasks/{task_id} returned task_id, name, status, and result. The retry columns added in transmission 026 stayed invisible — a caller had to query PostgreSQL directly to see how many times a task had been attempted or what the last error was.

The gateway’s fetch_task now selects the three retry columns:

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

The Pydantic model picks them up without any extra mapping:

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

The alternative was leaving retry state in the database and telling callers to query it themselves. That breaks the gateway pattern from section 3 — the API is the only public surface. Exposing the fields through the response keeps PostgreSQL private while making failure state visible.

49. Background PEL reclaim via XCLAIM

Section 38 left a hole: a message a worker dies on mid-task — after XREADGROUP delivers it but before the retry path can XADD or XACK — stays in the PEL forever. No other consumer receives it.

Transmission 027 closes the hole with a 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 lists every pending entry in the group:

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

For each entry idle longer than PEL_IDLE_THRESHOLD_MS (default 60 seconds), it claims ownership with XCLAIM, re-publishes the 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)

XCLAIM takes ownership of the existing entry the dead worker never acknowledged. XADD then publishes a fresh copy, and XACK clears the stale PEL record. The loop catches asyncio.CancelledError separately and returns immediately on cancel. In the finally block of run(), the task is cancelled and awaited before the PostgreSQL pool closes:

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

The alternative was leaving the XADD retry path as the only re-queue mechanism (section 44). That works when the worker survives long enough to run its own failure handler. It does nothing for a worker that dies mid-task. The background scan is the safety net for the crash case the retry path cannot reach.

50. Task idempotency and deduplication

At-least-once delivery means a client that times out and retries a POST can submit the same logical task twice. Nothing in transmission 026 stopped that. Transmission 027 adds a deduplication layer.

The schema gains one nullable column and one 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. Keyless submissions write NULL to the column, and NULL values fall outside the uniqueness check, so every ordinary request still creates a new row. Only rows with a real key are compared.

The gateway checks the database before it inserts. The Idempotency-Key HTTP header takes precedence over the request 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
    ...

fetch_task_by_idempotency_key runs a SELECT on the key and returns a dict shaped for TaskResponse. A hit returns 200 OK with the original row. A miss falls through to the normal INSERT + XADD path and returns 202 Accepted. The header wins over the body field so HTTP-level infrastructure — load balancers, API gateways — can inject or strip the key independently of the request body.

OptionWhat it isWhy not chosen
Unconditional insertEvery POST creates a new rowNo deduplication at all
Redis cache lockCheck a Redis key before insertingA second system to coordinate, when PostgreSQL already enforces uniqueness
Partial unique indexDB-level uniqueness on keyed rows onlyChosen — engine-level enforcement, keyless rows unaffected

51. The full event loop — end to end

Every step across all four transmissions, fully chained:

1. Client sends:
   POST http://localhost:8000/tasks
   {"name": "generate_report", "payload": {"user_id": 42, "target": "q3_metrics"}}

2. Uvicorn receives the request on port 8000.

3. FastAPI routes to create_task().

4. Pydantic validates: name present, string, 1–128 chars. payload is dict. Passes.

5. Gateway generates:
   task_id = "26fd9233-55a2-453b-8825-3b5ea4e479d2"
   accepted_at = "2026-08-27T06:34:57.439656+00:00"

6. Gateway calls db.create_task():
   INSERT INTO tasks (...) VALUES (...)
   Row created: status='pending', retry_count=0, max_retries=3.

7. Gateway calls redis_client.xadd("tasks", {task_id, name, payload, accepted_at}).
   Redis appends entry. Message now in stream.

8. Gateway returns HTTP 202:
   {"task_id": "26fd9233...", "status": "accepted", "accepted_at": "..."}

9. Worker (blocked on XREADGROUP) receives the message.

10. Worker calls db.fetch_task_retry_info("26fd9233..."):
    Returns {retry_count: 0, max_retries: 3}

11. Worker calls db.update_task_status("26fd9233...", "processing"):
    Row updated: status='processing'.

12. Worker calls process_task():
    "generate_report" — no simulated failure. Sleeps 0.1s.
    Returns {"processed": True, "name": "generate_report"}

13. Worker calls client.xack("tasks", "workers", msg_id):
    Message removed from PEL.

14. Worker calls db.update_task_status("26fd9233...", "completed", result=...):
    Row updated: status='completed', result={...}, updated_at=now.

15. Developer queries:
    SELECT task_id, name, status FROM tasks;
    -> 26fd9233... | generate_report | completed

Loop is closed.

52. Concepts map

CLIENT
  |
  | HTTP POST /tasks  (optional Idempotency-Key header)
  v
GATEWAY (FastAPI + Uvicorn, port 8000)
  | Pydantic validates the request body
  | Idempotency lookup: header or body key -> fetch_task_by_idempotency_key()
  |   -> hit: return original row, HTTP 200 OK (no new task)
  |   -> miss: fall through to insert
  | UUID generated (before any DB write)
  | db.create_task() -> INSERT INTO tasks (status='pending', retry_count=0, idempotency_key)
  | xadd() -> publishes to Redis Stream "tasks"
  | Returns HTTP 202 Accepted immediately
  v
REDIS (Stream: "tasks")                   <- in-memory, AOF persistence
  | Append-only log
  | Consumer group "workers" tracks delivery per-worker
  | Pending Entries List: delivered but not yet ACKed
  | reclaim_idle_pel: XPENDING scan every 30s for entries idle > 60s
  v
WORKER (plain Python script, asyncio event loop, no HTTP server)
  | init_pool() -> 10-attempt retry loop, 2s back-off, until PostgreSQL is ready
  | XREADGROUP blocks on empty stream (no CPU waste)
  | fetch_task_retry_info() -> reads retry_count and max_retries
  | update_task_status("processing") -> marks task in flight
  | process_task() -> business logic (placeholder: sleep 0.1s)
  |
  |-- SUCCESS path:
  |     XACK -> clears PEL entry
  |     update_task_status("completed") -> final state
  |
  |-- FAILURE, retries remaining (retry_count < max_retries):
  |     update_task_for_retry() -> status='pending', retry_count+1, error_message
  |     backoff_s = min(REQUEUE_BASE_DELAY_S * 2^max(retry_count-1,0), REQUEUE_MAX_DELAY_S)
  |     asyncio.sleep(backoff_s) -> exponential back-off, capped at 30s
  |     xadd() -> re-publishes same fields as new stream entry
  |     xack() -> clears original from PEL (new entry is authoritative)
  |
  |-- FAILURE, retries exhausted (retry_count >= max_retries):
  |     update_task_as_failed() -> status='failed', error_message
  |     xack() -> clears from PEL permanently
  |
  | reclaim_idle_pel background task -> XCLAIM + XADD + XACK orphaned PEL entries
  | SIGTERM -> _shutdown=True -> finishes current task -> cancels pel_task -> close_pool() -> exit 0
  v
POSTGRESQL                                <- relational, disk-persisted
  | tasks table: task_id, name, status, payload, result
  | retry_count, max_retries, error_message, idempotency_key
  | created_at, updated_at
  | partial unique index on idempotency_key WHERE idempotency_key IS NOT NULL
  | GET /tasks/{task_id} -> gateway queries this, returns retry_count, max_retries, error_message

Infrastructure:
  All four services on nexusflow_net (bridge network, Docker DNS)
  Named volumes: postgres_data, redis_data survive restarts
  Health checks gate startup: service_healthy not service_started
  Multi-stage Dockerfiles: builder installs, runtime runs, nothing extra
  COPY . . with .dockerignore: all service files included automatically
  Non-root user (appuser) in both application containers
  Layer caching: requirements.txt copied before source code
  urllib.request health check: no curl dependency in slim image

53. What comes next

The application layer is hardened. The next work moves the platform from Docker Compose to Kubernetes.

ItemWhy it matters
Kubernetes manifests (Deployments, Services, ConfigMaps)Moving the 4 services from Docker Compose to native Kubernetes objects.
AWS EKS Cluster ProvisioningDeploying to managed Kubernetes in the cloud with managed PostgreSQL (RDS) and Redis (ElastiCache).
Horizontal Pod Autoscaling (HPA) on queue lagScaling worker pods dynamically based on Redis Stream unread length rather than just CPU/memory.
Prometheus & Grafana telemetryExposing gateway throughput, task latency percentiles, and retry rates to Prometheus.
Helm chart packagingBundling the entire multi-service deployment into a configurable Helm chart.
Pod lifecycle hooks & preStopEnsuring in-flight tasks finish processing before Kubernetes terminates worker pods during rollouts.

References

Redis

FastAPI and Python async

asyncpg and PostgreSQL

Docker

Kubernetes (relevant to design decisions made in NexusFlow)

Distributed systems concepts

This guide covers every concept in transmissions 024, 025, 026, and 027. The application layer is hardened. The next transmission moves the platform to Kubernetes.