并发控制:有上限、有超时、能停下
状态: 已在 v0.1.76 发布(2026-09-27)—— 设计与落地记录见 StandMeet 仓库的 docs/design/event-bus-outbox-webhooks.md。
所有后台工作都跑在有上限、有超时、能优雅停下的 worker 里;应用代码里的游离 go func 不再存在。
队列与并发上限
flowchart LR
R[relay loop · every process · ≤ 200 rows per pass] --> QI
R --> QN
R --> QW
MCP[jobs.fetch_new] --> QF
subgraph pool["worker pool (each queue has its own limit, none starves another)"]
QI["index queue<br/>MaxWorkers 4 · timeout 30s"]
QN["notify queue<br/>MaxWorkers 2 · timeout 30s"]
QW["webhook queue<br/>MaxWorkers 8 · timeout 15s<br/>≤ 1 in flight per endpoint"]
QM["maintenance queue<br/>MaxWorkers 1<br/>periodic jobs"]
QF["fetch queue<br/>MaxWorkers 3<br/>one job per job source"]
end
QI --> MS[(Meili)]
QN --> SM[SMTP · suppliers]
QW --> EP[external endpoints]
QF --> JB[job boards]
| 队列 | MaxWorkers | 任务超时 | 跑什么 |
|---|---|---|---|
index |
4 | 30 秒(重建 10 分钟) | corpus.index、corpus.reindex、构建完成的订阅方 |
notify |
2 | 30 秒 | owner.notify、批准邮件与确认邮件、supplier.invoke |
webhook |
8 | 15 秒(扇出 30 秒) | webhook.fanout、webhook.deliver;每端点同时 ≤ 1 |
maintenance |
1 | 10 分钟 | 周期任务 |
fetch |
3 | 按种类 | jobs.fetch_source(P4 加入) |
数字是起始值,按实测调。
- 分队列,各有上限:一个慢的外部端点只会占满 webhook 队列,不会拖慢索引和邮件。有一条 UT 把 webhook 队列堵死,断言索引任务照样按时完成。
- 每端点同时 ≤ 1:一行租约。
webhook.deliver执行UPDATE webhook_endpoints SET busy_until = now() + 20 s WHERE id = $1 AND enabled AND (busy_until IS NULL OR busy_until < now()) RETURNING …。拿不到行就 snooze 2 秒(不计次数)。租约比任务超时长,worker 崩了租约会自己过期。多进程也成立。- 没用
pg_try_advisory_xact_lock:事务级的锁意味着 POST 期间一直占着事务和它的连接。
- 没用
- relay 每轮 ≤ 200 行,每个进程都跑,不选主:一次 2000 篇的导入不会变成一个巨大的事务。见 relay-claims-rows-not-cursor。
- 请求会等的队列每 100 毫秒轮询一次(
jobs.QueueAwaited:index、notify)。其余队列保持 River 的 1 秒。River 每个队列每个FetchCooldown(100 毫秒)只发一条插入通知。一次写入的第二个索引任务(wiki 那篇,比 raw 晚约 5 毫秒)发不出通知,worker 取完第一个后它要等 1 秒轮询,每次写入多花约 0.5 秒。现在回执的耗时是一次轮询(≤ 100 毫秒)加 River 批量完结器的周期:它每 250 毫秒记一次完结,不可配置。
连接池预算:网络调用绝不在事务里
sequenceDiagram participant W as webhook worker participant DB as Postgres participant E as external endpoint rect rgba(200,80,60,0.08) note over W,E: Wrong — HTTP inside a transaction W->>DB: BEGIN (holds a connection + locks) W->>E: POST (slow, 15 s) E-->>W: 200 W->>DB: COMMIT note over W,DB: 8 workers × 15 s = pool drained, visitor requests queue end rect rgba(80,140,90,0.08) note over W,E: Right — lease, read, release the connection, then HTTP W->>DB: take lease (one UPDATE, autocommit) W->>DB: read event + secret (short reads, connection returned at once) W->>E: POST (holds no connection) E-->>W: 200 W->>DB: settle (release lease, clear failing_since) end
pgxpool 的 MaxConns 是 40,一半留给请求。启动校验:各队列 MaxWorkers 之和 + relay(现在 4 + 2 + 8 + 1 + 3 + 1 = 19)超过 MaxConns 的一半(20)就拒绝启动。配置错误在启动时暴露,而不是线上排队时。
超时与取消
- 每个任务种类有硬超时(上表),到点 ctx 被取消,按 retryable 处理。
- webhook 的 HTTP 超时(10 秒)< 任务超时(15 秒),确保先得到明确的网络错误,而不是被整体杀掉。
- worker 里的 panic 被任务层捕获,记为一次失败,不会带崩进程。
优雅停机(升级会重建容器)
sequenceDiagram
participant D as docker / updater
participant S as backend
participant R as relay loop
participant Q as job workers
participant W as running job
D->>S: SIGTERM
S->>R: stop the relay first
S->>Q: stop claiming new jobs
Q->>W: wait for running jobs, up to 20 s
alt finished within 20 s
W-->>Q: completed / retryable
else timeout
Q->>W: cancel ctx
Note over Q,W: job stays running, rescued and retried after restart (at least once)
end
S-->>D: exit
请求内等待的背压
“写后等索引最多 2 秒”靠共享的 LISTEN 连接(pgstore.Listener:每个进程每个频道一条)分发通知,不是每个请求各开一个连接轮询。每个频道最多 64 个等待者,超了就直接返回 indexed: false,不排队。见 async-response-contract 和 completion-hooks。
不许再有游离 goroutine
- 门禁
check-no-bare-goroutine.sh:go语句只允许出现在internal/infra/**和cmd/server/**(测试除外),域代码和internal/routes里一律禁止;要后台做事就入队一个任务,或声明一个jobs.Periodic。没有排除清单。见 no-bypass-by-structure。 - 必须自己拥有 goroutine 的进程管道放在 infra:
detach.Go拥有一个 goroutine 并吸收它的 panic(block 挂载预热用它);hostsocket.ListenWith自己起 accept 循环。 - 上门禁前清掉的违规点:
routes/admin/obsidian.go(goroutine 删除)、plugin/adapters/invoke_background.go(删除)、routes/hostdesk/hostdesk.go和agentcore/hostops.go(accept 循环移进hostsocket)、plugin/mount/mounted_warm.go(改用detach.Go)。 - 门禁上线前,在临时副本里植入一个
go func证红过一次。