Service Bus Interview Questions and Answers Part 4 - Peek-Lock, Settlement and Redelivery

Service Bus Interview Q&A · Part 4 of 10

In this post we are going to cover peek-lock, settlement and redelivery in Azure Service Bus. These are 7 real Service Bus interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: a payments consumer writes to a database and must never charge a customer twice, and you design receive, write and settle.

In peek-lock mode, a consumer receives a locked message, does the work, and then settles it. If that settle doesn't happen in time, the message comes back. Interviewers ask about this because it's where duplicates come from: a slow consumer, a crash or a lost connection all end in a second delivery. If you can explain why, and how to make that second delivery harmless, you show that you've run consumers in production.

Part 4 of the Real Time Azure Service Bus Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's set up the project
3. Question 1: The same order was processed twice. Why does Service Bus deliver a message more than once?
4. Question 2: A consumer crashed while holding a message. What happens to the lock?
5. Question 3: Some messages take longer to process than the lock duration. How do you renew the lock?
6. Question 4: A teammate asks the difference between complete, abandon, dead-letter and defer. When do you use each?
7. Question 5: Someone asks what the delivery count tells you. How does it change on each attempt?
8. Question 6: Your manager asks how to make duplicate deliveries harmless. How do you design idempotent consumers?
9. Scenario 7: A payments consumer writes to a database and must never charge a customer twice. How do you design receive, write and settle?
10. The full script

The short answer

Say this first

In peek-lock mode Service Bus delivers at least once: a message comes back when its lock expires, the consumer crashes or it is abandoned; expiry and abandon raise the delivery count. Renew locks for slow work, pick the right settlement, and make consumers idempotent with a business key: write, commit, then complete.

Let's set up the project

We run Microsoft's official Service Bus emulator locally, with SQL Server next to it, and small Python scripts that use the azure-servicebus client, so every output is real. Features the emulator doesn't support are shown as code. The emulator reads its queues from Config.json. Most of them have a 5 second lock, so a lock expires quickly in the demo, and the poison queue has a max delivery count of 3. The consumers in questions 6 and 7 write to small SQLite databases.

Config.json

{
  "UserConfig": {
    "Namespaces": [
      {
        "Name": "sbemulatorns",
        "Queues": [
          {
            "Name": "orders",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "shipments",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "reports",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "settle",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT1M",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "poison",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT1M",
              "MaxDeliveryCount": 3,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "invoices",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "payments",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          }
        ],
        "Topics": []
      }
    ],
    "Logging": {
      "Type": "File"
    }
  }
}

Question 1: The same order was processed twice. Why does Service Bus deliver a message more than once?

Because in peek-lock mode, Service Bus delivers at least once. A message is removed only when the consumer completes it. If the complete doesn't arrive while the lock is held, the message is handed out again, and the work runs twice.

Let's try it. The orders queue has a 5 second lock. dup.py receives order 1001, works for 7 seconds, then tries to complete it, and receives again. We run python dup.py.

dup.py

import os, time
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.exceptions import MessageLockLostError

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("orders") as sender:
        sender.send_messages(ServiceBusMessage("order 1001"))
    with client.get_queue_receiver("orders") as receiver:  # lock: 5 s
        for attempt in [1, 2]:
            msg = receiver.receive_messages(max_wait_time=10)[0]
            print("processing", msg, "delivery_count",
                  msg.delivery_count)
            if attempt == 1:
                time.sleep(7)                    # slow work
            try:
                receiver.complete_message(msg)
                print("completed")
            except MessageLockLostError as e:
                print("complete failed:", type(e).__name__)
                print(" ", str(e).split(" TrackingId")[0])

Terminal

$ python dup.py
processing order 1001 delivery_count 0
complete failed: MessageLockLostError
  The lock on the message lock has expired.
processing order 1001 delivery_count 1
completed

The first try shows delivery_count 0. After 7 seconds of work, the complete fails with MessageLockLostError: The lock on the message lock has expired. So the next receive returns the same order, with delivery_count 1, and it's processed a second time.

The lock expiring is one cause. Here are all four:

• Lock expired - work slower than the lock
• Consumer crashed - before complete
• Connection lost - locks are lost with it
• Publisher retried - a second send

Microsoft says it plainly: design consumers to be idempotent.

Real project

