Postmortem: How a Claude 4 Code Generation Error Deployed a Bug to Production Affecting 200K Users

import json
import time
import logging
import redis
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from typing import Dict, Optional

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class PaymentRetryQueue:
    """
    Claude 4 Opus generated implementation of payment retry queue (BUGGY VERSION)
    Deployed to production on 2024-10-17 14:37 UTC
    Bug: Idempotency key uses second-precision timestamp, leading to duplicate key collisions
    and skipped retries for high-throughput checkout flows.
    """
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.redis_client = redis.from_url(redis_url, decode_responses=True)
        self.queue_key = "payment_retry_queue"
        self.idempotency_ttl = 86400  # 24 hours TTL for idempotency keys

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        retry=retry_if_exception_type(redis.RedisError)
    )
    def _enqueue_with_retry(self, payload: Dict) -> None:
        """Enqueue payment payload to Redis sorted set with retry timestamp as score"""
        try:
            # Claude 4 generated idempotency key logic (BUGGY LINE BELOW)
            idempotency_key = f"{payload['user_id']}_{int(time.time())}"
            # Check if idempotency key already exists (flawed: TTL not checked, key collision likely)
            existing = self.redis_client.get(idempotency_key)
            if existing:
                logger.warning(f"Duplicate idempotency key {idempotency_key}, skipping enqueue")
                return
            # Serialize payload and add to sorted set
            score = time.time() + payload.get("retry_delay", 60)
            self.redis_client.zadd(self.queue_key, {json.dumps(payload): score})
            # Set idempotency key with TTL (Claude forgot to set TTL initially, fixed in later edit but still buggy)
            self.redis_client.setex(idempotency_key, self.idempotency_ttl, json.dumps(payload))
            logger.info(f"Enqueued payment for user {payload['user_id']} with key {idempotency_key}")
        except KeyError as e:
            logger.error(f"Missing required field in payload: {e}")
            raise ValueError(f"Invalid payment payload: {e}") from e
        except redis.RedisError as e:
            logger.error(f"Redis error enqueueing payment: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error enqueueing payment: {e}")
            raise

    def process_retries(self) -> None:
        """Process pending retries from the queue"""
        try:
            current_time = time.time()
            # Get all pending retries where score (scheduled time) <= current time
            pending = self.redis_client.zrangebyscore(self.queue_key, 0, current_time)
            for item in pending:
                try:
                    payload = json.loads(item)
                    # Execute payment (simplified for example)
                    payment_result = self._execute_payment(payload)
                    if payment_result["success"]:
                        self.redis_client.zrem(self.queue_key, item)
                        logger.info(f"Processed successful retry for user {payload['user_id']}")
                    else:
                        # Reschedule retry with exponential backoff
                        new_score = current_time + (payload.get("retry_count", 0) + 1) * 60
                        self.redis_client.zadd(self.queue_key, {item: new_score})
                        # Update retry count in payload
                        payload["retry_count"] = payload.get("retry_count", 0) + 1
                        logger.warning(f"Rescheduling retry for user {payload['user_id']}, attempt {payload['retry_count']}")
                except json.JSONDecodeError as e:
                    logger.error(f"Invalid JSON in queue item: {e}")
                    self.redis_client.zrem(self.queue_key, item)
                except Exception as e:
                    logger.error(f"Error processing retry item: {e}")
        except redis.RedisError as e:
            logger.error(f"Redis error processing retries: {e}")
            raise

    def _execute_payment(self, payload: Dict) -> Dict:
        """Mock payment execution (simplified)"""
        # In production this would call Stripe/PayPal APIs
        return {"success": True, "transaction_id": f"txn_{int(time.time())}"}

