diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 5dcde88..72b775c 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -813,7 +813,7 @@ async def sync_minute(request: Request): """ import asyncio - from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot, LONG_JOB_TIMEOUT_S + from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot from app.api.data import invalidate_storage_cache from app.services.preferences import get_minute_sync_days from app.tickflow.capabilities import Cap @@ -836,7 +836,7 @@ async def sync_minute(request: Request): extend_flag = body.get("extend") # 分钟K全市场同步是长任务(数据量是日K的 ~240 倍),用更宽松的卡死阈值 - job_id, is_new = job_store.create(timeout_s=LONG_JOB_TIMEOUT_S) + job_id, is_new = job_store.create(long_running=True) if not is_new: return {"status": "reused", "job_id": job_id} diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 6528466..55f8bdf 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -350,6 +350,11 @@ class DataProvidersIn(BaseModel): financial_data_provider: str | None = None +class DataSourceJobTimeoutPrefs(BaseModel): + data_source_job_timeout_s: int = Field(ge=60) + data_source_long_job_timeout_s: int = Field(ge=60) + + class DatasetFieldMapItem(BaseModel): source: str target: str @@ -413,6 +418,8 @@ def get_preferences() -> dict: "minute_data_provider": preferences.get_minute_data_provider(), "realtime_data_provider": preferences.get_realtime_data_provider(), "financial_data_provider": preferences.get_financial_provider(), + "data_source_job_timeout_s": preferences.get_data_source_job_timeout_s(), + "data_source_long_job_timeout_s": preferences.get_data_source_long_job_timeout_s(), "realtime_watchlist_symbols": preferences.get_realtime_watchlist_symbols(), **preferences.get_realtime_quote_scope(), "pipeline_pull_a_share": preferences.get_pipeline_pull_a_share(), @@ -616,6 +623,14 @@ def update_data_providers(req: DataProvidersIn) -> dict: } +@router.put("/preferences/data-source-job-timeouts") +def update_data_source_job_timeouts(req: DataSourceJobTimeoutPrefs) -> dict: + """保存普通与长数据后台任务的卡死判定时间。""" + from app.services import preferences + preferences.save(req.model_dump()) + return req.model_dump() + + @router.get("/preferences/watchlist-columns") def get_watchlist_columns() -> dict: """返回自选列表列配置。""" diff --git a/backend/app/services/pipeline_jobs.py b/backend/app/services/pipeline_jobs.py index 53f3397..c4b99eb 100644 --- a/backend/app/services/pipeline_jobs.py +++ b/backend/app/services/pipeline_jobs.py @@ -27,7 +27,7 @@ JobStatus = Literal["pending", "running", "succeeded", "failed"] # 由 reap_stale() 在 /run 和 /jobs/{id} 轮询端点检查 — 保证卡死后能自愈, # 无需用户再次点击「同步」。 # -# 超时阈值按任务类型区分: +# 默认超时阈值按任务类型区分,可在 Web 数据源设置中调整: # - 普通任务(日K管道/扩展/修正/重算): 1200s (20 分钟) # - 长任务(分钟K全市场同步,数据量是日K的 ~240 倍): 1800s (30 分钟) # 分钟K即使流式落盘后仍可能跑十几到数十分钟(限速 sleep 是主因), @@ -105,7 +105,12 @@ class JobStore: # ===== lifecycle ===== - def create(self, timeout_s: int = DEFAULT_JOB_TIMEOUT_S) -> tuple[str, bool]: + def create( + self, + timeout_s: int | None = None, + *, + long_running: bool = False, + ) -> tuple[str, bool]: """单飞创建任务。返回 (job_id, is_new)。 去重条件为 **pending ∨ running**(而非仅 running):`/run` 先 create() 再在 @@ -115,9 +120,17 @@ class JobStore: is_new=False 表示复用了已有活跃任务,调用方**不得**再调度新的后台任务。 - timeout_s: reap_stale 判定卡死的阈值。普通任务默认 1200s; - 分钟K全市场同步等长任务传 LONG_JOB_TIMEOUT_S (1800s)。 + timeout_s: reap_stale 判定卡死的阈值。None 时读取用户配置。 + long_running: timeout_s 为 None 时,是否读取长任务配置;普通任务默认 + 1200s,分钟K全市场同步等长任务默认 1800s。 """ + if timeout_s is None: + from app.services import preferences + if long_running: + timeout_s = preferences.get_data_source_long_job_timeout_s() + else: + timeout_s = preferences.get_data_source_job_timeout_s() + with self._lock: if self._active_id: active = self._active_jobs.get(self._active_id) diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index 5671bf8..42d47af 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -186,6 +186,32 @@ def get_minute_sync_segment_days() -> int: # ===== 数据源选择 (默认 TickFlow；第一阶段仅日K切换入口) ===== _ALLOWED_DATA_PROVIDERS = {"tickflow"} +DATA_SOURCE_JOB_TIMEOUT_MIN_S = 60 + + +def get_data_source_job_timeout_s() -> int: + """返回普通数据后台任务的卡死判定时间(秒)。""" + from app.services.pipeline_jobs import DEFAULT_JOB_TIMEOUT_S + raw = load().get("data_source_job_timeout_s", DEFAULT_JOB_TIMEOUT_S) + try: + timeout_s = int(raw) + except (TypeError, ValueError): + timeout_s = DEFAULT_JOB_TIMEOUT_S + return max(DATA_SOURCE_JOB_TIMEOUT_MIN_S, timeout_s) + + +def get_data_source_long_job_timeout_s() -> int: + """返回分钟 K 全市场等长任务的卡死判定时间(秒)。""" + from app.services.pipeline_jobs import LONG_JOB_TIMEOUT_S + raw = load().get( + "data_source_long_job_timeout_s", + LONG_JOB_TIMEOUT_S, + ) + try: + timeout_s = int(raw) + except (TypeError, ValueError): + timeout_s = LONG_JOB_TIMEOUT_S + return max(DATA_SOURCE_JOB_TIMEOUT_MIN_S, timeout_s) def _allowed_data_providers() -> set[str]: diff --git a/backend/tests/test_pipeline_and_monitor_fixes.py b/backend/tests/test_pipeline_and_monitor_fixes.py index a3dd75e..3b3615c 100644 --- a/backend/tests/test_pipeline_and_monitor_fixes.py +++ b/backend/tests/test_pipeline_and_monitor_fixes.py @@ -8,7 +8,7 @@ import polars as pl import pytest from app.jobs import daily_pipeline -from app.services import pipeline_jobs, quote_service +from app.services import pipeline_jobs, preferences, quote_service from app.services.pipeline_jobs import JobStore from app.services.quote_service import QuoteService from app.strategy import monitor_rules @@ -16,12 +16,14 @@ from app.strategy.monitor import MonitorRuleEngine # ── JobStore 单飞 ──────────────────────────────────────────────────────── -def test_create_singleflight_dedupes_pending_window(tmp_path): +def test_create_singleflight_dedupes_pending_window(monkeypatch, tmp_path): """两次快速 create() 在 pending 窗口内应复用同一 job(is_new=False)。""" + monkeypatch.setattr(preferences, "load", lambda: {"data_source_job_timeout_s": 3600}) store = JobStore(store_dir=tmp_path / "jobs") jid1, new1 = store.create() assert new1 is True + assert store.get(jid1)["timeout_s"] == 3600 # 尚未 start(), job 仍是 pending —— 旧实现会在此另起新 job(并发双跑根因) jid2, new2 = store.create() @@ -35,10 +37,12 @@ def test_create_singleflight_dedupes_pending_window(tmp_path): assert new3 is False -def test_create_new_after_terminal(tmp_path): +def test_create_new_after_terminal(monkeypatch, tmp_path): """job 终态(succeed/fail)后, create() 应给出新 job。""" + monkeypatch.setattr(preferences, "load", lambda: {"data_source_long_job_timeout_s": 5400}) store = JobStore(store_dir=tmp_path / "jobs") - jid1, _ = store.create() + jid1, _ = store.create(long_running=True) + assert store.get(jid1)["timeout_s"] == 5400 store.start(jid1) store.succeed(jid1, {"ok": True}) diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 0d03ab9..c4f2e27 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -969,6 +969,8 @@ export interface Preferences { minute_data_provider?: string realtime_data_provider?: string financial_data_provider?: string + data_source_job_timeout_s: number + data_source_long_job_timeout_s: number realtime_watchlist_symbols?: string[] realtime_pull_stock?: boolean realtime_pull_etf?: boolean @@ -1126,6 +1128,17 @@ export const api = { '/api/settings/preferences/data-providers', { method: 'PUT', body: JSON.stringify(cfg) }, ), + updateDataSourceJobTimeouts: (dataSourceJobTimeoutS: number, dataSourceLongJobTimeoutS: number) => + request>( + '/api/settings/preferences/data-source-job-timeouts', + { + method: 'PUT', + body: JSON.stringify({ + data_source_job_timeout_s: dataSourceJobTimeoutS, + data_source_long_job_timeout_s: dataSourceLongJobTimeoutS, + }), + }, + ), updateMinuteSync: (enabled: boolean, days: number, segmentDays?: number) => request('/api/settings/preferences/minute-sync', { method: 'PUT', diff --git a/frontend/src/pages/settings/DataSources.tsx b/frontend/src/pages/settings/DataSources.tsx index dc57b08..9744fa3 100644 --- a/frontend/src/pages/settings/DataSources.tsx +++ b/frontend/src/pages/settings/DataSources.tsx @@ -1,8 +1,8 @@ import { useState } from 'react' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import { motion, AnimatePresence } from 'framer-motion' -import { Check, Database, Plus, RefreshCw, Zap, FileWarning } from 'lucide-react' -import { api, type DataSourceItem, type PluginDataSourceItem } from '@/lib/api' +import { Check, Clock3, Database, Plus, RefreshCw, Zap, FileWarning } from 'lucide-react' +import { api, type DataSourceItem, type PluginDataSourceItem, type Preferences } from '@/lib/api' import { QK } from '@/lib/queryKeys' import { usePreferences } from '@/lib/useSharedQueries' import { toast } from '@/components/Toast' @@ -15,12 +15,65 @@ const DATASET_LABEL: Record = { minute: '分钟', } +type TimeoutUnit = 'second' | 'minute' | 'hour' + +const TIMEOUT_UNIT_SECONDS: Record = { + second: 1, + minute: 60, + hour: 3600, +} + +function preferredTimeoutUnit(seconds: number): TimeoutUnit { + if (seconds >= 3600 && seconds % 1800 === 0) return 'hour' + if (seconds % 60 === 0) return 'minute' + return 'second' +} + +function formatTimeoutValue(seconds: number, unit: TimeoutUnit): string { + if (!Number.isFinite(seconds)) return '' + const value = seconds / TIMEOUT_UNIT_SECONDS[unit] + return String(Number(value.toFixed(4))) +} + export function SettingsDataSourcesPanel() { const qc = useQueryClient() const prefs = usePreferences() const sources = useQuery({ queryKey: QK.dataSources, queryFn: api.dataSources }) const [selected, setSelected] = useState('tickflow') // 当前在右侧编辑的源 name const [confirmDelete, setConfirmDelete] = useState(null) + const [timeoutDraft, setTimeoutDraft] = useState<{ regular: string; long: string } | null>(null) + const [regularUnitOverride, setRegularUnitOverride] = useState(null) + const [longUnitOverride, setLongUnitOverride] = useState(null) + + const currentRegularTimeout = prefs.data?.data_source_job_timeout_s ?? 1200 + const currentLongTimeout = prefs.data?.data_source_long_job_timeout_s ?? 1800 + const regularTimeoutUnit = regularUnitOverride ?? preferredTimeoutUnit(currentRegularTimeout) + const longTimeoutUnit = longUnitOverride ?? preferredTimeoutUnit(currentLongTimeout) + const regularTimeoutInput = timeoutDraft?.regular + ?? formatTimeoutValue(currentRegularTimeout, regularTimeoutUnit) + const longTimeoutInput = timeoutDraft?.long + ?? formatTimeoutValue(currentLongTimeout, longTimeoutUnit) + const regularInputNumber = Number(regularTimeoutInput) + const longInputNumber = Number(longTimeoutInput) + const regularTimeout = Math.round(regularInputNumber * TIMEOUT_UNIT_SECONDS[regularTimeoutUnit]) + const longTimeout = Math.round(longInputNumber * TIMEOUT_UNIT_SECONDS[longTimeoutUnit]) + const timeoutValuesValid = Number.isFinite(regularInputNumber) && regularInputNumber > 0 + && Number.isFinite(longInputNumber) && longInputNumber > 0 + && regularTimeout >= 60 && longTimeout >= 60 + const timeoutValuesChanged = regularTimeout !== currentRegularTimeout + || longTimeout !== currentLongTimeout + + const saveJobTimeouts = useMutation({ + mutationFn: () => api.updateDataSourceJobTimeouts(regularTimeout, longTimeout), + onSuccess: (saved) => { + qc.setQueryData(QK.preferences, current => ( + current ? { ...current, ...saved } : current + )) + setTimeoutDraft(null) + toast('任务超时配置已保存', 'success') + }, + onError: (e: Error) => toast(`保存失败: ${e.message}`, 'error'), + }) const reload = useMutation({ mutationFn: api.reloadDataSources, @@ -312,6 +365,93 @@ export function SettingsDataSourcesPanel() { +
+
+
+ +
+

数据任务超时

+

+ 后台任务运行超过对应时间后将判定为疑似卡死。保存时自动换算为秒，修改后对新建任务生效。 +

+
+
+ +
+ +
+ + + +
+
+ {/* ===== 下方: 编辑区 ===== */}