Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-workers/src/lexigram/ai/workers/dlq/worker.py: 16%

1""" 

2Dead Letter Queue (DLQ) worker for failed task recovery. 

3 

4Handles failed task recovery including: 

5- Retry mechanisms with exponential backoff 

6- Error classification and routing 

7- Failure analysis and reporting 

8- Manual retry triggers 

9- Permanent failure handling 

10 

11Uses Dependency Injection - TaskQueueProtocol is injected, not imported. 

12""" 

13 

14from __future__ import annotations 

15 

16import asyncio 

17import contextlib 

18from datetime import UTC, datetime, timedelta 

19from typing import TYPE_CHECKING, Any 

20 

21if TYPE_CHECKING: 

22 from collections.abc import Callable 

23 

24 from lexigram.contracts import JobProtocol, TaskQueueProtocol 

25 

26from lexigram.ai.workers.types import ( 

27 DLQItem, 

28 DLQStats, 

29 FailureCategory, 

30) 

31from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

32from lexigram.logging import ( 

33 get_logger, 

34) 

35from lexigram.result import Result 

36 

37logger = get_logger(__name__) 

38 

39 

40class ErrorClassifier: 

41 """Classify errors into failure categories.""" 

42 

43 @staticmethod 

44 def classify(error: str, _job: JobProtocol) -> FailureCategory: 

45 """ 

46 Classify error into a failure category. 

47 

48 Args: 

49 error: Error message 

50 job: Original job 

51 

52 Returns: 

53 Failure category 

54 """ 

55 error_lower = error.lower() 

56 

57 # Permanent failures 

58 if any( 

59 term in error_lower 

60 for term in [ 

61 "not found", 

62 "does not exist", 

63 "invalid", 

64 "malformed", 

65 "syntax error", 

66 ] 

67 ): 

68 return FailureCategory.PERMANENT 

69 

70 # Throttling 

71 if any( 

72 term in error_lower 

73 for term in [ 

74 "rate limit", 

75 "too many requests", 

76 "throttled", 

77 "quota exceeded", 

78 ] 

79 ): 

80 return FailureCategory.THROTTLED 

81 

82 # Invalid input 

83 if any( 

84 term in error_lower 

85 for term in [ 

86 "validation error", 

87 "invalid input", 

88 "bad request", 

89 ] 

90 ): 

91 return FailureCategory.INVALID_INPUT 

92 

93 # Transient failures 

94 if any( 

95 term in error_lower 

96 for term in [ 

97 "timeout", 

98 "connection", 

99 "network", 

100 "temporary", 

101 "unavailable", 

102 "503", 

103 "502", 

104 ] 

105 ): 

106 return FailureCategory.TRANSIENT 

107 

108 return FailureCategory.UNKNOWN 

109 

110 

111class DeadLetterQueueWorker: 

112 """ 

113 Worker for handling failed tasks and retry logic. 

114 

115 Monitors failed jobs and implements intelligent retry strategies 

116 based on failure categorization. 

117 

118 Example: 

119 ```python 

120 from lexigram.ai.workers import DeadLetterQueueWorker 

121 from lexigram.contracts.infra.tasks import TaskQueueProtocol 

122 

123 # Setup 

124 main_queue = TaskQueueProtocol() 

125 dlq_queue = TaskQueueProtocol() # Separate queue for DLQ items 

126 

127 worker = DeadLetterQueueWorker( 

128 main_queue=main_queue, 

129 dlq_queue=dlq_queue, 

130 worker_id="dlq", 

131 check_interval=60, # Check every minute 

132 ) 

133 

134 # Register error notification handler 

135 async def notify_on_permanent_failure(item: DLQItem): 

136 await send_alert(f"Permanent failure: {item.job_id}") 

137 

138 worker.set_notification_handler(notify_on_permanent_failure) 

139 

140 # Start worker 

141 await worker.start() 

142 

143 # Manually retry a failed job 

144 await worker.retry_item(job_id="abc123") 

145 

146 # Get DLQ statistics 

147 stats = await worker.get_stats() 

148 logger.info(f"Total failed items: {stats.total_items}") 

149 

150 # Stop worker 

151 await worker.stop() 

152 ``` 

153 """ 