if __name__ == "__main__":
    # Example usage that would trigger the bug in production
    queue = PaymentRetryQueue()
    test_payload = {
        "user_id": "usr_12345",
        "amount": 99.99,
        "currency": "USD",
        "payment_method_id": "pm_67890",
        "retry_count": 0
    }
    # Enqueue multiple times in the same second to trigger idempotency collision
    for _ in range(5):
        queue._enqueue_with_retry(test_payload)
        time.sleep(0.1)  # 100ms delay, still within same second

import json
import time
import logging
import uuid
import redis
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from typing import Dict, Optional, Tuple

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class FixedPaymentRetryQueue:
    """
    Postmortem fixed implementation of payment retry queue
    Addresses Claude 4 Opus idempotency key bug and adds validation
    Deployed to production on 2024-10-17 16:24 UTC after incident rollback
    """
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.redis_client = redis.from_url(redis_url, decode_responses=True)
        self.queue_key = "payment_retry_queue_v2"
        self.idempotency_ttl = 86400  # 24 hours TTL for idempotency keys
        self.max_retries = 5

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        retry=retry_if_exception_type(redis.RedisError)
    )
    def _enqueue_with_retry(self, payload: Dict) -> Tuple[bool, Optional[str]]:
        """
        Enqueue payment payload with collision-resistant idempotency key
        Returns (success: bool, idempotency_key: Optional[str])
        """
        required_fields = ["user_id", "amount", "currency", "payment_method_id"]
        for field in required_fields:
            if field not in payload:
                raise ValueError(f"Missing required field: {field}")

        try:
            # FIX 1: Use nanosecond-precision timestamp + UUID to eliminate collisions
            idempotency_key = f"{payload['user_id']}_{time.time_ns()}_{uuid.uuid4().hex[:8]}"
            # FIX 2: Use Redis pipeline for atomic check-and-set
            with self.redis_client.pipeline() as pipe:
                pipe.watch(idempotency_key)
                existing = pipe.get(idempotency_key)
                if existing:
                    # Check if existing key is still valid (TTL > 0)
                    ttl = pipe.ttl(idempotency_key)
                    if ttl > 0:
                        logger.warning(f"Valid idempotency key {idempotency_key} exists, skipping enqueue")
                        return False, None
                # Start transaction
                pipe.multi()
                # Add to sorted set with scheduled retry time
                score = time.time() + payload.get("retry_delay", 60)
                pipe.zadd(self.queue_key, {json.dumps(payload): score})
                # Set idempotency key with TTL
                pipe.setex(idempotency_key, self.idempotency_ttl, json.dumps(payload))
                # Execute transaction
                pipe.execute()
            logger.info(f"Enqueued payment for user {payload['user_id']} with key {idempotency_key}")
            return True, idempotency_key
        except redis.WatchError:
            logger.warning(f"Idempotency key {idempotency_key} modified during transaction, retrying")
            return self._enqueue_with_retry(payload)
        except KeyError as e:
            logger.error(f"Missing required field in payload: {e}")
            raise ValueError(f"Invalid payment payload: {e}") from e
        except redis.RedisError as e:
            logger.error(f"Redis error enqueueing payment: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error enqueueing payment: {e}")
            raise

    def process_retries(self) -> int:
        """Process pending retries from the queue, returns number of processed items"""
        processed_count = 0
        try:
            current_time = time.time()
            # Get all pending retries where score (scheduled time) <= current time
            pending = self.redis_client.zrangebyscore(self.queue_key, 0, current_time, withscores=True)
            for item, score in pending:
                try:
                    payload = json.loads(item)
                    # Validate payload before processing
                    if not all(k in payload for k in ["user_id", "amount"]):
                        logger.error(f"Invalid payload in queue: {item}")
                        self.redis_client.zrem(self.queue_key, item)
                        continue
                    # Execute payment
                    payment_result = self._execute_payment(payload)
                    if payment_result["success"]:
                        self.redis_client.zrem(self.queue_key, item)
                        processed_count += 1
                        logger.info(f"Processed successful retry for user {payload['user_id']}")
                    else:
                        # Check max retries
                        retry_count = payload.get("retry_count", 0) + 1
                        if retry_count > self.max_retries:
                            logger.error(f"Max retries exceeded for user {payload['user_id']}, moving to dead letter queue")
                            self.redis_client.zrem(self.queue_key, item)
                            self.redis_client.lpush("payment_dead_letter_queue", item)
                        else:
                            # Reschedule retry with exponential backoff
                            new_score = current_time + (retry_count) * 60
                            self.redis_client.zadd(self.queue_key, {item: new_score})
                            payload["retry_count"] = retry_count
                            logger.warning(f"Rescheduling retry for user {payload['user_id']}, attempt {retry_count}")
                except json.JSONDecodeError as e:
                    logger.error(f"Invalid JSON in queue item: {e}")
                    self.redis_client.zrem(self.queue_key, item)
                except Exception as e:
                    logger.error(f"Error processing retry item: {e}")
        except redis.RedisError as e:
            logger.error(f"Redis error processing retries: {e}")
            raise
        return processed_count

    def _execute_payment(self, payload: Dict) -> Dict:
        """Mock payment execution with simulated failure for testing"""
        import random
        # Simulate 10% failure rate for testing
        if random.random() < 0.1:
            return {"success": False, "error": "Simulated payment failure"}
        return {"success": True, "transaction_id": f"txn_{uuid.uuid4().hex}"}

