Kafka Idempotency Patterns for Resilient Asynchronous Systems

CREATE TYPE idempotency_status AS ENUM ('started', 'completed', 'failed');

CREATE TABLE idempotency_keys (
    -- The idempotency key itself, provided by the client.
    key VARCHAR(255) PRIMARY KEY,

    -- The current state of the operation.
    status idempotency_status NOT NULL,

    -- The HTTP status code and response body to be returned on subsequent requests.
    -- Storing this allows us to return the exact same response, making the endpoint truly idempotent.
    response_code INT,
    response_body JSONB,

    -- Timestamps for lifecycle management and debugging.
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- A lock timeout to prevent keys from being stuck in 'started' indefinitely.
    locked_at TIMESTAMPTZ
);

-- Index for efficient cleanup of old keys.
CREATE INDEX idx_idempotency_keys_created_at ON idempotency_keys(created_at);

package idempotency

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	_ "github.com/lib/pq"
)

// Represents the stored response for an idempotent operation.
type StoredResponse struct {
	StatusCode int
	Body       []byte
}

// The state of an operation stored in the database.
type IdempotencyRecord struct {
	Key          string
	Status       string
	ResponseCode sql.NullInt32
	ResponseBody []byte
	LockedAt     sql.NullTime
}

// BusinessLogicFunc is a placeholder for the actual work to be done.
type BusinessLogicFunc func(ctx context.Context) (*StoredResponse, error)

const lockTimeout = 5 * time.Minute // How long a lock is considered valid.

// Middleware handles the idempotency check for a Kafka message consumer.
func Middleware(db *sql.DB, idempotencyKey string, businessLogic BusinessLogicFunc) (*StoredResponse, error) {
	ctx := context.Background()

	// Step 1: Begin a transaction.
	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return nil, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback() // Rollback is a no-op if Commit() was called.

	// Step 2: Attempt to lock the idempotency key row.
	// `NOWAIT` causes the query to fail immediately if the row is locked by another transaction.
	record, err := getRecordForUpdate(ctx, tx, idempotencyKey)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			// Key does not exist, this is the first time we've seen it.
			// Insert a new record in the 'started' state.
			if err := insertNewRecord(ctx, tx, idempotencyKey); err != nil {
				return nil, err
			}
			// Commit this transaction to release the lock and make the 'started' record visible.
			if err := tx.Commit(); err != nil {
				return nil, fmt.Errorf("failed to commit initial insert: %w", err)
			}

			// Now execute the business logic outside of the initial lock transaction.
			return executeAndSaveResult(db, idempotencyKey, businessLogic)
		} else {
			// Any other error during SELECT FOR UPDATE is a problem.
			return nil, fmt.Errorf("failed to select for update: %w", err)
		}
	}

	// If we get here, the row exists. We need to check its state.
	switch record.Status {
	case "completed":
		// Operation already completed. Return the stored response.
		// No need to commit, we only performed a read.
		return &StoredResponse{
			StatusCode: int(record.ResponseCode.Int32),
			Body:       record.ResponseBody,
		}, nil
	case "started":
		// Another process started this operation but hasn't finished.
		// Check if the lock has expired.
		if record.LockedAt.Valid && time.Since(record.LockedAt.Time) > lockTimeout {
			// The lock has expired. We can take over.
			fmt.Printf("Lock for key %s expired. Taking over.\n", idempotencyKey)
			// Commit the read transaction.
			if err := tx.Commit(); err != nil {
				return nil, fmt.Errorf("failed to commit before takeover: %w", err)
			}
			return executeAndSaveResult(db, idempotencyKey, businessLogic)
		}
		// The lock is still valid. This is a concurrent request. Treat as a conflict.
		return nil, fmt.Errorf("operation with key %s is already in progress", idempotencyKey)

	case "failed":
		// The operation was previously attempted and failed permanently.
		// Depending on business requirements, you might retry or reject immediately.
		return nil, fmt.Errorf("operation with key %s has previously failed", idempotencyKey)
	}

	return nil, fmt.Errorf("unhandled idempotency status: %s", record.Status)
}

