[阶段7.1] 同步改为后台任务 + 进度上报,支持回补历史

趋势页提供了"一年"档,但库里只有 7 天数据,那一档形同虚设。
实测每天约 2.84 秒(一天要打 5 个端点),回补一年需要约 28 分钟,
远超任何 HTTP 超时能等的时间。

- sync_status 新增 progress_current / progress_total / started_at
- start_sync() 起后台线程并立即返回,sync_data 每 5 天写一次进度
  (写库便宜但不免费,而前端本来就是 2 秒一轮询)
- POST /api/garmin/sync 改为 202 立即返回,接受 days 参数并
  夹在 1..730;进度经 GET /status 轮询
- 一次新同步会清掉上一次的错误,避免旧错误一直挂在界面上

前端:
- 同步页给出 7 / 30 / 90 / 365 天四个选项,日常与首次回补分开
- 进度条显示"第 N / 共 M 天"与预计耗时,并说明可以离开本页
- 页面挂载时若发现正在同步会接着轮询 —— 回补比页面存活时间长,
  刷新后必须能接上进度

tests (+7, 共 299):
- start_sync 在工作完成前就返回,且返回前已把 total 写好
- 进度随同步推进,结束时等于总天数
- days 超范围被夹到 730
- 新同步清除上一次的错误

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
2026-08-23 21:31:50 +08:00
parent ad88ec7e41
commit 6a5cfa7806
7 changed files with 261 additions and 32 deletions

View File

