Nasiko Build Worker 全解：基于 Postgres 的 Agent 构建异步任务队列设计与实现-北京尧图网络

【免费下载链接】nasikoDeveloper Control Plane for your AI Agents项目地址https://gitcode.com/gh_mirrors/na/nasiko点击查看免费下载导读本文以 server/src/agents/BUILD_WORKER.md 为骨架深入剖析 nasikoDeveloper Control Plane for your AI Agents中负责 Agent 构建的核心组件 ——build_worker。它用一张 Postgres 表 FOR UPDATE SKIP LOCKED实现了一个零外部中间件无 Redis、无 Celery的进程内异步任务队列支撑 OCI 镜像推送、git clone buildkit 构建、版本回滚等耗时 530 分钟的操作。读完本文你将掌握该队列的表结构设计、唤醒机制、两阶段 panic 隔离执行、尝试计数与三层恢复策略、多副本安全模型以及如何向队列新增一种任务类型。为什么需要任务队列Agent 构建操作——OCI 镜像推送、git clone buildkit 调用、版本回滚——往往需要5 到 30 分钟。如果 HTTP handler 同步执行这类操作请求连接会被长时间占用这在生产环境中非常脆弱负载均衡器有超时限制、客户端可能随时断开。nasiko 的解法是引入一个Postgres 支撑的进程内任务队列HTTP handler 只负责把任务写入build_jobs表并立即返回202 Accepted真正的构建由后台的 build worker 异步执行前端通过轮询agent_builds状态列来获取进度。核心设计要点build_jobs表即队列没有独立的队列服务队列就存在于数据库行中天然具备持久化与事务能力。FOR UPDATE SKIP LOCKED即分布式锁多个 server replica 并发抢任务时只有第一个能锁定某一行其余自动跳过。零外部依赖不引入 Redis、Celery 或任何外部 brokerbuild_worker.rs 的依赖只有sqlx::PgPool、tokio::sync::mpsc与uuid。从源码注释build_worker.rs可以看到worker 主循环run在服务启动时被 spawn 一次每个 replica 一个接收新任务通知的同时以 5 秒为周期兜底轮询、以 10 分钟为周期清扫卡死任务。数据库 schema一张表就是一个队列build_jobs表定义在 migrations/0001_schema.sql核心 DDL 如下CREATE TABLE build_jobs ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), agent_id UUID REFERENCES agents(id) ON DELETE CASCADE, owner_id UUID NOT NULL, payload JSONB NOT NULL, -- BuildJobPayload (tagged enum) status TEXT NOT NULL DEFAULT pending CHECK (status IN (pending,in_progress,done,failed)), attempt INTEGER NOT NULL DEFAULT 0, error_msg TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), picked_at TIMESTAMPTZ, -- set when claimed; NULL pending or done completed_at TIMESTAMPTZ -- set when done or failed ); -- Partial index covers only the statuses the worker queries. CREATE INDEX build_jobs_work_queue ON build_jobs(status, created_at) WHERE status IN (pending, in_progress);几个值得注意的设计决策status用 TEXT CHECK 约束而不是 Postgres 枚举类型——0003_mcp.sql 的注释解释了这一惯例避免枚举迁移的麻烦同时保留约束力。部分索引partial index索引只覆盖pending和in_progress两种 worker 实际查询的状态done/failed的历史行不会拖累索引大小符合队列短、历史长的负载特征。picked_at语义任务被认领时写入为NULL表示从未被认领pending或已完成done。它是判断任务是否卡死的唯一时间依据。ON DELETE CASCADEAgent 在构建中途被删除时对应任务行随之消失。执行器随后对该任务的UPDATE影响 0 行no-op代码中对此做了明确容忍——build_worker.rs 的注释写着the UPDATE above is a no-op (0 rows affected) and thats fine。一个队列服务两类目标Agent 与 MCP Connectorbuild_jobs并不只服务 Agent 构建。在 0003_mcp.sql 中表被扩展用于 MCP Server 构建——MCP 构建没有agents行因此增加了connector_id列并靠一个 CHECK 约束保证每个任务恰好引用一个真实目标ALTER TABLE build_jobs ADD COLUMN connector_id UUID REFERENCES mcp_connectors(id) ON DELETE CASCADE, ADD CONSTRAINT chk_build_jobs_one_target CHECK ( (agent_id IS NOT NULL AND connector_id IS NULL) OR (agent_id IS NULL AND connector_id IS NOT NULL) ); CREATE INDEX idx_build_jobs_connector ON build_jobs(connector_id) WHERE connector_id IS NOT NULL;对应地worker 侧的结构体BuildJobbuild_worker.rs中agent_id与connector_id均为OptionUuid由该 CHECK 保证二者不会同时为Some。后续的恢复逻辑recover_stuck_jobs、reset_panicked_job也据此分支处理两种目标。任务负载变体BuildJobPayload标签化枚举任务的详细参数并不拆成多张表而是整体序列化进payload这一列 JSONB。BuildJobPayload是一个#[serde(tag kind)]枚举定义在 upload.rskind标签决定了 worker 的派发路径dispatch path所有字段都内嵌在 payload 中——worker 执行时不需要回查数据库。文档中的变体表如下VariantTriggerExecutorUploadNew agent uploaded via zipexecute_upload_and_deployinupload.rsUpdateAgent source updated (new version)execute_agent_updateinupdate.rsRollbackAgent rolled back to a previous versionexecute_agent_rollbackinupdate.rsStandaloneBuildBuild from a GitHub URL (no zip)execute_buildinbuild/routes.rs对照 upload.rs 的实际定义当前仓库中枚举实际包含7 个变体文档表格之外还有 3 个Variant触发场景执行器关键负载字段Upload通过 zip 上传新 AgentPOST /api/agents/upload-and-deployexecute_upload_and_deployupload.rszip_path、image_tag、ports、env、writable、writable_pathUpdateAgent 源码更新PUT /api/agents/{id}/updateexecute_agent_updateupdate.rsnew_version、prev_version、changelog、zip_path: OptionStringNone表示 GitHub 重新部署Rollback回滚到先前版本POST /api/agents/{id}/rollbackexecute_agent_rollbackupdate.rstarget_version、target_image_tag、reasonStandaloneBuild从 GitHub URL 构建镜像但不部署POST /api/build/buildsexecute_buildbuild/routes.rsgithub_url、source_key、version_tagClone旧的 GitHub clone-and-deployPOST /api/github/cloneexecute_clone_and_deploy源码注释明确legacy 变体实际已不再入队仅为兼容在途任务保留GithubCloneGitHub clone-and-deploy当前版本execute_github_clone_and_deployrepo_full_name、branch、version_override、prior_version/prior_image/prior_statusMcpServerUploadMCP Server 上传/从 GitHub 构建POST /api/mcp/connectors/upload等execute_mcp_server_buildmcp/build.rsconnector_id、source: McpBuildSourcePayloadZip/Github、加密的env每个变体都携带一个build_id: Uuid——即agent_builds或mcp_connector_builds表的主键。执行结束后worker回读*_builds表中的status来判断成败而不是依赖执行器的返回值执行器把结果写进*_builds行worker 再据此终结build_jobs行build_worker.rs。这一结果以数据库为准的设计让任务状态与构建状态天然一致。值得注意的负载字段设计细节Update::owner_id与Update::agent_owner_id是两个不同的字段前者是触发更新的调用者superuser 或 ACL 授权者可能更新他人的 Agent仅用于状态/审计追踪后者才是 Agent 的真正属主必须注入 LLM router JWT 并持久化到agent_deployments.owner_id保证 Agent 始终解析真实属主的 secrets 与 llm_configupload.rs。带#[serde(default)]的字段如writable_path保证在字段新增之前入队的旧任务行仍能正常反序列化体现向后兼容设计。McpServerUpload的env在落库前按属主密钥加密worker 执行时通过decrypt_build_secrets解密明文绝不以静止状态存在于 payload 中build_worker.rs与 Agent LLM secrets 执行时注入、不持久化RUN-5的既有原则一脉相承。端到端生命周期文档给出了完整的生命周期示意结合源码可以还原每一个环节HTTP handler │ INSERT INTO build_jobs (payload, statuspending) │ build_tx.send(()) ← wake notification (capacity-64 mpsc) │ return 202 Accepted ▼ build worker (background) ├─ select! { notify | 5s poll | 10min recovery_tick } │ └─ drain loop claim_next_job() ← SELECT … FOR UPDATE SKIP LOCKED UPDATE in_progress │ Ok(None) → break (queue empty) │ Ok(Some) → continue ├─ attempt cap check ← fail immediately if attempt ≥ MAX_ATTEMPTS (3) └─ tokio::task::spawn(execute_claimed_job()) │ Ok(()) → check for next job │ is_panic() → reset_panicked_job() immediately └─ cancelled → break (server shutting down)入队端写库唤醒立即返回以 Agent 更新为例update.rsHTTP handler 的入队动作是原子化的三步sqlx::query(INSERT INTO build_jobs (agent_id, owner_id, payload) VALUES ($1, $2, $3)) .bind(agent_id) .bind(owner_id) .bind(serde_json::to_value(payload).expect(serialize update payload)) .execute(state.db) .await?; let _ state.build_tx.send(()).await; // 唤醒 worker ( StatusCode::ACCEPTED, Json(UpdateAgentResponse { build_id, agent_id, new_version, previous_version: current_version, status: queued, }), ).into_response()客户端立刻收到202 Accepted和status: queued随后即可通过agent_builds轮询状态。当前仓库中所有入队点build_tx.send位于upload.rsUploadupdate.rsUpdate与 update.rsRollbackbuild/routes.rsStandaloneBuildmcp/build.rsMcpServerUpload原文档给出的行号为upload.rs:365、update.rs:287/697、build/routes.rs:211与当前仓库代码存在偏差应以实际代码位置为准。消费端select! 三路唤醒 drain 循环worker 主循环build_worker.rs用tokio::select!同时等待三种信号任一信号触发后都进入 drain 循环——连续认领任务直到队列为空而不是只处理一个就回去睡觉避免突发批量任务时每个任务都引入 5 秒延迟。drain 循环内部分两个阶段执行详见下文两阶段执行。唤醒机制三个信号兜底#信号机制作用1mpsc 通知build_tx.send(())channel 容量 64发送为 fire-and-forgetlet _ build_tx.send(()).awaitHTTP handler 入队后立即唤醒 workerchannel 满说明 worker 已在 drain丢弃唤醒也无妨25 秒兜底轮询tokio::time::sleep(Duration::from_secs(5))捕获丢失的通知如发送瞬间 channel 恰好满员把尾部延迟控制在 5 秒内无需心跳列310 分钟恢复 ticktokio::time::interval_at首 tick 在启动后 10 分钟启动时已内联执行过恢复MissedTickBehavior::Skip周期性调用recover_stuck_jobs让崩溃 replica 遗留的任务在多 replica 集群中也能被回收即使没有任何 replica 重启channel 与 spawn 的装配点在 state.rsmpsc::channel(64)创建(build_tx, build_rx)对随后tokio::spawn(crate::agents::build_worker::run(worker_state, build_rx))。两阶段执行claim 与 execute 的 panic 隔离构建执行器可能 panic——比如编解码器里的一个unwrap、原生代码中的坏指针。若不做隔离一个 panic 的构建会击穿整个 worker 协程把所有排队任务永久卡死。修复方案是把claim认领与execute执行分离Phase 1claim在 worker 外层任务中直接运行只做最小化的数据库读写不存在 panic 风险Phase 2execute被包进tokio::task::spawnpanic 只杀死这个子任务不会波及 worker 循环。核心代码build_worker.rs// Phase 1 — claim (outer task, no spawn) let job claim_next_job(state).await?; let job_id job.id; let old_attempt job.attempt; // pre-increment; DB now holds old_attempt 1 // Phase 2 — execute (spawned task; panic here does not kill the worker) match tokio::task::spawn(async move { execute_claimed_job(state_clone, job).await }).await { Ok(()) { /* done */ } Err(ref e) if e.is_panic() reset_panicked_job(state.db, job_id, old_attempt).await, Err(_) break, // task cancelled (server shutdown) }关键细节job_id和old_attempt在spawn 之前就被捕获到外层作用域因此 panic 分支可以立即基于它们执行恢复无需任何共享状态——这正是两阶段分离的价值恢复动作不需要等待 60 分钟的卡死阈值而是毫秒级即时响应。claim_next_job本身build_worker.rs是一个事务let mut tx state.db.begin().await?; let job sqlx::query_as::_, BuildJob( SELECT id, agent_id, connector_id, payload, attempt FROM build_jobs WHERE status pending ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1, ).fetch_optional(mut *tx).await?; let Some(job) job else { tx.rollback().await?; return Ok(None); // 队列为空 }; sqlx::query( UPDATE build_jobs SET status in_progress, picked_at now(), attempt attempt 1 WHERE id $1, ).bind(job.id).execute(mut *tx).await?; tx.commit().await?; Ok(Some(job))认领与标记in_progress 自增attempt在同一事务内完成保证认领即占用的原子性。按created_at排序实现 FIFO 公平调度。尝试计数attempt的完整语义build_jobs.attempt记录一个任务被认领的次数它的生命周期有三处关键节点递增claim_next_job在把任务标记为in_progress的同一事务中执行attempt attempt 1。返回的BuildJob.attempt是递增前的值此时数据库里已是attempt 1。上限检查MAX_ATTEMPTS 3build_worker.rs。认领后立即检查——若old_attempt 3任务被标记failed且不执行build_worker.rs。panic 重置reset_panicked_job接收old_attempt若old_attempt 1 3则永久失败否则重置为pending等待立即重试build_worker.rs。一个容易忽略的细节当任务因超过上限被永久标记failed后它会掉出recover_stuck_jobs的清扫范围该查询只扫描仍为in_progress的行因此 worker 必须同步把 Agent/MCP Connector 驱动到终态调用fail_agent_terminal/fail_mcp_connector_terminal否则目标会永远停留在building/deploying的非终态等待中的 SSE/轮询将无限期挂起。这正是 build_worker.rs 注释中 RUN-4 修复的回归点。恢复策略即时周期两层互补两套机制协同处理卡死任务即时恢复 ——reset_panicked_job由 drain 循环的is_panic()分支触发在 panic 发生后毫秒级动作old_attempt MAX_ATTEMPTS → mark failed otherwise → reset to pending (picked_at NULL)永久失败路径同样会调用fail_agent_terminal终结 Agent 状态避免agent_builds行停留在building。周期恢复 ——recover_stuck_jobs在启动时和每 10 分钟运行一次处理两类场景崩溃 replica 遗留的in_progress任务以及从未 panic 但也没完成的任务网络分区、OOM、构建中途 SIGKILL。阈值STUCK_JOB_MINS 60build_worker.rs是最大构建超时Docker 与 K8s 均默认 30 分钟的 2 倍给大镜像留足余量、避免误伤。对应 SQL-- Permanently fail exhausted jobs UPDATE build_jobs SET status failed, error_msg max attempts exceeded, completed_at now() WHERE status in_progress AND picked_at now() - make_interval(mins 60) AND attempt 3; -- Reset remaining stuck jobs for retry UPDATE build_jobs SET status pending, picked_at NULL WHERE status in_progress AND picked_at now() - make_interval(mins 60) AND attempt 3;实现要点build_worker.rs第一条查询带RETURNING agent_id, connector_id这样被永久失败的构建能立即驱动目标进入终态两条语句都用make_interval(mins $2)绑定STUCK_JOB_MINS常量阈值只存在于一处 Rust 常量中避免在两条 SQL 字符串里重复硬编码第二条查询用rows_affected()记录被重置的任务数仅在有实际影响时输出 warn 日志。恢复矩阵场景恢复途径延迟执行器 panicreset_panicked_jobis_panic()分支即时服务崩溃于构建中途下次启动时recover_stuck_jobs内联执行下次重启多 replicapeer 崩溃且无重启每 10 分钟recover_stuck_jobs≤ 70 分钟通知丢失channel 满5 秒兜底轮询≤ 5 秒多副本安全SKIP LOCKED 与幂等恢复FOR UPDATE SKIP LOCKED保证同一时刻只有一个 replica 能认领某个任务。两个 replica 同时调用claim_next_job时后到者会跳过已被锁定的行转而认领另一个 pending 任务或返回Ok(None)。配合事务先 SELECT 锁定、再 UPDATE 标记认领过程无竞态。recover_stuck_jobs的两条 UPDATE 也可以安全地并发执行多个 replica 会对同一批in_progress且超时的行执行相同的UPDATE … WHEREPostgres 按行 last-write-wins——第二个 replica 对第一个已重置的行执行更新是 no-op天然幂等。另外一个隐性保障worker 循环在认领阶段对数据库错误Err(e)的处理是记录 error 日志并 break等待下一轮信号再次尝试不会因单次 DB 抖动而崩溃。启动与关闭语义启动run()在 state.rs 被tokio::spawn一次每个 replica 一个实例。recover_stuck_jobs在进入第一个select!之前内联执行——确保上一个 replica 遗留的任务在第一个 HTTP 请求到达前就已被重新排队最大化恢复速度。关闭服务收到 SIGTERM 后Tokio 会 dropmpsc::Senderbuild_tx。worker 的notify.recv()分支返回None当前 drain 循环迭代结束后干净退出build_worker.rs。正在执行的tokio::task::spawn任务会运行至完成Tokio 默认关闭行为因此 SIGTERM 前一刻开始的构建仍会完成不会半途丢失。关键常量速查常量值用途MAX_ATTEMPTS3永久失败前的最大认领次数STUCK_JOB_MINS60判定任务卡死的时间阈值2× 最大构建超时Channel 容量64mpsc::channel(64)——worker 已在 drain 时丢弃唤醒信号兜底轮询5 stokio::time::sleep(5s)——捕获丢失的通知恢复间隔10 mininterval_atMissedTickBehavior::Skip首 tick 在启动后 10 分钟启动时已内联恢复这些常量集中在 build_worker.rs 文件顶部配以注释说明设计理由如STUCK_JOB_MINS必须超过 runtime 的build_timeout2× 余量避免大镜像误判。如何新增一种任务类型worker 循环本身是任务类型无关的——它只负责认领、派发、恢复不关心具体业务。新增类型只需 4 步BUILD_WORKER.md 原文源码印证在 upload.rs 的BuildJobPayload中添加一个变体#[serde(tag kind)]会自动写入新的kind标签记得给新增字段加#[serde(default)]以兼容旧行。编写一个 async 执行器函数接收AppState或其拆分字段从现有执行器如execute_agent_update、execute_github_clone_and_deploy可以看到执行器内部直接读写state.db、state.runtime、state.config等构建结果写回*_builds表。在 build_worker.rs 的execute_claimed_job中新增一个 match 分支调用该执行器分支开头可解构负载字段build_id用于最终回读构建状态。在触发构建的 HTTP handler 中INSERT INTO build_jobspayload 序列化为 JSONB随后state.build_tx.send(()).await唤醒 worker返回202 Accepted。不需要修改 worker 循环本身——认领、attempt 上限、panic 隔离、卡死恢复对任何新变体自动生效。单元测试层面可参照 build_worker.rs 中已有的两个测试mcp_server_upload_payload_round_trips_through_json验证 payload 经 JSONB 序列化往返后build_id与label不变decrypt_build_secrets_recovers_encrypted_values_and_drops_bad_ones验证加密 env 的解密与坏密文静默丢弃行为。小结nasiko 的 build worker 是一个以数据库为骨、以 Tokio 为肌的精巧设计用一张build_jobs表承载队列状态机用FOR UPDATE SKIP LOCKED化解多副本竞争用 mpsc 轮询周期清扫三路信号覆盖从即时唤醒到崩溃恢复的全场景用 claim/execute 两阶段分离实现 panic 隔离再用attempt计数与 60 分钟阈值把卡死收敛为可预测的终态。它证明了在 Agent 编排这类需要持久化、可靠性与多副本一致性的场景下Postgres 本身就可以是足够好的任务队列——不需要为队列引入额外的中间件。赞分享【免费下载链接】nasikoDeveloper Control Plane for your AI Agents项目地址https://gitcode.com/gh_mirrors/na/nasiko点击查看免费下载相关推荐PDFMathTranslate异步翻译实现基于Celery的任务队列设计PDFMathTranslate异步翻译实现基于Celery的任务队列设计在处理大型PDF学术论文翻译时同步翻译模式常导致页面卡顿和超时问题。本文将详解PAI 应用人工智能NLPOCRPostGraphile 后台任务Background Tasks实战指南基于 Graphile Worker 的异步任务队列方案PostGraphile 后台任务Background Tasks实战指南基于 Graphile Worker 的异步任务队列方案本文是 PostGra后端API网关MoneyPrinter 架构解析基于 Postgres 队列与 API/Worker 分离的视频生成流水线设计MoneyPrinter 架构解析基于 Postgres 队列与 API/Worker 分离的视频生成流水线设计导读本文以 MoneyPrinter 仓库后端人工智能大模型本地部署媒体生成音视频上一篇比原版 Chromium 更快更顺手三步上手优化浏览器 Thorium下一篇使用 reduxjs/rtk-codemods 的 createReducerBuilder 迁移 Redux Toolkit 的 createReducer 对象语法到 Builder 回调语法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Nasiko Build Worker 全解：基于 Postgres 的 Agent 构建异步任务队列设计与实现

