Idempotent technic

What's idempotent?

Idempotent is a situation where the outcome remains if the context provided, the logic, and the data do not change.

How is it used?

Idempotence helps prevent duplicate requests or transactions from a client to the server when the request data remains unchanged. When a request is sent to the server with an idempotent key, the server checks and stores this key to compare with future requests. This ensures that each request is only executed once and that the server returns the same result for multiple requests that have the same content.

Key benefits of Idempotency in distributed systems

• Data Consistency Guarantees
  Ensures operations produce the same result regardless of multiple executions
• Error Handling & Retry Capabilities
  Safely retry failed operations without side effects
  Handles network issues, timeouts, and system failures gracefully
• Distributed System Safety
  Maintains consistency across multiple servers/nodes
  Prevents duplicate processing in distributed environments
• Race Condition Prevention
  Eliminates concurrent processing issues
  Ensures only one process handles a specific operation
• Scalability Benefits
  Enables horizontal scaling without data inconsistencies
  Allows multiple instances to process requests safely
• System Resilience
  Handles system failures and recoveries smoothly
  Maintains system integrity during outages
• Monitoring & Debugging
  Tracks operation status and execution history
  Simplifies troubleshooting in complex systems

Key Implementation Principles

• Always use unique identifiers for operations
• Implement proper caching strategies
• Use distributed locks for coordination
• Handle edge cases and failures gracefully
• Maintain audit trails
• Implement proper cleanup mechanisms
• Monitor system behavior
• Use proper timeout mechanisms
• Implement circuit breakers for system protection
• Follow the principle of “exactly once” delivery semantics

Idempotent Flow in Distributed System

Idempotent Flow in Distributed System

Multiple Requests with the Same Idempotent Key Flow

How to create an idempotent key

import uuid
import hashlib
import time

class IdempotencyKeyGenerator:
    @staticmethod
    def generate_key_from_request(request_data, prefix=""):
        """create key from request data"""
        sorted_data = json.dumps(request_data, sort_keys=True)
        
        # create hash
        hash_object = hashlib.sha256(sorted_data.encode())
        hash_value = hash_object.hexdigest()
        
        # add prefix
        return f"{prefix}:{hash_value}" if prefix else hash_value

    @staticmethod
    def generate_unique_key(prefix=""):
        """Create unique key based on UUID"""
        unique_id = str(uuid.uuid4())
        timestamp = str(int(time.time()))
        return f"{prefix}:{timestamp}:{unique_id}" if prefix else f"{timestamp}:{unique_id}"

Manage Idempotency key with Reddis

import redis
from datetime import timedelta

class IdempotencyManager:
    def __init__(self, redis_client, ttl_seconds=86400):  
        self.redis = redis_client
        self.ttl = ttl_seconds

    def is_processed(self, idempotency_key):
        return bool(self.redis.exists(idempotency_key))

    def mark_as_processed(self, idempotency_key, result=None):
        pipeline = self.redis.pipeline()
        pipeline.set(idempotency_key, json.dumps(result) if result else "processed")
        pipeline.expire(idempotency_key, self.ttl)
        pipeline.execute()

    def get_result(self, idempotency_key):
        result = self.redis.get(idempotency_key)
        return json.loads(result) if result else None

Apply Idempotency key in Rest API

from fastapi import FastAPI, Header, HTTPException
from typing import Optional

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)
idempotency_manager = IdempotencyManager(redis_client)

@app.post("/api/payments")
async def create_payment(
    payment: dict,
    idempotency_key: Optional[str] = Header(None)
):
    # Check idempotency key
    if not idempotency_key:
        idempotency_key = IdempotencyKeyGenerator.generate_key_from_request(payment)

    # check request status
    if idempotency_manager.is_processed(idempotency_key):
        return idempotency_manager.get_result(idempotency_key)

    try:
        # Process payment
        result = process_payment(payment)
        
        # Store result
        idempotency_manager.mark_as_processed(idempotency_key, result)
        
        return result
    except Exception as e:
        # handle error without marked request be procced
        raise HTTPException(status_code=500, detail=str(e))

Handle concurrent requests

class IdempotencyManagerWithLock:
    def __init__(self, redis_client, ttl_seconds=86400):
        self.redis = redis_client
        self.ttl = ttl_seconds

    def acquire_lock(self, key, timeout=10):
        """Get lock to handle request"""
        lock_key = f"lock:{key}"
        return self.redis.set(lock_key, "1", ex=timeout, nx=True)

    def release_lock(self, key):
        """release lock"""
        lock_key = f"lock:{key}"
        self.redis.delete(lock_key)

    async def process_with_lock(self, key, process_func):
        """Handle request with lock"""
        if self.is_processed(key):
            return self.get_result(key)

        if not self.acquire_lock(key):
            # wait và retry
            await asyncio.sleep(0.1)
            return await self.process_with_lock(key, process_func)

        try:
            result = await process_func()
            self.mark_as_processed(key, result)
            return result
        finally:
            self.release_lock(key)

Clean strategy

class IdempotencyCleanup:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def cleanup_expired_keys(self, batch_size=1000):
        cursor = 0
        pattern = "*"  

        while True:
            cursor, keys = self.redis.scan(cursor, pattern, batch_size)
            
            for key in keys:
                if not self.redis.ttl(key):
                    self.redis.delete(key)
                    
            if cursor == 0:
                break

    async def schedule_cleanup(self, interval_hours=24):
        while True:
            await self.cleanup_expired_keys()
            await asyncio.sleep(interval_hours * 3600)