@@ -261,6 +261,13 @@ def _row_to_dict(row):
# nothing to a table that already exists, so new metrics need an explicit
# additive migration or they silently never appear in production.
MIGRATIONS = {
"sync_status": [
# A full backfill runs for many minutes, so the UI needs to show how
# far along it is rather than an indefinite spinner.
("progress_current", "INT"),
("progress_total", "INT"),
("started_at", "DATETIME"),
],
"health_data": [
# activity / energy
("distance_meters", "DOUBLE"),

View File

@@ -36,8 +36,16 @@ def sync():
400,
)
result = garmin_svc.sync_data(g.user_id, creds)
return jsonify(result)
days = request.get_json(silent=True).get("days") if request.is_json else None
try:
days = max(1, min(int(days), 730)) if days else None
except (TypeError, ValueError):
days = None
# Always run in the background: even a week takes ~20s, and a full
# backfill runs for many minutes. Progress is polled via /status.
result = garmin_svc.start_sync(g.user_id, creds, days)
return jsonify(result), 202
@bp.route("/auth-status", methods=["GET"])

View File

@@ -21,6 +21,7 @@ so passing it a date silently asks for activity number "2026-08-23".
"""
import datetime
import os
import threading
from db import execute, query_one
from config import DB_TYPE
@@ -62,6 +63,9 @@ def get_sync_status(user_id):
"lastSyncTime": row["last_sync_time"],
"recordsSynced": row["records_synced"],
"lastError": row["last_error"],
"progressCurrent": row.get("progress_current"),
"progressTotal": row.get("progress_total"),
"startedAt": row.get("started_at"),
}
@@ -411,6 +415,31 @@ def _sync_activities(client, user_id, start_date, end_date):
return stored
# Above this many days a sync is long enough that the caller must not block
# on it — a year takes roughly 20 minutes at ~3s per day.
BACKGROUND_THRESHOLD_DAYS = 14
def start_sync(user_id, creds, days=None):
"""Run a sync in the background and return immediately.
Progress lands in sync_status, which the UI polls; a full backfill runs
far longer than any sensible HTTP timeout.
"""
days = days or DEFAULT_SYNC_DAYS
now = datetime.datetime.utcnow().isoformat(timespec="seconds")
_set_sync_status(
user_id, "syncing", now,
records_synced=0, progress_current=0, progress_total=days,
started_at=now, last_error=None,
)
thread = threading.Thread(
target=sync_data, args=(user_id, creds, days), daemon=True
)
thread.start()
return {"status": "syncing", "days": days}
def sync_data(user_id, creds, days=None, client=None):
"""Pull the last `days` days from Garmin Connect into the local database.
@@ -418,7 +447,10 @@ def sync_data(user_id, creds, days=None, client=None):
"""
days = days or DEFAULT_SYNC_DAYS
now = datetime.datetime.utcnow().isoformat(timespec="seconds")
_set_sync_status(user_id, "syncing", now, records_synced=0)
_set_sync_status(
user_id, "syncing", now,
records_synced=0, progress_current=0, progress_total=days,
)
try:
client = client or _connect(creds, user_id)
@@ -451,6 +483,15 @@ def sync_data(user_id, creds, days=None, client=None):
health.upsert_health_daily(user_id, record)
days_synced += 1
# Reported every few days rather than every day: the write is cheap
# but not free, and the UI polls on a 2s cadence anyway.
if (i + 1) % 5 == 0 or i + 1 == days:
_set_sync_status(
user_id, "syncing", now,
records_synced=days_synced, progress_current=i + 1,
progress_total=days,
)
activities_synced = 0
try:
activities_synced = _sync_activities(
@@ -482,6 +523,7 @@ def sync_data(user_id, creds, days=None, client=None):
_set_sync_status(
user_id, "idle", now, records_synced=days_synced,
progress_current=days, progress_total=days,
last_error="; ".join(day_errors[:3]) if day_errors else None,
)
message = (

View File

@@ -527,3 +527,78 @@ class TestTimestampNormalisation:
def test_nonsense_value_does_not_raise(self):
assert garmin_svc._to_datetime(float("inf")) is None
class TestBackgroundSync:
"""A full backfill runs for ~20 minutes at ~3s per day, so the request
must not block on it and the UI needs progress rather than a spinner."""
def test_progress_is_reported_during_the_run(self, db, user):
garmin_svc.sync_data(user["id"], CREDS, days=10, client=StubClient())
status = garmin_svc.get_sync_status(user["id"])
assert status["progressTotal"] == 10
assert status["progressCurrent"] == 10
def test_progress_total_matches_the_requested_window(self, db, user):
garmin_svc.sync_data(user["id"], CREDS, days=3, client=StubClient())
assert garmin_svc.get_sync_status(user["id"])["progressTotal"] == 3
def test_start_sync_returns_immediately(self, db, user, monkeypatch):
import threading
release = threading.Event()
def slow(uid, creds, days=None, client=None):
release.wait(5)
monkeypatch.setattr(garmin_svc, "sync_data", slow)
out = garmin_svc.start_sync(user["id"], CREDS, days=365)
# Returns before the work finishes.
assert out["status"] == "syncing"
assert out["days"] == 365
assert garmin_svc.get_sync_status(user["id"])["status"] == "syncing"
release.set()
def test_start_sync_marks_total_before_any_work(self, db, user, monkeypatch):
import threading
release = threading.Event()
monkeypatch.setattr(
garmin_svc, "sync_data", lambda *a, **k: release.wait(5)
)
garmin_svc.start_sync(user["id"], CREDS, days=200)
status = garmin_svc.get_sync_status(user["id"])
assert status["progressTotal"] == 200
assert status["progressCurrent"] == 0
release.set()
def test_previous_error_is_cleared_when_a_new_sync_starts(
self, db, user, monkeypatch
):
garmin_svc.sync_data(
user["id"], CREDS, days=1, client=StubClient(fail_days=[day(0)])
)
assert garmin_svc.get_sync_status(user["id"])["lastError"]
import threading
release = threading.Event()
monkeypatch.setattr(garmin_svc, "sync_data", lambda *a, **k: release.wait(5))
garmin_svc.start_sync(user["id"], CREDS, days=7)
assert not garmin_svc.get_sync_status(user["id"])["lastError"]
release.set()
def test_endpoint_returns_202_without_waiting(self, client, auth, user, db, monkeypatch):
garmin_svc.save_token(user["id"], "blob", "g@example.com")
monkeypatch.setattr(garmin_svc, "start_sync", lambda *a, **k: {"status": "syncing", "days": 30})
r = client.post("/api/garmin/sync", headers=auth, json={"days": 30})
assert r.status_code == 202
assert r.get_json()["status"] == "syncing"
def test_days_is_clamped_to_a_sane_range(self, client, auth, user, db, monkeypatch):
garmin_svc.save_token(user["id"], "blob", "g@example.com")
seen = {}
monkeypatch.setattr(
garmin_svc, "start_sync",
lambda uid, creds, days=None: seen.setdefault("days", days) or {"status": "syncing"},
)
client.post("/api/garmin/sync", headers=auth, json={"days": 99999})
assert seen["days"] == 730