> ## Documentation Index
> Fetch the complete documentation index at: https://hyrex.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Error Handling & Recovery
> Implement robust error handling with retry logic and monitoring
Build resilient systems with comprehensive error handling, intelligent retry mechanisms, circuit breakers, and detailed error monitoring.
## Overview
This example demonstrates robust error handling patterns:
* **Comprehensive error classification** - Differentiate between retriable and non-retriable errors
* **Intelligent retry strategies** - Exponential backoff, circuit breakers, and custom retry logic
* **Error monitoring & alerting** - Track error rates, patterns, and system health
* **Graceful degradation** - Fallback mechanisms when services are unavailable
* **Error aggregation** - Collect and analyze error patterns for improvement
* **Recovery workflows** - Automatic and manual recovery processes
Perfect for building fault-tolerant systems, API integrations, or any application requiring high reliability.
## Task Definitions
```python Python theme={null}
from hyrex import HyrexRegistry, HyrexKV, get_hyrex_context
from pydantic import BaseModel
from typing import Dict, Any, Optional, List
import traceback
import logging
import time
import json
from datetime import datetime, timedelta
from enum import Enum
hy = HyrexRegistry()
logger = logging.getLogger(__name__)
class ErrorSeverity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class ErrorCategory(Enum):
VALIDATION = "validation"
NETWORK = "network"
DATABASE = "database"
EXTERNAL_API = "external_api"
BUSINESS_LOGIC = "business_logic"
SYSTEM = "system"
class MonitoredTaskContext(BaseModel):
operation: str
data: Dict[str, Any]
retry_config: Optional[Dict[str, Any]] = None
circuit_breaker_key: Optional[str] = None
class TaskError(Exception):
def __init__(self, message: str, category: ErrorCategory, severity: ErrorSeverity,
retriable: bool = True, details: Optional[Dict] = None):
super().__init__(message)
self.category = category
self.severity = severity
self.retriable = retriable
self.details = details or {}
class ValidationError(TaskError):
def __init__(self, message: str, details: Optional[Dict] = None):
super().__init__(message, ErrorCategory.VALIDATION, ErrorSeverity.MEDIUM, False, details)
class ExternalAPIError(TaskError):
def __init__(self, message: str, status_code: int, details: Optional[Dict] = None):
severity = ErrorSeverity.HIGH if status_code >= 500 else ErrorSeverity.MEDIUM
retriable = status_code >= 500 or status_code in [429, 408] # Server errors and rate limits
super().__init__(message, ErrorCategory.EXTERNAL_API, severity, retriable,
{**(details or {}), "status_code": status_code})
@hy.task(max_retries=5, retry_backoff=True)
def monitored_task(context: MonitoredTaskContext):
"""Task with comprehensive error handling and monitoring"""
start_time = time.time()
task_context = get_hyrex_context()
operation = context.operation
# Check circuit breaker if configured
if context.circuit_breaker_key:
circuit_state = check_circuit_breaker(context.circuit_breaker_key)
if circuit_state == "OPEN":
raise TaskError(
f"Circuit breaker OPEN for {context.circuit_breaker_key}",
ErrorCategory.SYSTEM,
ErrorSeverity.HIGH,
retriable=False
)
try:
# Track task execution metrics
update_task_metrics(operation, "started")
# Perform the actual operation
result = perform_operation(context.operation, context.data)
# Record success metrics
duration = time.time() - start_time
record_task_success(operation, duration, task_context)
# Reset circuit breaker on success
if context.circuit_breaker_key:
reset_circuit_breaker(context.circuit_breaker_key)
return {
"success": True,
"result": result,
"duration": duration,
"operation": operation
}
except ValidationError as e:
# Don't retry validation errors - log and return error response
error_details = log_task_error(operation, e, task_context, retriable=False)
return {
"success": False,
"error_type": "validation_error",
"error_message": str(e),
"details": e.details,
"error_id": error_details["error_id"]
}
except ExternalAPIError as e:
# Handle external API errors with circuit breaker logic
if context.circuit_breaker_key:
record_circuit_breaker_failure(context.circuit_breaker_key)
error_details = log_task_error(operation, e, task_context, retriable=e.retriable)
if e.retriable:
# Let Hyrex handle the retry
raise e
else:
return {
"success": False,
"error_type": "external_api_error",
"error_message": str(e),
"status_code": e.details.get("status_code"),
"error_id": error_details["error_id"]
}
except Exception as e:
# Handle unexpected errors
error_details = log_task_error(operation, e, task_context, retriable=True)
# Record for circuit breaker if configured
if context.circuit_breaker_key:
record_circuit_breaker_failure(context.circuit_breaker_key)
# Send critical error alert
if isinstance(e, TaskError) and e.severity == ErrorSeverity.CRITICAL:
send_critical_error_alert.send(operation, str(e), error_details)
raise e # Let Hyrex handle retries
@hy.task(max_retries=3)
def resilient_external_api_call(url: str, method: str = "GET", data: Optional[Dict] = None,
timeout: int = 30, circuit_breaker_key: Optional[str] = None):
"""Make external API call with comprehensive error handling"""
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
# Check circuit breaker
if circuit_breaker_key:
circuit_state = check_circuit_breaker(circuit_breaker_key)
if circuit_state == "OPEN":
raise ExternalAPIError("Service unavailable - circuit breaker OPEN", 503)
# Configure requests with retry strategy
session = requests.Session()
retry_strategy = Retry(
total=3,
status_forcelist=[429, 500, 502, 503, 504],
method_whitelist=["HEAD", "GET", "OPTIONS"],
backoff_factor=1
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
try:
response = session.request(
method=method,
url=url,
json=data,
timeout=timeout,
headers={"User-Agent": "HyrexApp/1.0"}
)
# Check for HTTP errors
if response.status_code >= 400:
raise ExternalAPIError(
f"API request failed with status {response.status_code}",
response.status_code,
{"url": url, "response_body": response.text[:1000]}
)
# Reset circuit breaker on success
if circuit_breaker_key:
reset_circuit_breaker(circuit_breaker_key)
return {
"status_code": response.status_code,
"data": response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text,
"headers": dict(response.headers)
}
except requests.exceptions.Timeout:
if circuit_breaker_key:
record_circuit_breaker_failure(circuit_breaker_key)
raise ExternalAPIError("Request timeout", 408, {"url": url, "timeout": timeout})
except requests.exceptions.ConnectionError as e:
if circuit_breaker_key:
record_circuit_breaker_failure(circuit_breaker_key)
raise ExternalAPIError("Connection failed", 503, {"url": url, "error": str(e)})
except requests.exceptions.RequestException as e:
if circuit_breaker_key:
record_circuit_breaker_failure(circuit_breaker_key)
raise ExternalAPIError(f"Request failed: {str(e)}", 500, {"url": url})
@hy.task
def error_recovery_workflow(error_id: str, recovery_strategy: str):
"""Handle error recovery workflows"""
# Get error details
error_details = get_error_details(error_id)
if not error_details:
raise ValueError(f"Error {error_id} not found")
try:
if recovery_strategy == "retry_with_fallback":
# Retry original operation with fallback data
fallback_data = get_fallback_data(error_details["operation"])
result = perform_operation(error_details["operation"], fallback_data)
elif recovery_strategy == "manual_intervention":
# Create ticket for manual intervention
ticket_id = create_support_ticket(error_details)
result = {"ticket_id": ticket_id, "status": "manual_intervention_required"}
elif recovery_strategy == "alternative_workflow":
# Execute alternative workflow
result = execute_alternative_workflow(error_details)
else:
raise ValueError(f"Unknown recovery strategy: {recovery_strategy}")
# Mark error as recovered
mark_error_recovered(error_id, recovery_strategy, result)
return {
"error_id": error_id,
"recovery_strategy": recovery_strategy,
"recovered": True,
"result": result
}
except Exception as e:
# Recovery failed
mark_recovery_failed(error_id, recovery_strategy, str(e))
raise TaskError(
f"Recovery failed for error {error_id}: {str(e)}",
ErrorCategory.SYSTEM,
ErrorSeverity.HIGH
)
def log_task_error(operation: str, error: Exception, task_context: Any, retriable: bool = True) -> Dict:
"""Log detailed error information"""
error_id = f"err_{int(time.time())}_{hash(str(error)) % 10000}"
error_data = {
"error_id": error_id,
"operation": operation,
"error_type": type(error).__name__,
"error_message": str(error),
"retriable": retriable,
"timestamp": datetime.now().isoformat(),
"task_id": getattr(task_context, 'task_id', 'unknown'),
"attempt_number": getattr(task_context, 'attempt_number', 0),
"traceback": traceback.format_exc()
}
# Add error-specific details
if isinstance(error, TaskError):
error_data.update({
"category": error.category.value,
"severity": error.severity.value,
"details": error.details
})
# Store error for analysis
HyrexKV.set(
f"error:{error_id}",
json.dumps(error_data),
expiry_seconds=604800 # 7 days
)
# Update error metrics
update_error_metrics(operation, error_data)
# Log to application logs
logger.error(f"Task error in {operation}: {error_data}")
return error_data
def update_error_metrics(operation: str, error_data: Dict):
"""Update error tracking metrics"""
# Update error counts
error_key = f"error_metrics:{operation}"
try:
metrics_data = HyrexKV.get(error_key)
metrics = json.loads(metrics_data) if metrics_data else {
"total_errors": 0,
"error_types": {},
"error_categories": {},
"hourly_counts": {}
}
except:
metrics = {"total_errors": 0, "error_types": {}, "error_categories": {}, "hourly_counts": {}}
# Update counters
metrics["total_errors"] += 1
error_type = error_data["error_type"]
metrics["error_types"][error_type] = metrics["error_types"].get(error_type, 0) + 1
if "category" in error_data:
category = error_data["category"]
metrics["error_categories"][category] = metrics["error_categories"].get(category, 0) + 1
# Update hourly counts for trend analysis
hour_key = datetime.now().strftime("%Y-%m-%d-%H")
metrics["hourly_counts"][hour_key] = metrics["hourly_counts"].get(hour_key, 0) + 1
# Store updated metrics
HyrexKV.set(error_key, json.dumps(metrics), expiry_seconds=2592000) # 30 days
def check_circuit_breaker(key: str) -> str:
"""Check circuit breaker state"""
cb_key = f"circuit_breaker:{key}"
try:
cb_data = HyrexKV.get(cb_key)
if not cb_data:
return "CLOSED"
cb_state = json.loads(cb_data)
# Check if circuit breaker should be reset
if cb_state["state"] == "OPEN":
if time.time() - cb_state["opened_at"] > cb_state.get("timeout", 60):
cb_state["state"] = "HALF_OPEN"
HyrexKV.set(cb_key, json.dumps(cb_state), expiry_seconds=3600)
return cb_state["state"]
except:
return "CLOSED"
def record_circuit_breaker_failure(key: str):
"""Record a failure for circuit breaker logic"""
cb_key = f"circuit_breaker:{key}"
try:
cb_data = HyrexKV.get(cb_key)
if cb_data:
cb_state = json.loads(cb_data)
else:
cb_state = {
"failures": 0,
"state": "CLOSED",
"threshold": 5,
"timeout": 60
}
cb_state["failures"] += 1
cb_state["last_failure"] = time.time()
# Open circuit breaker if threshold exceeded
if cb_state["failures"] >= cb_state["threshold"] and cb_state["state"] != "OPEN":
cb_state["state"] = "OPEN"
cb_state["opened_at"] = time.time()
# Send circuit breaker alert
send_circuit_breaker_alert.send(key, cb_state)
HyrexKV.set(cb_key, json.dumps(cb_state), expiry_seconds=3600)
except Exception as e:
logger.error(f"Failed to update circuit breaker {key}: {e}")
def reset_circuit_breaker(key: str):
"""Reset circuit breaker on successful operation"""
cb_key = f"circuit_breaker:{key}"
try:
cb_data = HyrexKV.get(cb_key)
if cb_data:
cb_state = json.loads(cb_data)
cb_state["failures"] = 0
cb_state["state"] = "CLOSED"
HyrexKV.set(cb_key, json.dumps(cb_state), expiry_seconds=3600)
except:
pass
@hy.task
def send_critical_error_alert(operation: str, error_message: str, error_details: Dict):
"""Send alert for critical errors"""
alert_message = f"""
🚨 CRITICAL ERROR ALERT 🚨
Operation: {operation}
Error: {error_message}
Error ID: {error_details.get('error_id')}
Task ID: {error_details.get('task_id')}
Time: {error_details.get('timestamp')}
Immediate attention required!
"""
# Send to multiple channels
send_slack_alert(alert_message, channel="#alerts")
send_email_alert("Critical System Error", alert_message, ["oncall@company.com"])
return {"alert_sent": True, "error_id": error_details.get('error_id')}
@hy.task
def send_circuit_breaker_alert(key: str, cb_state: Dict):
"""Send alert when circuit breaker opens"""
alert_message = f"""
âš¡ Circuit Breaker OPEN âš¡
Service: {key}
Failure Count: {cb_state['failures']}
Threshold: {cb_state['threshold']}
Opened At: {datetime.fromtimestamp(cb_state['opened_at']).isoformat()}
Service calls are being blocked to prevent cascade failures.
"""
send_slack_alert(alert_message, channel="#infrastructure")
return {"alert_sent": True, "circuit_breaker": key}
@hy.task
def generate_error_report(operation: Optional[str] = None, hours: int = 24):
"""Generate comprehensive error analysis report"""
end_time = datetime.now()
start_time = end_time - timedelta(hours=hours)
# Collect error data
if operation:
operations = [operation]
else:
operations = get_all_monitored_operations()
report_data = {
"period": {
"start": start_time.isoformat(),
"end": end_time.isoformat(),
"hours": hours
},
"operations": {},
"summary": {
"total_errors": 0,
"most_common_errors": {},
"error_trends": {},
"recovery_stats": {}
}
}
for op in operations:
error_metrics = get_error_metrics(op, start_time, end_time)
if error_metrics:
report_data["operations"][op] = error_metrics
report_data["summary"]["total_errors"] += error_metrics.get("total_errors", 0)
# Store report
report_id = f"error_report_{int(time.time())}"
HyrexKV.set(
f"report:{report_id}",
json.dumps(report_data),
expiry_seconds=2592000 # 30 days
)
# Send report to stakeholders if errors are above threshold
if report_data["summary"]["total_errors"] > 10:
send_error_report_email.send(report_id, report_data)
return {
"report_id": report_id,
"total_errors": report_data["summary"]["total_errors"],
"operations_analyzed": len(operations)
}
# Helper functions (simplified implementations)
def perform_operation(operation: str, data: Dict) -> Any:
"""Perform the actual operation - placeholder"""
if operation == "test_validation_error":
raise ValidationError("Invalid input data", {"field": "email"})
elif operation == "test_api_error":
raise ExternalAPIError("API service unavailable", 503)
return {"success": True, "data": data}
def update_task_metrics(operation: str, status: str):
"""Update task execution metrics"""
pass
def record_task_success(operation: str, duration: float, task_context: Any):
"""Record successful task execution"""
pass
def get_fallback_data(operation: str) -> Dict:
"""Get fallback data for recovery"""
return {"fallback": True}
def execute_alternative_workflow(error_details: Dict) -> Dict:
"""Execute alternative workflow"""
return {"alternative_executed": True}
```
```typescript TypeScript theme={null}
import { HyrexRegistry, HyrexKV, getHyrexContext } from 'hyrex';
import { z } from 'zod';
import axios, { AxiosError } from 'axios';
const hy = new HyrexRegistry();
enum ErrorSeverity {
LOW = "low",
MEDIUM = "medium",
HIGH = "high",
CRITICAL = "critical"
}
enum ErrorCategory {
VALIDATION = "validation",
NETWORK = "network",
DATABASE = "database",
EXTERNAL_API = "external_api",
BUSINESS_LOGIC = "business_logic",
SYSTEM = "system"
}
class TaskError extends Error {
constructor(
message: string,
public category: ErrorCategory,
public severity: ErrorSeverity,
public retriable: boolean = true,
public details: Record = {}
) {
super(message);
this.name = 'TaskError';
}
}
class ValidationError extends TaskError {
constructor(message: string, details: Record = {}) {
super(message, ErrorCategory.VALIDATION, ErrorSeverity.MEDIUM, false, details);
this.name = 'ValidationError';
}
}
class ExternalAPIError extends TaskError {
constructor(message: string, statusCode: number, details: Record = {}) {
const severity = statusCode >= 500 ? ErrorSeverity.HIGH : ErrorSeverity.MEDIUM;
const retriable = statusCode >= 500 || [429, 408].includes(statusCode);
super(message, ErrorCategory.EXTERNAL_API, severity, retriable,
{ ...details, statusCode });
this.name = 'ExternalAPIError';
}
}
const MonitoredTaskContextSchema = z.object({
operation: z.string(),
data: z.record(z.any()),
retryConfig: z.record(z.any()).optional(),
circuitBreakerKey: z.string().optional()
});
type MonitoredTaskContext = z.infer;
const monitoredTask = hy.task({
name: 'monitoredTask',
config: {
maxRetries: 5,
retryBackoff: true
},
argSchema: MonitoredTaskContextSchema,
func: async (context: MonitoredTaskContext) => {
const startTime = Date.now();
const taskContext = getHyrexContext();
const { operation } = context;
// Check circuit breaker if configured
if (context.circuitBreakerKey) {
const circuitState = await checkCircuitBreaker(context.circuitBreakerKey);
if (circuitState === "OPEN") {
throw new TaskError(
`Circuit breaker OPEN for ${context.circuitBreakerKey}`,
ErrorCategory.SYSTEM,
ErrorSeverity.HIGH,
false
);
}
}
try {
// Track task execution metrics
await updateTaskMetrics(operation, "started");
// Perform the actual operation
const result = await performOperation(context.operation, context.data);
// Record success metrics
const duration = (Date.now() - startTime) / 1000;
await recordTaskSuccess(operation, duration, taskContext);
// Reset circuit breaker on success
if (context.circuitBreakerKey) {
await resetCircuitBreaker(context.circuitBreakerKey);
}
return {
success: true,
result,
duration,
operation
};
} catch (error: any) {
if (error instanceof ValidationError) {
// Don't retry validation errors
const errorDetails = await logTaskError(operation, error, taskContext, false);
return {
success: false,
errorType: "validation_error",
errorMessage: error.message,
details: error.details,
errorId: errorDetails.errorId
};
}
if (error instanceof ExternalAPIError) {
// Handle external API errors with circuit breaker logic
if (context.circuitBreakerKey) {
await recordCircuitBreakerFailure(context.circuitBreakerKey);
}
const errorDetails = await logTaskError(operation, error, taskContext, error.retriable);
if (error.retriable) {
// Let Hyrex handle the retry
throw error;
} else {
return {
success: false,
errorType: "external_api_error",
errorMessage: error.message,
statusCode: error.details.statusCode,
errorId: errorDetails.errorId
};
}
}
// Handle unexpected errors
const errorDetails = await logTaskError(operation, error, taskContext, true);
// Record for circuit breaker if configured
if (context.circuitBreakerKey) {
await recordCircuitBreakerFailure(context.circuitBreakerKey);
}
// Send critical error alert
if (error instanceof TaskError && error.severity === ErrorSeverity.CRITICAL) {
await sendCriticalErrorAlert.send(operation, error.message, errorDetails);
}
throw error; // Let Hyrex handle retries
}
}
});
const resilientExternalApiCall = hy.task({
name: 'resilientExternalApiCall',
config: { maxRetries: 3 },
func: async (
url: string,
method: string = "GET",
data?: Record,
timeout: number = 30000,
circuitBreakerKey?: string
) => {
// Check circuit breaker
if (circuitBreakerKey) {
const circuitState = await checkCircuitBreaker(circuitBreakerKey);
if (circuitState === "OPEN") {
throw new ExternalAPIError("Service unavailable - circuit breaker OPEN", 503);
}
}
try {
const response = await axios({
method: method as any,
url,
data,
timeout,
headers: { "User-Agent": "HyrexApp/1.0" }
});
// Reset circuit breaker on success
if (circuitBreakerKey) {
await resetCircuitBreaker(circuitBreakerKey);
}
return {
statusCode: response.status,
data: response.data,
headers: response.headers
};
} catch (error: any) {
if (circuitBreakerKey) {
await recordCircuitBreakerFailure(circuitBreakerKey);
}
if (error.code === 'ECONNABORTED') {
throw new ExternalAPIError("Request timeout", 408, { url, timeout });
} else if (error.code === 'ECONNREFUSED' || error.code === 'ENOTFOUND') {
throw new ExternalAPIError("Connection failed", 503, { url, error: error.message });
} else if (error.response) {
throw new ExternalAPIError(
`API request failed with status ${error.response.status}`,
error.response.status,
{ url, responseBody: error.response.data }
);
} else {
throw new ExternalAPIError(`Request failed: ${error.message}`, 500, { url });
}
}
}
});
const errorRecoveryWorkflow = hy.task({
name: 'errorRecoveryWorkflow',
func: async (errorId: string, recoveryStrategy: string) => {
// Get error details
const errorDetails = await getErrorDetails(errorId);
if (!errorDetails) {
throw new Error(`Error ${errorId} not found`);
}
try {
let result: any;
switch (recoveryStrategy) {
case "retry_with_fallback":
// Retry original operation with fallback data
const fallbackData = await getFallbackData(errorDetails.operation);
result = await performOperation(errorDetails.operation, fallbackData);
break;
case "manual_intervention":
// Create ticket for manual intervention
const ticketId = await createSupportTicket(errorDetails);
result = { ticketId, status: "manual_intervention_required" };
break;
case "alternative_workflow":
// Execute alternative workflow
result = await executeAlternativeWorkflow(errorDetails);
break;
default:
throw new Error(`Unknown recovery strategy: ${recoveryStrategy}`);
}
// Mark error as recovered
await markErrorRecovered(errorId, recoveryStrategy, result);
return {
errorId,
recoveryStrategy,
recovered: true,
result
};
} catch (error: any) {
// Recovery failed
await markRecoveryFailed(errorId, recoveryStrategy, error.message);
throw new TaskError(
`Recovery failed for error ${errorId}: ${error.message}`,
ErrorCategory.SYSTEM,
ErrorSeverity.HIGH
);
}
}
});
const sendCriticalErrorAlert = hy.task({
name: 'sendCriticalErrorAlert',
func: async (operation: string, errorMessage: string, errorDetails: any) => {
const alertMessage = `
🚨 CRITICAL ERROR ALERT 🚨
Operation: ${operation}
Error: ${errorMessage}
Error ID: ${errorDetails.errorId}
Task ID: ${errorDetails.taskId}
Time: ${errorDetails.timestamp}
Immediate attention required!
`;
// Send to multiple channels
await sendSlackAlert(alertMessage, "#alerts");
await sendEmailAlert("Critical System Error", alertMessage, ["oncall@company.com"]);
return { alertSent: true, errorId: errorDetails.errorId };
}
});
const generateErrorReport = hy.task({
name: 'generateErrorReport',
func: async (operation?: string, hours: number = 24) => {
const endTime = new Date();
const startTime = new Date(endTime.getTime() - hours * 60 * 60 * 1000);
// Collect error data
const operations = operation ? [operation] : await getAllMonitoredOperations();
const reportData = {
period: {
start: startTime.toISOString(),
end: endTime.toISOString(),
hours
},
operations: {} as Record,
summary: {
totalErrors: 0,
mostCommonErrors: {},
errorTrends: {},
recoveryStats: {}
}
};
for (const op of operations) {
const errorMetrics = await getErrorMetrics(op, startTime, endTime);
if (errorMetrics) {
reportData.operations[op] = errorMetrics;
reportData.summary.totalErrors += errorMetrics.totalErrors || 0;
}
}
// Store report
const reportId = `error_report_${Date.now()}`;
await HyrexKV.set(
`report:${reportId}`,
JSON.stringify(reportData),
2592000 // 30 days
);
// Send report to stakeholders if errors are above threshold
if (reportData.summary.totalErrors > 10) {
await sendErrorReportEmail.send(reportId, reportData);
}
return {
reportId,
totalErrors: reportData.summary.totalErrors,
operationsAnalyzed: operations.length
};
}
});
async function logTaskError(
operation: string,
error: Error,
taskContext: any,
retriable: boolean = true
): Promise {
const errorId = `err_${Date.now()}_${Math.floor(Math.random() * 10000)}`;
const errorData = {
errorId,
operation,
errorType: error.constructor.name,
errorMessage: error.message,
retriable,
timestamp: new Date().toISOString(),
taskId: taskContext.taskId || 'unknown',
attemptNumber: taskContext.attemptNumber || 0,
stack: error.stack
};
// Add error-specific details
if (error instanceof TaskError) {
Object.assign(errorData, {
category: error.category,
severity: error.severity,
details: error.details
});
}
// Store error for analysis
await HyrexKV.set(
`error:${errorId}`,
JSON.stringify(errorData),
604800 // 7 days
);
// Update error metrics
await updateErrorMetrics(operation, errorData);
// Log to console
console.error(`Task error in ${operation}:`, errorData);
return errorData;
}
async function updateErrorMetrics(operation: string, errorData: any) {
const errorKey = `error_metrics:${operation}`;
try {
const metricsData = await HyrexKV.get(errorKey);
const metrics = metricsData ? JSON.parse(metricsData) : {
totalErrors: 0,
errorTypes: {},
errorCategories: {},
hourlyCounts: {}
};
// Update counters
metrics.totalErrors += 1;
const errorType = errorData.errorType;
metrics.errorTypes[errorType] = (metrics.errorTypes[errorType] || 0) + 1;
if (errorData.category) {
metrics.errorCategories[errorData.category] = (metrics.errorCategories[errorData.category] || 0) + 1;
}
// Update hourly counts for trend analysis
const hourKey = new Date().toISOString().slice(0, 13); // YYYY-MM-DDTHH
metrics.hourlyCounts[hourKey] = (metrics.hourlyCounts[hourKey] || 0) + 1;
await HyrexKV.set(errorKey, JSON.stringify(metrics), 2592000); // 30 days
} catch (error) {
console.error('Failed to update error metrics:', error);
}
}
async function checkCircuitBreaker(key: string): Promise {
const cbKey = `circuit_breaker:${key}`;
try {
const cbData = await HyrexKV.get(cbKey);
if (!cbData) return "CLOSED";
const cbState = JSON.parse(cbData);
// Check if circuit breaker should be reset
if (cbState.state === "OPEN") {
if (Date.now() / 1000 - cbState.openedAt > (cbState.timeout || 60)) {
cbState.state = "HALF_OPEN";
await HyrexKV.set(cbKey, JSON.stringify(cbState), 3600);
}
}
return cbState.state;
} catch {
return "CLOSED";
}
}
async function recordCircuitBreakerFailure(key: string) {
const cbKey = `circuit_breaker:${key}`;
try {
const cbData = await HyrexKV.get(cbKey);
const cbState = cbData ? JSON.parse(cbData) : {
failures: 0,
state: "CLOSED",
threshold: 5,
timeout: 60
};
cbState.failures += 1;
cbState.lastFailure = Date.now() / 1000;
// Open circuit breaker if threshold exceeded
if (cbState.failures >= cbState.threshold && cbState.state !== "OPEN") {
cbState.state = "OPEN";
cbState.openedAt = Date.now() / 1000;
// Send circuit breaker alert
await sendCircuitBreakerAlert.send(key, cbState);
}
await HyrexKV.set(cbKey, JSON.stringify(cbState), 3600);
} catch (error) {
console.error(`Failed to update circuit breaker ${key}:`, error);
}
}
async function resetCircuitBreaker(key: string) {
const cbKey = `circuit_breaker:${key}`;
try {
const cbData = await HyrexKV.get(cbKey);
if (cbData) {
const cbState = JSON.parse(cbData);
cbState.failures = 0;
cbState.state = "CLOSED";
await HyrexKV.set(cbKey, JSON.stringify(cbState), 3600);
}
} catch {
// Ignore errors
}
}
// Helper functions (placeholder implementations)
async function performOperation(operation: string, data: any): Promise {
if (operation === "test_validation_error") {
throw new ValidationError("Invalid input data", { field: "email" });
} else if (operation === "test_api_error") {
throw new ExternalAPIError("API service unavailable", 503);
}
return { success: true, data };
}
async function updateTaskMetrics(operation: string, status: string) {
// Implementation placeholder
}
async function recordTaskSuccess(operation: string, duration: number, taskContext: any) {
// Implementation placeholder
}
async function getErrorDetails(errorId: string) {
const errorData = await HyrexKV.get(`error:${errorId}`);
return errorData ? JSON.parse(errorData) : null;
}
async function getFallbackData(operation: string) {
return { fallback: true };
}
async function executeAlternativeWorkflow(errorDetails: any) {
return { alternativeExecuted: true };
}
async function sendSlackAlert(message: string, channel: string) {
// Implementation placeholder
}
async function sendEmailAlert(subject: string, message: string, recipients: string[]) {
// Implementation placeholder
}
```
## Usage Examples
### Basic Error Handling
```bash theme={null}
# Process task with error handling
curl -X POST http://localhost:8000/tasks/monitored \
-H "Content-Type: application/json" \
-d '{
"operation": "process_user_data",
"data": {"user_id": 123, "email": "test@example.com"},
"circuit_breaker_key": "user_service"
}'
# Make resilient API call
curl -X POST http://localhost:8000/tasks/api-call \
-H "Content-Type: application/json" \
-d '{
"url": "https://api.external.com/data",
"method": "GET",
"circuit_breaker_key": "external_api"
}'
```
### Error Recovery
```bash theme={null}
# Trigger error recovery workflow
curl -X POST http://localhost:8000/recovery/start \
-H "Content-Type: application/json" \
-d '{
"error_id": "err_1234567890_5678",
"recovery_strategy": "retry_with_fallback"
}'
# Generate error report
curl -X POST http://localhost:8000/reports/errors \
-H "Content-Type: application/json" \
-d '{
"operation": "process_payments",
"hours": 48
}'
```
## Circuit Breaker Configuration
```python theme={null}
# Configure circuit breaker thresholds
CIRCUIT_BREAKER_CONFIG = {
"payment_service": {
"threshold": 3, # Open after 3 failures
"timeout": 30, # Stay open for 30 seconds
"half_open_max": 1 # Allow 1 request in half-open state
},
"external_api": {
"threshold": 5,
"timeout": 60,
"half_open_max": 2
}
}
```
## Monitoring Dashboard
```python theme={null}
@hy.task
def health_check_dashboard():
"""Generate system health dashboard data"""
services = ["payment_service", "user_service", "external_api"]
dashboard_data = {
"timestamp": datetime.now().isoformat(),
"services": {}
}
for service in services:
# Check circuit breaker status
cb_status = check_circuit_breaker(service)
# Get error metrics
error_metrics = get_recent_error_metrics(service, hours=1)
dashboard_data["services"][service] = {
"circuit_breaker_status": cb_status,
"error_rate": error_metrics.get("error_rate", 0),
"recent_errors": error_metrics.get("total_errors", 0),
"availability": calculate_availability(service)
}
return dashboard_data
```
## Production Considerations
* **Error classification**: Properly categorize errors for appropriate handling
* **Circuit breaker tuning**: Configure thresholds based on service characteristics
* **Alert fatigue**: Implement intelligent alerting to avoid noise
* **Error aggregation**: Use tools like Sentry or Rollbar for error tracking
* **Performance impact**: Monitor overhead of error handling mechanisms
* **Recovery testing**: Regularly test error recovery workflows
## Next Steps
Add human oversight to error recovery
Handle errors in batch operations