release: bump version to 0.68.0

backend/app/services/data_jobs.py

@@ -54,6 +54,9 @@ DATA_WRITE_JOB_TYPES = (JOB_TYPE_COLLECT, JOB_TYPE_CLEAR_DATA, JOB_TYPE_CLEAR_CA
 SOURCE_LOCK_JOB_STATUSES = (JOB_STATUS_RUNNING, JOB_STATUS_CANCELLING)
 QUEUE_POLL_SECONDS = 0.35
 JOB_STALE_LOCK_MINUTES = 90
+ORPHAN_CANCELLING_GRACE_SECONDS = 30
+JOB_RECOVERY_SWEEP_SECONDS = 15
+DATA_DELETE_BATCH_SIZE = 50_000
 DEFAULT_WORKER_CONCURRENCY = 2
 
 RUNNING_DATA_JOB_TASKS: dict[int, asyncio.Task[Any]] = {}
@@ -281,6 +284,7 @@ class DataJobWorker:
         self._task: asyncio.Task[None] | None = None
         self._stop_event: asyncio.Event | None = None
         self._running: set[asyncio.Task[Any]] = set()
+        self._last_recovery_sweep_at: datetime | None = None
 
     def start(self) -> None:
         if self._task and not self._task.done():
@@ -303,6 +307,11 @@ class DataJobWorker:
         await self._recover_stale_running_jobs()
         while not self._stop_event.is_set():
             self._running = {task for task in self._running if not task.done()}
+            if (
+                self._last_recovery_sweep_at is None
+                or (_utcnow() - self._last_recovery_sweep_at).total_seconds() >= JOB_RECOVERY_SWEEP_SECONDS
+            ):
+                await self._recover_stale_running_jobs()
             if len(self._running) >= self.concurrency:
                 await asyncio.sleep(QUEUE_POLL_SECONDS)
                 continue
@@ -316,7 +325,9 @@ class DataJobWorker:
             self._running.add(runner)
 
     async def _recover_stale_running_jobs(self) -> None:
+        self._last_recovery_sweep_at = _utcnow()
         cutoff = _utcnow() - timedelta(minutes=JOB_STALE_LOCK_MINUTES)
+        orphan_cancelling_cutoff = _utcnow() - timedelta(seconds=ORPHAN_CANCELLING_GRACE_SECONDS)
         async with async_session_factory() as db:
             result = await db.execute(
                 select(CollectionTask)
@@ -332,6 +343,21 @@ class DataJobWorker:
                 job.error_message = "Marked failed after stale data job lock timeout"
             if stale_jobs:
                 await db.commit()
+            orphan_result = await db.execute(
+                select(CollectionTask)
+                .where(CollectionTask.status == JOB_STATUS_CANCELLING)
+                .where(CollectionTask.locked_at.is_(None))
+                .where(CollectionTask.requested_cancel_at.is_not(None))
+                .where(CollectionTask.requested_cancel_at < orphan_cancelling_cutoff)
+            )
+            for job in orphan_result.scalars().all():
+                if job.id in RUNNING_DATA_JOB_TASKS:
+                    continue
+                await _cancel_task_without_runner(
+                    db,
+                    job,
+                    reason=job.cancel_reason or "cancelled_after_orphaned_runner",
+                )
 
     async def _claim_next_job(self) -> int | None:
         async with async_session_factory() as db:
@@ -477,15 +503,22 @@ async def _run_clear_data_job(db: AsyncSession, task: CollectionTask) -> None:
     await db.commit()
     await _broadcast_task_update(task)
 
-    count_result = await db.execute(
-        select(CollectedData.id).where(CollectedData.source == source)
+    deleted_count = await _delete_table_rows_by_source(
+        db,
+        task,
+        table_name="collected_data",
+        source_column="source",
+        source=source,
+    )
+    derived_deleted_counts = await _clear_derived_datasource_data_in_batches(
+        db,
+        task,
+        source,
+        progress_offset=deleted_count,
     )
-    collected_ids = [row[0] for row in count_result.all()]
-    derived_deleted_counts = await clear_derived_datasource_data(db, source)
-    if collected_ids:
-        await db.execute(CollectedData.__table__.delete().where(CollectedData.id.in_(collected_ids)))
-    deleted_count = len(collected_ids)
     derived_deleted_count = sum(derived_deleted_counts.values())
+    if any(key.startswith("ais_") for key in derived_deleted_counts):
+        await db.execute(text("ANALYZE ais_raw_observations"))
 
     task.records_processed = deleted_count + derived_deleted_count
     task.total_records = task.records_processed
@@ -504,10 +537,99 @@ async def _run_clear_data_job(db: AsyncSession, task: CollectionTask) -> None:
     task.phase = "completed"
     task.phase_message = "数据库数据已清理"
     task.completed_at = _utcnow()
+    datasource = await db.get(DataSource, task.datasource_id)
+    if datasource is not None:
+        datasource.last_status = JOB_STATUS_SUCCESS
+        datasource.last_run_at = task.completed_at
+    await db.execute(
+        DataSnapshot.__table__.update()
+        .where(DataSnapshot.source == source)
+        .values(is_current=False)
+    )
     await db.commit()
     await _broadcast_task_update(task)
 
 
+async def _delete_table_rows_by_source(
+    db: AsyncSession,
+    task: CollectionTask,
+    *,
+    table_name: str,
+    source_column: str,
+    source: str,
+    progress_offset: int = 0,
+) -> int:
+    deleted = 0
+    while True:
+        result = await db.execute(
+            text(
+                f"""
+                WITH doomed AS (
+                    SELECT ctid
+                    FROM {table_name}
+                    WHERE {source_column} = :source
+                    LIMIT :batch_size
+                ),
+                deleted_rows AS (
+                    DELETE FROM {table_name}
+                    USING doomed
+                    WHERE {table_name}.ctid = doomed.ctid
+                    RETURNING 1
+                )
+                SELECT COUNT(*) FROM deleted_rows
+                """
+            ),
+            {"source": source, "batch_size": DATA_DELETE_BATCH_SIZE},
+        )
+        batch_deleted = max(int(result.scalar_one() or 0), 0)
+        if batch_deleted <= 0:
+            break
+        deleted += batch_deleted
+        task.records_processed = progress_offset + deleted
+        task.phase_current = task.records_processed
+        task.phase_unit = "records"
+        task.phase_message = f"正在删除数据：{task.records_processed} 条"
+        await db.commit()
+        await _broadcast_task_update(task)
+    return deleted
+
+
+async def _clear_derived_datasource_data_in_batches(
+    db: AsyncSession,
+    task: CollectionTask,
+    source: str,
+    progress_offset: int = 0,
+) -> dict[str, int]:
+    deleted_counts: dict[str, int] = {}
+    if source in {"barentswatch_vessels", "aisstream_vessels"}:
+        deleted_counts["ais_conflict_records"] = await _delete_table_rows_by_source(
+            db,
+            task,
+            table_name="ais_conflict_records",
+            source_column="selected_source",
+            source=source,
+            progress_offset=progress_offset + sum(deleted_counts.values()),
+        )
+        deleted_counts["ais_source_health"] = await _delete_table_rows_by_source(
+            db,
+            task,
+            table_name="ais_source_health",
+            source_column="source",
+            source=source,
+            progress_offset=progress_offset + sum(deleted_counts.values()),
+        )
+        deleted_counts["ais_raw_observations"] = await _delete_table_rows_by_source(
+            db,
+            task,
+            table_name="ais_raw_observations",
+            source_column="source",
+            source=source,
+            progress_offset=progress_offset + sum(deleted_counts.values()),
+        )
+        return deleted_counts
+    return await clear_derived_datasource_data(db, source)
+
+
 async def _run_clear_cache_job(db: AsyncSession, task: CollectionTask) -> None:
     source = str(task.source or (task.payload or {}).get("source") or "").strip()
     if not source: