Spaces:
Sleeping
Sleeping
File size: 3,761 Bytes
3493993 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | from __future__ import annotations
import asyncio
import random
from datetime import datetime, timedelta, timezone
from app.analytics.service import AnalyticsDomainService
from app.core.logger import get_logger
from app.social.database import SocialDatabase
logger = get_logger(__name__)
class AnalyticsSyncWorker:
"""Claims durable sync runs and executes bounded provider synchronization."""
def __init__(
self,
analytics: AnalyticsDomainService,
database: SocialDatabase,
*,
interval_seconds: int,
concurrency: int = 2,
) -> None:
self.analytics = analytics
self.database = database
self.interval_seconds = max(1, interval_seconds)
self.concurrency = max(1, min(concurrency, 8))
self._task: asyncio.Task[None] | None = None
self._stop = asyncio.Event()
async def start(self) -> None:
if self._task is not None or not self.analytics.ready:
return
self._stop.clear()
self._task = asyncio.create_task(self._run(), name="analytics-sync-worker")
async def stop(self) -> None:
self._stop.set()
if self._task is not None:
self._task.cancel()
await asyncio.gather(self._task, return_exceptions=True)
self._task = None
async def run_once(self) -> None:
if not self.analytics.ready:
return
async with self.database.worker_boundary():
runs = await self.analytics.repository.claim_due_syncs(limit=self.concurrency)
await asyncio.gather(*(self._execute(run) for run in runs))
async def _execute(self, run) -> None:
try:
latest = await self.analytics.repository.get_sync(run.workspace_id, run.id)
if latest.status == "cancelled":
return
await self.analytics.sync_once(run)
except asyncio.CancelledError:
raise
except Exception as exc:
retryable = run.attempt_count < 3
if retryable:
delay = min(900, 30 * (2 ** max(0, run.attempt_count - 1))) + random.randint(0, 10)
await self.analytics.repository.update_sync(
run.id,
status="queued",
error_code=getattr(exc, "code", "ANALYTICS_SYNC_RETRY"),
error_message="Analytics synchronization will be retried.",
next_attempt_at=datetime.now(timezone.utc) + timedelta(seconds=delay),
)
else:
await self.analytics.repository.update_sync(
run.id,
status="failed",
error_code=getattr(exc, "code", "ANALYTICS_SYNC_FAILED"),
error_message="Analytics synchronization failed.",
)
await self.analytics.audit.record(
workspace_id=run.workspace_id,
event_type="analytics.sync_failed",
metadata={"sync_run_id": run.id, "error_code": getattr(exc, "code", None)},
)
logger.warning(
"analytics sync execution failed",
extra={"sync_run_id": run.id, "attempt": run.attempt_count},
)
async def _run(self) -> None:
while not self._stop.is_set():
try:
await self.run_once()
except asyncio.CancelledError:
raise
except Exception:
logger.exception("analytics sync worker iteration failed")
try:
await asyncio.wait_for(self._stop.wait(), timeout=self.interval_seconds)
except TimeoutError:
pass
|