Data Consistency and Distributed Transactions Questions

from flask import Flask, request, jsonify
import redis
import time
import uuid
import json

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# Config
LOCK_TTL_MS = 30_000      # lock expires in 30s
DONE_TTL_SECONDS = 7 * 24 * 3600  # keep processed result for 7 days

# Lua script: atomically set done if absent and return 1/0
SET_DONE_IF_ABSENT = """
if redis.call('EXISTS', KEYS[1]) == 0 then
  redis.call('HSET', KEYS[1], 'status', ARGV[1], 'result', ARGV[2], 'ts', ARGV[3])
  redis.call('EXPIRE', KEYS[1], ARGV[4])
  return 1
else
  return 0
end
"""
set_done_if_absent = r.register_script(SET_DONE_IF_ABSENT)

def process_message(payload):
    # Placeholder for real, idempotent processing (side-effects must tolerate retries)
    time.sleep(0.5)  # simulate work
    return {"processed_at": time.time(), "echo": payload}

@app.route('/consume', methods=['POST'])
def consume():
    data = request.get_json(force=True)
    message_id = data.get('message_id')
    payload = data.get('payload')

    if not message_id:
        return jsonify({"error": "missing message_id"}), 400

    done_key = f"msg:done:{message_id}"
    lock_key = f"msg:lock:{message_id}"

    # 1) fast-path: if already done, return stored result
    done = r.hgetall(done_key)
    if done:
        return jsonify({"message_id": message_id, "status": done.get('status'), "result": json.loads(done.get('result')) if done.get('result') else None}), 200

    # 2) try to acquire lock (unique token helps safe release)
    lock_token = str(uuid.uuid4())
    acquired = r.set(lock_key, lock_token, nx=True, px=LOCK_TTL_MS)
    if not acquired:
        # Someone else is processing. Wait briefly and re-check done (simple backoff)
        time.sleep(0.2)
        done = r.hgetall(done_key)
        if done:
            return jsonify({"message_id": message_id, "status": done.get('status'), "result": json.loads(done.get('result')) if done.get('result') else None}), 200
        else:
            # Optionally return 202 Accepted so sender can retry later
            return jsonify({"message_id": message_id, "status": "processing"}), 202

    try:
        # re-check after acquiring lock to avoid race where done was set between checks
        done = r.hgetall(done_key)
        if done:
            return jsonify({"message_id": message_id, "status": done.get('status'), "result": json.loads(done.get('result')) if done.get('result') else None}), 200

        # 3) process message
        result_obj = process_message(payload)
        result_json = json.dumps(result_obj)

        # 4) atomically set done metadata only if absent (protect against double-writes)
        ok = set_done_if_absent(keys=[done_key], args=['success', result_json, str(time.time()), str(DONE_TTL_SECONDS)])
        if ok == 1:
            return jsonify({"message_id": message_id, "status": "success", "result": result_obj}), 200
        else:
            # another worker wrote success in the meantime
            done = r.hgetall(done_key)
            return jsonify({"message_id": message_id, "status": done.get('status'), "result": json.loads(done.get('result')) if done.get('result') else None}), 200
    finally:
        # safe release: only delete lock if we own it (avoid deleting another's lock after TTL rollover)
        release_script = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
          return redis.call('DEL', KEYS[1])
        else
          return 0
        end
        """
        r.eval(release_script, 1, lock_key, lock_token)

if __name__ == '__main__':
    app.run(port=5000)