Pulse 工作流的检查点存储与恢复
Pulse Durable 为 Pulse 工作流提供检查点存储与恢复。你仍然使用 Pulse 定义图、节点和输入输出，通过 Durable 的 Run 创建一次运行，通过 Resume 恢复已有运行。
完整节点链的检查点被存储确认后，输出才会发布给下游。恢复会复用有效作用域内已确认的节点成果，未确认的节点从函数开头重新执行。Durable 不保存 Go 调用栈，也不自动撤销数据库写入、HTTP 请求等外部副作用。
本文以目标 Durable v0.1.0 为 API 基线，需要 Go 1.26 或更高版本，依赖 Pulse v0.5.0。
发版准备中，尚不称已发布。 tag 发布状态以 GitHub Releases 为准；相应 tag 发布前，以下安装命令尚不可用。
go get github.com/Luo-root/pulse-durable@v0.1.0
# 可选后端，按需安装：
go get github.com/Luo-root/pulse-durable/sqlitestore@v0.1.0
go get github.com/Luo-root/pulse-durable/postgresstore@v0.1.0两个后端是独立模块。三 tag 命名、发布前开发 pin 备选及依赖核对见安装；完整使用见 SQLite 与 PostgreSQL，源码验证见贡献指南。
下面的 main.go 声明一个输入和一个出口，运行“加一”节点，再读取本次执行的出口。MemoryStore 仅用于最小演示；跨进程恢复请使用持久存储。
package main
import (
"context"
"fmt"
"log"
"github.com/Luo-root/pulse"
durable "github.com/Luo-root/pulse-durable"
)
func main() {
if err := run(); err != nil {
log.Fatal(err)
}
}
func run() error {
runtime, err := durable.New(durable.NewMemoryStore(), durable.NewCodecs())
if err != nil {
return err
}
input := pulse.NewKey[int]("input")
output := pulse.NewKey[int]("output")
graph, err := pulse.New("quickstart", "Add one")
if err != nil {
return err
}
if err := graph.DeclareInput(pulse.NewPort(input)); err != nil {
return err
}
if err := graph.DeclareOutput(pulse.NewPort(output)); err != nil {
return err
}
if err := graph.Add(pulse.NewNode("add", "Add one",
pulse.Requires(input), pulse.Provides(output),
func(rc *pulse.RunCtx) error {
value, err := pulse.Get(rc, input)
if err != nil {
return err
}
return pulse.Set(rc, output, value+1)
})); err != nil {
return err
}
var inputs pulse.Inputs
if err := pulse.Seed(&inputs, input, 41); err != nil {
return err
}
execution, err := runtime.Run(context.Background(), graph, inputs,
durable.RunOptions{RunID: "quickstart-001", Version: "increment/1"})
if err != nil {
return err
}
value, state, err := pulse.ReadOutput(execution, output)
if err != nil {
return err
}
if state != pulse.SlotStateReady {
return fmt.Errorf("unexpected output state: %s", state)
}
fmt.Println("result:", value, state)
return nil
}在已安装依赖的模块中执行：
go run .预期输出：
result: 42 ready
从空目录开始的完整步骤见快速开始。要观察真正退出进程再恢复的效果，请继续进程退出后恢复；该教程使用文件存储和两个独立进程。
在线文档：中文 · English。以下链接指向仓库内的文档源文件。
| 任务 | 文档 |
|---|---|
| 了解定位与阅读顺序 | 指南概览 |
| 安装 v0.1.0 并检查依赖与发布状态 | 安装 |
| 完成第一次运行并读取出口 | 快速开始 |
| 保留检查点并在新进程中恢复 | 进程退出后恢复 |
| 组合子图并理解 Retry 恢复 | 子图与重试 |
| 保存业务结构体与自定义类型 | 值编码 |
| 使用文件存储或实现 Store | 存储 |
| 使用 SQLite 并在新进程中恢复 | SQLite 教程 |
| 使用 PostgreSQL 并管理连接与权限 | PostgreSQL 教程 |
| 估算 1–10000 步的运行、存储与恢复成本，复跑同任务对照 | 规模与性能 · 评测源码与方法 |
| 核对 API、记录和错误规则 | 参考概览、执行契约、数据模型、错误处理 |
- 支持普通值、受管
pulse.Stream[T]、静态子图，以及节点 Retry 和消息重试预算的保存与恢复。Resume延续剩余预算；已失败的父尝试下的子成果不会复用。 - 流消息先可靠追加再交付；
HandleMessage的输入确认、显式SetMessageState状态和输出在同一次 Save 中确认。消息贡献的普通值在恢复时认回完整节点暂存，仍等完整节点确认才发布，不重跑已 Ack 输入。活跃来源须声明WithStreamSource(id, true)，按StreamWriter.Position()与SendAt续读；已完成来源会重建未确认尾部而不重跑业务体。 RunID和业务Version必须是非空 UTF-8 文本，非法文本在访问 Store 前作为参数错误拒绝；Version不要求 SemVer 格式。恢复需要相同的定义、输入和编码契约；修改业务函数、闭包或切面语义时，需要升级业务版本。filestore面向本机单宿主，依赖文件系统的独占锁和同目录原子替换。Acquire 持锁后清理本运行遗留的严格匹配暂存文件，清理错误会返回；JSON 记录始终是唯一权威。它不提供分布式租约、跨机器调度或断电持久性保证。MemoryStore只验证进程内协议。- 未确认节点的外部副作用可能重复。业务应使用幂等键、事务、去重或 outbox。Durable 不承诺 exactly-once 或 deterministic replay；宿主可按组合指南登记业务等待并 Cancel/Resume，库不内置唤醒服务。不支持 channel、函数或
unsafe.Pointer值。 - 新建记录使用格式
2；普通格式1仍可恢复，下一次保存升级为2。旧 Runtime 拒绝新格式。Codec 编码消息元素、游标和显式状态，不编码流端点；普通值中嵌套 channel 的拒绝不变。 RunOptions.BeforeStart在 Start 前供宿主登记pulse.SubscribeStream并启动读取任务；Run/Resume 仍阻塞 Wait 并持有 Session，读取任务须确认消息或显式停止。- 检查点包含编码后的业务输出，并非加密容器。存储访问控制、备份与数据保留由宿主管理。
在仓库根目录、未启用本地工作区替换的环境中执行：
go build ./...
go vet ./...
go test -race -count=1 ./...节点、子图与受管流的可运行示例集中在 example_test.go；真实进程退出与重启测试见 integration/recovery_test.go。
提交修改前请阅读贡献指南，社区交流遵守行为准则。安全漏洞请按安全政策私密上报。
本项目采用 MIT 许可，Copyright © 2026 Luo-root。Logo 中的字体轮廓素材保留第三方字体许可。
为 Pulse 工作流提供检查点存储与恢复能力。
如果你在做长流程、需要跨进程恢复的执行，Star 一下能帮它被更多人看到 —— 对个人维护的开源项目帮助很大。