Idempotency Patterns in Event-Driven Systems via Distributed Locks

# WARNING: This is a naive and flawed implementation

def process_event_naive(event):
    event_id = event['id']

    # 1. CHECK
    if database.has_processed(event_id):
        print(f"Event {event_id} already processed.")
        return

    # --- RACE CONDITION WINDOW ---
    # Two concurrent processes can pass the check above for the same event_id
    # before either has a chance to write the record.

    # 2. ACT
    result = execute_business_logic(event['data'])

    # 3. WRITE
    database.mark_as_processed(event_id, result)

    return result

import redis
import json
import time
import functools
from uuid import uuid4

# Configure your Redis client
# In a real application, this would come from configuration management
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

# Constants
LOCK_TTL_SECONDS = 60  # How long the lock is held (should exceed max processing time)
RESULT_TTL_SECONDS = 3600 # How long to cache the final result (e.g., 1 hour)

class IdempotencyException(Exception):
    pass

class RequestInProgress(IdempotencyException):
    pass

def idempotent(key_arg_name='idempotency_key'):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            idempotency_key = kwargs.get(key_arg_name)
            if not idempotency_key:
                raise ValueError(f"Missing required keyword argument: {key_arg_name}")

            lock_key = f"idempotency:lock:{idempotency_key}"
            result_key = f"idempotency:result:{idempotency_key}"

            # 1. Atomically try to acquire the lock
            # SET key value NX PX ttl -> Set key to value if key does not exist, with a millisecond TTL
            if redis_client.set(lock_key, "in_progress", nx=True, px=LOCK_TTL_SECONDS * 1000):
                try:
                    # 2. Lock acquired, execute the business logic
                    print(f"[{idempotency_key}] Lock acquired. Executing function.")
                    result = func(*args, **kwargs)
                    
                    # 3. Store the result and set its TTL
                    # We use a transaction (pipeline) to ensure atomicity
                    pipe = redis_client.pipeline()
                    pipe.set(result_key, json.dumps(result), px=RESULT_TTL_SECONDS * 1000)
                    # 4. Release the lock explicitly
                    pipe.delete(lock_key)
                    pipe.execute()

                    print(f"[{idempotency_key}] Execution successful. Result stored.")
                    return result
                except Exception as e:
                    # 5. On failure, release the lock to allow retries
                    print(f"[{idempotency_key}] Execution failed. Releasing lock.")
                    redis_client.delete(lock_key)
                    raise e # Re-raise the exception
            else:
                # 6. Lock could not be acquired
                print(f"[{idempotency_key}] Lock conflict. Checking for result.")
                # Check if a result is already cached
                cached_result = redis_client.get(result_key)
                if cached_result:
                    print(f"[{idempotency_key}] Returning cached result.")
                    return json.loads(cached_result)
                else:
                    # This is the critical edge case: another process holds the lock but hasn't finished.
                    # You must choose a strategy: fail fast, wait, or retry.
                    print(f"[{idempotency_key}] Request in progress by another worker.")
                    raise RequestInProgress("This request is currently being processed.")

        return wrapper
    return decorator

# --- Example Usage ---

@idempotent(key_arg_name='request_id')
def process_payment(request_id: str, amount: float, currency: str):
    """A mock function that simulates a long-running, critical operation."""
    print(f"Processing payment of {amount} {currency} for request {request_id}...")
    time.sleep(5)  # Simulate network latency or heavy computation
    return {"status": "success", "transaction_id": str(uuid4()), "amount_processed": amount}

if __name__ == '__main__':
    # Simulate two identical requests coming in at the same time
    # In a real system, these would be from two different threads/processes/servers
    from threading import Thread

    request_id_1 = f"req-{uuid4()}"

    def call_processor(req_id):
        try:
            result = process_payment(request_id=req_id, amount=100.0, currency="USD")
            print(f"Thread finished with result: {result}")
        except RequestInProgress:
            print("Thread caught RequestInProgress. Request is being handled elsewhere.")
        except Exception as e:
            print(f"Thread caught an unexpected error: {e}")

    print("--- Scenario 1: Concurrent identical requests ---")
    t1 = Thread(target=call_processor, args=(request_id_1,))
    t2 = Thread(target=call_processor, args=(request_id_1,))

    t1.start()
    time.sleep(0.1) # Ensure t1 gets the lock first
    t2.start()

    t1.join()
    t2.join()

    print("\n--- Scenario 2: A third request arrives after the first has completed ---")
    # This call should be very fast and return the cached result
    call_processor(request_id_1)

