- 后端
- 前端
- 搜索引擎
- 网页爬虫
- 数据工程
- AI 应用
- 可观测性
【免费下载链接】freehire
freehire — the open-source search engine for job seekers
本文以 freehire 仓库中的变更提案 proposal.md 为主线，完整拆解enrichment_outbox（富化队列）从“串行、按插入 id 顺序、不区分岗位状态”的消费方式，改造为“最新开放岗位优先 + 有界并发消费”的全过程。读者读完可以掌握该改造的动机、SQL 领取语义、并发与租约（lease）的正确配合方式、相关环境变量的真实默认值与边界行为，以及如何在无数据库迁移的前提下安全上线和回滚。配套的设计决策与验收标准分别记录在同一变更目录的 design.md、tasks.md 与 ai-enrichment 规格 中。
改造背景：一个永远追不上的积压队列
提案开篇给出的问题陈述非常具体，也是整个改造的出发点：
enrichment_outbox中积压了约313k条待富化任务；cmd/enrich严格串行消费，每条任务约45 秒的 LLM 调用，折算吞吐约1,900 条/天；- 与此同时 ingest 仍在持续写入新岗位，产出速度高于消费速度，因此积压无限增长；
- 更糟的是，领取（claim）按插入 id 排序，而非按岗位新鲜度排序，也从未排除已关闭的岗位——于是新鲜、仍然开放的岗位排在一条长长的“陈旧 + 已死”尾巴之后，LLM 预算被花在用户永远看不到的空缺上。
这三个问题分别对应三类修复：排序语义、状态过滤、并发模型。提案在 “What Changes” 中把它们压缩成四条要点，并在 “Capabilities” 中说明这不是新增能力，而是对既有能力ai-enrichment的细化——具体表现为入队与领取排除关闭岗位、claim 按新鲜度排序、批处理命令以可配置并发消费整波任务。
设计目标与非目标
在动手改 SQL 之前，变更先划定了边界（见 design.md）：
目标：
- 优先富化更新鲜、仍开放的岗位；
- 永不富化已关闭岗位；
- 通过有界并发把吞吐提升约 4 倍，同时不破坏“每次运行中每条任务只被领取一次”的保证；
- 不引入任何新的数据库迁移。
非目标（同样重要，避免范围蔓延）：
- 不清理 outbox 中已经存在的关闭岗位残留（claim 时跳过即可，清理扫描被列为后续独立变更）；
- 不引入每轮运行上限，也不改变“一直消费到队列为空”的语义；
- 不做并发度自动调优，也不针对 LLM 端点做自适应限流；
- 不把多条岗位拼进同一个 prompt。
理解这些非目标，才能正确解读后面“关闭岗位会残留、但无害”的取舍。
改造一：claim 按岗位新鲜度排序，并排除关闭岗位
SQL 的真实形态
提案的表述是：claim 关联jobs表，并按posted_at DESC NULLS LAST, id DESC（用户在信息流中看到的顺序）排序。而仓库当前的实现比提案更进一步，实际查询位于 enrichment.sql：
WITH claimable AS ( SELECT o.id FROM enrichment_outbox o JOIN jobs j ON j.id = o.job_id WHERE o.failed_at IS NULL AND (o.claimed_at IS NULL OR o.claimed_at < now() - make_interval(secs => sqlc.arg(lease_seconds)::int)) AND j.closed_at IS NULL AND j.duplicate_of IS NULL ORDER BY COALESCE(j.posted_at, j.created_at) DESC, j.id DESC FOR UPDATE OF o SKIP LOCKED LIMIT sqlc.arg(batch_size) ) UPDATE enrichment_outbox o SET claimed_at = now() FROM claimable c WHERE o.id = c.id RETURNING o.id, o.job_id, o.target_version;这段 SQL 里每一个子句都对应一条设计决策：
JOIN jobs：排序依据与状态过滤都必须来自岗位表，而 outbox 只保存job_id与target_version，不复制岗位字段。o.failed_at IS NULL：已进入死信（dead-letter）的任务不再被领取。claimed_at租约谓词：claimed_at IS NULL或已早于now() - lease_seconds的任务才可领取。这同时充当崩溃回收器与跨运行/跨实例去重闸门。j.closed_at IS NULL：关闭岗位不领取。j.duplicate_of IS NULL：重复投递（repost）不是可检索的规范行，富化它等于为一条不可见记录付费。ORDER BY COALESCE(j.posted_at, j.created_at) DESC, j.id DESC：新鲜度排序。FOR UPDATE OF o SKIP LOCKED：只锁 outbox 行，并发 claim 各自拿到不相交的行。
为什么用COALESCE(posted_at, created_at)而不是NULLS LAST
这是设计文档中被反复强调、也最容易被“简化”掉的一处。提案写的是posted_at DESC NULLS LAST，设计文档则明确改为COALESCE(posted_at, created_at)，理由是数据分布：
posted_at是面向用户的新鲜度信号，与信息流顺序一致，但对某些来源为 NULL（如 telegram、linksource 以及部分 ATS）；- 在纯
NULLS LAST语义下，这些无日期岗位会排在所有有日期岗位之后；当 ingest 在数天的积压清空期间不断追加有日期的岗位时，它们会被无限期饿死； - 回退到
created_at（NOT NULL，即写入时间）后，无日期岗位按“入库新鲜度”参与排序：对有日期岗位仍然是“最新优先”，对无日期岗位则是不被饿死的公平排序。
设计文档还记录了两个被否决的替代方案：不做 COALESCE 的NULLS LAST（会导致上述饿死）、以及在 outbox 上冗余一列posted_at（需持续同步，相较 join 无收益）。
为什么必须是FOR UPDATE OF o
带 join 的FOR UPDATE在 PostgreSQL 中默认会锁定涉及的每一张表。如果直接写FOR UPDATE，jobs行也会被锁住，并发 claim 波次就会在热门岗位行上互相争用，SKIP LOCKED的意义随之消失。限定OF o后，只有 outbox 行被锁定，SKIP LOCKED才能真正让并发消费者取到互不相交的任务。
关于索引的一处事实校正
提案与设计文档都提到复用jobs_posted_at_id_idx (posted_at DESC NULLS LAST, id DESC)并依赖 outbox 的UNIQUE (job_id, target_version)所提供的job_id索引。但按当前仓库的迁移快照，情况已有变化，阅读时需注意区分：
- 0116_drop_jobs_posted_at_id_idx.sql 已将该索引删除（生产上 692 MB），原因是 jobs 列表排序早已迁移到 Meilisearch；
- 而 0001_init.sql 中存在一个与本次 claim 排序完全同形的部分索引：
CREATE INDEX jobs_open_enrich_freshness_idx ON public.jobs (COALESCE(posted_at, created_at) DESC, id DESC) WHERE (closed_at IS NULL);从索引定义可以直接推断：ORDER BY COALESCE(posted_at, created_at) DESC, id DESC加上closed_at IS NULL的过滤，正是该部分索引的服务形态。此外 0093_jobs_similar_backfill_idx.sql 的注释也明确把这个索引称为“富化 claim 使用的、查找未完成工作且最新优先”的同一形状。
同时，设计文档承认新排序不再由enrichment_outbox_claimable_idx（(id) WHERE failed_at IS NULL，见 0001_init.sql）服务，大积压时可能出现排序或嵌套循环探测。其结论是：该工作负载是LLM 延迟受限的，最坏情况每波的全排序耗时不到它所要等待的“每岗位数秒 LLM 时间”的 2%，因此按 YAGNI 不新增索引，等生产EXPLAIN ANALYZE显示真正 DB-bound 时再议。
改造二：在入队侧就挡住关闭岗位
排序只解决“谁先被处理”，过滤才解决“该不该花预算”。变更同时修改了入队侧。当前 EnqueuePendingJobs 的实际门槛比提案更完整：
INSERT INTO enrichment_outbox (job_id, target_version) SELECT id, sqlc.arg(target_version)::int FROM jobs WHERE closed_at IS NULL AND duplicate_of IS NULL AND (enriched_at IS NULL OR enrichment_version < sqlc.arg(target_version)::int) AND is_tech IS TRUE AND description <> '' ON CONFLICT (job_id, target_version) DO NOTHING;各条件的含义（可由查询注释与相邻查询交叉确认）：
| 条件 | 作用 |
|---|---|
closed_at IS NULL | 本次变更新增：关闭岗位不入队 |
duplicate_of IS NULL | 重复投递不入队，与 claim 侧过滤对齐 |
enriched_at IS NULL OR enrichment_version < target_version | 仅未富化或版本落后的岗位入队，支持版本升级重跑 |
is_tech IS TRUE | 只对已确认技术类岗位花费 LLM 预算；is_tech IS NULL（未判定）也被排除 |
description <> '' | 空描述无可提取内容 |
ON CONFLICT ... DO NOTHING | 每个(job_id, target_version)至多一条，天然幂等 |
规格文件对这一行为的表述是强制性的：入队 SHALL 只为开放岗位创建条目；对已入队的(job_id, target_version)重复入队 SHALL NOT 产生重复行（见 spec.md）。
提案同时明确指出：ingest 的事务内入队（UpsertJob路径）保持不变，因为刚写入或刚重开的岗位在构造上就是开放的。这条路径在仓库中对应 EnqueueJobEnrichment，与岗位 upsert 处于同一事务，其注释说明了对duplicate_of IS NULL、is_tech IS TRUE、非空描述的完整门槛；调用点见 cmd/ingest/store.go，其中if !deduped的守卫确保重复行不进入富化队列。
值得强调的是提案坦白的一处“已知缝隙”：已经躺在 outbox 里的关闭岗位行不会被领取，但也不会被删除，它们作为无害残留存在。设计文档把它明确列为非目标与未来清理扫描的候选。
改造三：以ENRICH_CONCURRENCY为界并发消费
波次大小 = 并发度，这是租约算术的结果
提案的关键洞察不是“开 goroutine”，而是把 claim 波次的大小绑定到并发度。设计文档给出的算术非常直观：
- 每条任务最慢的 LLM 调用约85 秒；
- 若波次等于并发度 N、池子也是 N，则每条被领取的任务立刻开始处理，其租约窗口 ≈ 一次最慢调用（约 85 秒），远小于
LeaseSeconds（300 秒）； - 若沿用
BatchSize=50而并发 4，则尾部任务会“已租未跑”长达50/4 × 45s ≈ 560s，超过 300 秒租约；此时重叠的 cron 运行会把这些任务再次领取，造成重复富化。
这也是为什么ENRICH_BATCH_SIZE被移除、波次改由并发度决定——两个旋钮在语义上本就是同一个东西。规格对此的表述是：命令 SHALL 以可配置的 worker 数并发处理每个 claim 波次，并 SHALL 将波次大小设置为该并发度，使条目在被处理前保持租约的时间远小于租约时长（见 spec.md）。
运行时实现：共享的outbox.RunPool
当前代码中，这一循环已被抽到共享包 internal/platform/outbox/runner.go，Runner.Run只需把并发度同时传给BatchSize与Concurrency：
s, err := outbox.RunPool(ctx, r.Store, outbox.RunOptions{ // The wave is sized to the concurrency, then drained in parallel: each entry // starts processing at once, so its lease window stays ≈ one LLM call. BatchSize: opt.Concurrency, LeaseSeconds: opt.LeaseSeconds, Concurrency: opt.Concurrency, OnWave: func(cum outbox.Stats) { /* 每波心跳日志 */ }, }, rn.process)见 internal/ai/enrich/runner.go。波次执行体runPoolWave用带缓冲 channel 作为信号量，保证同时在飞的处理数不超过并发度，并在汇总统计时加锁：
func runPoolWaveC any { var mu sync.Mutex var wg sync.WaitGroup slots := make(chan struct{}, concurrency) for _, item := range batch { wg.Add(1) go func(it C) { defer wg.Done() slots <- struct{}{} defer func() { <-slots }() outcome := process(ctx, it) mu.Lock() tally(stats, outcome) mu.Unlock() }(item) } wg.Wait() }见 internal/platform/outbox/runner.go。一个容易被忽略的细节是：Concurrency <= 1时该函数根本不启动 goroutine，而是退回顺序循环，以便需要严格顺序语义的调用方（如邮件链接处理）仍然获得真实的顺序保证——注释解释了为什么“大小 1 的信号量池”不能替代字面循环。
单条任务的失败语义没有被并发破坏
rn.process对每条领取项的处理路径保持不变（runner.go）：
Store.Job读取岗位；若返回损坏页错误（XX001），立即以maxAttempts=1死信，因为不可读行重试永远失败；rn.enrich调用 provider：传输类错误进程内重试一次，errUnparseableResponse则不重试（同样的调用会复现同样结果）；结果先Sanitize()丢弃词表外枚举，再Validate()，校验失败视为确定性错误；Store.Complete在同一事务中写入富化结果并删除 outbox 行（见 cmd/enrich/store.go）；- 任一步失败都走
failN记录，不中断整轮运行。
失败边界分为两类（这与提案的“一条失败不影响整轮”直接相关，也是当前实现的延伸）：postingAtFault判定的“岗位自身问题”按尝试次数封顶（ENRICH_MAX_ATTEMPTS），其余（网关错误、超时、数据库不可达）按条目在队列中的年龄封顶（ENRICH_UPSTREAM_GRACE_DAYS）。enrichment.sql 中的RecordEnrichmentFailure注释解释了原因：租约到期即可重领，队首条目在网关宕机时大约每小时消耗十二次尝试，三次尝试只等于十五分钟；而两次长时间网关故障曾把 172,875 条可富化岗位永久打成死信。
dbStore的并发安全性
设计文档专门用一节说明：dbStore包装的是*pgxpool.Pool，本身对并发使用安全；Job、Complete、Fail各自从池中获取连接，因此除了 runner 之外无需改动。唯一的例外是测试替身：internal/ai/enrich/runner_test.go中的fakeStore与funcProvider必须加互斥锁才能通过-race——当前代码中确实能看到mu sync.Mutex字段与加锁的Complete/Fail/calls（见 runner_test.go）。这让“并发正确性”从注释承诺变成了可验证断言。
配置项：真实默认值与边界行为
提案只给出一个新变量（ENRICH_CONCURRENCY，默认 4）并删除ENRICH_BATCH_SIZE。当前实现位于 internal/platform/config/enrich.go，可整理的完整参数如下：
| 环境变量 | 结构体字段 | 默认值 | 含义与边界 |
|---|---|---|---|
ENRICH_CONCURRENCY | Concurrency | 4 | 同时在飞的 LLM 调用数，同时作为 claim 波次大小；小于 1 会被压到 1 |
ENRICH_LEASE_SECONDS | LeaseSeconds | 300 | 一次领取的租约时长，过期后可被重新领取（兼作崩溃回收） |
ENRICH_MAX_ATTEMPTS | MaxAttempts | 3 | 岗位自身导致失败时的尝试上限 |
ENRICH_UPSTREAM_GRACE_DAYS | UpstreamGraceDays | 14 | 非岗位自身失败时，按条目队列年龄封顶的天数 |
ENRICH_BATCH_SIZE | — | 已移除 | 波次大小改由并发度决定；残留该变量会被忽略 |
非正值为何要 floor 到 1，代码注释写得很直接：LIMIT 0会让 worker 变成静默空转（cron 看起来健康，实际一条都不富化），而负值会把非法LIMIT传给 PostgreSQL。测试 enrich_test.go 对这一行为有明确断言：默认值为concurrency=4 / lease=300 / max=3、ENRICH_CONCURRENCY=8被正确采纳、0与-3均被压到 1。同一测试还校验UpstreamGraceDays默认 14 天且必须显著长于 48 小时，理由是它要覆盖多日网关故障。
envInt的取值规则也值得注意：未设置、空串、无法解析为整数，三种情况都回退默认值，而不是报错——这让一个写错格式的环境变量退化为“保守默认”，而不是让 worker 直接起不来（真正必须存在的 LLM 配置另由Require()快速失败）。
上线步骤、验证与回滚
提案的 Impact 与设计文档的 Migration Plan 给出的落地顺序非常明确，且没有数据库迁移：
- 改查询 → 生成代码：修改 enrichment.sql（Claim + Enqueue），执行
make sqlc重新生成internal/db。Makefile 中该目标通过容器运行固定版本的 sqlc，无需本地安装。 - 发布代码：在 enrich worker 的环境中设置
ENRICH_CONCURRENCY（未设置即默认 4），并从部署配置里删除ENRICH_BATCH_SIZE（留着也会被忽略）。 - 无迁移、可回滚：outbox 表结构未变，回滚方式就是重新部署上一个二进制；旧 worker 与新 worker 甚至可以针对同一队列并行运行。
验证依据（对应当前仓库中的测试）：
- 领取顺序与关岗过滤：
TestEnrichmentClaimPriority，包含“更新的posted_at先被领取”“无日期但刚入库的先于陈旧有日期岗位”“关闭岗位不入队”“已入队后关闭的岗位不被领取”四个子测试（见 enrichment_integration_test.go）； - 租约与不相交：同一文件的
TestEnrichmentQueue用两次 claim 断言不会拿到同一 outbox id； - 并发波次：
TestRunPool_RunsUpToConcurrencyItemsInParallel用阻塞 channel 证明并发度内的条目会同时进入处理（见 outbox/runner_test.go）； - 并发统计与失败不中断：
TestRun_oneFailureDoesNotAbortBatch（见 runner_test.go）； - 全量校验：
go test ./...、go test -race、go test -tags=integration ./internal/db/。
风险与取舍
提案与设计文档列出的风险几乎都可以在当前代码里找到对应处置：
- 并发 4 触发 LLM 端点 429：默认值保守，可按环境调低；失败调用已有“重试一次、再计一次失败”的既有行为，瞬时 429 不会丢条目，租约到期后下次运行重试。共享 outbox 包还预留了
Deferred结果，用于表达“对方要求降速”而不消耗条目尝试次数。 - 关闭岗位残留成为 outbox cruft：这是提案明确承认的“known seam”——claim 过滤会跳过它们，因此无害，清理留给后续变更。
- 并发打乱日志顺序：每条目仍打印自己的
job=… ok/FAILED in …s，并保留每波心跳；批次 worker 允许交错。 - 无日期岗位被饿死：由
COALESCE(posted_at, created_at)解决。 - 非正并发度：
LoadEnrich压到 1，避免静默空转。 - claim 不再由
enrichment_outbox_claimable_idx服务：工作负载是 LLM 延迟受限，最坏全排序开销占比不足 2%，不新增索引，以生产EXPLAIN ANALYZE为触发条件。 - 连接池与并发不匹配：
Complete每条目从共享pgxpool开启短事务（默认MaxConns = max(4, numCPU)），默认并发 4 足够；若把ENRICH_CONCURRENCY调到远高于池大小，写回会在pool.Begin上排队，需要同步提高池大小。
后续演进：把“无害残留”真正清掉
提案把关闭岗位残留列为非目标，而当前代码已经补上了这一步：Runner.Run在入队之前先调用Store.Reap（runner.go），对应查询DeleteIneligibleEnrichmentOutbox（enrichment.sql），并通过常量reapLimit = 20_000限制单轮删除量。查询注释记录了一次生产测量：1,118,601 条条目中 486,178 条位于关闭岗位之后、216,283 条位于重复项之后、19 条无对应岗位，仅 484,651 条可领取——57% 的队列是不可达残留，却被深度指标当成待办工作。
这次演进同样保持了原有取舍的一致性：删除不是有损的（EnqueuePendingJobs会在岗位重新符合条件时自动重建条目）；已死信行被刻意保留，作为freehire_queue_dead_letters指标的证据；单轮限量避免长事务持锁；并且刻意不序列化“岗位重新符合条件”与删除之间的竞态，因为窗口只有一条语句，且下一轮入队会重新评估整个目录——最坏情况是一条岗位延迟一个 cron 周期。
从“串行 1,900 条/天”到“波次等于并发度、租约窗口 ≈ 一次 LLM 调用”，这套改造的价值不在于引入了多少新组件，而在于把排序语义、状态过滤、并发度与租约这四件事对齐到同一个不变量上：每一波被领取的任务都会立刻开始处理，因此租约永远够用；只有仍然开放、非重复、最新优先的岗位才会消耗 LLM 预算。这条不变量既写在了提案的四条变更里，也落实在一条 SQL、一个共享池和一个默认值为 4 的环境变量中。
- 后端
- 前端
- 搜索引擎
- 网页爬虫
- 数据工程
- AI 应用
- 可观测性
【免费下载链接】freehire
freehire — the open-source search engine for job seekers
相关推荐
freehire enrichment 队列优化：新鲜度优先认领与有界并发排空
freehire enrichment 队列优化：新鲜度优先认领与有界并发排空 本文以 freehire 仓库中 enrich priority concurr
后端前端搜索引擎网页爬虫数据工程AI 应用可观测性freehire 富化队列提速实战：按新鲜度优先出队、剔除已关闭职位与并发排空
freehire 富化队列提速实战：按新鲜度优先出队、剔除已关闭职位与并发排空 本文以 freehire 仓库中 2026 06 14 enrich prior
后端前端搜索引擎网页爬虫数据工程AI 应用可观测性CouchDB JSON 结构参考大全：数据库、文档、变更流、复制与请求对象的完整字段速查手册
CouchDB JSON 结构参考大全：数据库、文档、变更流、复制与请求对象的完整字段速查手册 导读 CouchDB 是一个以 HTTP/JSON API 为核
后端前端搜索引擎网页爬虫数据工程AI 应用可观测性
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
