fix(release): verify and repair new-api user quota

game-runtime/tools/newapi_pool_provision.py

@@ -141,6 +141,10 @@ class Postgres:
         line = out.splitlines()[0].strip() if out else ""
         return int(line) if line else None
 
+    def set_user_quota(self, uid: int, quota: int) -> None:
+        """修复当前 new-api 版本会静默忽略管理 API quota 字段的兼容路径。"""
+        self._run(f"UPDATE users SET quota={quota} WHERE id={uid};", "set-user-quota")
+
     def set_access_token(self, uid: int, access_token: str) -> None:
         # access_token 无法经 API 设，必须 DB 直写（S0 坐实的关键步）
         self._run(f"UPDATE users SET access_token='{access_token}' WHERE id={uid};", "set-access-token")
@@ -247,6 +251,30 @@ def wait_for_db_visibility(query, attempts: int = 20, interval_seconds: float =
     return None
 
 
+def ensure_user_quota(pg: Postgres, root_token: str, uid: int, username: str, grant: int) -> str:
+    """设置并核验用户额度；API 漂移时经同一受控数据库通道完成补偿。"""
+    if pg.query_user_quota(uid) == grant:
+        return "reused"
+    _http_json("PUT", "/api/user/", root_token, ROOT_UID, {
+        "id": uid,
+        "username": username,
+        "display_name": username[:20],
+        "quota": grant,
+        "group": "default",
+    })
+    visible = wait_for_db_visibility(
+        lambda: grant if pg.query_user_quota(uid) == grant else None,
+        attempts=5,
+        interval_seconds=0.1,
+    )
+    if visible == grant:
+        return "api"
+    pg.set_user_quota(uid, grant)
+    if pg.query_user_quota(uid) != grant:
+        raise RuntimeError(f"user.quota 补偿后仍不一致 uid={uid}")
+    return "database-repair"
+
+
 def validate_provision_request(start: int, count: int, prefix: str, hostname: str) -> list[str]:
     """校验离线补池边界，并返回确定性的目标用户名。
 
@@ -325,18 +353,13 @@ def provision_one(pg: Postgres, root_token: str, username: str, grant: int,
     token_id, token_key = tok
 
     # 步骤 4：设 user.quota（root 令牌；账户自洽，权威余额闸仍是 token.remain_quota）
-    cur_quota = pg.query_user_quota(uid)
-    if cur_quota != grant:
-        _http_json("PUT", "/api/user/", root_token, ROOT_UID, {
-            "id": uid,
-            "username": username,
-            "display_name": username[:20],
-            "quota": grant,
-            "group": "default",
-        })
-        log.info("[4/4] 设 user.quota uid=%d → %d", uid, grant)
-    else:
+    quota_result = ensure_user_quota(pg, root_token, uid, username, grant)
+    if quota_result == "reused":
         log.info("[4/4] user.quota uid=%d 已=%d（复用）", uid, grant)
+    elif quota_result == "api":
+        log.info("[4/4] 管理 API 设置并核验 user.quota uid=%d → %d", uid, grant)
+    else:
+        log.warning("[4/4] 管理 API 未兑现 quota，数据库补偿并核验 uid=%d → %d", uid, grant)
 
     # 组装池表条目（§3.5 契约字段）
     return {

game-runtime/tools/test_newapi_pool_provision.py

@@ -84,6 +84,25 @@ class ProvisionRequestTest(unittest.TestCase):
         self.assertEqual(run.call_args.kwargs["input"],
                          "SELECT id FROM users WHERE username='neice_051';")
 
+    def test_repairs_quota_when_management_api_does_not_persist_it(self) -> None:
+        pg = mock.Mock()
+        pg.query_user_quota.side_effect = [0, 0, 0, 0, 0, 0, 6849315]
+        with mock.patch.object(MODULE, "_http_json") as request, \
+                mock.patch.object(MODULE.time, "sleep"):
+            result = MODULE.ensure_user_quota(pg, "root-token", 67, "neice_051", 6849315)
+        self.assertEqual(result, "database-repair")
+        request.assert_called_once()
+        pg.set_user_quota.assert_called_once_with(67, 6849315)
+
+    def test_reuses_verified_user_quota(self) -> None:
+        pg = mock.Mock()
+        pg.query_user_quota.return_value = 6849315
+        with mock.patch.object(MODULE, "_http_json") as request:
+            result = MODULE.ensure_user_quota(pg, "root-token", 67, "neice_051", 6849315)
+        self.assertEqual(result, "reused")
+        request.assert_not_called()
+        pg.set_user_quota.assert_not_called()
+
 
 if __name__ == "__main__":
     unittest.main()