I Crashed My AI Worker Twice: Azure Service Bus for AI Apps

AI calls are slow and traffic comes in bursts, so your API should never wait for the model. This guide puts Azure Service Bus between them and tests every claim on a local emulator, including two real crashes. One request came back after 60.7 seconds. The other was gone.

You don't need an Azure subscription or any messaging background: Python and Docker are enough. Every section ends with a one-line Remember, every animation replays a real run, and the whole guide fits in 8 lines at the end.

The guide has three parts:

• Part 1 · Understand it: why the API shouldn't wait, the five parts, and the patterns, sections 1–7.
• Part 2 · Never lose a request: the crash test, settling messages and poison messages, sections 8–10.
• Part 3 · Do it yourself: the emulator, all the code, and moving to Azure, section 11.

If you read the Redis Streams guide, this is the same set of ideas as a managed Azure service.

1. Why your API shouldn't wait for the model

Picture a contract-analysis app. A client uploads a contract, and an LLM pulls out the parties, dates and amounts. Each call takes 5 to 30 seconds. At 9 a.m. a few hundred contracts arrive in a few minutes, but the model can only work on a few at a time.

If the API calls the model directly, it has to hold the connection open the whole time. Threads fill up, clients time out, and anything in flight is lost when a server restarts.

The fix is a message broker: a middleman that takes the work, stores it safely and hands it out. Think of a restaurant ticket rail. The waiter clips the order and goes back to the tables. The kitchen takes tickets at its own pace. Nobody stands at the stove waiting.

Animated12 requests · illustrative

Left: 2 served, 10 lost. Right: 12 served, 0 lost.

←Replay→step 4 of 4

Azure Service Bus is Microsoft's managed broker. You create a namespace (your Service Bus account, with one address like contoso-ai.servicebus.windows.net), and inside it you create queues and topics. Clients talk to it over AMQP, which the Python SDK handles for you.

2. The five parts you'll use

Everything in this guide is one of five parts, and each keeps its colour in every figure:

• Producer: your API. It sends messages and moves on.
• Queue: holds messages. Each one goes to one receiver.
• Worker: takes a message, runs the model, says "done".
• Topic + subscriptions: gives every subscription its own copy.
• Dead-letter queue: a side queue where broken messages get parked.

The mental modelfive parts, one colour each

3. Load leveling: the queue absorbs the burst

When the burst arrives, the queue takes all of it. Workers drain it at a steady pace. You size your GPU workers for an average hour, not the busiest minute of the day. That's load leveling.

The producer and the worker also don't need to be online at the same time. When I sent 6 requests with no worker running, all 6 waited in the queue. That's temporal decoupling, and it's why a model deployment or a GPU node restart doesn't lose customer requests.

Animated10 requests · 1 worker · 0.3 s a job · when I ran it

worker finished 10, queue depth now 0

←Replay→step 7 of 7

When I ran it (p01_queue_patterns.py, step 3):

--- 3) Load leveling: a burst of 10 arrives, one worker drains it at a steady pace ---
API accepted all 10 in a single send call; nothing was rejected
worker finished  2, queue depth now 8
worker finished  4, queue depth now 6
worker finished  6, queue depth now 4
worker finished  8, queue depth now 2
worker finished 10, queue depth now 0
Worker drained the burst in 3.6 s at its own pace (10 x 0.3 s of work + overhead)

Queue depth (the active message count) tells you if you have enough workers. Rising for minutes? Add workers. Always zero? You may have too many. Azure Monitor charts it, and Azure Container Apps or Azure Functions can scale on it automatically with KEDA.

4. Competing consumers: more workers, no extra code

Start several workers on the same queue. Service Bus hands each message to exactly one of them. That's the competing consumers pattern. To go faster, run more copies of the same worker, on Container Apps, AKS or Functions. If one crashes, the others carry on.

Animated6 documents · 2 workers · when I ran it

Total 6 of 6, no document processed twice

←Replay→step 7 of 7

worker-A processed 3: ['doc-001', 'doc-003', 'doc-005']
worker-B processed 3: ['doc-002', 'doc-004', 'doc-006']
Total 6 of 6, no document processed twice

5. Topics and filters: one result, many services

When an analysis finishes, several services care: one notifies the user, one writes an audit record, one updates a dashboard. Sending to a queue would give the result to only one of them.

A topic fixes that. You send once, and every subscription on the topic gets its own copy. Each subscription behaves like a private queue, so if the audit service is down for 10 minutes, its copies wait for it. You can add a new subscriber next month without touching the sender.

A subscription can also have a filter, which Service Bus evaluates itself. My high-priority subscription has the SQL filter priority = 'high', so it only receives messages whose priority property is high. A subscription with no rules gets everything. For simple equality checks, a correlation filter is faster than SQL.