I first compare the lock duration with the real processing time. A lock shorter than the slowest message is the most common source of duplicates I see.

Question 2: A consumer crashed while holding a message. What happens to the lock?

Nothing is lost. The lock simply runs out. Until then, the message stays invisible to other receivers, and when the lock duration passes, it goes back to the front of the queue for the next receive.

Let's crash a worker. crash.py sends two shipments, and worker A locks shipment 1 and dies, with no complete and no clean close. Then worker B starts at once. We run python crash.py and python recover.py.

crash.py

import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("shipments") as sender:
        sender.send_messages([ServiceBusMessage(f"shipment {n}",
                              message_id=f"s-{n}") for n in [1, 2]])
    with client.get_queue_receiver("shipments") as receiver:
        msg = receiver.receive_messages(max_wait_time=5)[0]
        print("worker A locked", msg.message_id, "and crashes")
        os._exit(1)                   # no complete, no clean close

recover.py

import os
from azure.servicebus import ServiceBusClient

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("shipments") as receiver:
    peeked = receiver.peek_messages(10, sequence_number=1)
    print("peek:", [(m.message_id, m.delivery_count) for m in peeked])
    for check in range(4):                   # about 12 s
        for msg in receiver.receive_messages(max_wait_time=3):
            print("worker B got", msg.message_id,
                  "delivery_count", msg.delivery_count)
            receiver.complete_message(msg)

Terminal

$ python crash.py
worker A locked s-1 and crashes
$ python recover.py
peek: [('s-1', 0), ('s-2', 0)]
worker B got s-2 delivery_count 0
worker B got s-1 delivery_count 1

worker A locked s-1 and crashes. The peek still shows both shipments, [('s-1', 0), ('s-2', 0)], because peek takes no lock. Worker B gets s-2 straight away. Shipment 1 stays locked by the dead worker until the 5 second lock expires, and then it comes back: worker B got s-1 delivery_count 1.

On Azure, these are the lock settings.

Lock duration on Azure	Value
default	1 minute
maximum	5 minutes
set it	above normal processing time

Set it above your normal processing time, but not far above it: after a crash, that's how long the message waits.

And locks are volatile:

• Lost connection - the lock goes with it
• Receiver closed before settling - lock expires, redelivered

The rule

The consumer must always expect a second copy.

Question 3: Some messages take longer to process than the lock duration. How do you renew the lock?

Renew it before it runs out. Each renewal gives us the full lock duration again, and the message stays ours. A lock that has already expired can't be renewed.

Let's try both ways on a 5 second lock. Report 1 takes 9 seconds, and the script calls renew_message_lock every 3 seconds. Report 2 uses the AutoLockRenewer, passed to the receiver, through 12 seconds of work. We run python slow.py.

slow.py

import os, time
from azure.servicebus import (AutoLockRenewer, ServiceBusClient,
                              ServiceBusMessage)

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("reports") as sender:
        sender.send_messages([ServiceBusMessage(f"report {n}")
                              for n in [1, 2]])
    with client.get_queue_receiver("reports") as receiver: # lock 5 s
        msg = receiver.receive_messages(max_wait_time=5)[0]
        for chunk in [1, 2, 3]:                 # 9 s of work
            time.sleep(3)
            receiver.renew_message_lock(msg)    # 5 s again
        receiver.complete_message(msg)
        print(msg, "| 9 s, 3 manual renewals | completed,",
              "delivery_count", msg.delivery_count)
    renewer = AutoLockRenewer(max_lock_renewal_duration=60)
    with client.get_queue_receiver(
            "reports", auto_lock_renewer=renewer) as receiver:
        msg = receiver.receive_messages(max_wait_time=5)[0]
        time.sleep(12)                          # 12 s of work
        receiver.complete_message(msg)
        renewer.close()                         # before the receiver
        print(msg, "| 12 s, AutoLockRenewer | completed,",
              "delivery_count", msg.delivery_count)

Terminal

$ python slow.py
report 1 | 9 s, 3 manual renewals | completed, delivery_count 0
report 2 | 12 s, AutoLockRenewer | completed, delivery_count 0

report 1 | 9 s, 3 manual renewals | completed, delivery_count 0, and report 2 | 12 s, AutoLockRenewer | completed, delivery_count 0. No redelivery for either. The AutoLockRenewer renews every message the receiver hands out, in the background.