154 

155 def __init__( 

156 self, 

157 main_queue: TaskQueueProtocol, 

158 dlq_queue: TaskQueueProtocol | None = None, 

159 worker_id: str = "dlq", 

160 check_interval: int = 60, 

161 max_retries: int = 5, 

162 base_backoff: int = 60, 

163 ): 

164 """ 

165 Initialize DLQ worker. 

166 

167 Args: 

168 main_queue: Main task queue to monitor 

169 dlq_queue: Optional separate queue for DLQ items 

170 worker_id: Unique worker identifier 

171 check_interval: Interval in seconds to check for failed jobs 

172 max_retries: Maximum retry attempts per job 

173 base_backoff: Base delay in seconds for exponential backoff 

174 """ 

175 self.main_queue = main_queue 

176 self.dlq_queue = dlq_queue or main_queue 

177 self.worker_id = worker_id 

178 self.check_interval = check_interval 

179 self.max_retries = max_retries 

180 self.base_backoff = base_backoff 

181 

182 # DLQ storage 

183 self._items: dict[str, DLQItem] = {} 

184 self._items_lock = asyncio.Lock() 

185 self.dead_letter_count = 0 

186 

187 # Error classifier 

188 self._classifier = ErrorClassifier() 

189 

190 # Notification handler 

191 self._notification_handler: Callable[[DLQItem], Any] | None = None 

192 

193 # Worker state 

194 self._running = False 

195 self._worker_task: asyncio.Task | None = None 

196 

197 async def start(self) -> None: 

198 """Start the DLQ worker.""" 

199 if self._running: 

200 logger.warning("Worker %s already running", self.worker_id) 

201 return 

202 

203 self._running = True 

204 self._worker_task = asyncio.create_task(self._dlq_loop()) 

205 

206 logger.info( 

207 "Started DLQ worker", 

208 worker_id=self.worker_id, 

209 check_interval=self.check_interval, 

210 max_retries=self.max_retries, 

211 ) 

212 

213 async def stop(self) -> None: 

214 """Stop the DLQ worker.""" 

215 if not self._running: 

216 return 

217 

218 self._running = False 

219 

220 if self._worker_task: 

221 self._worker_task.cancel() 

222 with contextlib.suppress(asyncio.CancelledError): 

223 await self._worker_task 

224 

225 logger.info("Stopped DLQ worker", worker_id=self.worker_id) 

226 

227 def set_notification_handler( 

228 self, 

229 handler: Callable[[DLQItem], Any], 

230 ) -> None: 

231 """ 

232 Set notification handler for permanent failures. 

233 

234 Args: 

235 handler: Async callable that receives DLQItem 

236 """ 

237 self._notification_handler = handler 

238 logger.info("Set DLQ notification handler") 

239 

240 async def add_failed_job(self, job: JobProtocol, error: str) -> None: 

241 """ 

242 Add a failed job to the DLQ. 

243 

244 Args: 

245 job: Failed job 

246 error: Error message 

247 """ 

248 async with self._items_lock: 

249 if job.id in self._items: 

250 # Update existing item 

251 item = self._items[job.id] 

252 item.failure_count += 1 

253 item.last_failure = datetime.now(UTC) 

254 item.last_error = error 

255 else: 

256 # Create new DLQ item 

257 now = datetime.now(UTC) 

258 failure_category = self._classifier.classify(error, job) 

259 

260 item = DLQItem( 

261 job_id=job.id, 

262 original_job=job, 

263 failure_count=1, 

264 first_failure=now, 

265 last_failure=now, 

266 last_error=error, 

267 failure_category=failure_category, 

268 max_retries=self.max_retries, 

269 ) 

270 

271 self._items[job.id] = item 

272 self.dead_letter_count += 1 

273 

274 logger.info( 

275 "Added job to DLQ", 

276 job_id=job.id, 

277 failure_count=item.failure_count, 

278 category=item.failure_category.value, 

279 ) 

280 

