1"""Single ingest worker: claims jobs from Postgres, holds the lake writer lock, runs the task registry.2Also hosts the scheduler loop (idempotent enqueue with dedupe keys) so there is exactly one writer process."""3from __future__ import annotations45import signal6import time7import traceback8from datetime import UTC, datetime9from zoneinfo import ZoneInfo1011from findata_core.config import get_settings12from findata_core.logging import configure_logging, get_logger13from findata_ingest import queue14from findata_ingest.lake import Lake, WriterLock15from findata_ingest.tasks import REGISTRY1617ET = ZoneInfo("America/New_York")18_stop = False192021def _handle(sig, frame): # noqa: ANN00122 global _stop23 _stop = True242526def schedule_tick(log) -> None:27 """Enqueue recurring jobs (dedupe keys make this idempotent). Times in America/New_York."""28 now = datetime.now(ET)29 day = now.date().isoformat()30 # vendor daily update lands ~23:30–03:00 ET → poll from 02:00 every 30 min (incremental skips when vendor unchanged)31 if now.weekday() < 6 and (now.hour >= 2 or now.hour < 1):32 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day"}, priority=20,33 dedupe_key=f"incremental:stock:day:{day}:{now.hour}:{now.minute // 30}")34 if now.weekday() == 5 and now.hour == 6: # Saturday 06:00 ET — week archive repairs any missed day35 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "week", "force": True}, priority=25,36 dedupe_key=f"incremental:stock:week:{day}")37 if now.weekday() == 6 and now.hour == 7:38 queue.enqueue("metadata", None, {"types": ["stock", "etf"]}, priority=30, dedupe_key=f"metadata:{day}")39 queue.enqueue("compaction", "market_bars.stock.1m.raw", {}, priority=60, dedupe_key=f"compaction:1m:{day}")40 queue.enqueue("compaction", "market_bars.stock.5m.raw", {}, priority=61, dedupe_key=f"compaction:5m:{day}")41 if now.hour == 5 and now.minute < 30:42 queue.enqueue("fundamentals", "fundamentals.financials", {"tracked": True}, priority=50, dedupe_key=f"fundamentals:{day}")43 if now.hour == 4 and now.minute < 30 and get_settings().backup_target:44 queue.enqueue("backup", None, {}, priority=70, dedupe_key=f"backup:{day}")45 if now.hour == 6 and now.minute < 30:46 queue.enqueue("quality", "market_bars.stock.1m.raw", {}, priority=80, dedupe_key=f"quality:{day}")47 queue.enqueue("gc", None, {}, priority=90, dedupe_key=f"gc:{day}")484950def main(once: bool = False, kinds: list[str] | None = None, with_scheduler: bool = True) -> None:51 configure_logging()52 log = get_logger("worker")53 signal.signal(signal.SIGTERM, _handle)54 signal.signal(signal.SIGINT, _handle)55 lake = Lake()56 requeued = queue.requeue_stale()57 if requeued:58 log.warning("worker.requeued_stale", n=requeued)59 last_tick = 0.060 with WriterLock(lake.root):61 log.info("worker.start", lake=str(lake.root), free_gb=round(lake.free_gb(), 1))62 while not _stop:63 if with_scheduler and time.monotonic() - last_tick > 300:64 try:65 schedule_tick(log)66 except Exception as e: # noqa: BLE00167 log.error("scheduler.error", error=str(e))68 last_tick = time.monotonic()69 job = queue.claim_next(kinds)70 if job is None:71 if once:72 break73 time.sleep(5)74 continue75 jl = log.bind(job_id=str(job["id"]), kind=job["kind"], dataset=job.get("dataset_id"), attempt=job["attempts"])76 jl.info("job.start")77 t0 = time.monotonic()78 try:79 fn = REGISTRY[job["kind"]]80 result = fn(job, lake, jl)81 queue.finish(job, True, {**(result or {}), "seconds": round(time.monotonic() - t0, 1)})82 jl.info("job.done", seconds=round(time.monotonic() - t0, 1))83 except Exception as e: # noqa: BLE00184 err = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}"85 queue.finish(job, False, error=err)86 jl.error("job.failed", error=str(e)[:300], seconds=round(time.monotonic() - t0, 1))87 if once:88 break89 log.info("worker.stop")90