The previous article in Programmer Craft in the AI Era discussed databases. A database is the hardest constraint layer in a system, but not every piece of work should be pushed into the database and the request path. Many features become stable only when caches, queues, and asynchronous jobs are designed properly.

AI often writes slow operations directly into API handlers: a request arrives, the code queries the database, calls an external service, generates a file, writes history records, sends a notification, and finally returns a response. The code looks straightforward. The user waits too long. The whole system can also be dragged down by one slow downstream dependency.

Do not put slow work inside the request. But moving slow work elsewhere does not make the problem disappear. Caches have consistency problems. Queues can pile up. Async jobs may run more than once. Retries can create side effects. Logs and history records can even slow the main workflow down.

What Caches, Queues, and Async Jobs Are Responsible For

These three tools often appear together, but they solve different problems.

Component	Main problem it solves	Most common mistake
Cache	Reduces repeated computation and repeated reads, lowering latency	Treating it as the source of truth
Queue	Smooths traffic spikes, decouples components, serializes work, delays processing	Treating it as a trash bin that never fills up
Async job	Moves slow operations out of the request path and finishes them in the background	Forgetting idempotency, retries, progress, and failure recovery

Take “a user uploads a CSV, and the system imports it and generates a report” as an example.

The request should only do a few things:

1. Authenticate the user.
2. Save the file.
3. Create an import job.
4. Return the job ID.

Parsing, validation, batch writes, report generation, and user notification should move into the background workflow.

At that point, a cache may store job progress or intermediate computation results. A queue hands the job to a worker. The async job performs the import. Each component has a boundary. This is not simply “add Redis”, “add an MQ”, or “start a background thread”.

When asking AI to write such a feature, if you only say “handle it asynchronously”, it may casually start a thread. A better instruction is:

The HTTP request only creates a job and returns job_id.
File parsing, data validation, batch writes, and report generation are all executed by a background worker.
Job status must be queryable. Failures must include an error code and an error summary.
If a worker restarts, jobs that have been running for more than 30 minutes can be reclaimed.
Each job must execute idempotently by job_id.

This instruction defines the execution boundary, state recovery, and idempotency.

A Cache Is Not a Second Database

The temptation of a cache is that it is fast. So fast that people unconsciously start using it as a database.

A common cache read looks like this:

async def get_user_profile(user_id: str):
    cached = await redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    profile = await db.get_user_profile(user_id)
    await redis.set(f"user:{user_id}", json.dumps(profile), ex=300)
    return profile

This works as a read cache, but questions immediately appear:

• When does the cache expire after the user profile is updated?
• If the database returns nothing, should the empty result be cached too?
• If many requests miss at the same time, will all of them hit the database?
• If the serialized fields change, can old cache values still be read?
• If Redis is down, should the API fail or degrade to reading the database?

A cache design should answer at least four questions: key, value, TTL, and invalidation strategy.

Question	Example
How is the key designed?	user_profile:{user_id}:v2
What is stored as the value?	A response model, not an internal ORM object
How long is the TTL?	Five minutes, one hour, or based on business update frequency
How is it invalidated?	Delete after write, update after write, natural expiry, versioned keys

Cache keys should usually include a version. When the field structure changes, you can switch directly to v2 and avoid deserialization failures on old values:

cache_key = f"user_profile:v2:{user_id}"

Caching empty values also matters. For example, if a user does not exist and every request penetrates to the database, malicious requests can overload the database:

if profile is None:
    await redis.set(cache_key, "__null__", ex=60)
    return None

The TTL for empty values should be shorter, otherwise newly created data may remain invisible for too long.

Cache Breakdown and Singleflight

The biggest cache problem is not a miss. It is many concurrent misses for the same hot key.

Suppose the homepage configuration cache expires every five minutes. Exactly when it expires, 1,000 requests arrive. If every request queries the database or calls a configuration service, the cache becomes a traffic amplifier.

Inside one process, you can use the singleflight idea: at any moment, only one loading task is allowed for the same key. Other requests wait for the result.

class SingleFlightCache:
    def __init__(self):
        self._locks: dict[str, asyncio.Lock] = {}

    async def get_or_load(self, key: str, loader):
        cached = await redis.get(key)
        if cached is not None:
            return decode(cached)

        lock = self._locks.setdefault(key, asyncio.Lock())
        async with lock:
            cached = await redis.get(key)
            if cached is not None:
                return decode(cached)

            value = await loader()
            await redis.set(key, encode(value), ex=300)
            return value