281 async def retry_item(self, job_id: str) -> bool: 

282 """ 

283 Manually retry a DLQ item. 

284 

285 Args: 

286 job_id: JobProtocol ID to retry 

287 

288 Returns: 

289 True if retry was initiated, False otherwise 

290 """ 

291 async with self._items_lock: 

292 if job_id not in self._items: 

293 logger.warning("JobProtocol not found in DLQ", job_id=job_id) 

294 return False 

295 

296 item = self._items[job_id] 

297 

298 if not item.can_retry(): 

299 logger.warning( 

300 "JobProtocol cannot be retried", 

301 job_id=job_id, 

302 retry_count=item.retry_count, 

303 max_retries=item.max_retries, 

304 ) 

305 return False 

306 

307 # Retry the job 

308 await self._retry_job(item) 

309 return True 

310 

311 async def remove_item(self, job_id: str) -> bool: 

312 """ 

313 Remove item from DLQ (permanent deletion). 

314 

315 Args: 

316 job_id: JobProtocol ID to remove 

317 

318 Returns: 

319 True if removed, False if not found 

320 """ 

321 async with self._items_lock: 

322 if job_id in self._items: 

323 del self._items[job_id] 

324 logger.info("Removed job from DLQ", job_id=job_id) 

325 return True 

326 return False 

327 

328 async def archive_item(self, job_id: str) -> bool: 

329 """ 

330 Archive item (mark as permanently failed). 

331 

332 Args: 

333 job_id: JobProtocol ID to archive 

334 

335 Returns: 

336 True if archived, False if not found 

337 """ 

338 async with self._items_lock: 

339 if job_id in self._items: 

340 item = self._items[job_id] 

341 item.failure_category = FailureCategory.PERMANENT 

342 item.metadata["archived"] = True 

343 item.metadata["archived_at"] = datetime.now(UTC).isoformat() 

344 

345 logger.info("Archived job in DLQ", job_id=job_id) 

346 return True 

347 return False 

348 

349 async def get_stats(self) -> DLQStats: 

350 """Get DLQ statistics.""" 

351 async with self._items_lock: 

352 stats = DLQStats(total_items=len(self._items)) 

353 

354 for item in self._items.values(): 

355 category = item.failure_category.value 

356 stats.by_category[category] = stats.by_category.get(category, 0) + 1 

357 

358 if item.retry_count > 0: 

359 stats.retried_count += 1 

360 

361 if item.failure_category == FailureCategory.PERMANENT: 

362 stats.permanent_failures += 1 

363 

364 if item.metadata.get("archived"): 

365 stats.archived_count += 1 

366 

367 return stats 

368 

369 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

370 """Report the health of this worker. 

371 

372 Args: 

373 timeout: Unused; present for protocol conformance. 

374 

375 Returns: 

376 HEALTHY when the worker is running, UNHEALTHY otherwise. 

377 """ 

378 status = HealthStatus.HEALTHY if self._running else HealthStatus.UNHEALTHY 

379 stats = await self.get_stats() 

380 return HealthCheckResult( 

381 component=f"worker.dlq.{self.worker_id}", 

382 status=status, 

383 details=stats.to_dict(), 

384 ) 

385 

386 async def get_items( 

387 self, 

388 category: FailureCategory | None = None, 

389 limit: int = 100, 

390 ) -> list[dict[str, Any]]: 

391 """ 

392 Get DLQ items. 

393 

394 Args: 

395 category: Optional filter by failure category 

396 limit: Maximum items to return 

397 

398 Returns: 

399 List of DLQ items as dictionaries 

400 """ 

401 async with self._items_lock: 

402 items = list(self._items.values()) 

403 

404 if category: 

405 items = list( 

406 filter(lambda item: item.failure_category == category, items), 

407 ) 

408 

409 # Sort by last failure (most recent first) 

410 items.sort(key=lambda x: x.last_failure, reverse=True) 

411 

412 return [item.to_dict() for item in items[:limit]] 

413 

414 async def _dlq_loop(self) -> None: 

415 """Main DLQ processing loop.""" 

416 while self._running: 