Animated5 results published once · when I ran it

published doc-005 priority=low

←Replay→step 6 of 6

When I ran it (p02_topic_fanout.py):

notifications  holds 5 message(s)
high-priority  holds 2 message(s)
--- 3) Each consumer reads its own subscription ---
notifications  received 5: ['doc-001(standard)', 'doc-002(high)', 'doc-003(standard)', 'doc-004(high)', 'doc-005(low)']
high-priority  received 2: ['doc-002(high)', 'doc-004(high)']
--- 4) Completing in one subscription never touches the other ---
after notifications completed its copy, high-priority still holds 1

The last line matters: finishing a copy in one subscription never touches the copy in another.

6. What goes in a message

A Service Bus message has three parts:

• Body: the work itself. For AI requests, JSON with the prompt, model, temperature and document ID.
• Application properties: your own tags, like priority or model_name. Filters and routers read these without opening the body.
• System properties: message_id, correlation_id, content_type, time_to_live and more.

message = ServiceBusMessage(
    body=json.dumps({"prompt": "Extract the parties and effective date.",
                     "model": "gpt-4o", "temperature": 0.1, "document_id": "doc-2025-0142"}),
    content_type="application/json",
    message_id=str(uuid.uuid4()),        # unique for every message
    correlation_id="req-abc-12345",      # the same ID across the whole journey
    application_properties={"model_name": "gpt-4o", "priority": "standard"},
    time_to_live=timedelta(minutes=5),   # stale requests expire
)

Two IDs, two jobs. message_id must be unique, because duplicate detection drops a second message with the same ID. correlation_id is created once, when the client submits, and copied onto every message and log line after that. When something fails, one search for that ID shows the request's whole path.

Time-to-live is how long a message may wait. A real-time classification that's 10 minutes old is useless, so give it a short TTL. In my run, a message with a 2-second TTL and no worker landed in the dead-letter queue with reason TTLExpiredException. I also asked for 24 hours on another message and got 1:00:00, because the queue's own limit (1 hour in my config) wins.

One Python surprise: custom properties come back as bytes.

application_properties: {'model_name': 'gpt-4o', 'priority': 'standard', 'document_type': 'contract'}
  raw, as delivered   : {b'model_name': b'gpt-4o', b'priority': b'standard', b'document_type': b'contract'}

So msg.application_properties["priority"] raises KeyError. Decode keys and values first; the props() helper in section 11 does it.

Sending many small requests, like one per page of a document? Use a batch. My 40 page requests of about 9 KB each went out in 2 network calls instead of 40.

7. Big payloads: send the address, not the parcel

The Standard tier caps a message at 256 KB. Premium goes to 100 MB. A scanned contract, an image or a batch of embeddings can be bigger than either.

The claim-check pattern works on any tier. Upload the file to Azure Blob Storage, then send a tiny message holding only its address. The worker reads the address and downloads the file. It's a coat-check ticket: you hand over the coat once and carry a small token around.

Animated300 KB document · 256 KB limit · when I ran it

The worker downloads the document from the address: 300,000 bytes.

←Replay→step 5 of 5

sending 300 KB directly -> MessageSizeExceededError (limit here is 256 KB)
message body was 100 bytes; worker fetched 300,000 bytes from blob_store/doc-2025-0142.pdf

In my script a local folder stands in for Blob Storage. On Azure it's upload_blob() on one side and a download on the other.

8. The crash test: peek-lock vs receive-and-delete

A worker can receive a message in one of two receive modes:

• Receive-and-delete: the message is removed from the queue the moment it's delivered.
• Peek-lock (the default): the worker gets a locked copy. The message stays in the queue, hidden from other workers, until this worker calls complete_message(). If it never does, the lock expires (after 1 minute by default, 5 at most) and the message becomes available again.

I wanted to see this happen, so I wrote a test where worker A receives req-42 and then its process really dies with os._exit(1): no "done", no clean shutdown. Then worker B waits up to 90 seconds.

Animateda real os._exit crash · when I ran it

Peek-lock (default)Receive-and-delete

The lock expires. Worker B receives req-42 60.7 s after it was sent, with delivery_count=1, and completes it.

←Replay→step 5 of 5

When I ran it (p05_crash_test.py):

--- PEEK-LOCK ---
  worker A got 'req-42' ... running inference ... CRASH
  worker B got 'req-42' 60.7 s after it was sent (delivery_count=1)
--- RECEIVE-AND-DELETE ---
  worker A got 'req-42' ... running inference ... CRASH
  worker B waited 90 s: nothing. req-42 is gone for good.