func getRecordForUpdate(ctx context.Context, tx *sql.Tx, key string) (*IdempotencyRecord, error) {
	query := `SELECT key, status, response_code, response_body, locked_at FROM idempotency_keys WHERE key = $1 FOR UPDATE NOWAIT`
	row := tx.QueryRowContext(ctx, query, key)

	record := &IdempotencyRecord{}
	err := row.Scan(&record.Key, &record.Status, &record.ResponseCode, &record.ResponseBody, &record.LockedAt)
	if err != nil {
		return nil, err
	}
	return record, nil
}

func insertNewRecord(ctx context.Context, tx *sql.Tx, key string) error {
	query := `INSERT INTO idempotency_keys (key, status, locked_at) VALUES ($1, 'started', NOW())`
	_, err := tx.ExecContext(ctx, query, key)
	if err != nil {
		return fmt.Errorf("failed to insert new idempotency record: %w", err)
	}
	return nil
}

func executeAndSaveResult(db *sql.DB, key string, businessLogic BusinessLogicFunc) (*StoredResponse, error) {
	// Execute the actual business logic.
	resp, err := businessLogic(context.Background())

	// Regardless of the outcome, we update the idempotency record.
	tx, txErr := db.BeginTx(context.Background(), nil)
	if txErr != nil {
		return nil, fmt.Errorf("failed to begin final update transaction: %w", txErr)
	}
	defer tx.Rollback()

	if err != nil {
		// Business logic failed. Mark the key as 'failed'.
		// This prevents retries for non-transient errors.
		query := `UPDATE idempotency_keys SET status = 'failed', updated_at = NOW() WHERE key = $1`
		_, updateErr := tx.ExecContext(context.Background(), query, key)
		if updateErr != nil {
			return nil, fmt.Errorf("failed to mark key as failed: %w (original error: %w)", updateErr, err)
		}
		if err := tx.Commit(); err != nil {
			return nil, fmt.Errorf("failed to commit failed status: %w", err)
		}
		return nil, err // Return the original business logic error.
	}

	// Business logic succeeded. Mark as 'completed' and store the response.
	jsonBody, jsonErr := json.Marshal(resp.Body)
	if jsonErr != nil {
		return nil, fmt.Errorf("failed to marshal response body: %w", jsonErr)
	}

	query := `UPDATE idempotency_keys SET status = 'completed', response_code = $2, response_body = $3, updated_at = NOW() WHERE key = $1`
	_, updateErr := tx.ExecContext(context.Background(), query, key, resp.StatusCode, jsonBody)
	if updateErr != nil {
		return nil, fmt.Errorf("failed to update idempotency record to completed: %w", updateErr)
	}

	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("failed to commit completed status: %w", err)
	}

	return resp, nil
}

import redis
import json
import time
import uuid
from typing import Callable, Dict, Any, Tuple

# Connect to Redis
# In production, use redis.from_url() and connection pooling
r = redis.Redis(decode_responses=True)

# The TTL for a key in a 'started' state (e.g., 5 minutes)
STARTED_STATE_TTL_SECONDS = 300
# The TTL for a key in a 'completed' state (e.g., 24 hours)
COMPLETED_STATE_TTL_SECONDS = 86400

# Lua script for atomic check-and-set
# KEYS[1] = idempotency_key
# ARGV[1] = 'started' status string
# ARGV[2] = TTL in seconds for the 'started' state
# Returns:
# - nil if the key was successfully set to 'started'
# - the existing value of the key if it already exists
ATOMIC_CHECK_AND_SET_LUA = """
local key_exists = redis.call('get', KEYS[1])
if not key_exists then
    redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2])
    return nil
else
    return key_exists
end
"""

