From 4936db3c4737b74765e3d3254802b0fa51f9d9b8 Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 17:59:04 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20docker=20=E9=BB=98=E8=AE=A4=20runner=20?= =?UTF-8?q?=E6=94=AF=E6=8C=81=20env=20+=20=E5=B9=B6=E5=8F=91=E5=BC=80?= =?UTF-8?q?=E9=80=9A=E7=AB=9E=E6=80=81=E5=A4=8D=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 翻转实弹暴露的两个 #88 缺口： - _docker_default_runner 不接受 env，_call_runner 遇 env 即 loud fail， 导致 sub2api 路径下所有 docker 任务失败（xcmapi 路径不传 env， 此前未暴露）。默认 runner 现接受 env 并与进程环境合并后传给 docker CLI 子进程（-e <名> 由此解析，值绝不进 argv）。 - 同一用户两任务并发开通时建号撞 email 已存在：现轮询复用 胜出者入库的行，等不到再原错抛出。 测试：sub2api 16、docker 沙箱 29；全量 cloud 789（4 跳过）、 host 65、deploy 19 全绿。 --- cloud/sandbox.py | 16 ++++++++++--- cloud/sub2api.py | 18 +++++++++++++-- cloud/tests/test_docker_sandbox.py | 25 +++++++++++++++++++++ cloud/tests/test_sub2api.py | 36 ++++++++++++++++++++++++++++++ 4 files changed, 90 insertions(+), 5 deletions(-) diff --git a/cloud/sandbox.py b/cloud/sandbox.py index d690da7..a05ecf0 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -1382,10 +1382,20 @@ def _parse_host_result(task_id, fake): } -def _docker_default_runner(cmd, timeout=None): - """docker 的默认 runner：subprocess.run，支持 timeout。""" +def _docker_default_runner(cmd, timeout=None, env=None): + """docker 的默认 runner：subprocess.run，支持 timeout 与 env。 + + env 非空时与当前进程环境合并后传给 docker CLI 子进程 + （-e <名> 的密钥值由此解析，绝不进 argv；只替换不合并会 + 丢 PATH/HOME，docker CLI 本身都跑不起来）。#88 只给 + _call_runner 加了 env 透传却没给默认 runner 加 env 参数， + 导致生产 docker 任务一律 loud fail，本补丁补上。 + """ + full_env = dict(os.environ) + if env: + full_env.update(env) return subprocess.run(cmd, capture_output=True, text=True, - timeout=timeout) + timeout=timeout, env=full_env) class CloudExecutor(Executor): diff --git a/cloud/sub2api.py b/cloud/sub2api.py index d6a51b6..3fddbe7 100644 --- a/cloud/sub2api.py +++ b/cloud/sub2api.py @@ -309,7 +309,8 @@ def ensure_user_gateway_key(conn, user_id, admin=None, request=None): 入库失败时尽力清理刚建的 key 与账号（避免孤儿 key），再抛原错； 授余额失败同样走这条清理（余额为 0 的 key 在网关侧直接 403， - 留着没有意义）。调用方负责 conn 的提交语义：本函数在入库后 + 留着没有意义）。同一用户并发开通时建号撞 email 已存在， + 则轮询复用胜出者入库的行。调用方负责 conn 的提交语义：本函数在入库后 commit（开通是独立事务，不跟调用方的业务事务绑在一起）。 """ ensure_gateway_keys_table(conn) @@ -324,7 +325,20 @@ def ensure_user_gateway_key(conn, user_id, admin=None, request=None): email = "codex-%s@codex.local" % user_id username = "codex-" + str(user_id)[:12] password = secrets.token_hex(16) - sub_id = admin.create_user(email, password, username) + try: + sub_id = admin.create_user(email, password, username) + except Sub2apiError as exc: + if "already exists" not in str(exc).lower(): + raise + # 并发开通竞态：同一用户的另一路已建号（email 唯一）， + # 等它入库后复用那一行；等不到则原错抛出（任务失败后 + # 可重跑，孤儿号场景下重跑仍撞号需管理员介入清理）。 + for _ in range(5): + time.sleep(2) + row = get_user_gateway_key(conn, user_id) + if row is not None: + return row["api_key"] + raise key_id = None try: user_token, refresh = user_login(origin, email, password, diff --git a/cloud/tests/test_docker_sandbox.py b/cloud/tests/test_docker_sandbox.py index d02b26f..f8e5a8e 100644 --- a/cloud/tests/test_docker_sandbox.py +++ b/cloud/tests/test_docker_sandbox.py @@ -336,3 +336,28 @@ class TestDockerExtraMounts(DockerCommandTestCase): DockerExecutor("/fake/host.py", extra_mounts=["/tmp:/app"]) with self.assertRaises(ValueError): DockerExecutor("/fake/host.py", extra_mounts=["/nonexistent-xyz:/x"]) + + +class TestDockerDefaultRunnerEnv(unittest.TestCase): + def test_env_merged_not_replaced(self): + from cloud.sandbox import _docker_default_runner + out = _docker_default_runner( + ["python3", "-c", + "import os,sys;sys.stdout.write(" + "os.environ.get('PROBE_X','')+':'+" + "('1' if os.environ.get('PATH') else '0'))"], + env={"PROBE_X": "42"}) + self.assertEqual(out.returncode, 0) + self.assertEqual(out.stdout, "42:1") + + def test_call_runner_passes_env_to_default_runner(self): + seen = [] + + def recording(cmd, timeout=None, env=None): + seen.append(env) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + ex = DockerExecutor("/fake/host.py", runner=recording, + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + ex._call_runner(["docker", "version"], env={"SUB2API_API_KEY": "x"}) + self.assertEqual(seen, [{"SUB2API_API_KEY": "x"}]) diff --git a/cloud/tests/test_sub2api.py b/cloud/tests/test_sub2api.py index 16b5341..7533c1b 100644 --- a/cloud/tests/test_sub2api.py +++ b/cloud/tests/test_sub2api.py @@ -9,6 +9,7 @@ import sqlite3 import sys import unittest from pathlib import Path +from unittest import mock try: from cloud import sub2api @@ -38,6 +39,8 @@ class FakeApi: self.fail_refresh_once = False self.balance_calls = [] self.fail_balance = False + self.fail_create_user = False + self.create_user_hook = None def __call__(self, method, url, body=None, token=None): path = url.split("28084", 1)[-1] if "28084" in url else url @@ -62,6 +65,10 @@ class FakeApi: return 200, {"data": {"balance": body.get("balance")}} if path == "/api/v1/admin/users" and method == "POST": self.next_user_id += 1 + if self.create_user_hook is not None: + self.create_user_hook(self.next_user_id) + if self.fail_create_user: + return 400, {"message": "email already exists"} return 200, {"data": {"id": self.next_user_id}} if path == "/api/v1/admin/groups" and method == "GET": return 200, {"data": self.groups} @@ -180,6 +187,35 @@ class TestEnsureGatewayKey(unittest.TestCase): os.environ["CLOUD_SUB2API_OPENING_BALANCE_USD"] = old self.assertEqual(self.api.balance_calls, []) + def test_provision_race_reuses_winner_row(self): + conn = _conn() + + def hook(_new_id): + conn.execute( + "INSERT INTO user_gateway_keys(user_id, sub2api_user_id," + " api_key, refresh_token, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?)", + ("user-race", 99, "sk-race", "ref-race", 0, 0)) + conn.commit() + + self.api.create_user_hook = hook + self.api.fail_create_user = True + key = ensure_user_gateway_key(conn, "user-race", admin=self.admin, + request=self.api) + self.assertEqual(key, "sk-race") + + def test_provision_race_no_row_reraises(self): + self.api.fail_create_user = True + conn = _conn() + with mock.patch("time.sleep", return_value=None): + with self.assertRaises(sub2api.Sub2apiError): + ensure_user_gateway_key(conn, "user-race2", + admin=self.admin, request=self.api) + row = conn.execute( + "SELECT COUNT(*) FROM user_gateway_keys" + " WHERE user_id = ?", ("user-race2",)).fetchone() + self.assertEqual(row[0], 0) + class TestProviderCommand(unittest.TestCase): def test_server_command_sub2api(self):