This is only an illustration. Production code also needs to clean up _locks, handle loader exceptions, and control waiting timeouts. If the service runs across multiple processes or machines, you need to consider distributed locks, background refresh, early refresh, or randomized TTLs to reduce simultaneous expiry.

When asking AI to write cache logic, you can require:

Hot keys need cache-breakdown protection.
Use singleflight inside one process so concurrent misses do not hit the database at the same time.
Add 10% random jitter to TTLs to avoid many keys expiring together.
If Redis fails, degrade to database reads, but record metrics and limit concurrency.

A cache is not just a get/set wrapper. It is another read path. Once there are multiple read paths, you must think about consistency and failure modes.

A Queue Is Not a Trash Bin

Queues are often used for “decoupling”. The caller writes a task into the queue. Workers process it slowly. The main flow looks lighter.

But queues also have capacity. Workers also have speed limits. Downstream systems may also be slow. A queue only moves pressure from the request side to the background side. Without backpressure, the background side will eventually break.

A job message should contain at least:

{
  "job_id": "job_123",
  "job_type": "import_csv",
  "dedupe_key": "user_1:file_abc",
  "attempt": 0,
  "created_at": "2026-07-03T10:00:00+08:00",
  "payload": {
    "file_id": "file_abc",
    "user_id": "user_1"
  }
}

job_id is for tracing. dedupe_key is for deduplication. attempt is for retry control. Do not put only the business payload into the message.

Queue design should answer these questions first:

• How many messages can the queue hold at most?
• How many messages can workers process per second?
• How much concurrency can downstream services tolerate?
• How are failed tasks retried?
• After how many retries does a task enter dead letter handling?
• Should tasks from the same user be rate-limited?
• Should tasks for the same business object preserve order?
• Can the frontend continue submitting tasks when the queue is backed up?

If these are not designed, AI-generated consumers often look like this:

async def worker():
    while True:
        message = await queue.get()
        await process(message)
        await queue.ack(message)

This code has no timeout, no failure classification, no retry, no dead letter handling, no concurrency limit, and no shutdown logic. It can run, but it is not enough.

Idempotent Consumption Is More Important Than Retry

Queue systems usually promise “at least delivery as much as possible”. They do not promise “exactly once”. Messages can be duplicated. A worker may process successfully but fail to ack. After a consumer restarts, it may receive the same message again.

So consumers must be idempotent.

A processing-record table can look like this:

create table processed_events (
  event_id text primary key,
  status text not null,
  processed_at timestamptz not null default now(),
  error_code text
);

The consumer first claims the event, then executes it:

async def consume(event: Event):
    inserted = await processed_events.try_insert(event.event_id, status="running")
    if not inserted:
        return

    try:
        await handle_event(event)
        await processed_events.mark_done(event.event_id)
    except RetryableError as exc:
        await processed_events.mark_retryable(event.event_id, exc.code)
        raise
    except PermanentError as exc:
        await processed_events.mark_failed(event.event_id, exc.code)
        await dead_letters.add(event, reason=exc.code)

If an external side effect is not idempotent by itself, such as sending an SMS, charging money, or creating a ticket, you must use a business idempotency key on the external API, or persist the side-effect state in your own database.

For example, notification sending can be protected with:

create unique index idx_notifications_dedupe
on notifications (user_id, template_id, dedupe_key);

For the same user, template, and deduplication key, only one notification can be created. Even if a message is duplicated, it will not send repeatedly.

When asking AI to write a consumer, add hard constraints:

The consumer must be idempotent by event_id.
External side effects must have a business dedupe_key.
Retries may execute the same message more than once, so the code must be safe.
Please write a test that delivers the same event_id repeatedly.

Without idempotency, retry only amplifies accidents.

Retries Need Error Types

“Retry on failure” is a dangerous sentence.

Errors should at least be split into two types:

Type	Example	Handling
Temporary error	Timeout, connection failure, downstream 503, rate limit	Retry after a delay
Permanent error	Invalid parameter, missing resource, insufficient permission, invalid format	Do not retry; send to dead letter handling or mark failed

If permanent errors are retried, the queue will be stuck with bad messages. If temporary errors are marked failed immediately, the system loses tasks that could have recovered.

Retries also need backoff:

