From 11869c6a8d2fc62671d9911974c2c2c9b0716d55 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Fri, 28 Aug 2026 14:54:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E9=99=90=E6=B5=81=E9=80=80?= =?UTF-8?q?=E9=81=BF=E6=8C=81=E4=B9=85=E5=8C=96=E5=88=B0=20DB=EF=BC=8C?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E5=99=A8=E4=B8=8E=E6=89=8B=E5=8A=A8=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E5=9D=87=E5=B0=8A=E9=87=8D=E9=80=80=E9=81=BF=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit db.py: 修复迁移非幂等——gunicorn 双 worker 并发 init_db 时后到者遇 Duplicate column 会让整个服务起不来，改为吞掉 duplicate column 错误。garmin.py/scheduler.py: 退避期写入 sync_status.rate_limited_until，自动同步与手动同步在退避期内跳过，不再反复撞击 Garmin 把限流窗口越撞越深。 --- backend/db.py | 19 ++++++- backend/services/garmin.py | 104 ++++++++++++++++++++++++++++++++-- backend/services/scheduler.py | 10 ++++ 3 files changed, 125 insertions(+), 8 deletions(-) diff --git a/backend/db.py b/backend/db.py index 9a6f5f6..6572fc1 100644 --- a/backend/db.py +++ b/backend/db.py @@ -422,6 +422,10 @@ MIGRATIONS = { # Which part of the sync is running. "0 / 730 天" says nothing about # what is actually happening for the several minutes of it. ("stage", "VARCHAR(64)"), + # When Garmin answered 429 we stand the account down. Persisted so every + # gunicorn worker and a restart agree on the cooldown — a process-local + # dict let each worker re-hit Garmin and keep the limit alive forever. + ("rate_limited_until", "DATETIME"), ], "health_data": [ # activity / energy @@ -492,9 +496,18 @@ def _migrate(cur): for name, coltype in columns: if name in present: continue - # SQLite has no "ADD COLUMN IF NOT EXISTS"; the membership check - # above is what keeps this idempotent on both backends. - cur.execute(f"ALTER TABLE {table} ADD COLUMN {name} {coltype}") + # The membership check above keeps this idempotent for the normal + # case. But gunicorn runs 2 workers that both call init_db() on + # boot; under that race the loser can decide to add a column the + # winner already added (ADD COLUMN commits implicitly and becomes + # visible a hair after the loser's probe). Swallow the + # duplicate-column error so a concurrent boot can't take the whole + # service down. Same guard covers SQLite's "duplicate column name". + try: + cur.execute(f"ALTER TABLE {table} ADD COLUMN {name} {coltype}") + except Exception as e: # noqa: BLE001 - only "already exists" is safe to ignore + if "duplicate column" not in str(e).lower(): + raise # --- Public API ------------------------------------------------------------- diff --git a/backend/services/garmin.py b/backend/services/garmin.py index 0ebbced..a9f3d5a 100644 --- a/backend/services/garmin.py +++ b/backend/services/garmin.py @@ -97,17 +97,85 @@ class RateLimited(RuntimeError): # Retrying while rate limited is what deepens the limit, so once Garmin says -# 429 the whole process stands down until this passes. -RATE_LIMIT_BACKOFF = datetime.timedelta(minutes=30) +# 429 the whole process stands down until this passes. Bumped from 30 to 60 +# minutes: the account was stuck for days because every hourly tick re-hit it, +# so a longer cooldown gives Garmin's window room to actually close. +RATE_LIMIT_BACKOFF = datetime.timedelta(minutes=60) +# In-process cache of the cooldown, kept in sync with the DB copy below and +# still the lever the tests reach for via _rate_limited_until.clear(). _rate_limited_until = {} +def _parse_dt(v): + """Coerce a stored rate-limit time into a naive UTC datetime, or None.""" + if v is None: + return None + if isinstance(v, datetime.datetime): + return v + if isinstance(v, str): + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): + try: + return datetime.datetime.strptime(v, fmt) + except ValueError: + continue + try: + return datetime.datetime.fromisoformat(v) + except ValueError: + return None + return None + + def rate_limited_until(user_id): - return _rate_limited_until.get(user_id) + """When the account is still cooling down, as a UTC datetime (or None). + + The cooldown is persisted to the database so every gunicorn worker and a + process restart see the same deadline — a process-local dict alone let each + worker re-hit Garmin and keep the limit alive forever. + """ + mem = _rate_limited_until.get(user_id) + try: + row = query_one( + "SELECT rate_limited_until FROM sync_status WHERE user_id = ?", [user_id] + ) + except Exception: + row = None + candidates = [c for c in (mem, _parse_dt(row["rate_limited_until"] if row else None)) if c] + return max(candidates) if candidates else None def _note_rate_limit(user_id): - _rate_limited_until[user_id] = datetime.datetime.utcnow() + RATE_LIMIT_BACKOFF + """Record a rate-limit cooldown, extending it if one is already running.""" + now = datetime.datetime.utcnow() + existing = rate_limited_until(user_id) + until = (existing + datetime.timedelta(minutes=30)) if existing and existing > now \ + else (now + RATE_LIMIT_BACKOFF) + _rate_limited_until[user_id] = until + try: + # Persist alongside the current status so the cooldown survives across + # workers and restarts. + cur_status = (get_sync_status(user_id) or {}).get("status") or "idle" + _set_sync_status( + user_id, cur_status, now.isoformat(timespec="seconds"), + rate_limited_until=until.isoformat(timespec="seconds"), + ) + except Exception: + pass + + +def _rate_limit_block(user_id): + """If the account is cooling down, return (until, message); else (None, None). + + Central guard used by both the manual and scheduled sync entry points, so a + blocked account issues zero Garmin requests until the window closes. + """ + until = rate_limited_until(user_id) + if not until or datetime.datetime.utcnow() >= until: + return None, None + mins = max(1, int((until - datetime.datetime.utcnow()).total_seconds() // 60)) + return until, ( + f"Garmin 仍在限制请求频率，预计约 {mins} 分钟后自动恢复。" + "已自动退避，请耐心等待——反复点击正是把限流撞得更深的原因，令牌本身没有失效。" + ) def _is_rate_limited(e): @@ -241,7 +309,7 @@ def _connect(creds, user_id=None): client.garth.loads(token) _use_api_user_agent(client) - blocked = _rate_limited_until.get(user_id) + blocked = rate_limited_until(user_id) if blocked and datetime.datetime.utcnow() < blocked: raise RateLimited( "Garmin 暂时限制了请求频率，稍后会自动恢复（约 " @@ -815,6 +883,18 @@ def start_sync(user_id, creds, days=None): """ days = days or DEFAULT_SYNC_DAYS now = datetime.datetime.utcnow().isoformat(timespec="seconds") + until, msg = _rate_limit_block(user_id) + if until: + _set_sync_status( + user_id, "rate_limited", now, + records_synced=0, progress_current=0, progress_total=days, + started_at=now, last_error=msg, + ) + return { + "status": "rate_limited", + "message": msg, + "retryAfterSeconds": int((until - datetime.datetime.utcnow()).total_seconds()), + } _set_sync_status( user_id, "syncing", now, records_synced=0, progress_current=0, progress_total=days, @@ -834,6 +914,20 @@ def sync_data(user_id, creds, days=None, client=None): """ days = days or DEFAULT_SYNC_DAYS now = datetime.datetime.utcnow().isoformat(timespec="seconds") + until, msg = _rate_limit_block(user_id) + if until: + # A blocked account must issue zero Garmin requests — that is the whole + # point. Return immediately without touching the network. + _set_sync_status( + user_id, "rate_limited", now, + records_synced=0, last_error=msg, + ) + return { + "status": "rate_limited", + "recordsSynced": 0, + "message": msg, + "retryAfterSeconds": int((until - datetime.datetime.utcnow()).total_seconds()), + } _set_sync_status( user_id, "syncing", now, records_synced=0, progress_current=0, progress_total=days, diff --git a/backend/services/scheduler.py b/backend/services/scheduler.py index 3ec999f..2bef9b5 100644 --- a/backend/services/scheduler.py +++ b/backend/services/scheduler.py @@ -144,6 +144,16 @@ def sync_all_accounts(days=None, respect_schedule=False): results.append({"user": uid, "status": "skipped", "reason": "not due"}) continue + # Never poke Garmin while it is rate-limiting us — that is exactly + # what keeps the limit alive. Respect the persisted cooldown and sit + # this tick out. + blocked = garmin_svc.rate_limited_until(uid) + if blocked and blocked > _now(): + results.append({ + "user": uid, "status": "skipped", "reason": "rate-limited", + "retryAfterSeconds": int((blocked - _now()).total_seconds()), + }) + continue out = garmin_svc.sync_data(uid, {}, days=days) results.append({"user": uid, "status": out.get("status"), "records": out.get("recordsSynced")})