sync

apps/api/app/community_scheduler.py

@@ -42,13 +42,15 @@ def run_source(source_system: str, *, now: datetime | None = None) -> bool:
         source = session.get(DataSource, source_system)
         if source is None or not source.enabled:
             return False
+        # Lock before reading cooldown: committing the reservation makes it visible
+        # to the next contender before releasing this transaction lock.
+        if session.bind and session.bind.dialect.name == "postgresql" and not session.scalar(text("select pg_try_advisory_xact_lock(hashtext(:key))"), {"key": f"community:{source_system}"}):
+            return False
         recent = list(session.scalars(select(CommunityImportRun).where(CommunityImportRun.source_system == source_system).order_by(CommunityImportRun.started_at.desc()).limit(8)))
         latest = recent[0].started_at if recent else None
         delay = retry_delay([run.status for run in recent])
         if latest and (latest if latest.tzinfo else latest.replace(tzinfo=timezone.utc)) > current - timedelta(seconds=delay):
             return False
-        if session.bind and session.bind.dialect.name == "postgresql" and not session.scalar(text("select pg_try_advisory_xact_lock(hashtext(:key))"), {"key": f"community:{source_system}"}):
-            return False
         run = CommunityImportRun(source_system=source_system, source_url=url, started_at=current, status="running")
         session.add(run); session.commit()
         try:
@@ -57,6 +59,7 @@ def run_source(source_system: str, *, now: datetime | None = None) -> bool:
             created, updated = stage_observations(session, [asdict(item) for item in records])
             run.status, run.rows_seen, run.rows_created, run.rows_updated = "success", len(records), created, updated
         except Exception as exc:
+            session.rollback()
             run.status, run.error_summary = "failed", f"{type(exc).__name__}: {str(exc)[:500]}"
             logger.exception("community import failed", extra={"event":"community_import_failed", "source_system":source_system})
         run.finished_at = datetime.now(timezone.utc); session.commit()