支持并发agent
This commit is contained in:
Sirius
@@ -25,7 +25,7 @@ Command Code OpenAI Bridge 是一个仅在本机运行的 OpenAI 兼容服务。
|
||||
- 支持中文、Unicode、Markdown、代码块和较长文本。
|
||||
- 模型名称通过 YAML 配置映射到实际 Command Code 模型。
|
||||
- 请求内容通过 stdin 传输，不受命令行参数长度限制。
|
||||
- 同时只运行一个 Command Code 请求；其余请求进入有限 FIFO 队列，队列已满时返回 HTTP 429。
|
||||
- 可配置同时运行的 Command Code 请求数；其余请求进入共享的有限 FIFO 队列，队列已满时返回 HTTP 429。
|
||||
- 客户端断开、请求取消、总超时和 Ctrl+C 会终止当前子进程组。
|
||||
- 单次 Command Code 失败后，HTTP 服务可继续处理后续请求。
|
||||
- Authorization 请求头会被忽略，服务强制绑定 `127.0.0.1`。
|
||||
@@ -48,7 +48,7 @@ Command Code OpenAI Bridge 是一个仅在本机运行的 OpenAI 兼容服务。
|
||||
```text
|
||||
src/
|
||||
├── index.ts # 程序入口、服务启动和信号处理
|
||||
├── server.ts # HTTP 路由、单请求状态和错误处理
|
||||
├── server.ts # HTTP 路由、共享并发协调和错误处理
|
||||
├── openai.ts # 请求校验、消息转换和响应生成
|
||||
├── command-code.ts # CLI 检查、子进程管理和最终结果提取
|
||||
├── tool-calling.ts # 外部 function calling 决策解析与参数校验
|
||||
@@ -92,25 +92,56 @@ curl http://127.0.0.1:18000/v1/chat/completions \
|
||||
<!-- gitnexus:start -->
|
||||
# GitNexus — Code Intelligence
|
||||
|
||||
This project is indexed by GitNexus as **command-code-openai-bridge** (428 symbols, 1186 relationships, 33 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
|
||||
This project is indexed by GitNexus as **command-code-openai-bridge** (320 symbols, 917 relationships, 27 execution flows). Use the GitNexus MCP tools to understand code, assess impact, and navigate safely.
|
||||
|
||||
> Index stale? Run `node .gitnexus/run.cjs analyze` from the project root — it auto-selects an available runner. No `.gitnexus/run.cjs` yet? `npx gitnexus analyze` (npm 11 crash → `npm i -g gitnexus`; #1939).
|
||||
> If any GitNexus tool warns the index is stale, run `npx gitnexus analyze` in terminal first.
|
||||
|
||||
## Always Do
|
||||
|
||||
- **MUST run impact analysis before editing any symbol.** Before modifying a function, class, or method, run `impact({target: "symbolName", direction: "upstream"})` and report the blast radius (direct callers, affected processes, risk level) to the user.
|
||||
- **MUST run `detect_changes()` before committing** to verify your changes only affect expected symbols and execution flows. For regression review, compare against the default branch: `detect_changes({scope: "compare", base_ref: "main"})`.
|
||||
- **MUST run impact analysis before editing any symbol.** Before modifying a function, class, or method, run `gitnexus_impact({target: "symbolName", direction: "upstream"})` and report the blast radius (direct callers, affected processes, risk level) to the user.
|
||||
- **MUST run `gitnexus_detect_changes()` before committing** to verify your changes only affect expected symbols and execution flows.
|
||||
- **MUST warn the user** if impact analysis returns HIGH or CRITICAL risk before proceeding with edits.
|
||||
- When exploring unfamiliar code, use `query({search_query: "concept"})` to find execution flows instead of grepping. It returns process-grouped results ranked by relevance.
|
||||
- When you need full context on a specific symbol — callers, callees, which execution flows it participates in — use `context({name: "symbolName"})`.
|
||||
- For security review, `explain({target: "fileOrSymbol"})` lists taint findings (source→sink flows; needs `analyze --pdg`).
|
||||
- When exploring unfamiliar code, use `gitnexus_query({query: "concept"})` to find execution flows instead of grepping. It returns process-grouped results ranked by relevance.
|
||||
- When you need full context on a specific symbol — callers, callees, which execution flows it participates in — use `gitnexus_context({name: "symbolName"})`.
|
||||
|
||||
## When Debugging
|
||||
|
||||
1. `gitnexus_query({query: "<error or symptom>"})` — find execution flows related to the issue
|
||||
2. `gitnexus_context({name: "<suspect function>"})` — see all callers, callees, and process participation
|
||||
3. `READ gitnexus://repo/command-code-openai-bridge/process/{processName}` — trace the full execution flow step by step
|
||||
4. For regressions: `gitnexus_detect_changes({scope: "compare", base_ref: "main"})` — see what your branch changed
|
||||
|
||||
## When Refactoring
|
||||
|
||||
- **Renaming**: MUST use `gitnexus_rename({symbol_name: "old", new_name: "new", dry_run: true})` first. Review the preview — graph edits are safe, text_search edits need manual review. Then run with `dry_run: false`.
|
||||
- **Extracting/Splitting**: MUST run `gitnexus_context({name: "target"})` to see all incoming/outgoing refs, then `gitnexus_impact({target: "target", direction: "upstream"})` to find all external callers before moving code.
|
||||
- After any refactor: run `gitnexus_detect_changes({scope: "all"})` to verify only expected files changed.
|
||||
|
||||
## Never Do
|
||||
|
||||
- NEVER edit a function, class, or method without first running `impact` on it.
|
||||
- NEVER edit a function, class, or method without first running `gitnexus_impact` on it.
|
||||
- NEVER ignore HIGH or CRITICAL risk warnings from impact analysis.
|
||||
- NEVER rename symbols with find-and-replace — use `rename` which understands the call graph.
|
||||
- NEVER commit changes without running `detect_changes()` to check affected scope.
|
||||
- NEVER rename symbols with find-and-replace — use `gitnexus_rename` which understands the call graph.
|
||||
- NEVER commit changes without running `gitnexus_detect_changes()` to check affected scope.
|
||||
|
||||
## Tools Quick Reference
|
||||
|
||||
| Tool | When to use | Command |
|
||||
|------|-------------|---------|
|
||||
| `query` | Find code by concept | `gitnexus_query({query: "auth validation"})` |
|
||||
| `context` | 360-degree view of one symbol | `gitnexus_context({name: "validateUser"})` |
|
||||
| `impact` | Blast radius before editing | `gitnexus_impact({target: "X", direction: "upstream"})` |
|
||||
| `detect_changes` | Pre-commit scope check | `gitnexus_detect_changes({scope: "staged"})` |
|
||||
| `rename` | Safe multi-file rename | `gitnexus_rename({symbol_name: "old", new_name: "new", dry_run: true})` |
|
||||
| `cypher` | Custom graph queries | `gitnexus_cypher({query: "MATCH ..."})` |
|
||||
|
||||
## Impact Risk Levels
|
||||
|
||||
| Depth | Meaning | Action |
|
||||
|-------|---------|--------|
|
||||
| d=1 | WILL BREAK — direct callers/importers | MUST update these |
|
||||
| d=2 | LIKELY AFFECTED — indirect deps | Should test |
|
||||
| d=3 | MAY NEED TESTING — transitive | Test if critical path |
|
||||
|
||||
## Resources
|
||||
|
||||
@@ -121,6 +152,32 @@ This project is indexed by GitNexus as **command-code-openai-bridge** (428 symbo
|
||||
| `gitnexus://repo/command-code-openai-bridge/processes` | All execution flows |
|
||||
| `gitnexus://repo/command-code-openai-bridge/process/{name}` | Step-by-step execution trace |
|
||||
|
||||
## Self-Check Before Finishing
|
||||
|
||||
Before completing any code modification task, verify:
|
||||
1. `gitnexus_impact` was run for all modified symbols
|
||||
2. No HIGH/CRITICAL risk warnings were ignored
|
||||
3. `gitnexus_detect_changes()` confirms changes match expected scope
|
||||
4. All d=1 (WILL BREAK) dependents were updated
|
||||
|
||||
## Keeping the Index Fresh
|
||||
|
||||
After committing code changes, the GitNexus index becomes stale. Re-run analyze to update it:
|
||||
|
||||
```bash
|
||||
npx gitnexus analyze
|
||||
```
|
||||
|
||||
If the index previously included embeddings, preserve them by adding `--embeddings`:
|
||||
|
||||
```bash
|
||||
npx gitnexus analyze --embeddings
|
||||
```
|
||||
|
||||
To check whether embeddings exist, inspect `.gitnexus/meta.json` — the `stats.embeddings` field shows the count (0 means no embeddings). **Running analyze without `--embeddings` will delete any previously generated embeddings.**
|
||||
|
||||
> Claude Code users: A PostToolUse hook handles this automatically after `git commit` and `git merge`.
|
||||
|
||||
## CLI
|
||||
|
||||
| Task | Read this skill file |
|
||||
|
||||
@@ -92,6 +92,7 @@ command_code_executable: command-code
|
||||
command_code_working_directory: .
|
||||
timeout_seconds: 1800
|
||||
max_request_bytes: 20971520
|
||||
max_concurrent_requests: 1
|
||||
max_queue_size: 8
|
||||
max_turns: 100
|
||||
stream_thinking: false
|
||||
@@ -105,7 +106,7 @@ models:
|
||||
effort: max
|
||||
```
|
||||
|
||||
`command_code_working_directory` 决定 Command Code 能看到和操作的项目目录。`response_store_directory` 保存 `store: true` 的 Response；两个相对路径都以配置文件所在目录为基准。默认存储目录是隐藏目录 `.command-code-openai-bridge/responses`。`max_queue_size` 是等待执行的生成请求上限，不包含当前运行中的请求；设为 `0` 可恢复忙碌时立即返回 429 的行为。`stream_thinking` 控制 Chat Completions 流是否把 Command Code 思考事件包装成 `<think>` 块；默认关闭。`models.<name>.effort` 会传给 Command Code 的 `--effort`，Responses 请求中的 `reasoning.effort` 优先。可用 effort 由对应底层模型决定。服务强制只监听 `127.0.0.1`。客户端只能选择配置中的模型名，不能注入额外 CLI 参数。
|
||||
`command_code_working_directory` 决定 Command Code 能看到和操作的项目目录。`response_store_directory` 保存 `store: true` 的 Response；两个相对路径都以配置文件所在目录为基准。默认存储目录是隐藏目录 `.command-code-openai-bridge/responses`。`max_concurrent_requests` 是同时运行的 Command Code 请求上限，范围 `1..64`，默认 `1`。`max_queue_size` 是等待执行的生成请求上限，不包含正在运行的请求；设为 `0` 可恢复并发槽满时立即返回 429 的行为。`stream_thinking` 控制 Chat Completions 流是否把 Command Code 思考事件包装成 `<think>` 块；默认关闭。`models.<name>.effort` 会传给 Command Code 的 `--effort`，Responses 请求中的 `reasoning.effort` 优先。可用 effort 由对应底层模型决定。服务强制只监听 `127.0.0.1`。客户端只能选择配置中的模型名，不能注入额外 CLI 参数。
|
||||
|
||||
### Obsidian Copilot 流式输出
|
||||
|
||||
@@ -136,7 +137,7 @@ export OPENCODE_CONFIG="/path/to/command-code-openai-bridge/examples/opencode.js
|
||||
|
||||
OpenCode 1.18.15 实际发送的 Chat 请求包含 `stream: true`、`stream_options.include_usage: true`、`max_tokens`（默认约 32000）、`store: false`、`tools`（约 24 个内置工具）、`tool_choice: auto`。Bridge 接受 `store` 与采样/输出限制参数但不改变无状态行为；未映射参数通过 `X-Command-Code-Ignored-Parameters` 标出。工具 `parameters` 中的 JSON Schema `$schema`（draft 2020-12）会在校验前剥离元数据字段，避免 AJV 拒绝 OpenCode 工具定义。
|
||||
|
||||
OpenCode 还会在会话标题生成等场景发送无 `tools` 的压缩/摘要请求（`stream: true`、较长 system 提示、无 `store` 或 `store: false`）。这些请求与主 agent 请求共用同一 provider。
|
||||
OpenCode 还会在会话标题生成、子 agent 和后台任务等场景发起独立请求。这些请求与主 agent 请求共用同一 provider；将 `max_concurrent_requests` 设为大于 `1` 可让它们与多个会话并行执行。
|
||||
|
||||
已知限制：Bridge 不实现 OpenCode hosted tools、图片/音频输入、Computer Use 或 Responses `/v1/responses` 路径；OpenCode 自定义 provider 走 `/v1/chat/completions`。`temperature`、`max_tokens` 等采样参数不会传给 Command Code CLI。
|
||||
|
||||
@@ -155,7 +156,7 @@ npm run build
|
||||
|
||||
首次执行 `scripts/install.sh` 时会自动安装依赖并完成一次编译。
|
||||
|
||||
空闲时按 Ctrl+C 停止。请求运行时第一次 Ctrl+C 取消当前 Command Code 子进程组并保留服务；请求结束后再按 Ctrl+C 停止服务。重启就是再次执行启动脚本。
|
||||
空闲时按 Ctrl+C 停止。请求运行时第一次 Ctrl+C 取消所有活动 Command Code 子进程组并保留服务；请求结束后再按 Ctrl+C 停止服务。重启就是再次执行启动脚本。
|
||||
|
||||
运行中的客户端断开连接会先给整个 Command Code 子进程组发送 SIGINT，随后按需升级为 SIGTERM 和 SIGKILL；排队中的客户端断开连接只会从等待队列移除。总超时由 `timeout_seconds` 控制。
|
||||
|
||||
@@ -169,7 +170,7 @@ npm run build
|
||||
- `DELETE /v1/responses/:response_id`
|
||||
- `GET /v1/responses/:response_id/input_items`
|
||||
|
||||
`GET /health` 额外返回 `busy`、`queue_length` 和 `queue_capacity`，可用于观察当前执行槽和等待队列。
|
||||
`GET /health` 额外返回 `active_requests`、`max_concurrent_requests`、`queue_length` 和 `queue_capacity`；`busy` 作为兼容字段保留，等价于 `active_requests > 0`。
|
||||
|
||||
Bearer Token 会被忽略。所有生成接口只接受文本内容。图片、音频、文件输入和其他未实现字段会返回 OpenAI 格式的 400，`code` 为 `unsupported_parameter`；服务不会静默忽略会改变行为的参数。
|
||||
|
||||
@@ -328,7 +329,7 @@ command-code \
|
||||
|
||||
## 有限队列和错误
|
||||
|
||||
同一时间只运行一个 Command Code 实例，其余生成请求按到达顺序进入等待队列。默认最多等待 8 个请求，队列已满时返回 HTTP 429 和 `code: queue_full`。流式请求进入队列后会立即建立 SSE 连接；非流式请求保持等待。客户端在排队期间断开会立即移出队列。`timeout_seconds` 从请求取得执行位置后开始计算。
|
||||
同一时间最多运行 `max_concurrent_requests` 个 Command Code 实例，默认值 `1` 保持原来的单并发行为。其余 Chat Completions 和 Responses 生成请求共享同一条 FIFO 等待队列；任一执行槽释放后都会按队首顺序补齐空闲槽。默认最多等待 8 个请求，队列已满时返回 HTTP 429 和 `code: queue_full`。流式请求进入队列后会立即建立 SSE 连接；非流式请求保持等待。客户端在排队期间断开会立即移出队列，活动请求取消并结束后会推进队首请求。`timeout_seconds` 从请求取得执行位置后开始计算。
|
||||
|
||||
请求体超过 `max_request_bytes` 返回 413。未知模型和非文本内容返回 4xx。CLI 的详细错误留在服务终端，客户端只收到简洁的 OpenAI 格式错误。
|
||||
|
||||
@@ -336,11 +337,12 @@ command-code \
|
||||
|
||||
## 本机实际检查结果
|
||||
|
||||
检查日期：Chat Completions 原有检查为 2026-08-04；Responses 文本子集检查为 2026-08-05；Chat function calling 协议检查为 2026-08-06；Responses function calling 检查为 2026-08-12；OpenCode 1.18.15 Chat Completions E2E 检查为 2026-08-12。没有编写测试用例。原有条目来自真实 CLI 和真实 HTTP/SDK 客户端；有限队列改动完成了类型检查、生产构建和协调器运行时冒烟检查，尚未重新运行真实 CLI 并发检查。Chat function calling 完成了内部协议冒烟检查、Obsidian Copilot 当前依赖 `@langchain/openai 1.2.2` 的双轮 wire compatibility 检查，以及真实 Command Code 与 OpenAI Node.js SDK 的双轮 HTTP/SSE 调用；尚未在 Obsidian UI 中运行完整检查。Responses function calling 完成了类型检查、生产构建、真实 HTTP 双轮检查、流式 function call SSE 检查和 OpenAI Node.js SDK 双轮检查。OpenCode 1.18.15 完成了真实 `opencode run` 文本回复、`glob` 工具循环、并行 `glob`、流式 `tool_calls`/`stop` SSE、`store: false` 与 `max_tokens` 接受、工具 Schema `$schema` 兼容，以及客户端断开后的取消恢复检查。
|
||||
检查日期：Chat Completions 原有检查为 2026-08-04；Responses 文本子集检查为 2026-08-05；Chat function calling 协议检查为 2026-08-06；Responses function calling、OpenCode 1.18.15 Chat Completions E2E 与可配置并发检查为 2026-08-12。没有编写测试用例。原有条目来自真实 CLI 和真实 HTTP/SDK 客户端；可配置并发改动完成了类型检查、生产构建和真实 Command Code 多进程 HTTP 冒烟检查。Chat function calling 完成了内部协议冒烟检查、Obsidian Copilot 当前依赖 `@langchain/openai 1.2.2` 的双轮 wire compatibility 检查，以及真实 Command Code 与 OpenAI Node.js SDK 的双轮 HTTP/SSE 调用；尚未在 Obsidian UI 中运行完整检查。Responses function calling 完成了类型检查、生产构建、真实 HTTP 双轮检查、流式 function call SSE 检查和 OpenAI Node.js SDK 双轮检查。OpenCode 1.18.15 完成了真实 `opencode run` 文本回复、`glob` 工具循环、并行 `glob`、流式 `tool_calls`/`stop` SSE、`store: false` 与 `max_tokens` 接受、工具 Schema `$schema` 兼容，以及客户端断开后的取消恢复检查。
|
||||
|
||||
- TypeScript 严格类型检查和生产构建通过。
|
||||
- npm 生产依赖审计：0 个已知漏洞。
|
||||
- `/health` 返回 Command Code v1.10.0、已登录、空闲。
|
||||
- `max_concurrent_requests: 2` 时实际观察到 2 个 Command Code CLI 子进程并行运行，后续请求保持 FIFO；取消排队请求会移出队列，取消活动请求后队首请求会被推进，最终健康状态恢复为 `active_requests: 0`、`queue_length: 0`。
|
||||
- `/v1/models` 返回两个本地映射。
|
||||
- curl 中文请求返回 HTTP 200，客户端只收到 `中文接口成功` 和标准 completion 字段。
|
||||
- 字符串 content 与 text part 数组均通过；非文本 part 返回 HTTP 400。
|
||||
@@ -383,7 +385,7 @@ command-code \
|
||||
- 普通文本 Chat Completions 会实时转发 Command Code 文本 delta；外部 function calling 和 Responses 结构化输出需要等待 Bridge 校验完成。
|
||||
- Chat Completions 和 Responses 都只实现外部 function tools，不实现图片、音频、文件输入、Computer Use、web search 等 hosted tools 或 MCP hosted tool。非 function 工具类型返回 400 `unsupported_parameter`。Responses 不实现原生 reasoning item、加密 reasoning 或隐藏思维过程。
|
||||
- Chat 与 Responses 的外部 function calling 都是 Bridge 通过提示协议、JSON 解析、工具参数 Schema 校验和一次修复实现的兼容层；Command Code CLI 没有公开原生 function calling 输出接口，因此连续两次输出不合格时返回 HTTP 502 `invalid_tool_decision`。
|
||||
- Responses 的本地文件存储只供本 Bridge 使用，没有跨进程锁、队列或多实例一致性保证。项目本身仍严格单并发。
|
||||
- Responses 的本地文件存储只供本 Bridge 使用，没有跨进程锁或多实例一致性保证；并发请求应避免同时更新同一条 Response 链。
|
||||
- Responses Structured Outputs 是 Bridge 层约束，底层 Command Code 模型仍可能连续两次输出不合格 JSON；此时状态为 incomplete。
|
||||
- `usage` 使用 Command Code 最终结果提供的真实 input/output token；Responses 缺失时返回 `null`，Chat Completions 为兼容旧行为返回 0。
|
||||
- 长文本受 HTTP 请求体上限和 Command Code 模型上下文上限共同限制，不会由桥接服务自行截断。
|
||||
|
||||
@@ -4,6 +4,8 @@ command_code_executable: command-code
|
||||
command_code_working_directory: .
|
||||
timeout_seconds: 1800
|
||||
max_request_bytes: 20971520
|
||||
# 同时运行的 Command Code 请求数，范围 1..64。
|
||||
max_concurrent_requests: 1
|
||||
max_queue_size: 8
|
||||
max_turns: 100
|
||||
stream_thinking: false
|
||||
|
||||
@@ -4,6 +4,8 @@ command_code_executable: command-code
|
||||
command_code_working_directory: .
|
||||
timeout_seconds: 1800
|
||||
max_request_bytes: 20971520
|
||||
# 同时运行的 Command Code 请求数，范围 1..64。
|
||||
max_concurrent_requests: 1
|
||||
max_queue_size: 8
|
||||
max_turns: 100
|
||||
stream_thinking: true
|
||||
|
||||
@@ -15,6 +15,7 @@ const configSchema = z.object({
|
||||
command_code_working_directory: z.string().min(1).default("."),
|
||||
timeout_seconds: z.number().int().positive().default(1800),
|
||||
max_request_bytes: z.number().int().positive().default(20 * 1024 * 1024),
|
||||
max_concurrent_requests: z.number().int().min(1).max(64).default(1),
|
||||
max_queue_size: z.number().int().min(0).max(1000).default(8),
|
||||
max_turns: z.number().int().positive().default(100),
|
||||
stream_thinking: z.boolean().default(false),
|
||||
|
||||
+4
-3
@@ -30,12 +30,13 @@ async function main(): Promise<void> {
|
||||
process.stdout.write(`工作目录: ${config.resolvedWorkingDirectory}\n`);
|
||||
process.stdout.write(`Command Code: ${bridge.installation.version ?? "unavailable"}\n`);
|
||||
process.stdout.write(`权限模式: ${config.dangerously_skip_permissions ? "yolo" : config.permission_mode}\n`);
|
||||
process.stdout.write("空闲时按 Ctrl+C 停止；请求运行中第一次 Ctrl+C 只取消当前请求。\n");
|
||||
process.stdout.write("空闲时按 Ctrl+C 停止；请求运行中第一次 Ctrl+C 取消所有活动请求。\n");
|
||||
|
||||
let shuttingDown = false;
|
||||
const shutdown = async (signal: NodeJS.Signals) => {
|
||||
if (bridge.abortActive()) {
|
||||
process.stderr.write(`\n收到 ${signal}，正在取消当前 Command Code 请求。\n`);
|
||||
const abortedRequests = bridge.abortActive();
|
||||
if (abortedRequests > 0) {
|
||||
process.stderr.write(`\n收到 ${signal}，正在取消 ${abortedRequests} 个活动 Command Code 请求。\n`);
|
||||
return;
|
||||
}
|
||||
if (shuttingDown) return;
|
||||
|
||||
+32
-17
@@ -9,13 +9,20 @@ interface QueuedRequest {
|
||||
}
|
||||
|
||||
export class RequestCoordinator {
|
||||
private activeController: AbortController | undefined;
|
||||
private readonly activeControllers = new Set<AbortController>();
|
||||
private readonly queue: QueuedRequest[] = [];
|
||||
|
||||
constructor(private readonly maxQueueSize: number) {}
|
||||
constructor(
|
||||
private readonly maxConcurrentRequests: number,
|
||||
private readonly maxQueueSize: number,
|
||||
) {}
|
||||
|
||||
get busy(): boolean {
|
||||
return this.activeController !== undefined;
|
||||
return this.activeControllers.size > 0;
|
||||
}
|
||||
|
||||
get activeRequests(): number {
|
||||
return this.activeControllers.size;
|
||||
}
|
||||
|
||||
get queueLength(): number {
|
||||
@@ -24,8 +31,8 @@ export class RequestCoordinator {
|
||||
|
||||
begin(): RequestLease | undefined {
|
||||
const controller = new AbortController();
|
||||
if (!this.activeController) {
|
||||
this.activeController = controller;
|
||||
if (this.activeControllers.size < this.maxConcurrentRequests) {
|
||||
this.activeControllers.add(controller);
|
||||
return { controller, ready: Promise.resolve(true) };
|
||||
}
|
||||
if (this.queue.length >= this.maxQueueSize) return undefined;
|
||||
@@ -39,7 +46,7 @@ export class RequestCoordinator {
|
||||
}
|
||||
|
||||
cancel(controller: AbortController, reason = "request cancelled"): boolean {
|
||||
if (this.activeController === controller) {
|
||||
if (this.activeControllers.has(controller)) {
|
||||
controller.abort(reason);
|
||||
return true;
|
||||
}
|
||||
@@ -54,25 +61,33 @@ export class RequestCoordinator {
|
||||
}
|
||||
|
||||
finish(controller: AbortController): void {
|
||||
if (this.activeController !== controller) return;
|
||||
this.activeController = undefined;
|
||||
if (!this.activeControllers.delete(controller)) return;
|
||||
this.fillAvailableSlots();
|
||||
}
|
||||
|
||||
while (this.queue.length > 0) {
|
||||
abortActive(reason = "bridge interrupted"): number {
|
||||
let aborted = 0;
|
||||
for (const controller of this.activeControllers) {
|
||||
if (controller.signal.aborted) continue;
|
||||
controller.abort(reason);
|
||||
aborted += 1;
|
||||
}
|
||||
return aborted;
|
||||
}
|
||||
|
||||
private fillAvailableSlots(): void {
|
||||
while (
|
||||
this.activeControllers.size < this.maxConcurrentRequests
|
||||
&& this.queue.length > 0
|
||||
) {
|
||||
const next = this.queue.shift();
|
||||
if (!next) return;
|
||||
if (next.controller.signal.aborted) {
|
||||
next.resolve(false);
|
||||
continue;
|
||||
}
|
||||
this.activeController = next.controller;
|
||||
this.activeControllers.add(next.controller);
|
||||
next.resolve(true);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
abortActive(reason = "bridge interrupted"): boolean {
|
||||
if (!this.activeController) return false;
|
||||
this.activeController.abort(reason);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
+7
-2
@@ -33,7 +33,7 @@ import {
|
||||
export interface BridgeServer {
|
||||
app: FastifyInstance;
|
||||
installation: InstallationStatus;
|
||||
abortActive: () => boolean;
|
||||
abortActive: () => number;
|
||||
}
|
||||
|
||||
export async function createServer(config: BridgeConfig): Promise<BridgeServer> {
|
||||
@@ -43,7 +43,10 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
|
||||
bodyLimit: config.max_request_bytes,
|
||||
requestTimeout: 0,
|
||||
});
|
||||
const coordinator = new RequestCoordinator(config.max_queue_size);
|
||||
const coordinator = new RequestCoordinator(
|
||||
config.max_concurrent_requests,
|
||||
config.max_queue_size,
|
||||
);
|
||||
|
||||
app.addHook("onRequest", async (request, reply) => {
|
||||
const requestedHeaders = request.headers["access-control-request-headers"];
|
||||
@@ -90,6 +93,8 @@ export async function createServer(config: BridgeConfig): Promise<BridgeServer>
|
||||
status: installation.installed && installation.authenticated ? "ok" : "degraded",
|
||||
command_code: installation,
|
||||
busy: coordinator.busy,
|
||||
active_requests: coordinator.activeRequests,
|
||||
max_concurrent_requests: config.max_concurrent_requests,
|
||||
queue_length: coordinator.queueLength,
|
||||
queue_capacity: config.max_queue_size,
|
||||
}));
|
||||
|
||||
Reference in New Issue
Block a user