AutoLockRenewer	Value
max_lock_renewal_duration	default 300 s
register	each message, or pass it to the receiver

Its default limit is five minutes, and here we set 60 seconds. In Azure Functions, a trigger that takes one message at a time renews for you, and the host setting maxAutoLockRenewalDuration defaults to five minutes.

host.json (Azure Functions, not run)

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "maxAutoLockRenewalDuration": "00:05:00"
    }
  }
}

My rule

Renewal is a safety net, not a design. Keep the lock close to normal processing time, renew in the background for the slow ones, and split really long jobs into smaller messages.

Question 4: A teammate asks the difference between complete, abandon, dead-letter and defer. When do you use each?

These are the four ways to settle a locked message. Complete means done. Abandon releases the lock now, for another try. Dead-letter moves the message to the dead-letter queue with a reason. Defer sets it aside, and we get it back only by sequence number.

Settlement	What happens	Use it when
complete	removed from the queue	processing worked
abandon	lock released now, delivered again at once	a brief glitch: no back-off
dead-letter	moved to the dead-letter queue, with a reason	a retry can never help
defer	stays, set aside: receive it by sequence number	it arrived before its data

Let's settle four payments, one each way. settle.py completes pay-1, abandons pay-2, dead-letters pay-3 with a reason and defers pay-4, saving its sequence number to a file. Then later.py receives normally, receives the deferred one and reads the dead-letter queue. We run python settle.py and python later.py.

settle.py

import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
ids = ["pay-1", "pay-2", "pay-3", "pay-4"]
with client, client.get_queue_sender("settle") as sender, \
        client.get_queue_receiver("settle", max_wait_time=2) as r:
    sender.send_messages([ServiceBusMessage(i, message_id=i)
                          for i in ids])
    got = {m.message_id: m for m in r.receive_messages(10, 2)}
    r.complete_message(got["pay-1"])                   # done
    r.abandon_message(got["pay-2"])                    # try again
    r.dead_letter_message(got["pay-3"], reason="BadCard",
                          error_description="card number is invalid")
    r.defer_message(got["pay-4"])                      # not yet
    seq = got["pay-4"].sequence_number                 # keep it
    open("deferred.txt", "w").write(str(seq))
    for m in r.peek_messages(10, sequence_number=1):
        print("peek:", m.message_id, m.state.name)

later.py

import os
from azure.servicebus import ServiceBusClient, ServiceBusSubQueue

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
seq = int(open("deferred.txt").read())
with client, client.get_queue_receiver(
        "settle", max_wait_time=2) as r:
    for m in r.receive_messages(10, 2):              # normal receive
        print("receive:", m.message_id, "delivery_count",
              m.delivery_count)
        r.complete_message(m)
    m = r.receive_deferred_messages([seq])[0]        # by number
    print(f"receive_deferred_messages([{seq}]):", m.message_id)
    r.complete_message(m)
    with client.get_queue_receiver("settle", max_wait_time=2,
            sub_queue=ServiceBusSubQueue.DEAD_LETTER) as dlq:
        for m in dlq.receive_messages(10, 2):
            print("dead-letter queue:", m.message_id,
                  m.dead_letter_reason, "|",
                  m.dead_letter_error_description)
            dlq.complete_message(m)

Terminal

$ python settle.py
peek: pay-2 ACTIVE
peek: pay-4 DEFERRED
$ python later.py
receive: pay-2 delivery_count 1
receive_deferred_messages([4]): pay-4
dead-letter queue: pay-3 BadCard | card number is invalid

The peek shows pay-2 ACTIVE and pay-4 DEFERRED. pay-1 is gone, and pay-3 has left the main queue.

A normal receive now returns only pay-2, with delivery_count 1, because an abandon counts. pay-4 comes back only through receive_deferred_messages([4]), with the saved sequence number. The dead-letter queue holds pay-3, with our reason and description: BadCard | card number is invalid.

The pitfall with defer

Losing the sequence number. Store it durably: a normal receive never returns a deferred message.

Question 5: Someone asks what the delivery count tells you. How does it change on each attempt?

The delivery count says how many times the message was handed out without being completed. It goes up when a lock expires or the consumer abandons it. In this Python client it reads zero on the first try.

Max delivery count	Value
Azure default	10
this demo queue	3
turn it off	not possible: set it higher

