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
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 NoneApply 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)
