From a2d6a5c57f0cf2f9e5fc1c881b44e0d68b997a51 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Tue, 1 Sep 2026 15:46:42 +0800 Subject: [PATCH] =?UTF-8?q?fix(ai):=20=E9=98=9F=E5=88=97=E4=BC=9A=E6=8A=8A?= =?UTF-8?q?=E5=85=B1=E7=94=A8=E7=9A=84=E7=BD=91=E5=85=B3=E6=89=93=E5=88=B0?= =?UTF-8?q?=20502=EF=BC=8C=E5=8A=A0=E5=B9=B6=E5=8F=91=E4=B8=8A=E9=99=90?= =?UTF-8?q?=E5=92=8C=E9=97=B4=E9=9A=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 排查生产上一直生成不出来，发现网关在返 502。上去看：机器好好的、systemd 说 active、5100 端口在监听——但它是 `gunicorn -w 1 --threads 4`，全部并发 就四个，而且 fam-edge 和摄像头项目也在用同一个。 我们这边一个请求占一个线程 2~5 分钟，NVIDIA 链重试起来最坏十七分钟（它自己 README 已知问题 #3）。而我写的 worker 是跑完一个立刻拉下一个，同步后还有八 个 scope 排队——等于拿满线程不撒手。这个 502 大概率是我打出来的，而且顺带 把另外两个项目也打下线了。 - 并发按整个部署计算，不是每个 gunicorn worker 一个：claim 前先数全局 running（两个 worker 各跑「一个」就是两个并发） - 每跑完一个任务停 20 秒，不只是空闲时才停 - 两个都可用环境变量调，注释里写清楚调大的代价是什么 补齐的历史数据晚二十分钟到没有任何人受影响；网关不响应是三个项目一起受影响。 Co-Authored-By: Claude Opus 5 --- backend/.env.example | 13 +++++++++++++ backend/services/jobs.py | 36 +++++++++++++++++++++++++++++++----- backend/tests/test_coach.py | 32 ++++++++++++++++++++++++++++++++ 3 files changed, 76 insertions(+), 5 deletions(-) diff --git a/backend/.env.example b/backend/.env.example index e775e68..4e4c1e0 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -89,3 +89,16 @@ AI_MAX_TOKENS=1024 # before the answer starts, and at 1024 the reply was all thinking with the # JSON truncated away. AI_COACH_MAX_TOKENS=4000 + +# --- AI coach job queue --- +# The gateway runs `gunicorn -w 1 --threads 4` and is shared with fam-edge and +# the camera project: four concurrent requests for everyone, while one of ours +# holds a thread for 2-5 minutes. So this consumer runs one job at a time +# across the whole deployment (not one per Gunicorn worker) and waits between +# jobs. Raising either of these makes the backfill finish sooner at the cost of +# the shared box — a 502 there is a 502 for the other two projects as well. +AI_JOB_CONCURRENCY=1 +AI_JOB_GAP_SECONDS=20 +# Set AI_JOBS=false to stop consuming entirely (screens then show the computed +# figures with no model reading). +AI_JOBS=true diff --git a/backend/services/jobs.py b/backend/services/jobs.py index 2c12c55..6ef4db1 100644 --- a/backend/services/jobs.py +++ b/backend/services/jobs.py @@ -37,6 +37,20 @@ POLL_SECONDS = float(os.environ.get("AI_JOB_POLL_SECONDS") or 2) MAX_ATTEMPTS = int(os.environ.get("AI_JOB_MAX_ATTEMPTS") or 3) +# The gateway is not ours alone. It runs `gunicorn -w 1 --threads 4` on the +# Oracle box and is shared with fam-edge and the camera project, so it can +# serve four requests at a time in total — while one of ours occupies a thread +# for two to five minutes, and up to seventeen when its NVIDIA chain retries +# (its own README, known issue #3). +# +# So this consumer deliberately runs one job at a time across the whole +# deployment, not one per Gunicorn worker, and leaves a gap between jobs. A +# backfill of eight scopes is background work; taking the shared box down to +# finish it sooner is not a trade worth making — and a 502 there is a 502 for +# the other two projects too. +MAX_CONCURRENT = int(os.environ.get("AI_JOB_CONCURRENCY") or 1) +GAP_SECONDS = float(os.environ.get("AI_JOB_GAP_SECONDS") or 20) + # How long a job that exhausted its attempts stays given up on before it is # tried again. Without this a transient upstream outage is permanent: three # quick failures while the gateway is unreachable would retire that screen's @@ -200,6 +214,18 @@ def _claim_next(): the one that has waited longest goes first. """ cutoff = _iso(_now() - datetime.timedelta(seconds=CLAIM_TIMEOUT_SECONDS)) + + # Concurrency is counted across the deployment, not per process: two + # Gunicorn workers each running "one job" is two concurrent calls into a + # gateway that has four threads for three projects. + running = query_one( + "SELECT COUNT(*) AS n FROM ai_jobs WHERE status = 'running' " + "AND claimed_at IS NOT NULL AND claimed_at >= ?", + [cutoff], + ) + if (running or {}).get("n", 0) >= MAX_CONCURRENT: + return None + rows = query_all( "SELECT * FROM ai_jobs WHERE status = 'pending' " "OR (status = 'running' AND (claimed_at IS NULL OR claimed_at < ?)) " @@ -254,11 +280,11 @@ def run_once(): def _loop(): while True: try: - # Straight on to the next job when one was just run: after a - # sync there is a whole backfill waiting, and sleeping between - # each would add hours to it for no reason. - if not run_once(): - time.sleep(POLL_SECONDS) + # A gap after each job, not just when idle. Running them + # back-to-back is what saturates the shared gateway: a backfill + # finishing twenty minutes later costs nobody anything, a gateway + # that stops answering costs all three projects. + time.sleep(GAP_SECONDS if run_once() else POLL_SECONDS) except Exception as e: # noqa: BLE001 - the loop must outlive any failure print(f"[ai-jobs] worker error: {e}") time.sleep(POLL_SECONDS) diff --git a/backend/tests/test_coach.py b/backend/tests/test_coach.py index a8175a5..c1475b4 100644 --- a/backend/tests/test_coach.py +++ b/backend/tests/test_coach.py @@ -1073,3 +1073,35 @@ class TestHighlightsReadAlone: f"{name}: 折叠态只显示 detail，但它没提到 {title!r}: " f"{first['detail']!r}" ) + + +class TestGatewayCourtesy: + """The gateway runs one worker with four threads and is shared with two + other projects. This consumer must not be able to saturate it.""" + + def test_only_one_job_runs_at_a_time_across_the_deployment(self, db, user): + jobs.enqueue(user["id"], "health", "a") + jobs.enqueue(user["id"], "sleep", "b") + assert jobs._claim_next() is not None + assert jobs._claim_next() is None, \ + "a second Gunicorn worker must not start a second gateway call" + + def test_a_finished_job_frees_the_slot(self, db, user): + jobs.enqueue(user["id"], "health", "a") + jobs.enqueue(user["id"], "sleep", "b") + first = jobs._claim_next() + jobs._finish(first["id"]) + assert jobs._claim_next() is not None + + def test_an_abandoned_claim_does_not_block_the_queue_forever(self, db, user): + jobs.enqueue(user["id"], "health", "a") + jobs.enqueue(user["id"], "sleep", "b") + jobs._claim_next() + stale = jobs._now() - jobs.datetime.timedelta( + seconds=jobs.CLAIM_TIMEOUT_SECONDS + 60) + db.execute("UPDATE ai_jobs SET claimed_at = ?", [jobs._iso(stale)]) + assert jobs._claim_next() is not None + + def test_there_is_a_gap_between_jobs(self): + """Back-to-back is what saturates a four-thread box.""" + assert jobs.GAP_SECONDS > 0