Let's try it. The poison queue has a max delivery count of 3, and poison.py is a consumer that always fails and abandons. We run python poison.py.

poison.py

import os
from azure.servicebus import (ServiceBusClient, ServiceBusMessage,
                              ServiceBusSubQueue)

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("poison") as sender:   # max 3
        sender.send_messages(ServiceBusMessage("order 2001"))
    with client.get_queue_receiver("poison", max_wait_time=2) as r:
        while msgs := r.receive_messages(max_wait_time=2):
            print("delivery_count", msgs[0].delivery_count,
                  "-> fails, abandon")
            r.abandon_message(msgs[0])
    print("main queue: empty")
    with client.get_queue_receiver("poison", max_wait_time=2,
            sub_queue=ServiceBusSubQueue.DEAD_LETTER) as dlq:
        msg = dlq.receive_messages(max_wait_time=2)[0]
        print("dead-letter queue:", msg, "| delivery_count",
              msg.delivery_count)
        print("reason:", msg.dead_letter_reason)
        print("description:", msg.dead_letter_error_description)
        dlq.complete_message(msg)

Terminal

$ python poison.py
delivery_count 0 -> fails, abandon
delivery_count 1 -> fails, abandon
delivery_count 2 -> fails, abandon
main queue: empty
dead-letter queue: order 2001 | delivery_count 3
reason: MaxDeliveryCountExceeded
description: Message could not be consumed after 3 delivery attempts.

The order is abandoned three times: delivery_count 0, 1 and 2. After the third failed delivery, Service Bus moves it to the dead-letter queue by itself. The copy there shows delivery_count 3, reason MaxDeliveryCountExceeded, and the description Message could not be consumed after 3 delivery attempts.

On Azure, we can't turn this off, only set it higher. Read the count in your consumer too: a message on its fifth try deserves a log line.

Real project

I pick the max delivery count from how many retries a transient error really needs. Too high, and a poison message keeps a consumer busy for nothing.

Question 6: Your manager asks how to make duplicate deliveries harmless. How do you design idempotent consumers?

Processing a message twice must leave the same result as once. The standard way: a business key, like the order ID, under a unique constraint, in the same database transaction as the work.

• Receive - peek-lock
• Insert the key - order_id: unique
• Do the work - only if the row is new
• Complete - after the commit

Let's build one. idem.py inserts the invoice row keyed by order ID, with ON CONFLICT DO NOTHING, so a repeat writes nothing. Order 7 is slow, so its lock expires. Order 8 is sent twice, as two different messages. We run python idem.py.

idem.py

import os, sqlite3, time
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.exceptions import MessageLockLostError

db = sqlite3.connect("shop.db")
db.execute("CREATE TABLE IF NOT EXISTS invoices"
           " (order_id TEXT PRIMARY KEY)")
SQL = "INSERT INTO invoices VALUES (?) ON CONFLICT DO NOTHING"
client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("invoices") as sender, \
        client.get_queue_receiver("invoices", max_wait_time=8) as r:
    sender.send_messages([ServiceBusMessage(o) for o in
                          ["order-7", "order-8", "order-8"]]) # resent
    while msgs := r.receive_messages(max_wait_time=8):
        msg, order_id = msgs[0], str(msgs[0])        # business key
        with db:                                     # one transaction
            new = db.execute(SQL, (order_id,)).rowcount
        print(order_id, "delivery_count", msg.delivery_count, "->",
              "invoice written" if new else "duplicate: skipped")
        if order_id == "order-7" and msg.delivery_count == 0:
            time.sleep(7)                    # too slow: lock lost
        try:
            r.complete_message(msg)
        except MessageLockLostError:
            print("  complete failed: lock lost")
rows = db.execute("SELECT COUNT(*) FROM invoices").fetchone()[0]
print("invoices in the database:", rows)

Terminal

$ python idem.py
order-7 delivery_count 0 -> invoice written
  complete failed: lock lost
order-7 delivery_count 1 -> duplicate: skipped
order-8 delivery_count 0 -> invoice written
order-8 delivery_count 0 -> duplicate: skipped
invoices in the database: 2

order-7 is written, then complete failed: lock lost. It comes back with delivery_count 1, and the insert skips it: duplicate: skipped. The second order-8 is skipped too. invoices in the database: 2.

Duplicate detection doesn't replace this. It only drops a resend with the same message ID inside its time window.

