Coverage for src/qdrant_loader/webhooks/worker.py: 47%

1"""Background worker that processes queued webhook events.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import os 

7 

8from qdrant_loader.core.worker.job_types import JobType 

9from qdrant_loader.utils.logging import LoggingConfig 

10from qdrant_loader.webhooks.event_processor import process_change_event 

11from qdrant_loader.webhooks.queue_backend import ( 

12 FULL_SCAN, 

13 QueueBackendManager, 

14 parse_job_payload, 

15) 

16 

17logger = LoggingConfig.get_logger(__name__) 

18 

19WEBHOOK_WORKER_POLL_SECONDS = float(os.getenv("WEBHOOK_WORKER_POLL_SECONDS", "0.5")) 

20WEBHOOK_WORKER_LEASE_SECONDS = int(os.getenv("WEBHOOK_WORKER_LEASE_SECONDS", "120")) 

21 

22# Job types handled exclusively by the webhook worker. 

23# The ingestion QueueWorkerPool claims BULK_INGEST / INCREMENTAL_PULL; these two 

24# sets must remain disjoint so neither worker steals the other's jobs. 

25WEBHOOK_JOB_TYPES = [ 

26 JobType.SINGLE_UPSERT.value, 

27 JobType.SINGLE_DELETE.value, 

28 FULL_SCAN, 

29] 

30 

31 

32async def run_webhook_worker(stop_event: asyncio.Event) -> None: 

33 """Poll the durable queue and process webhook jobs until stopped.""" 

34 job_queue = QueueBackendManager.get_job_queue() 

35 logger.info( 

36 "Webhook worker started", 

37 poll_seconds=WEBHOOK_WORKER_POLL_SECONDS, 

38 lease_seconds=WEBHOOK_WORKER_LEASE_SECONDS, 

39 ) 

40 

41 while not stop_event.is_set(): 

42 job = await job_queue.claim_next( 

43 lease_seconds=WEBHOOK_WORKER_LEASE_SECONDS, 

44 job_types=WEBHOOK_JOB_TYPES, 

45 ) 

46 if job is None: 

47 try: 

48 await asyncio.wait_for( 

49 stop_event.wait(), 

50 timeout=WEBHOOK_WORKER_POLL_SECONDS, 

51 ) 

52 except TimeoutError: 

53 pass 

54 continue 

55 

56 try: 

57 event = parse_job_payload(job) 

58 logger.info( 

59 "Processing webhook job", 

60 job_id=job.id, 

61 operation=event.operation, 

62 source=event.source, 

63 entity_id=event.entity_id, 

64 ) 

65 await process_change_event(event) 

66 await job_queue.mark_done(job.id, claim_attempt=job.attempts) 

67 except Exception as exc: 

68 logger.exception( 

69 "Webhook job failed", 

70 job_id=job.id, 

71 error=str(exc), 

72 ) 

73 try: 

74 await job_queue.mark_failed( 

75 job.id, 

76 error_message=str(exc), 

77 claim_attempt=job.attempts, 

78 ) 

79 except Exception as mark_exc: 

80 logger.exception( 

81 "Failed to mark webhook job as failed", 

82 job_id=job.id, 

83 error=str(mark_exc), 

84 ) 

85 

86 logger.info("Webhook worker stopped")