def next_retry_delay(attempt: int) -> int:
    delays = [10, 30, 120, 300, 900]
    return delays[min(attempt, len(delays) - 1)]

You can also add random jitter to avoid many tasks retrying at the same time.

A job table usually needs these fields:

attempt integer not null default 0,
max_attempts integer not null default 5,
next_run_at timestamptz,
last_error_code text,
last_error_message text

When a worker fails, update the state:

update jobs
set attempt = attempt + 1,
    status = case
      when attempt + 1 >= max_attempts then 'failed'
      else 'pending'
    end,
    next_run_at = :next_run_at,
    last_error_code = :error_code,
    last_error_message = :error_message
where id = :job_id;

The transaction boundary matters here too. Job-state updates must be reliable. A logging failure must not prevent the job state from being updated.

Dead Letters Are Not a Failure Dump

Dead letter queues are often created and then ignored. At that point, they are just another trash pile.

Dead letter handling should be a failure entry point that can be investigated, replayed, and measured.

A dead letter record should include at least:

• Original message.
• Business keys, such as job_id, user_id, and order_id.
• Failure type and error code.
• Last error summary.
• Attempt count.
• First failure time and last failure time.
• Whether it can be replayed manually.
• Handling status.

create table dead_letters (
  id uuid primary key,
  queue_name text not null,
  message_key text not null,
  payload jsonb not null,
  error_code text not null,
  error_message text,
  attempts integer not null,
  first_failed_at timestamptz not null,
  last_failed_at timestamptz not null,
  status text not null check (status in ('open', 'ignored', 'replayed', 'resolved'))
);

After the table exists, there must also be operational actions: query, filter, replay, ignore, batch handling, and alerting. Otherwise dead letters only pile up quietly in the database.

When asking AI to write dead letter logic, require:

Dead letters must be queryable, replayable, and markable as handled.
Dead letter records retain the original payload, error code, attempt count, and business key.
Replay must generate a new replay_id, but retain the original event_id for idempotency checks.

The value of dead letters is not “where to put failures”. It is “how to recover after failure”.

Batch Processing Is Much More Stable Than Writing One Row at a Time

Background jobs often write logs, history records, statistics, and synchronized status. AI easily writes code that inserts into the database after every single processed item.

If a job has 200,000 records, and each record writes one log and commits once, performance will be poor. It may also make the database slow enough to drag the worker backward.

A steadier pattern is an in-memory queue plus background batch writes:

class AsyncLogWriter:
    def __init__(self, batch_size: int = 1000, flush_interval: float = 5.0):
        self.queue = asyncio.Queue(maxsize=50_000)
        self.batch_size = batch_size
        self.flush_interval = flush_interval

    async def write(self, record: dict):
        try:
            self.queue.put_nowait(record)
        except asyncio.QueueFull:
            if record.get("level") == "DEBUG":
                return
            raise

    async def run(self):
        batch = []
        while True:
            try:
                item = await asyncio.wait_for(self.queue.get(), timeout=self.flush_interval)
                batch.append(item)
                if len(batch) < self.batch_size:
                    continue
            except asyncio.TimeoutError:
                pass

            if batch:
                await insert_logs_batch(batch)
                batch.clear()

Several details matter here:

• The execution process only puts logs into an in-memory queue.
• A background task writes to the database in batches.
• The queue has an upper bound.
• When the queue is full, low-value DEBUG logs can be dropped or sampled.
• A slow database must not apply unlimited backpressure to the main execution flow.

This pattern is not only for logs. Many history records, telemetry details, low-priority audit records, and statistical events can be handled similarly. The key is to distinguish “business state that must be persisted synchronously” from “observability data that can be written asynchronously in batches”.

When asking AI to design this kind of write path, require:

Do not synchronously write one database row at a time inside the main execution loop.
The stdout/stderr reader only puts logs into an in-memory queue.
A background task writes to the DB in batches of 1000-5000 rows or every 5 seconds.
When the queue is full, DEBUG logs are sampled or summarized and dropped; they must not backpressure the execution process.
Business state updates must still be persisted reliably.

These constraints can significantly change system throughput.

Backpressure Is the System Saying “I Cannot Keep Up”

A queue can smooth spikes, but it cannot eliminate them. If the producer is faster than the consumer for long enough, the queue will eventually fill.

Backpressure means the system tells upstream callers: I cannot keep up.

Common backpressure strategies:

