Troubleshooting and Error Handling¶

{
  "timestamp": "2024-12-15T10:30:45Z",
  "level": "ERROR",
  "service": "payment-service",
  "environment": "production",
  "requestId": "req-abc-123",
  "correlationId": "trace-xyz-789",
  "message": "Payment processing failed",
  "sourceLocation": {
    "file": "payment_processor.py",
    "line": 145,
    "function": "process_payment"
  }
}

{
  "userId": "user_12345",
  "tenantId": "tenant_abc",
  "errorDetails": {
    "code": "PAYMENT_GATEWAY_ERROR",
    "type": "GatewayTimeoutError",
    "stack": "..."
  },
  "performance": {
    "duration": 2500,
    "memoryUsage": 128000000
  }
}

import threading
import uuid
import time
from typing import Dict, Any, Optional

class RequestContextManager:
    def __init__(self):
        # Thread-local storage for context
        self._context = threading.local()

    def set_context(self, **kwargs):
        """Set context values for the current request/thread"""
        if not hasattr(self._context, 'data'):
            self._context.data = {}
        self._context.data.update(kwargs)

    def get_context(self, key: Optional[str] = None, default=None):
        """Retrieve context values"""
        if not hasattr(self._context, 'data'):
            return default
        return self._context.data.get(key, default) if key else self._context.data

    def clear_context(self):
        """Clear context for the current request/thread"""
        if hasattr(self._context, 'data'):
            del self._context.data

    def create_context_decorator(self):
        """Decorator to manage context for function calls"""
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                request_id = str(uuid.uuid4())
                self.set_context(
                    request_id=request_id,
                    start_time=time.time()
                )
                try:
                    return func(*args, **kwargs)
                finally:
                    self.clear_context()
            return wrapper
        return decorator

# Global context manager
request_context = RequestContextManager()

# Usage example
@request_context.create_context_decorator()
def process_order(order_id):
    context = request_context.get_context()
    logger.info(f"Processing order {order_id}",
        extra={
            'request_id': context.get('request_id'),
            'start_time': context.get('start_time')
        })

import asyncio
import random
from typing import TypeVar, Callable, Awaitable
from dataclasses import dataclass

T = TypeVar('T')

@dataclass
class RetryConfig:
    max_attempts: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0
    jitter: float = 0.1

class RetryableError(Exception):
    """Errors that should trigger retry logic"""
    pass

async def retry_with_backoff(
    func: Callable[..., Awaitable[T]],
    config: RetryConfig = RetryConfig(),
    *args,
    **kwargs
) -> T:
    """
    Execute function with exponential backoff retry logic
    """
    attempt = 0
    delay = config.initial_delay

    while True:
        attempt += 1
        try:
            return await func(*args, **kwargs)

        except RetryableError as e:
            if attempt >= config.max_attempts:
                logger.error(
                    "Max retry attempts reached",
                    attempts=attempt,
                    error=str(e)
                )
                raise

            # Calculate delay with jitter
            jitter_amount = delay * config.jitter
            actual_delay = delay + random.uniform(-jitter_amount, jitter_amount)

            logger.warning(
                "Operation failed, retrying",
                attempt=attempt,
                delay=actual_delay,
                error=str(e)
            )

            await asyncio.sleep(actual_delay)

            # Increase delay for next attempt
            delay = min(
                delay * config.exponential_base,
                config.max_delay
            )

# Usage Example
async def fetch_user_data(user_id: str):
    """Fetch user data with automatic retry"""
    async def _fetch():
        async with aiohttp.ClientSession() as session:
            async with session.get(f'/api/users/{user_id}') as response:
                if response.status >= 500:
                    raise RetryableError(f"Server error: {response.status}")
                return await response.json()

    return await retry_with_backoff(_fetch)

from datetime import datetime, timedelta
from enum import Enum
from typing import Callable, TypeVar