417 try: 

418 # Check for items ready to retry 

419 async with self._items_lock: 

420 items_to_retry = [ 

421 item for item in self._items.values() if item.can_retry() 

422 ] 

423 

424 # Retry eligible items 

425 for item in items_to_retry: 

426 try: 

427 await self._retry_job(item) 

428 except Exception as e: 

429 logger.exception( 

430 "Error retrying job", 

431 job_id=item.job_id, 

432 error=str(e), 

433 ) 

434 

435 # Check for permanent failures to notify 

436 async with self._items_lock: 

437 permanent_failures = [ 

438 item 

439 for item in self._items.values() 

440 if ( 

441 item.failure_category == FailureCategory.PERMANENT 

442 and not item.metadata.get("notified") 

443 ) 

444 ] 

445 

446 # Send notifications 

447 for item in permanent_failures: 

448 try: 

449 await self._notify_permanent_failure(item) 

450 item.metadata["notified"] = True 

451 except Exception as e: 

452 logger.exception( 

453 "Error sending notification", 

454 job_id=item.job_id, 

455 error=str(e), 

456 ) 

457 

458 # Wait for next check 

459 await asyncio.sleep(self.check_interval) 

460 

461 except asyncio.CancelledError: 

462 break 

463 except Exception as e: 

464 logger.exception( 

465 "Error in DLQ loop", 

466 error=str(e), 

467 ) 

468 await asyncio.sleep(self.check_interval) 

469 

470 async def _retry_job(self, item: DLQItem) -> None: 

471 """Retry a failed job.""" 

472 # Calculate backoff delay 

473 backoff = item.calculate_backoff(self.base_backoff) 

474 item.next_retry = datetime.now(UTC) + timedelta(seconds=backoff) 

475 item.retry_count += 1 

476 

477 logger.info( 

478 "Retrying job", 

479 job_id=item.job_id, 

480 retry_count=item.retry_count, 

481 next_retry=item.next_retry.isoformat(), 

482 ) 

483 

484 # Re-enqueue the job using queue's enqueue(job_type, data, priority) API 

485 job_name = item.original_job.name 

486 if job_name is None: 

487 logger.error("Original job name missing; cannot retry", job_id=item.job_id) 

488 return 

489 

490 # Narrow type for mypy: job_name is Optional[str] on JobProtocol, assert non-None 

491 assert job_name is not None # noqa: S101 # type-narrowing only; log branch returned above 

492 

493 priority = ( 

494 item.original_job.priority if item.original_job.priority is not None else 0 

495 ) 

496 

497 new_job_data: dict[str, Any] = { 

498 "name": job_name, 

499 "args": item.original_job.args or [], 

500 "kwargs": item.original_job.kwargs or {}, 

501 "priority": priority, 

502 "max_retries": 1, # Only one retry per DLQ attempt 

503 "timeout": item.original_job.timeout, 

504 } 

505 

506 enqueue_result = await self.dlq_queue.enqueue( 

507 { 

508 "name": job_name, 

509 "args": tuple(item.original_job.args or []), 

510 "kwargs": new_job_data, 

511 "priority": priority, 

512 } 

513 ) 

514 if isinstance(enqueue_result, Result) and enqueue_result.is_err(): 

515 logger.warning( 

516 "dlq_retry_enqueue_failed", 

517 job_id=item.job_id, 

518 error=str(enqueue_result.unwrap_err()), 

519 ) 

520 return 

521 

522 if not item.metadata.get("replayed"): 

523 item.metadata["replayed"] = True 

524 self.dead_letter_count -= 1 

525 

526 async def _notify_permanent_failure(self, item: DLQItem) -> None: 

527 """Notify about permanent failure.""" 

528 if not self._notification_handler: 

529 return 

530 

531 logger.info( 

532 "Notifying permanent failure", 

533 job_id=item.job_id, 

534 error=item.last_error, 

535 ) 

536 

537 if asyncio.iscoroutinefunction(self._notification_handler): 

538 await self._notification_handler(item) 

539 else: 

540 await asyncio.to_thread(self._notification_handler, item)