Files
codex-app-server/cloud/tests/test_server_retry.py
host d51c06a900 Worker concurrency (capacity enforced), survival (heartbeat/sweep/drain), retry endpoint
1. Concurrency: capacity is now a real limit, not display-only.
   run_all with capacity>1 runs a manual thread pool (threading,
   not concurrent.futures: a test in this repo puts cloud/ on
   sys.path, shadowing stdlib queue and breaking
   concurrent.futures' SimpleQueue import). Claim stays CAS, so
   two threads racing one task still yield one owner; max_tasks
   counts dispatched+running to avoid overshoot; stop signals
   stop new claims only, running tasks are drained by run_loop.
   Running state moved from single fields to a _running dict
   (proc/container per task) so cancel kills the right task.
   Docker timeout is now a per-call argument, never a mutated
   shared executor field. CostStore opens a fresh connection per
   record/report (shared sqlite connection is thread-bound and
   raised ProgrammingError under the pool; :memory: keeps the
   shared connection because a fresh one is an empty DB).
   Final status write is CAS on status='running' so a background
   thread cannot overwrite an interrupted task with completed.
2. Survival: run_loop starts a daemon heartbeat thread (every
   30s, refreshes running tasks' last_heartbeat) and sweeps stale
   tasks every round (threshold 300s, best-effort); live tasks
   never go stale now, dead workers' tasks are reclaimed without
   manual /admin/sweep. On exit, run_loop waits up to
   drain_timeout (CLI --drain-timeout, default 60s, inside
   systemd's 90s stop window) for running tasks before killing
   leftovers and marking interrupted.
3. Retry: POST /tasks/{id}/retry (maintainer+, 404/403/409):
   failed/interrupted/merge-failed go back to queued with
   result/error/model cleared, prompt/model/deps kept, CAS
   update, audit task_retried.
Tests: 5 concurrency tests (barrier overlap, serial capacity=1,
max_tasks, wait_for_running, CAS final write), 3 retry tests;
full suites pass.
2026-10-26 06:46:20 +08:00

137 lines
4.7 KiB
Python

"""失败任务重跑端点测试:POST /tasks/{id}/retry。
只测状态机本身,不执行真实任务:失败/中断任务重回 queued,
上一轮结果清空,prompt 与模型原样保留;权限与状态码语义
与 merge 端点对齐(maintainer 以上、404/401/403/409)。
"""
import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import unittest
from pathlib import Path
try:
from cloud.auth import AuthStore
from cloud.server import create_server
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.server import create_server
class RetryTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.contrib_id, self.contrib_key = self.auth.create_user("contrib")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
status, self.project = self._req(
"POST", "/projects", {"name": "demo"}, self.owner_key)
self.assertEqual(status, 201)
self.auth.set_member(
self.project["id"], self.owner_id, self.contrib_id, "contributor")
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
try:
self.auth.conn.close()
except Exception:
pass
self.tmp.cleanup()
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
hdrs = {}
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = None
return resp.status, parsed
def _failed_task(self):
status, task = self._req(
"POST", "/tasks",
{"project_id": self.project["id"], "prompt": "x"},
self.owner_key)
self.assertEqual(status, 201)
conn = sqlite3.connect(self.db_path)
try:
conn.execute(
"UPDATE tasks SET status = 'failed',"
" error_text = 'old boom', actual_model = 'm1'"
" WHERE id = ?",
(task["id"],))
conn.commit()
finally:
conn.close()
return task
def test_retry_failed_by_owner_returns_queued_and_clears_result(self):
task = self._failed_task()
status, body = self._req(
"POST", "/tasks/%s/retry" % task["id"], {}, self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(body["status"], "queued")
self.assertIsNone(body.get("error_text"))
self.assertIsNone(body.get("actual_model"))
conn = sqlite3.connect(self.db_path)
try:
actions = [r[0] for r in conn.execute(
"SELECT action FROM audit_log WHERE project_id = ?",
(self.project["id"],)).fetchall()]
finally:
conn.close()
self.assertIn("task_retried", actions)
def test_retry_non_failed_409(self):
status, task = self._req(
"POST", "/tasks",
{"project_id": self.project["id"], "prompt": "x"},
self.owner_key)
self.assertEqual(status, 201)
status, body = self._req(
"POST", "/tasks/%s/retry" % task["id"], {}, self.owner_key)
self.assertEqual(status, 409)
def test_retry_by_contributor_forbidden_and_missing_404(self):
task = self._failed_task()
status, _ = self._req(
"POST", "/tasks/%s/retry" % task["id"], {}, self.contrib_key)
self.assertEqual(status, 403)
status, _ = self._req(
"POST", "/tasks/nope/retry", {}, self.owner_key)
self.assertEqual(status, 404)
status, _ = self._req(
"POST", "/tasks/%s/retry" % task["id"], {})
self.assertEqual(status, 401)
if __name__ == "__main__":
unittest.main()