T = TypeVar('T')

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreakerError(Exception):
    """Raised when circuit breaker is open"""
    pass

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        reset_timeout: float = 60.0,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.half_open_max_calls = half_open_max_calls

        self.state = CircuitState.CLOSED
        self.failures = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    async def call(
        self,
        func: Callable[..., Awaitable[T]],
        *args,
        **kwargs
    ) -> T:
        """Execute function through circuit breaker"""

        # Check if we should transition to HALF_OPEN
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitBreakerError("Circuit breaker is OPEN")

        # Check HALF_OPEN call limit
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.half_open_max_calls:
                raise CircuitBreakerError("Circuit breaker HALF_OPEN limit reached")
            self.half_open_calls += 1

        # Execute the function
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result

        except Exception as e:
            self._on_failure()
            raise e

    def _should_attempt_reset(self) -> bool:
        """Check if enough time has passed to attempt reset"""
        if not self.last_failure_time:
            return False

        elapsed = (datetime.now() - self.last_failure_time).total_seconds()
        return elapsed >= self.reset_timeout

    def _on_success(self):
        """Handle successful call"""
        if self.state == CircuitState.HALF_OPEN:
            # Success in HALF_OPEN -> back to CLOSED
            self.state = CircuitState.CLOSED
            self.failures = 0
            logger.info("Circuit breaker closed after successful test")

    def _on_failure(self):
        """Handle failed call"""
        self.failures += 1
        self.last_failure_time = datetime.now()

        if self.state == CircuitState.HALF_OPEN:
            # Failure in HALF_OPEN -> back to OPEN
            self.state = CircuitState.OPEN
            logger.warning("Circuit breaker reopened after failure in HALF_OPEN")

        elif self.failures >= self.failure_threshold:
            # Too many failures -> OPEN
            self.state = CircuitState.OPEN
            logger.error(
                "Circuit breaker opened",
                failures=self.failures,
                threshold=self.failure_threshold
            )

# Usage Example
payment_circuit = CircuitBreaker(
    failure_threshold=5,
    reset_timeout=30.0,
    half_open_max_calls=3
)

async def process_payment(order_id: str):
    """Process payment with circuit breaker protection"""
    try:
        return await payment_circuit.call(
            payment_service.charge,
            order_id
        )
    except CircuitBreakerError:
        logger.warning("Payment service unavailable (circuit open)")
        # Return to fallback or queue for later
        await queue_payment_for_retry(order_id)
        raise ServiceUnavailableError("Payment processing temporarily unavailable")

from enum import Enum
from typing import Optional, Callable, Any

class FeatureState(Enum):
    ENABLED = "enabled"
    DISABLED = "disabled"
    DEGRADED = "degraded"

class FeatureFlag:
    def __init__(
        self,
        name: str,
        default_state: FeatureState = FeatureState.ENABLED,
        fallback_function: Optional[Callable] = None
    ):
        self.name = name
        self.state = default_state
        self.fallback = fallback_function
        self.error_count = 0
        self.last_error_time = None

    def should_degrade(self) -> bool:
        """Check if feature should enter degraded mode"""
        # Degrade after 5 errors
        if self.error_count >= 5:
            return True

        # Degrade if recent errors
        if self.last_error_time:
            time_since_error = (
                datetime.now() - self.last_error_time
            ).total_seconds()
            return time_since_error < 60

        return False

    async def execute(
        self,
        main_function: Callable,
        *args,
        **kwargs
    ) -> Any:
        """Execute feature with fallback support"""

        if self.state == FeatureState.DISABLED:
            raise FeatureDisabledError(f"Feature {self.name} is disabled")

        try:
            # Use fallback if degraded
            if self.should_degrade() and self.fallback:
                logger.warning(
                    f"Feature {self.name} degraded, using fallback"
                )
                return await self.fallback(*args, **kwargs)

            # Normal execution
            result = await main_function(*args, **kwargs)

            # Reset error count on success
            if self.error_count > 0:
                self.error_count = max(0, self.error_count - 1)

            return result

        except Exception as e:
            self.error_count += 1
            self.last_error_time = datetime.now()

            # Try fallback if available
            if self.fallback:
                logger.error(
                    f"Error in feature {self.name}, using fallback",
                    error=str(e)
                )
                return await self.fallback(*args, **kwargs)

            raise

# Usage Example
async def get_user_recommendations(user_id: str):
    """Get personalized recommendations with fallback"""

    # Fallback: popular items instead of personalized
    async def fallback_recommendations(user_id: str):
        return await get_popular_items(limit=10)

    recommendations_feature = FeatureFlag(
        name="user_recommendations",
        fallback_function=fallback_recommendations
    )

    async def get_ml_recommendations(user_id: str):
        # Complex ML-based recommendation logic
        return await ml_service.get_recommendations(user_id)

    return await recommendations_feature.execute(
        get_ml_recommendations,
        user_id
    )

{
  "error": {
    "code": "PAYMENT_FAILED",
    "message": "Unable to process payment",
    "details": {
      "reason": "Insufficient funds",
      "action": "Please add funds and try again"
    },
    "request_id": "req_abc123",
    "timestamp": "2025-10-23T14:30:00Z"
  }
}