60.7 seconds is the 1-minute lock plus a little overhead. delivery_count=1 shows Service Bus knew it was a second delivery. With receive-and-delete, the request was deleted on delivery, so when the worker died there was nothing left to give anyone.

Peek-lock costs one extra "done" call per message. Next to seconds of inference, that's nothing. Use receive-and-delete only for data you can afford to lose, like telemetry pings.

The catch: peek-lock gives you at-least-once delivery. If a worker finishes the model call and dies just before saying "done", the request runs again on another worker. Save results under the document ID or correlation ID, so a repeat overwrites the same record instead of creating a second one.

Long jobs: if inference can take longer than the lock, the lock expires mid-work and a second worker starts the same job. Renew it with receiver.renew_message_lock(msg), or let AutoLockRenewer do it in the background:

renewer = AutoLockRenewer()
renewer.register(receiver, msg, max_lock_renewal_duration=600)   # keep it locked up to 10 min
run_long_inference(msg)
receiver.complete_message(msg)
renewer.close()

9. Four ways to finish a message

In peek-lock mode the worker must settle each message. Pick the call that matches the kind of failure:

try:
    job = json.loads(str(msg))
    save_result(job["document_id"], run_inference(job))   # keyed by ID: safe to repeat
    receiver.complete_message(msg)
except json.JSONDecodeError:
    receiver.dead_letter_message(msg, reason="MalformedPayload", error_description="Body is not JSON")
except TimeoutError:
    receiver.abandon_message(msg)                           # transient: try again

10. Poison messages and the dead-letter queue

A poison message fails every time. Retried forever, it would loop and burn GPU time. Service Bus counts deliveries, and after max_delivery_count (default 10; my queue uses 5) it moves the message to the dead-letter queue by itself, with reason MaxDeliveryCountExceeded.

Animatedreq-5 always times out · max delivery count 5 · when I ran it

req-5: reason=MaxDeliveryCountExceeded, delivery_count=5

←Replay→step 7 of 7

Every queue and subscription has a dead-letter queue already. You don't create it; you read it with sub_queue=ServiceBusSubQueue.DEAD_LETTER. In my run (p04_reliable_processing.py) it held three messages, each with a reason:

queued: ['req-1', 'req-2', 'req-3', 'req-4', 'req-5']
--- 2) A peek-lock worker settles every message (complete/abandon/dead-letter/defer) ---
req-1: COMPLETE
req-2: DEAD-LETTER (bad JSON - retrying can never fix it)
req-3: DEAD-LETTER (unknown model gpt-9)
req-4: DEFER (waiting for a dependency), sequence_number=63
req-5: ABANDON (timeout, try again)  delivery_count=0
req-5: ABANDON (timeout, try again)  delivery_count=1
req-5: ABANDON (timeout, try again)  delivery_count=2
req-5: ABANDON (timeout, try again)  delivery_count=3
req-5: ABANDON (timeout, try again)  delivery_count=4
--- 3) Later: the dependency is ready, fetch the deferred message by number ---
req-4: deferred message fetched and COMPLETED
--- 4) Inspect the dead-letter queue ---
req-2: reason=MalformedPayload

Two of those reasons are mine (I dead-lettered them in code). The third was written by Service Bus. To reprocess, fix the cause first, then send a new copy to the main queue and complete the dead-letter copy. I "deployed" gpt-9, resubmitted only req-3, and it went through. Resubmit before fixing and the message just fails again and comes back.

Put an Azure Monitor alert on the dead-letter count. A sudden jump usually means a broken model deployment or a new input format your worker doesn't understand.

11. Run it yourself

Microsoft ships a Service Bus emulator that runs in Docker. Your Python code uses the same SDK and the same calls as on Azure; only the connection line changes. You need Python 3.12+ and Docker Desktop.

mkdir -p ~/sb-course/emulator ~/sb-course/practice && cd ~/sb-course
python3 -m venv .venv && source .venv/bin/activate
pip install azure-servicebus==7.14.1 azure-identity==1.21.0

Save these two files in emulator/. Config.json creates the queue (max delivery count 5, dead-letter on expiry), the topic, both subscriptions and the filter.

# emulator/docker-compose.yaml
name: sb-emulator
services:
  emulator:
    container_name: servicebus-emulator
    image: mcr.microsoft.com/azure-messaging/servicebus-emulator:latest
    pull_policy: always
    volumes:
      - ./Config.json:/ServiceBus_Emulator/ConfigFiles/Config.json
    ports:
      - "5672:5672"       # AMQP: what your Python code connects to
      - "5300:5300"       # health check endpoint
    environment:
      SQL_SERVER: sqledge
      MSSQL_SA_PASSWORD: "Practice!Passw0rd"
      ACCEPT_EULA: "Y"
      SQL_WAIT_INTERVAL: "15"
    depends_on:
      - sqledge
  sqledge:
    container_name: servicebus-sqledge
    image: mcr.microsoft.com/azure-sql-edge:latest
    environment:
      ACCEPT_EULA: "Y"
      MSSQL_SA_PASSWORD: "Practice!Passw0rd"

