Serial order enforced by message queue.py

# Producer (web request handler)
function request_purchase(productId):
  enqueue(topic="purchases", key=productId, value="BUY")

# Consumer (single-threaded per key-partition)
function purchase_worker():
  loop:
    messages = poll(topic="purchases", timeout_ms=200)
    for msg in messages:           # all messages for same key arrive in order
      productId = msg.key

      stock = query("SELECT stock FROM products WHERE id = ?", productId)
      if stock <= 0:
        continue

      exec("UPDATE products SET stock = stock - 1 WHERE id = ?", productId)