Duplicate	Detection	Business key
resend, same MessageId, in window	dropped	skipped
resend, new MessageId	delivered	skipped
redelivery: lock lost, crash	delivered	skipped

A redelivery is the same message again, so only the consumer can catch it.

Take the key from the business

Not from Service Bus. The Python library doesn't set a message ID for you, so a resend without one looks like a brand new message.

Scenario 7: A payments consumer writes to a database and must never charge a customer twice. How do you design receive, write and settle?

Three steps, in this order:

• Receive - peek-lock
• Write and commit - one database transaction, keyed by payment_id
• Complete - only after the commit

If anything fails in between, the message comes back, and the key makes the second write a no-op. Complete before the commit, and a crash loses the payment.

Let's crash it. send_payments.py queues three payments. With CRASH_AFTER=pay-2, pay.py exits right after the commit for pay-2, before its complete. Then we restart it. We run python send_payments.py, CRASH_AFTER=pay-2 python pay.py and python pay.py.

send_payments.py

import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage

PAYMENTS = ["pay-1 cust-a 2500", "pay-2 cust-b 4000",
            "pay-3 cust-c 1500"]                # id, customer, cents
client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("payments") as sender:
    sender.send_messages([ServiceBusMessage(
        p, message_id=p.split()[0]) for p in PAYMENTS])
print("queued", len(PAYMENTS), "payments")

pay.py

import os, sqlite3
from azure.servicebus import ServiceBusClient

CRASH_AFTER = os.environ.get("CRASH_AFTER")       # simulate a crash
db = sqlite3.connect("payments.db")
db.execute("CREATE TABLE IF NOT EXISTS charges"
           " (payment_id TEXT PRIMARY KEY, customer TEXT, cents INT)")
SQL = "INSERT INTO charges VALUES (?, ?, ?) ON CONFLICT DO NOTHING"
client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("payments",
                                       max_wait_time=8) as receiver:
    while msgs := receiver.receive_messages(max_wait_time=8):
        msg = msgs[0]                                  # 1. receive
        payment_id, customer, cents = str(msg).split()
        with db:                                       # 2. write
            row = (payment_id, customer, int(cents))
            new = db.execute(SQL, row).rowcount        # keyed, once
        print(payment_id, "delivery_count", msg.delivery_count, "->",
              "charged" if new else "already charged: skip")
        if payment_id == CRASH_AFTER:
            print("crash after the commit, before complete")
            os._exit(1)
        receiver.complete_message(msg)                 # 3. settle
rows = db.execute("SELECT COUNT(*) FROM charges").fetchone()[0]
print("charges in the database:", rows)

Terminal

$ python send_payments.py
queued 3 payments
$ CRASH_AFTER=pay-2 python pay.py
pay-1 delivery_count 0 -> charged
pay-2 delivery_count 0 -> charged
crash after the commit, before complete
$ python pay.py
pay-3 delivery_count 0 -> charged
pay-2 delivery_count 1 -> already charged: skip
charges in the database: 3

The first consumer charges pay-1 and pay-2, then prints crash after the commit, before complete. The restarted consumer charges pay-3. When the dead consumer's lock expires, pay-2 comes back with delivery_count 1, and the insert skips it: already charged: skip. charges in the database: 3, and nobody is charged twice.

Why not one transaction across Service Bus and the database? Service Bus transactions don't enlist a database, and the Python library has no transaction API at all. The key in the database is what makes this safe.

One more step

If an external payment provider supports an idempotency key, pass it the same payment ID, so a retry there is safe too.

Key points to remember

• Peek-lock = at least once: lock expiry, crashes and lost connections redeliver
• A crashed consumer's lock runs out; then the message comes back
• Renew locks for slow work: renew_message_lock or AutoLockRenewer
• Complete, abandon, dead-letter, defer; past the max delivery count: DLQ
• Idempotent consumers: business key, write and commit, then complete

The full script

This is a Python script that writes every file and runs every command above in order against the Service Bus emulator: start the emulator with the Config.json above, run pip install azure-servicebus first, and set SERVICEBUS_CONNECTION_STRING.

Show the full script that writes the files and runs every step

Full script - writes the files and runs every step against the Service Bus emulator (pip install azure-servicebus), run it with Python