from typing import Dict, Any, Optional
from datetime import datetime

class ErrorResponseBuilder:
    @staticmethod
    def build_response(
        error: ApplicationError,
        request_id: str,
        include_details: bool = True
    ) -> Dict[str, Any]:
        """Build standardized error response"""

        response = {
            "error": {
                "code": error.error_code,
                "message": error.message,
                "request_id": request_id,
                "timestamp": datetime.utcnow().isoformat()
            }
        }

        # Only include details if appropriate
        if include_details and error.details:
            # Sanitize details for user consumption
            response["error"]["details"] = ErrorResponseBuilder._sanitize_details(
                error.details
            )

        return response

    @staticmethod
    def _sanitize_details(details: Dict) -> Dict:
        """Remove sensitive information from error details"""
        safe_keys = {'reason', 'action', 'field', 'constraint'}
        return {
            k: v for k, v in details.items()
            if k in safe_keys
        }

# Usage in API handler
@app.exception_handler(ApplicationError)
async def handle_application_error(request, exc: ApplicationError):
    response = ErrorResponseBuilder.build_response(
        error=exc,
        request_id=request.state.request_id,
        include_details=True
    )

    # Log internally with full context
    logger.error(
        "Application error occurred",
        error_code=exc.error_code,
        category=exc.category.value,
        severity=exc.severity.value,
        details=exc.details,
        request_id=request.state.request_id
    )

    # Determine HTTP status code
    status_code = {
        ErrorCategory.VALIDATION: 400,
        ErrorCategory.BUSINESS_LOGIC: 422,
        ErrorCategory.SECURITY: 403,
        ErrorCategory.INTEGRATION: 503,
        ErrorCategory.INFRASTRUCTURE: 503
    }.get(exc.category, 500)

    return JSONResponse(
        content=response,
        status_code=status_code
    )

def process_order(order: Order) -> OrderResult:
    # Breakpoint 1: Verify input
    validate_order(order)  # Check validation logic

    # Breakpoint 2: Before external call
    result = payment_processor.charge(order)  # Check before payment

    # Breakpoint 3: After business logic
    updated_order = update_order_status(result)  # Verify state change

    # Breakpoint 4: Before return
    return create_response(updated_order)  # Check final output

# Break only when specific conditions are met
for item in items:
    # Condition: item.price > 1000
    process_item(item)  # Breakpoint here with condition

# Condition: user_id == "debug_user"
user_data = fetch_user(user_id)  # Breakpoint with user condition

// Group related logs
console.group('User Authentication');
console.log('Username:', username);
console.time('loginDuration');

// Display data in table format
console.table(userData);

// Show call stack
console.trace('Auth flow');

console.timeEnd('loginDuration');
console.groupEnd();

// Conditional logging
console.assert(user.isValid, 'User validation failed', user);

// Count occurrences
console.count('API calls');
console.countReset('API calls');

// Performance markers
performance.mark('startOperation');
// ... operation code ...
performance.mark('endOperation');
performance.measure('operationDuration', 'startOperation', 'endOperation');

// Enhanced fetch with debugging
async function debugFetch(url, options = {}) {
  console.group(`${options.method || 'GET'} ${url}`);
  console.time('request');

  try {
    const response = await fetch(url, options);

    console.log('Status:', response.status);
    console.log('Headers:', Object.fromEntries(response.headers));

    const data = await response.json();
    console.table(data);

    console.timeEnd('request');
    console.groupEnd();

    return data;
  } catch (error) {
    console.error('Request failed:', error);
    console.timeEnd('request');
    console.groupEnd();
    throw error;
  }
}

// Usage
const userData = await debugFetch('/api/user/123');

// Take heap snapshots
if (window.performance.memory) {
  console.log('Memory usage:', {
    totalJSHeapSize: window.performance.memory.totalJSHeapSize / 1048576 + ' MB',
    usedJSHeapSize: window.performance.memory.usedJSHeapSize / 1048576 + ' MB',
    jsHeapSizeLimit: window.performance.memory.jsHeapSizeLimit / 1048576 + ' MB'
  });
}

// Profile function execution
console.profile('heavyComputation');
heavyComputation();
console.profileEnd('heavyComputation');

import logging
import contextvars
import time
from typing import Optional, Dict, Any

request_id = contextvars.ContextVar('request_id', default=None)

