shivamka1

# 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

shivamka1

function acquire_distributed_lock(key, ttl_ms) -> (ok, token):
  # e.g. SET key token NX PX ttl_ms in Redis
  ...

function release_distributed_lock(key, token):
  # delete key only if stored token matches
  ...

function purchase(productId):
  (ok, token) = acquire_distributed_lock("lock:product:" + productId, ttl_ms=5000)

shivamka1

locks = Map<productId, LockObject>()

function get_lock(productId):
  if productId not in locks:
    locks[productId] = new LockObject()
  return locks[productId]

function purchase(productId):
  lock = get_lock(productId)

shivamka1

function purchase(productId):
  BEGIN TRANSACTION (ISOLATION = SERIALIZABLE)

  stock = query("SELECT stock FROM products WHERE id = ?", productId)

  if stock <= 0:
    ROLLBACK
    return "Out of stock"

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

shivamka1

INSERT INTO orders (order_id, user_id) VALUES (2001, 42);
-- Must reference an existing user in another shard

shivamka1

INSERT INTO orders (order_id, customer_id) 
VALUES (nextval('global_seq'), 101);

shivamka1

BEGIN;
SELECT balance FROM accounts WHERE user_id = 1;
-- client logic checks balance > 0
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
COMMIT;

shivamka1

INSERT INTO ledger (account_id, amount, txn_time)
VALUES (42, -100, NOW());

shivamka1

INSERT INTO users (email, name)
VALUES ('alice@example.com', 'Alice');

shivamka1

BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;
COMMIT;