if __name__ == "__main__":
    # Example usage of fixed queue
    queue = FixedPaymentRetryQueue()
    test_payload = {
        "user_id": "usr_12345",
        "amount": 99.99,
        "currency": "USD",
        "payment_method_id": "pm_67890",
        "retry_count": 0
    }
    # Enqueue multiple times rapidly - no collisions due to nanosecond + UUID key
    for i in range(10):
        success, key = queue._enqueue_with_retry(test_payload)
        if success:
            print(f"Enqueued attempt {i} with key {key}")
        time.sleep(0.05)  # 50ms delay
    # Process pending retries
    processed = queue.process_retries()
    print(f"Processed {processed} retries")

import ast
import re
import sys
import logging
import os
from typing import List, Tuple

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class AICodeValidator:
    """
    Pre-commit validator to detect common AI-generated code errors
    Developed post-incident to catch patterns like weak idempotency keys,
    missing error handling, and untested edge cases.
    """
    # Patterns to detect weak idempotency key generation (Claude 4 common error)
    IDEMPOTENCY_PATTERNS = [
        r"int\(time\.time\(\)\)",  # Second-precision timestamp
        r"uuid\.uuid4\(\)\.hex$",  # UUID without additional entropy
        r"f\".*{user_id}.*time\.time\(\)\""  # String formatting with low-precision time
    ]
    # Required exception types to catch in try/except blocks
    REQUIRED_EXCEPTIONS = ["KeyError", "ValueError", "RedisError", "json.JSONDecodeError"]
    # Minimum test coverage for AI-generated modules
    MIN_COVERAGE = 85

    def __init__(self, target_path: str):
        self.target_path = target_path
        self.errors: List[Tuple[str, str, int]] = []  # (file, error, line)

    def validate_file(self, file_path: str) -> bool:
        """Validate a single Python file for AI-generated code errors"""
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                source = f.read()
            # Parse AST to analyze code structure
            tree = ast.parse(source)
            # Check for weak idempotency key patterns
            self._check_idempotency_patterns(source, file_path)
            # Check for required error handling
            self._check_error_handling(tree, file_path)
            # Check for missing docstrings in public methods
            self._check_docstrings(tree, file_path)
            return len(self.errors) == 0
        except SyntaxError as e:
            self.errors.append((file_path, f"Syntax error: {e}", e.lineno or 0))
            return False
        except Exception as e:
            self.errors.append((file_path, f"Validation error: {e}", 0))
            return False

    def _check_idempotency_patterns(self, source: str, file_path: str) -> None:
        """Check for weak idempotency key generation patterns"""
        for pattern in self.IDEMPOTENCY_PATTERNS:
            matches = re.finditer(pattern, source)
            for match in matches:
                line_no = source[:match.start()].count("\n") + 1
                self.errors.append((
                    file_path,
                    f"Weak idempotency key pattern detected: {pattern}",
                    line_no
                ))

    def _check_error_handling(self, tree: ast.AST, file_path: str) -> None:
        """Check that try/except blocks catch required exceptions"""
        for node in ast.walk(tree):
            if isinstance(node, ast.Try):
                caught_exceptions = []
                for handler in node.handlers:
                    if handler.type:
                        if isinstance(handler.type, ast.Name):
                            caught_exceptions.append(handler.type.id)
                        elif isinstance(handler.type, ast.Tuple):
                            caught_exceptions.extend([e.id for e in handler.type.elts if isinstance(e, ast.Name)])
                # Check if required exceptions are caught
                for req_exc in self.REQUIRED_EXCEPTIONS:
                    if req_exc not in caught_exceptions and "Exception" not in caught_exceptions:
                        self.errors.append((
                            file_path,
                            f"Try block missing required exception handler: {req_exc}",
                            node.lineno
                        ))

    def _check_docstrings(self, tree: ast.AST, file_path: str) -> None:
        """Check that public methods have docstrings"""
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
                if not node.name.startswith("_"):  # Public members only
                    if not ast.get_docstring(node):
                        self.errors.append((
                            file_path,
                            f"Public {node.__class__.__name__} {node.name} missing docstring",
                            node.lineno
                        ))

    def run_validation(self) -> bool:
        """Run validation on all Python files in target path"""
        if os.path.isfile(self.target_path) and self.target_path.endswith(".py"):
            return self.validate_file(self.target_path)
        elif os.path.isdir(self.target_path):
            all_passed = True
            for root, _, files in os.walk(self.target_path):
                for file in files:
                    if file.endswith(".py"):
                        file_path = os.path.join(root, file)
                        if not self.validate_file(file_path):
                            all_passed = False
            return all_passed
        return False

    def print_errors(self) -> None:
        """Print all validation errors"""
        if not self.errors:
            print("No validation errors found.")
            return
        print(f"Found {len(self.errors)} validation errors:")
        for file_path, error, line in self.errors:
            print(f"  {file_path}:{line} - {error}")

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python validate_ai_code.py ")
        sys.exit(1)
    validator = AICodeValidator(sys.argv[1])
    if validator.run_validation():
        logger.info("All files passed validation.")
        sys.exit(0)
    else:
        validator.print_errors()
        logger.error("Validation failed.")
        sys.exit(1)

def test_idempotency_key_collisions():
    """Test that idempotency keys do not collide under high throughput"""
    queue = FixedPaymentRetryQueue()
    payload = {"user_id": "usr_test", "amount": 10.0, "currency": "USD", "payment_method_id": "pm_test"}
    keys = []
    # Simulate 1000 enqueue requests in 100ms window
    for _ in range(1000):
        success, key = queue._enqueue_with_retry(payload)
        if success:
            keys.append(key)
    # Check for duplicate keys
    assert len(keys) == len(set(keys)), f"Found {len(keys) - len(set(keys))} duplicate idempotency keys"

# Regex pattern to detect second-precision timestamp in idempotency keys
IDEMPOTENCY_ANTI_PATTERN = re.compile(r"int\(time\.time\(\)\)")
# Check if pattern exists in source code
if IDEMPOTENCY_ANTI_PATTERN.search(source_code):
    errors.append("Weak idempotency key: uses second-precision timestamp")

apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: payment-retry-queue
spec:
  replicas: 10
  strategy:
    canary:
      steps:
        - setWeight: 1
        - pause: {duration: 30m}
        - setWeight: 10
        - pause: {duration: 30m}
        - setWeight: 50
        - pause: {duration: 15m}
        - setWeight: 100