# Register the script with Redis for better performance (avoids sending the script every time)
# In a real app, you'd do this once at startup.
script_sha = r.script_load(ATOMIC_CHECK_AND_SET_LUA)

class IdempotencyException(Exception):
    pass

class OperationInProgress(IdempotencyException):
    pass

class OperationPreviouslyFailed(IdempotencyException):
    pass

# A placeholder for your actual business logic
def process_payment(amount: float, currency: str) -> Dict[str, Any]:
    print(f"Processing payment of {amount} {currency}...")
    time.sleep(1) # Simulate work
    if amount < 0:
        raise ValueError("Amount cannot be negative")
    return {"status": "success", "transaction_id": str(uuid.uuid4())}

def redis_idempotency_wrapper(idempotency_key: str, business_logic: Callable[..., Any], *args, **kwargs) -> Any:
    """Wraps a function call with Redis-based idempotency checks."""

    # Keys used in Redis
    status_key = f"idempotency:status:{idempotency_key}"
    result_key = f"idempotency:result:{idempotency_key}"

    # Step 1: Atomic check-and-set
    try:
        existing_status = r.evalsha(script_sha, 1, status_key, "started", STARTED_STATE_TTL_SECONDS)
    except redis.exceptions.NoScriptError:
        # Fallback if script is not loaded (e.g., after a Redis flush)
        print("Script not found, loading and retrying...")
        script_sha = r.script_load(ATOMIC_CHECK_AND_SET_LUA)
        existing_status = r.evalsha(script_sha, 1, status_key, "started", STARTED_STATE_TTL_SECONDS)

    if existing_status is not None:
        # Key already existed
        if existing_status == "completed":
            print(f"Key {idempotency_key}: Operation already completed. Returning cached result.")
            cached_result = r.get(result_key)
            return json.loads(cached_result) if cached_result else None
        elif existing_status == "started":
            print(f"Key {idempotency_key}: Operation already in progress.")
            raise OperationInProgress(f"Operation with key {idempotency_key} is in progress.")
        elif existing_status == "failed":
            print(f"Key {idempotency_key}: Operation previously failed.")
            raise OperationPreviouslyFailed(f"Operation with key {idempotency_key} previously failed.")
        else:
            raise IdempotencyException(f"Unknown status: {existing_status}")

    # Step 2: If we are here, we have acquired the lock. Execute business logic.
    print(f"Key {idempotency_key}: Acquired lock. Executing business logic.")
    try:
        result = business_logic(*args, **kwargs)
        
        # Step 3: Store the result and set status to 'completed'
        # Use a pipeline for atomicity
        pipe = r.pipeline()
        pipe.set(result_key, json.dumps(result), ex=COMPLETED_STATE_TTL_SECONDS)
        pipe.set(status_key, "completed", ex=COMPLETED_STATE_TTL_SECONDS)
        pipe.execute()
        
        print(f"Key {idempotency_key}: Business logic successful. Result cached.")
        return result
    except Exception as e:
        # Business logic failed
        print(f"Key {idempotency_key}: Business logic failed: {e}")
        # Mark as 'failed' or simply delete the key to allow retries
        # Deleting allows full retry, setting 'failed' prevents it.
        # Let's choose deletion for retriability of transient errors.
        r.delete(status_key)
        raise # Re-raise the original exception

# --- Example Usage ---
if __name__ == "__main__":
    key1 = str(uuid.uuid4())
    
    # First call - should execute
    try:
        result1 = redis_idempotency_wrapper(key1, process_payment, amount=100.0, currency="USD")
        print("Result 1:", result1)
    except Exception as e:
        print("Error on first call:", e)

    print("\n" + "-"*20 + "\n")

    # Second call - should be idempotent and return cached result
    try:
        result2 = redis_idempotency_wrapper(key1, process_payment, amount=100.0, currency="USD")
        print("Result 2:", result2)
    except Exception as e:
        print("Error on second call:", e)