# ========================================================================
# Service Bus Interview Q&A #4 - Peek-lock, settlement and redelivery
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Azure Service Bus emulator (Docker image, version 2.0.1) with
# SQL Server 2022 and azure-servicebus 7.15.0; the outputs in the video and the blog post are the real outputs.
# To run it yourself:
#   1. Start the emulator with Microsoft's installer (github.com/Azure/azure-service-bus-emulator-installer),
#      using the Config.json this script writes first (Step 0) as the emulator's config file; it holds every
#      queue, topic, subscription, rule and setting (lock duration, max delivery count, duplicate detection...)
#      the steps depend on.
#   2. pip install azure-servicebus
#   3. Set SERVICEBUS_CONNECTION_STRING to the emulator connection string (Endpoint=sb://localhost;
#      SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;)
#      then run python ServiceBus_Interview_QA_04_Code.py.
# Run it in an empty folder and on a freshly started emulator: it creates its files (and any SQLite
# databases) there, and a second run on old files or old messages gives different output.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================

import os, subprocess, time

def write(path, text):
    if text is None:
        if os.path.exists(path): os.remove(path)
        return
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(text.strip("\n") + "\n")

def run(cmd, expect_error=False, allow_error=False, env=None):
    print("\n$ " + cmd, flush=True)
    rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
    if rc != 0 and not (expect_error or allow_error):
        raise SystemExit("stopped: this command was expected to succeed")

# Step 0: the emulator config (mount it as ServiceBus_Emulator/ConfigFiles/Config.json, then start the emulator)
write('Config.json', """\
{
  "UserConfig": {
    "Namespaces": [
      {
        "Name": "sbemulatorns",
        "Queues": [
          {
            "Name": "orders",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "shipments",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "reports",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "settle",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT1M",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "poison",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT1M",
              "MaxDeliveryCount": 3,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "invoices",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          },
          {
            "Name": "payments",
            "Properties": {
              "DeadLetteringOnMessageExpiration": false,
              "DefaultMessageTimeToLive": "PT1H",
              "DuplicateDetectionHistoryTimeWindow": "PT20S",
              "ForwardDeadLetteredMessagesTo": "",
              "ForwardTo": "",
              "LockDuration": "PT5S",
              "MaxDeliveryCount": 10,
              "RequiresDuplicateDetection": false,
              "RequiresSession": false
            }
          }
        ],
        "Topics": []
      }
    ],
    "Logging": {
      "Type": "File"
    }
  }
}
""")

# Step 1: A 5 s lock, 7 s of work: the complete fails and the order comes again
write('dup.py', """\
import os, time
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.exceptions import MessageLockLostError

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("orders") as sender:
        sender.send_messages(ServiceBusMessage("order 1001"))
    with client.get_queue_receiver("orders") as receiver:  # lock: 5 s
        for attempt in [1, 2]:
            msg = receiver.receive_messages(max_wait_time=10)[0]
            print("processing", msg, "delivery_count",
                  msg.delivery_count)
            if attempt == 1:
                time.sleep(7)                    # slow work
            try:
                receiver.complete_message(msg)
                print("completed")
            except MessageLockLostError as e:
                print("complete failed:", type(e).__name__)
                print(" ", str(e).split(" TrackingId")[0])
""")
run('python dup.py')

# Step 2: Worker A crashes holding a lock; worker B gets the message after the lock expires
write('crash.py', """\
import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("shipments") as sender:
        sender.send_messages([ServiceBusMessage(f"shipment {n}",
                              message_id=f"s-{n}") for n in [1, 2]])
    with client.get_queue_receiver("shipments") as receiver:
        msg = receiver.receive_messages(max_wait_time=5)[0]
        print("worker A locked", msg.message_id, "and crashes")
        os._exit(1)                   # no complete, no clean close
""")
write('recover.py', """\
import os
from azure.servicebus import ServiceBusClient

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("shipments") as receiver:
    peeked = receiver.peek_messages(10, sequence_number=1)
    print("peek:", [(m.message_id, m.delivery_count) for m in peeked])
    for check in range(4):                   # about 12 s
        for msg in receiver.receive_messages(max_wait_time=3):
            print("worker B got", msg.message_id,
                  "delivery_count", msg.delivery_count)
            receiver.complete_message(msg)
""")
run('python crash.py', allow_error=True)
run('python recover.py', allow_error=True)

