Merge pull request #92 from openqj/feat-sub2api-hotfix

fix: docker 默认 runner 支持 env + 并发开通竞态复用

cloud/sandbox.py

@@ -1382,10 +1382,20 @@ def _parse_host_result(task_id, fake):
     }
 
 
-def _docker_default_runner(cmd, timeout=None):
+def _docker_default_runner(cmd, timeout=None, env=None):
-    """docker 的默认 runner：subprocess.run，支持 timeout。"""
+    """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):

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,

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"}])

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):