{
  "UserConfig": {
    "Namespaces": [
      {
        "Name": "sbemulatorns",
        "Queues": [
          {
            "Name": "inference-requests",
            "Properties": {
              "DeadLetteringOnMessageExpiration": true,
              "DefaultMessageTimeToLive": "PT1H",
              "LockDuration": "PT1M",
              "MaxDeliveryCount": 5,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          }
        ],
        "Topics": [
          {
            "Name": "inference-results",
            "Properties": {
              "DefaultMessageTimeToLive": "PT1H",
              "RequiresDuplicateDetection": false
            },
            "Subscriptions": [
              {
                "Name": "notifications",
                "Properties": {
                  "DeadLetteringOnMessageExpiration": true,
                  "DefaultMessageTimeToLive": "PT1H",
                  "LockDuration": "PT1M",
                  "MaxDeliveryCount": 5,
                  "RequiresSession": false
                },
                "Rules": []
              },
              {
                "Name": "high-priority",
                "Properties": {
                  "DeadLetteringOnMessageExpiration": true,
                  "DefaultMessageTimeToLive": "PT1H",
                  "LockDuration": "PT1M",
                  "MaxDeliveryCount": 5,
                  "RequiresSession": false
                },
                "Rules": [
                  {
                    "Name": "filter-high-priority",
                    "Properties": {
                      "FilterType": "Sql",
                      "SqlFilter": {
                        "SqlExpression": "priority = 'high'"
                      }
                    }
                  }
                ]
              }
            ]
          }
        ]
      }
    ],
    "Logging": {
      "Type": "File"
    }
  }
}

Start it. The first start downloads about 1–2 GB.

cd ~/sb-course/emulator && docker compose up -d
docker logs servicebus-emulator 2>&1 | grep "Successfully Up"

Then save the scripts below in practice/ and run each with python <file> from inside that folder. sb_connection.py uses the emulator by default and switches to Azure (signed in with your az login identity, no keys) when SERVICE_BUS_FQDN is set.

# practice/sb_connection.py
"""
sb_connection.py - the ONLY file that knows where Service Bus is.

  SERVICE_BUS_FQDN not set -> LOCAL emulator in Docker (fixed connection string)
  SERVICE_BUS_FQDN set     -> AZURE Service Bus namespace (Microsoft Entra ID login)

Every practice script does:  from sb_connection import get_client, QUEUE, TOPIC
"""
import os
from azure.servicebus import ServiceBusClient, ServiceBusReceiveMode, ServiceBusSubQueue

QUEUE = "inference-requests"
TOPIC = "inference-results"

# The emulator always uses this connection string (it is printed in its log).
EMULATOR = ("Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;"
            "SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;")


def get_client() -> ServiceBusClient:
    fqdn = os.getenv("SERVICE_BUS_FQDN", "").strip()
    if not fqdn:
        print("Using LOCAL Service Bus emulator (localhost:5672)")
        return ServiceBusClient.from_connection_string(EMULATOR)

    from azure.identity import DefaultAzureCredential     # your 'az login' identity
    print(f"Using AZURE Service Bus namespace {fqdn}")
    return ServiceBusClient(fully_qualified_namespace=fqdn, credential=DefaultAzureCredential())


def props(msg) -> dict:
    """application_properties arrive as bytes (b'priority': b'high') - convert to normal text."""
    clean = {}
    for key, value in (msg.application_properties or {}).items():
        key = key.decode() if isinstance(key, bytes) else key
        value = value.decode() if isinstance(value, bytes) else value
        clean[key] = value
    return clean


def empty(client, queue=None, topic=None, subscription=None, dead_letter=False) -> int:
    """Delete leftovers from earlier runs so every run starts from an empty queue."""
    sub_queue = ServiceBusSubQueue.DEAD_LETTER if dead_letter else None
    mode = ServiceBusReceiveMode.RECEIVE_AND_DELETE
    if topic:
        receiver = client.get_subscription_receiver(topic, subscription, sub_queue=sub_queue,
                                                    receive_mode=mode, max_wait_time=2)
    else:
        receiver = client.get_queue_receiver(queue or QUEUE, sub_queue=sub_queue,
                                             receive_mode=mode, max_wait_time=2)
    removed = 0
    with receiver:
        for _ in receiver:
            removed += 1
    return removed

# practice/p01_queue_patterns.py
"""p01 - Units 2-3: a queue as a buffer - temporal decoupling, competing consumers, load leveling."""
import json
import time
from azure.servicebus import ServiceBusMessage
from sb_connection import get_client, empty, QUEUE

client = get_client()
with client:
    empty(client)

    print("\n--- 1) Temporal decoupling: the API sends while NO worker is running ---")
    with client.get_queue_sender(QUEUE) as sender:
        for i in range(1, 7):
            body = {"document_id": f"doc-{i:03}", "prompt": "Extract the parties.", "model": "gpt-4o"}
            sender.send_messages(ServiceBusMessage(json.dumps(body), content_type="application/json"))
            print(f"API: queued doc-{i:03} and answered the user at once")
    with client.get_queue_receiver(QUEUE) as peeker:
        waiting = peeker.peek_messages(max_message_count=50)        # look, don't take
    print(f"Messages waiting safely in the queue: {len(waiting)}")

    print("\n--- 2) Competing consumers: two workers share ONE queue ---")
    seen = {"worker-A": [], "worker-B": []}
    with client.get_queue_receiver(QUEUE, max_wait_time=3) as worker_a, \
         client.get_queue_receiver(QUEUE, max_wait_time=3) as worker_b:
        while True:
            took = False
            for name, worker in (("worker-A", worker_a), ("worker-B", worker_b)):
                for msg in worker.receive_messages(max_message_count=1, max_wait_time=2):
                    seen[name].append(json.loads(str(msg))["document_id"])
                    worker.complete_message(msg)            # done -> removed from the queue
                    took = True
            if not took:
                break
    for name, docs in seen.items():
        print(f"{name} processed {len(docs)}: {docs}")
    print(f"Total {sum(len(d) for d in seen.values())} of 6, no document processed twice")

    print("\n--- 3) Load leveling: a burst of 10 arrives, one worker drains it at a steady pace ---")
    with client.get_queue_sender(QUEUE) as sender:
        sender.send_messages([ServiceBusMessage(f"burst-{i}") for i in range(1, 11)])  # one call
    print("API accepted all 10 in a single send call; nothing was rejected")
    start = time.perf_counter()
    with client.get_queue_receiver(QUEUE, max_wait_time=3) as worker:
        done = 0
        for msg in worker:
            time.sleep(0.3)                                 # pretend inference takes 0.3 s
            worker.complete_message(msg)
            done += 1
            last_done = time.perf_counter()
            if done % 2 == 0:
                left = len(worker.peek_messages(max_message_count=50))
                print(f"worker finished {done:2}, queue depth now {left}")
    print(f"Worker drained the burst in {last_done - start:.1f} s at its own pace (10 x 0.3 s of work + overhead)")
    print("\nDone - queue is empty again.")

# practice/p02_topic_fanout.py
"""p02 - Unit 3: a topic fans out to subscriptions; a SQL filter routes by priority."""
import json
from azure.servicebus import ServiceBusMessage
from sb_connection import get_client, empty, props, TOPIC

SUBSCRIPTIONS = ["notifications", "high-priority"]        # created by Config.json / azdeploy.py

client = get_client()
with client:
    for sub in SUBSCRIPTIONS:
        empty(client, topic=TOPIC, subscription=sub)

    print("\n--- 1) Publish 5 inference results to the topic (nobody is reading yet) ---")
    with client.get_topic_sender(TOPIC) as sender:
        for i, priority in enumerate(["standard", "high", "standard", "high", "low"], start=1):
            result = {"document_id": f"doc-{i:03}", "status": "completed", "confidence": 0.95}
            sender.send_messages(ServiceBusMessage(
                json.dumps(result),
                content_type="application/json",
                application_properties={"priority": priority},   # what the SQL filter reads
            ))
            print(f"published doc-{i:03}  priority={priority}")

    print("\n--- 2) Each subscription has its OWN copy, waiting even though no consumer was connected ---")
    for sub in SUBSCRIPTIONS:
        with client.get_subscription_receiver(TOPIC, sub) as peeker:
            print(f"{sub:<14} holds {len(peeker.peek_messages(max_message_count=50))} message(s)")

    print("\n--- 3) Each consumer reads its own subscription ---")
    for sub in SUBSCRIPTIONS:
        with client.get_subscription_receiver(TOPIC, sub, max_wait_time=3) as receiver:
            got = []
            for msg in receiver:
                got.append(f"{json.loads(str(msg))['document_id']}({props(msg)['priority']})")
                receiver.complete_message(msg)
        print(f"{sub:<14} received {len(got)}: {got}")

    print("\n--- 4) Completing in one subscription never touches the other ---")
    with client.get_topic_sender(TOPIC) as sender:
        sender.send_messages(ServiceBusMessage("{}", application_properties={"priority": "high"}))
    with client.get_subscription_receiver(TOPIC, "notifications", max_wait_time=3) as r:
        for msg in r:
            r.complete_message(msg)
    with client.get_subscription_receiver(TOPIC, "high-priority") as peeker:
        print("after notifications completed its copy, high-priority still holds",
              len(peeker.peek_messages(max_message_count=50)))
    empty(client, topic=TOPIC, subscription="high-priority")
    print("\nDone - both subscriptions are empty again.")

# practice/p03_message_structure.py
"""p03 - Unit 4: anatomy of an AI message, correlation IDs, claim-check, TTL and batches."""
import json
import os
import time
import uuid
from datetime import timedelta
from azure.servicebus import ServiceBusMessage, ServiceBusSubQueue
from azure.servicebus.exceptions import MessageSizeExceededError
from sb_connection import get_client, empty, props, QUEUE

client = get_client()
with client:
    empty(client)
    empty(client, dead_letter=True)

    print("\n--- 1) Build a well-structured inference request ---")
    request = {"prompt": "Extract the contract parties and effective date.",
               "model": "gpt-4o", "temperature": 0.1, "max_tokens": 2000,
               "document_id": "doc-2025-0142"}
    correlation_id = "req-abc-12345"
    message = ServiceBusMessage(
        body=json.dumps(request),
        content_type="application/json",
        message_id=str(uuid.uuid4()),
        correlation_id=correlation_id,
        application_properties={"model_name": "gpt-4o", "priority": "standard",
                                "document_type": "contract"},
        time_to_live=timedelta(hours=24),
    )
    with client.get_queue_sender(QUEUE) as sender:
        sender.send_messages(message)

    with client.get_queue_receiver(QUEUE, max_wait_time=5) as receiver:
        msg = receiver.receive_messages(max_wait_time=5)[0]
        print("BODY (decoded)        :", json.loads(str(msg)))
        print("content_type          :", msg.content_type)
        print("message_id            :", msg.message_id[:8] + "...  (a UUID)")
        print("correlation_id        :", msg.correlation_id)
        print("application_properties:", props(msg))
        print("  raw, as delivered   :", msg.application_properties)
        print("sequence_number       :", msg.sequence_number, " (set by Service Bus)")
        print("time_to_live          :", msg.time_to_live)
        receiver.complete_message(msg)

    print("\n--- 2) Too big for one message? Claim-check pattern ---")
    document = b"x" * 300_000
    try:
        with client.get_queue_sender(QUEUE) as sender:
            sender.send_messages(ServiceBusMessage(document))
    except MessageSizeExceededError:
        print("sending 300 KB directly -> MessageSizeExceededError (limit here is 256 KB)")
    os.makedirs("blob_store", exist_ok=True)
    path = os.path.join("blob_store", "doc-2025-0142.pdf")
    with open(path, "wb") as f:
        f.write(document)
    claim = {"blob_uri": path, "document_id": "doc-2025-0142", "operation": "extract"}
    with client.get_queue_sender(QUEUE) as sender:
        sender.send_messages(ServiceBusMessage(json.dumps(claim), content_type="application/json",
                                               correlation_id=correlation_id,
                                               application_properties={"pattern": "claim-check"}))
    with client.get_queue_receiver(QUEUE, max_wait_time=5) as receiver:
        msg = receiver.receive_messages(max_wait_time=5)[0]
        ticket = json.loads(str(msg))
        with open(ticket["blob_uri"], "rb") as f:
            data = f.read()
        print(f"message body was {len(str(msg))} bytes; worker fetched {len(data):,} bytes from {ticket['blob_uri']}")
        receiver.complete_message(msg)
        os.remove(path)

    print("\n--- 3) Time-to-live: a stale request expires into the dead-letter queue ---")
    with client.get_queue_sender(QUEUE) as sender:
        sender.send_messages(ServiceBusMessage("realtime classification", time_to_live=timedelta(seconds=2)))
    print("sent with a 2-second TTL, waiting 4 seconds with no worker...")
    time.sleep(4)
    with client.get_queue_receiver(QUEUE, max_wait_time=2) as receiver:
        print("main queue now has:", len(receiver.receive_messages(max_wait_time=2)), "message(s)")
    with client.get_queue_receiver(QUEUE, sub_queue=ServiceBusSubQueue.DEAD_LETTER, max_wait_time=5) as dlq:
        for msg in dlq.receive_messages(max_wait_time=5):
            print("dead-letter queue has it, reason:", msg.dead_letter_reason)
            dlq.complete_message(msg)

    print("\n--- 4) Batches: many requests, few network round trips ---")
    pages = [{"document_id": "doc-2025-0142", "page": p, "text": "lorem " * 1500} for p in range(1, 41)]
    batches = 0
    with client.get_queue_sender(QUEUE) as sender:
        batch = sender.create_message_batch()
        for page in pages:
            message = ServiceBusMessage(json.dumps(page), content_type="application/json")
            try:
                batch.add_message(message)
            except MessageSizeExceededError:
                sender.send_messages(batch)
                batches += 1
                batch = sender.create_message_batch()
                batch.add_message(message)
        if len(batch) > 0:
            sender.send_messages(batch)
            batches += 1
    print(f"40 page requests (~9 KB each) sent in {batches} batch call(s) instead of 40 calls")
    print("messages now in the queue:", empty(client), "(removed again to tidy up)")

# practice/p04_reliable_processing.py
"""p04 - Unit 5: peek-lock settlement, poison messages, the dead-letter queue, reprocessing."""
import json
from azure.servicebus import ServiceBusMessage, ServiceBusReceiveMode, ServiceBusSubQueue
from sb_connection import get_client, empty, QUEUE

SUPPORTED_MODELS = {"gpt-4o", "gpt-4o-mini"}

client = get_client()
with client:
    empty(client)
    empty(client, dead_letter=True)

    print("\n--- 1) Five requests arrive, each will take a different path ---")
    requests = [
        ("req-1", json.dumps({"model": "gpt-4o", "prompt": "Summarize"})),
        ("req-2", "not valid json: [broken"),
        ("req-3", json.dumps({"model": "gpt-9", "prompt": "Summarize"})),
        ("req-4", json.dumps({"model": "gpt-4o", "prompt": "Summarize", "needs": "contract-template"})),
        ("req-5", json.dumps({"model": "gpt-4o-mini", "prompt": "flaky"})),
    ]
    with client.get_queue_sender(QUEUE) as sender:
        for cid, body in requests:
            sender.send_messages(ServiceBusMessage(body, correlation_id=cid))
    print("queued:", [cid for cid, _ in requests])

    print("\n--- 2) A peek-lock worker settles every message (complete/abandon/dead-letter/defer) ---")
    deferred = []
    with client.get_queue_receiver(QUEUE, receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
                                   max_wait_time=3) as receiver:
        for msg in receiver:
            cid = msg.correlation_id
            try:
                job = json.loads(str(msg))
            except json.JSONDecodeError:
                receiver.dead_letter_message(msg, reason="MalformedPayload",
                                             error_description="Message body is not valid JSON")
                print(f"{cid}: DEAD-LETTER (bad JSON - retrying can never fix it)")
                continue
            if job["model"] not in SUPPORTED_MODELS:
                receiver.dead_letter_message(msg, reason="UnsupportedModel",
                                             error_description=f"Model {job['model']} is not available")
                print(f"{cid}: DEAD-LETTER (unknown model {job['model']})")
            elif job.get("needs") == "contract-template":
                deferred.append(msg.sequence_number)
                receiver.defer_message(msg)
                print(f"{cid}: DEFER (waiting for a dependency), sequence_number={msg.sequence_number}")
            elif job["prompt"] == "flaky":
                receiver.abandon_message(msg)
                print(f"{cid}: ABANDON (timeout, try again)  delivery_count={msg.delivery_count}")
            else:
                receiver.complete_message(msg)
                print(f"{cid}: COMPLETE")

        print("\n--- 3) Later: the dependency is ready, fetch the deferred message by number ---")
        for msg in receiver.receive_deferred_messages(deferred):
            receiver.complete_message(msg)
            print(f"{msg.correlation_id}: deferred message fetched and COMPLETED")

    print("\n--- 4) Inspect the dead-letter queue ---")
    with client.get_queue_receiver(QUEUE, sub_queue=ServiceBusSubQueue.DEAD_LETTER,
                                   max_wait_time=3) as dlq:
        broken = []
        for msg in dlq:
            print(f"{msg.correlation_id}: reason={msg.dead_letter_reason}")
            print(f"        description={msg.dead_letter_error_description}")
            print(f"        delivery_count={msg.delivery_count}")
            broken.append(msg)

        print("\n--- 5) Fix the root cause (gpt-9 is now deployed) and resubmit only that one ---")
        SUPPORTED_MODELS.add("gpt-9")
        with client.get_queue_sender(QUEUE) as sender:
            for msg in broken:
                if msg.dead_letter_reason == "UnsupportedModel":
                    sender.send_messages(ServiceBusMessage(str(msg), correlation_id=msg.correlation_id))
                    print(f"{msg.correlation_id}: resubmitted to {QUEUE}")
                dlq.complete_message(msg)
    with client.get_queue_receiver(QUEUE, max_wait_time=3) as receiver:
        for msg in receiver:
            try:
                json.loads(str(msg))
            except json.JSONDecodeError:
                receiver.dead_letter_message(msg, reason="MalformedPayload",
                                             error_description="Still not valid JSON")
                print(f"{msg.correlation_id}: failed again -> back in the dead-letter queue")
                continue
            receiver.complete_message(msg)
            print(f"{msg.correlation_id}: processed successfully on the second attempt")

    print("\n--- 6) Receive-and-delete: fast, but the message is gone the moment you get it ---")
    with client.get_queue_sender(QUEUE) as sender:
        sender.send_messages(ServiceBusMessage("telemetry ping"))
    with client.get_queue_receiver(QUEUE, receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE,
                                   max_wait_time=3) as receiver:
        got = receiver.receive_messages(max_wait_time=3)
        print("received:", str(got[0]), "| crash now and it is lost: nothing left to retry")
    print("messages left in the queue:", empty(client))

# practice/p05_crash_test.py  (takes about 2.5 minutes: it waits for the lock)
"""p05 - extra: a worker process REALLY crashes mid-task (os._exit). Peek-lock vs receive-and-delete."""
import os, subprocess, sys, time
from azure.servicebus import ServiceBusMessage, ServiceBusReceiveMode
from sb_connection import get_client, empty, QUEUE

if len(sys.argv) > 1:                                   # child: the worker that crashes
    mode = (ServiceBusReceiveMode.PEEK_LOCK if sys.argv[1] == "peek"
            else ServiceBusReceiveMode.RECEIVE_AND_DELETE)
    c = get_client()
    r = c.get_queue_receiver(QUEUE, receive_mode=mode, max_wait_time=5)
    r.__enter__()
    m = r.receive_messages(max_wait_time=5)[0]
    print(f"  worker A got '{m}' ... running inference ... CRASH", flush=True)
    os._exit(1)                                         # no complete, no close, process gone

client = get_client()
with client:
    for mode in ("peek", "delete"):
        empty(client)
        label = "PEEK-LOCK" if mode == "peek" else "RECEIVE-AND-DELETE"
        print(f"\n--- {label} ---", flush=True)
        with client.get_queue_sender(QUEUE) as s:
            s.send_messages(ServiceBusMessage("req-42"))
        t0 = time.perf_counter()
        subprocess.run([sys.executable, __file__, mode])
        with client.get_queue_receiver(QUEUE, max_wait_time=90) as worker_b:
            got = worker_b.receive_messages(max_wait_time=90)
            if got:
                print(f"  worker B got '{got[0]}' {time.perf_counter() - t0:.1f} s after it was sent "
                      f"(delivery_count={got[0].delivery_count})")
                worker_b.complete_message(got[0])
            else:
                print(f"  worker B waited 90 s: nothing. req-42 is gone for good.")

Moving to Azure: create a Standard namespace (topics need Standard or higher), the same queue, topic, subscriptions and filter, and give your account the Azure Service Bus Data Owner role. Then export SERVICE_BUS_FQDN=<name>.servicebus.windows.net and run the same scripts. Delete the resource group when you're done; the namespace is billed while it exists.

The whole guide in 8 lines

1. Your API puts work on a queue and answers at once; the model is never in the request path.
2. The queue soaks up bursts, and queue depth tells you when to add workers.
3. Many workers on one queue share the work, and each message goes to exactly one.
4. A topic gives every subscription its own copy, and filters route by property.
5. JSON body, properties for routing, a unique message_id, one correlation_id per request.
6. Too big for 256 KB? Store the file in Blob Storage and send its address.
7. Peek-lock means a crash loses nothing (my request came back after 60.7 s); make the work safe to repeat.
8. Max delivery count parks poison messages in the dead-letter queue; fix the cause, then resubmit.

---

Checked against azure-servicebus 7.14.1 (Python) and the Azure Service Bus emulator (latest image), as of October 2026, following Microsoft Learn AI-200 Learning Path 7, Module 1. Every output block is real output from my laptop on 19 October 2026; sequence numbers, IDs and timings will differ on yours. The crash in section 8 is a real process killed with os._exit(1). The animations replay those runs with time compressed; the burst comparison in section 1 is illustrative. The emulator starts delivery_count at 0 on the first delivery; Azure may show 1. The Azure steps in section 11 were not run for this post. Limits (256 KB Standard, 100 MB Premium, 1-minute default lock, max delivery count 10) are from the Service Bus docs.