# Step 3: Manual renew_message_lock and AutoLockRenewer on a 5 s lock
write('slow.py', """\
import os, time
from azure.servicebus import (AutoLockRenewer, ServiceBusClient,
                              ServiceBusMessage)

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("reports") as sender:
        sender.send_messages([ServiceBusMessage(f"report {n}")
                              for n in [1, 2]])
    with client.get_queue_receiver("reports") as receiver: # lock 5 s
        msg = receiver.receive_messages(max_wait_time=5)[0]
        for chunk in [1, 2, 3]:                 # 9 s of work
            time.sleep(3)
            receiver.renew_message_lock(msg)    # 5 s again
        receiver.complete_message(msg)
        print(msg, "| 9 s, 3 manual renewals | completed,",
              "delivery_count", msg.delivery_count)
    renewer = AutoLockRenewer(max_lock_renewal_duration=60)
    with client.get_queue_receiver(
            "reports", auto_lock_renewer=renewer) as receiver:
        msg = receiver.receive_messages(max_wait_time=5)[0]
        time.sleep(12)                          # 12 s of work
        receiver.complete_message(msg)
        renewer.close()                         # before the receiver
        print(msg, "| 12 s, AutoLockRenewer | completed,",
              "delivery_count", msg.delivery_count)
""")
run('python slow.py')

# Step 4: complete, abandon, dead_letter and defer side by side
write('settle.py', 'import os\nfrom azure.servicebus import ServiceBusClient, ServiceBusMessage\n\nclient = ServiceBusClient.from_connection_string(\n    os.environ["SERVICEBUS_CONNECTION_STRING"])\nids = ["pay-1", "pay-2", "pay-3", "pay-4"]\nwith client, client.get_queue_sender("settle") as sender, \\\n        client.get_queue_receiver("settle", max_wait_time=2) as r:\n    sender.send_messages([ServiceBusMessage(i, message_id=i)\n                          for i in ids])\n    got = {m.message_id: m for m in r.receive_messages(10, 2)}\n    r.complete_message(got["pay-1"])                   # done\n    r.abandon_message(got["pay-2"])                    # try again\n    r.dead_letter_message(got["pay-3"], reason="BadCard",\n                          error_description="card number is invalid")\n    r.defer_message(got["pay-4"])                      # not yet\n    seq = got["pay-4"].sequence_number                 # keep it\n    open("deferred.txt", "w").write(str(seq))\n    for m in r.peek_messages(10, sequence_number=1):\n        print("peek:", m.message_id, m.state.name)\n')
write('later.py', """\
import os
from azure.servicebus import ServiceBusClient, ServiceBusSubQueue

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
seq = int(open("deferred.txt").read())
with client, client.get_queue_receiver(
        "settle", max_wait_time=2) as r:
    for m in r.receive_messages(10, 2):              # normal receive
        print("receive:", m.message_id, "delivery_count",
              m.delivery_count)
        r.complete_message(m)
    m = r.receive_deferred_messages([seq])[0]        # by number
    print(f"receive_deferred_messages([{seq}]):", m.message_id)
    r.complete_message(m)
    with client.get_queue_receiver("settle", max_wait_time=2,
            sub_queue=ServiceBusSubQueue.DEAD_LETTER) as dlq:
        for m in dlq.receive_messages(10, 2):
            print("dead-letter queue:", m.message_id,
                  m.dead_letter_reason, "|",
                  m.dead_letter_error_description)
            dlq.complete_message(m)
""")
run('python settle.py')
run('python later.py')

# Step 5: Delivery count grows on each abandon until MaxDeliveryCount moves it to the DLQ
write('poison.py', """\
import os
from azure.servicebus import (ServiceBusClient, ServiceBusMessage,
                              ServiceBusSubQueue)

client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client:
    with client.get_queue_sender("poison") as sender:   # max 3
        sender.send_messages(ServiceBusMessage("order 2001"))
    with client.get_queue_receiver("poison", max_wait_time=2) as r:
        while msgs := r.receive_messages(max_wait_time=2):
            print("delivery_count", msgs[0].delivery_count,
                  "-> fails, abandon")
            r.abandon_message(msgs[0])
    print("main queue: empty")
    with client.get_queue_receiver("poison", max_wait_time=2,
            sub_queue=ServiceBusSubQueue.DEAD_LETTER) as dlq:
        msg = dlq.receive_messages(max_wait_time=2)[0]
        print("dead-letter queue:", msg, "| delivery_count",
              msg.delivery_count)
        print("reason:", msg.dead_letter_reason)
        print("description:", msg.dead_letter_error_description)
        dlq.complete_message(msg)
""")
run('python poison.py')