import boto3
import json
import time
import functools
from botocore.exceptions import ClientError
from uuid import uuid4

# Configure Boto3 client
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('IdempotencyStore') # Make sure you've created this table

# Constants
IN_PROGRESS_TTL_SECONDS = 60
COMPLETED_TTL_SECONDS = 3600

class IdempotencyException(Exception):
    pass

class RequestInProgress(IdempotencyException):
    pass

def idempotent_dynamodb(key_arg_name='idempotency_key'):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            idempotency_key = kwargs.get(key_arg_name)
            if not idempotency_key:
                raise ValueError(f"Missing required keyword argument: {key_arg_name}")

            current_time = int(time.time())
            in_progress_expiry = current_time + IN_PROGRESS_TTL_SECONDS

            try:
                # 1. Atomically create the record if it doesn't exist
                # This is our lock acquisition step.
                table.put_item(
                    Item={
                        'IdempotencyKey': idempotency_key,
                        'Status': 'IN_PROGRESS',
                        'ExpiryTimestamp': in_progress_expiry
                    },
                    ConditionExpression='attribute_not_exists(IdempotencyKey)'
                )
                print(f"[{idempotency_key}] Lock acquired via DynamoDB. Executing function.")

            except ClientError as e:
                if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
                    # 2. Record already exists, lock was not acquired
                    print(f"[{idempotency_key}] Lock conflict. Checking item status.")
                    # We need to fetch the existing item to check its status
                    response = table.get_item(Key={'IdempotencyKey': idempotency_key})
                    item = response.get('Item')

                    if not item or item.get('Status') == 'IN_PROGRESS':
                        # Item might have expired or is still being processed
                        print(f"[{idempotency_key}] Request in progress by another worker.")
                        raise RequestInProgress("This request is currently being processed.")
                    
                    elif item.get('Status') == 'COMPLETED':
                        # Return the cached result
                        print(f"[{idempotency_key}] Returning cached result from DynamoDB.")
                        return json.loads(item.get('ResponseData', '{}'))
                else:
                    raise # Re-raise other DynamoDB errors

            try:
                # 3. Execution of the core business logic
                result = func(*args, **kwargs)
                result_json = json.dumps(result)
                completed_expiry = int(time.time()) + COMPLETED_TTL_SECONDS

                # 4. Update the record to COMPLETED and store the result
                table.update_item(
                    Key={'IdempotencyKey': idempotency_key},
                    UpdateExpression='SET #status = :status, #data = :data, #expiry = :expiry',
                    ExpressionAttributeNames={
                        '#status': 'Status',
                        '#data': 'ResponseData',
                        '#expiry': 'ExpiryTimestamp'
                    },
                    ExpressionAttributeValues={
                        ':status': 'COMPLETED',
                        ':data': result_json,
                        ':expiry': completed_expiry
                    }
                )
                print(f"[{idempotency_key}] Execution successful. Result stored in DynamoDB.")
                return result

            except Exception as e:
                # 5. On failure, delete the record to allow retries
                print(f"[{idempotency_key}] Execution failed. Deleting idempotency record.")
                table.delete_item(Key={'IdempotencyKey': idempotency_key})
                raise e

        return wrapper
    return decorator

# --- Example Usage ---

@idempotent_dynamodb(key_arg_name='request_id')
def create_order(request_id: str, user_id: str, items: list):
    """Simulates creating an order in a database."""
    print(f"Creating order for user {user_id} with items: {items}")
    time.sleep(5) # Simulate database writes and other operations
    order_id = f"order_{uuid4()}"
    return {"status": "created", "order_id": order_id, "user_id": user_id}

# You would need to set up AWS credentials and create the DynamoDB table
# for this to run. The simulation logic is the same as the Redis example.
# For brevity, the threading simulation is omitted here but would work identically.