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