# Step 6: An idempotent consumer with a business key in SQLite
write('idem.py', 'import os, sqlite3, time\nfrom azure.servicebus import ServiceBusClient, ServiceBusMessage\nfrom azure.servicebus.exceptions import MessageLockLostError\n\ndb = sqlite3.connect("shop.db")\ndb.execute("CREATE TABLE IF NOT EXISTS invoices"\n           " (order_id TEXT PRIMARY KEY)")\nSQL = "INSERT INTO invoices VALUES (?) ON CONFLICT DO NOTHING"\nclient = ServiceBusClient.from_connection_string(\n    os.environ["SERVICEBUS_CONNECTION_STRING"])\nwith client, client.get_queue_sender("invoices") as sender, \\\n        client.get_queue_receiver("invoices", max_wait_time=8) as r:\n    sender.send_messages([ServiceBusMessage(o) for o in\n                          ["order-7", "order-8", "order-8"]]) # resent\n    while msgs := r.receive_messages(max_wait_time=8):\n        msg, order_id = msgs[0], str(msgs[0])        # business key\n        with db:                                     # one transaction\n            new = db.execute(SQL, (order_id,)).rowcount\n        print(order_id, "delivery_count", msg.delivery_count, "->",\n              "invoice written" if new else "duplicate: skipped")\n        if order_id == "order-7" and msg.delivery_count == 0:\n            time.sleep(7)                    # too slow: lock lost\n        try:\n            r.complete_message(msg)\n        except MessageLockLostError:\n            print("  complete failed: lock lost")\nrows = db.execute("SELECT COUNT(*) FROM invoices").fetchone()[0]\nprint("invoices in the database:", rows)\n')
run('python idem.py')

# Step 7: Payments: write keyed by payment_id, then complete; a crash between them
write('send_payments.py', """\
import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage

PAYMENTS = ["pay-1 cust-a 2500", "pay-2 cust-b 4000",
            "pay-3 cust-c 1500"]                # id, customer, cents
client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("payments") as sender:
    sender.send_messages([ServiceBusMessage(
        p, message_id=p.split()[0]) for p in PAYMENTS])
print("queued", len(PAYMENTS), "payments")
""")
write('pay.py', """\
import os, sqlite3
from azure.servicebus import ServiceBusClient

CRASH_AFTER = os.environ.get("CRASH_AFTER")       # simulate a crash
db = sqlite3.connect("payments.db")
db.execute("CREATE TABLE IF NOT EXISTS charges"
           " (payment_id TEXT PRIMARY KEY, customer TEXT, cents INT)")
SQL = "INSERT INTO charges VALUES (?, ?, ?) ON CONFLICT DO NOTHING"
client = ServiceBusClient.from_connection_string(
    os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("payments",
                                       max_wait_time=8) as receiver:
    while msgs := receiver.receive_messages(max_wait_time=8):
        msg = msgs[0]                                  # 1. receive
        payment_id, customer, cents = str(msg).split()
        with db:                                       # 2. write
            row = (payment_id, customer, int(cents))
            new = db.execute(SQL, row).rowcount        # keyed, once
        print(payment_id, "delivery_count", msg.delivery_count, "->",
              "charged" if new else "already charged: skip")
        if payment_id == CRASH_AFTER:
            print("crash after the commit, before complete")
            os._exit(1)
        receiver.complete_message(msg)                 # 3. settle
rows = db.execute("SELECT COUNT(*) FROM charges").fetchone()[0]
print("charges in the database:", rows)
""")
run('python send_payments.py', allow_error=True)
run('CRASH_AFTER=pay-2 python pay.py', allow_error=True)
run('python pay.py', allow_error=True)

Next in this series

Part 5: Dead-Letter Queues and Poison Messages

The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on the official Azure Service Bus emulator (2.0.1) with SQL Server and azure-servicebus 7.15.0. Features the emulator does not support are shown as code. Part of the Real Time Azure Service Bus Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Azure Service Bus documentation (learn.microsoft.com/azure/service-bus-messaging) for anything you quote in an interview.