class DiagnosticLogger:
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.start_time = None

    def start_operation(self, operation_name: str, **context):
        """Start timing an operation with context"""
        self.start_time = time.time()
        self.logger.debug(
            f"Starting {operation_name}",
            extra={
                'operation': operation_name,
                'request_id': request_id.get(),
                **context
            }
        )

    def end_operation(self, operation_name: str, **context):
        """End timing with results"""
        if self.start_time:
            duration = time.time() - self.start_time
            self.logger.debug(
                f"Completed {operation_name}",
                extra={
                    'operation': operation_name,
                    'duration_ms': duration * 1000,
                    'request_id': request_id.get(),
                    **context
                }
            )

    def debug_state(self, obj: object, attributes: list):
        """Log object state for debugging"""
        state = {
            attr: getattr(obj, attr, None)
            for attr in attributes
        }
        self.logger.debug(
            f"State: {obj.__class__.__name__}",
            extra={
                'object_type': obj.__class__.__name__,
                'state': state,
                'request_id': request_id.get()
            }
        )

# Usage Example
logger = DiagnosticLogger(__name__)

def process_payment(payment_data: Dict):
    logger.start_operation('payment_processing', 
                          amount=payment_data['amount'])

    try:
        # Processing logic
        result = charge_payment(payment_data)
        logger.debug_state(result, ['status', 'transaction_id', 'amount'])

        logger.end_operation('payment_processing',
                           status='success',
                           transaction_id=result.transaction_id)
        return result

    except Exception as e:
        logger.end_operation('payment_processing',
                           status='failed',
                           error=str(e))
        raise

import debugpy
from contextlib import contextmanager
import logging
from datetime import datetime, timedelta

class SecureRemoteDebugger:
    def __init__(self, host: str = 'localhost', port: int = 5678):
        self.host = host
        self.port = port
        self.logger = logging.getLogger(__name__)
        self.session_timeout = timedelta(minutes=30)

    def setup(self, require_auth: bool = True):
        """Configure remote debugging with security"""
        try:
            if require_auth:
                # Implement authentication here
                if not self._authenticate():
                    raise PermissionError("Authentication required")

            debugpy.listen((self.host, self.port))
            self.logger.info(
                f"Debug server listening on {self.host}:{self.port}"
            )

            # Log security event
            self._audit_log("debug_server_started")

        except Exception as e:
            self.logger.error(f"Failed to start debug server: {e}")
            raise

    @contextmanager
    def debug_session(self, session_id: str, timeout: int = 300):
        """Create temporary debugging session with timeout"""
        try:
            self.logger.info(f"Starting debug session: {session_id}")
            self._audit_log("debug_session_started", session_id=session_id)

            # Wait for debugger with timeout
            debugpy.wait_for_client(timeout)

            yield

        finally:
            self.logger.info(f"Ending debug session: {session_id}")
            self._audit_log("debug_session_ended", session_id=session_id)
            debugpy.disconnect()

    def _authenticate(self) -> bool:
        """Authenticate debug session"""
        # Implement actual authentication logic
        return True

    def _audit_log(self, event: str, **context):
        """Log security audit events"""
        self.logger.info(
            f"Security Audit: {event}",
            extra={
                'event_type': 'security_audit',
                'event': event,
                'timestamp': datetime.utcnow().isoformat(),
                **context
            }
        )

# Usage Example (with feature flag)
if feature_flags.is_enabled('remote_debugging'):
    debugger = SecureRemoteDebugger()
    debugger.setup(require_auth=True)

    with debugger.debug_session('incident_investigation_123'):
        # Debug critical operation
        investigate_issue()

import cProfile
import pstats
import io
from functools import wraps
from typing import Callable, Any

class CodeProfiler:
    def __init__(self, enabled: bool = True):
        self.enabled = enabled
        self.profiler = cProfile.Profile()

    def profile(self, func: Callable) -> Callable:
        """Decorator for profiling functions"""
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            if not self.enabled:
                return func(*args, **kwargs)

            self.profiler.enable()
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                self.profiler.disable()
                self._print_stats()

        return wrapper

    def _print_stats(self):
        """Print profiling statistics"""
        s = io.StringIO()
        stats = pstats.Stats(self.profiler, stream=s)
        stats.sort_stats('cumulative')
        stats.print_stats(20)  # Top 20 functions

        print("\n" + "="*80)
        print("PROFILING RESULTS")
        print("="*80)
        print(s.getvalue())

# Usage Example
profiler = CodeProfiler()

@profiler.profile
def expensive_operation(data):
    # Complex computation
    result = process_large_dataset(data)
    return result