Scenario	Strategy
Queue length exceeds a threshold	Reject new tasks and ask callers to retry later
A user submits too quickly	Rate-limit by user
Too many low-priority tasks	Lower priority or delay execution
Downstream is slow	Reduce worker concurrency
Log queue is full	Drop DEBUG logs and keep ERROR summaries
Batch processing falls behind	Pause the import entry point or reduce batch size

Backpressure is not failure. It protects the system.

For example, before creating a task, check the number of pending jobs for the user:

select count(*)
from jobs
where user_id = :user_id
  and status in ('pending', 'running');

If the number exceeds the limit, reject the request:

{
  "error": {
    "code": "TOO_MANY_PENDING_JOBS",
    "message": "There are many queued jobs right now. Please try again later.",
    "retryable": true
  }
}

Without backpressure, the system looks very “hardworking”: APIs keep accepting tasks, queues keep piling up, workers keep chasing, the database keeps slowing down, and finally everyone times out together.

Async Jobs Need a State Machine Too

An async job is not just some code running in the background. It should have a state machine.

A common set of states:

State	Meaning
pending	Created and waiting to run
running	Claimed by a worker and executing
success	Completed successfully
failed	Unrecoverable failure or retry limit exceeded
retrying	Waiting for the next retry
cancelled	Cancelled by the user or system
timeout	Execution exceeded its lease

Workers should not claim tasks with a read followed by a separate update. Use an atomic update or skip locked:

select *
from jobs
where status = 'pending'
  and run_at <= now()
order by priority desc, run_at asc
limit 100
for update skip locked;

After claiming jobs, write locked_by and locked_at. If a worker crashes, a background sweep can reclaim jobs that are still running and whose locked_at is too old.

update jobs
set status = 'pending',
    locked_by = null,
    locked_at = null,
    run_at = now()
where status = 'running'
  and locked_at < now() - interval '30 minutes';

This is recoverability. Without a state machine and leases, jobs may remain stuck in running forever after a worker restart.

A Cache-and-Queue Prompt for AI

When asking AI to write cache, queue, or async-job code, you can start with this instruction:

Do not write code yet.

Please design a cache, queue, and asynchronous job plan for this feature:
1. Which operations must finish inside the request, and which must be asynchronous.
2. Cache key, value, TTL, invalidation strategy, empty-value caching, and cache-breakdown protection.
3. Queue message schema: job_id/event_id, business key, attempt, payload, trace_id.
4. Worker concurrency limits, downstream rate limits, task claiming method, and lease design.
5. Idempotency design: repeated messages, repeated jobs, and external side effects.
6. Error classification: temporary errors, permanent errors, retry count, and backoff strategy.
7. Dead letter design: fields, query, replay, ignore, and alerting.
8. Batch writes: which logs or history records can be written asynchronously in batches, and what batch size and flush interval to use.
9. Backpressure strategy: how to protect the system when the queue is full, one user has too many jobs, or downstream becomes slow.
10. Status query API, test cases, and key metrics.

After the plan is confirmed, then write the code.

The purpose of this prompt is to make AI acknowledge that systems can be slow, full, duplicated, and broken. Once those facts are acknowledged, the code has a chance to be stable.

Questions to Ask Before Writing

Before coding, ask at least once:

• Does this operation really need to finish inside the request?
• When the cache expires, can it overload downstream systems at the same time?
• Can the data in the cache be older than the database?
• If a message is delivered more than once, can it create side effects more than once?
• Are temporary errors and permanent errors separated?
• Can dead letters be viewed, replayed, and closed by someone?
• When the queue is full, do we reject, degrade, or keep piling up?
• Can batched logs or history records backpressure the main workflow?
• After a worker crashes, can the job be reclaimed?
• Are there metrics for queue length, consumer lag, retries, and dead letters?

Caches, queues, and async jobs do not exist to make code look sophisticated. Their value is that they let the system keep its boundaries when things are slow, full, wrong, or unstable.

AI can quickly write Redis access, queue consumers, and background workers. The programmer’s job is to turn “run it in the background” into a recoverable, observable, rate-limited, and explainable system.

WeChat

Follow ZiCode on WeChat

If this post was useful, you can follow later updates on WeChat as well.

X / Twitter

Follow @ax2_zicode

Faster technical notes, short thoughts, and new-post alerts are posted on X.

X Open X Profile @ax2_zicode