相关推荐

VS Code 与 IDEA 集成 Claude Code 实战指南：用 TaoToken 统一 Key 打通智谱 AI 辅助编码环境

保姆级教程：2026年OpenClaw（Clawdbot）华为云搭建攻略集——TaoToken统一Key接入与config.toml配置骨架

STC32G ARM转型困局：8051工程师如何跨越架构鸿沟

更多资讯

锐评32个AI编程工具：Cursor估值逼近500亿美元登顶，谁在“夯”谁在“拉”？TaoToken统一Key接入实测

华为云Flexus+DeepSeek征文｜用 DeepSeek-R1 当裁判：Dify Agent 自动化回归测试流水线配置实战

基于LSTM的Python日志异常检测：从解析到部署实战

初探来会会OpenClaw这只虾：用TaoToken统一Key跑通Agent配置

小白程序员必看！收藏这份A2A大模型协作秘籍，用TaoToken统一Key轻松跑通AI Agent之间的“社交舞”

建业V8自动变速箱维修靠谱吗，技术实力怎么样

今日更新

BunnyScholar硕士论文跨章节检查：长文档双栏如何梳理因果链

维普标红教育学SEM量表怎么办：助研君修改与RMSEA指标复核

知网标红有限元仿真与网格划分怎么办：助研君按字修改与参数复核

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？