Documentation
¶
Overview ¶
board.go —— taskboard.Board 的 PostgreSQL 参考实现。
判据落库形态 ¶
- **认领是读-判-写不可分割**:一个事务里 SELECT … FOR UPDATE 锁任务行, 依次判(不存在 / 终态 / 依赖 / 被活持有),通过才写——两个进程同时认领 同一任务时,输家会在行锁上等到赢家提交,然后读到 claimed 状态 ⇒ ErrTaken。并发正确性由 taskboard.ConcurrentClaim 的共享套件判。
- **令牌与接管代次由存储铸造**:令牌用随机 128 位(跨进程不复用不重合, ABA 封死);代次 = 该任务已见最大值 + 1,与令牌同一次写落库(同源)。
- **释放 / 完成 / 续期只认令牌**:条件更新 WHERE token = $2,影响行数为 0 即 ErrLeaseLost(迟到持有者放不掉别人的闸)。
到期时刻由**存储自己的时钟**算(可注入):调用方的时钟与存储的时钟可能 对不上,跨进程时更不能假设一致。
Package pgstore 副作用闸 / 任务认领面 / 进度快照的 **PostgreSQL 参考实现**。
生命周期: Experimental(v0.33.0 定级)——参考实现,随契约套件与参考部署 演进,不承诺任何兼容,升级风险自负;定级理由见 api-lifecycle.yaml,分级 规则见 docs/api-compat.md。
它是什么、不是什么 ¶
这是 examples 里的**参考实现**,不是内核依赖:内核(sideeffect / taskboard / checkpoint)依旧只给契约与判据,不给存储。本包演示的是「把同样的判据落到 一个真实数据库的原子操作上」长什么样——宿主照抄它,比照着契约散文重写, 漂移的概率小一个量级。
它与 examples/reference-runtime(内存版参考实现)是一对:
- 内存版证明「判据本身是对的」(进程内互斥锁即可);
- 本包证明「判据在**跨进程**下依然成立」(判据归存储,不归进程)—— 用内核的共享契约套件(sidetest / boardtest / RunSinkContract)在 真实 PostgreSQL 上原样跑通,不为任何实现开后门。
三套契约各自的落库形态 ¶
- sideeffect.Store:Acquire 的「三判合一」(已完成 > 结果未知 > 执行中 > 资源被占 > 授予)在一个事务里完成——SELECT … FOR UPDATE 行锁把 「读状态 → 判 → 写」钉成跨进程原子;执行意图与授予(令牌 + 持有者身份, sideeffect.MintHolder 在同一事务里铸造)写在同一事务里。Complete / Renew / MarkUncertain / Abandon / Release 全部三项联查(令牌 + 持有者 + 资源一致),缺一即拒(ErrFenceStale / ErrNotHolder / ErrResourceMismatch)。
- sideeffect.FencedSink:一条带条件的 INSERT … ON CONFLICT DO UPDATE (WHERE seen_fence <= presented)先原子预占,写入体在同一事务里执行; 写入体报错 ⇒ 整个事务回滚 ⇒ 不得部分写。
- taskboard.Board:认领是一个事务里的读-判-写(行锁),令牌与接管代次 在同一次操作里铸造;释放 / 完成 / 续期只认令牌。
- checkpoint.StoreCAS:一条带条件的 upsert(WHERE 持久化序号 < 本次序号), 影响行数为 0 即判陈旧。
时钟 ¶
过期判定读可注入时钟(sidetest / boardtest 的硬要求):缺省 time.Now, 经 UseClock 注入。跨进程部署时所有进程共享真实时钟即可;测试注入假时钟时, 存进数据库的时刻与比较用的「当前时刻」同源,判据不变。
数据库连接 ¶
构造函数直接收 *pgxpool.Pool 与 schema 名:本包不管理连接串与凭据, 那是宿主的接线责任。隔离方案是**每个实例一个 schema**(EnsureSchema 建 CREATE TABLE IF NOT EXISTS 一套表),契约套件每个子测试建一份全新实现 ——与「factory 每次构造全新实现」的套件约定对齐。
部署与验证入口 ¶
- 契约 + 故障恢复用例:GAK_PGREF_DSN=… go test ./examples/pgstore/… (见 examples/docker-compose.yml 与 Makefile 的 integration 目标);
- 冒烟部署:docker compose -f examples/docker-compose.yml up(pgsmoke 服务)。
effect.go —— sideeffect.Store 的 PostgreSQL 参考实现:类型、事务骨架与行读写。
三条判据在库里的落法(与内存版 effectStore 逐条对应) ¶
- **三项联查**:Complete / Release / Abandon / Renew / MarkUncertain 都在 同一个事务里校验「令牌仍是最新代 + 持有者身份一致 + 租约与动作的资源 一致」。校验与写入在同一事务 ⇒ 任何一步失败都发生在**任何状态被改之前** ——这正是 MarkUncertain「标记写失败必须保住执行意图」的落库形态。
- **执行意图**:Acquire 的授予(令牌 + 持有者身份)与意图登记写在 同一事务里;意图过期未裁决 ⇒ 同一 Key 转入未知态而非重新授予。
- **对账占用**:BeginReconcile 与裁决分表字段,占用过期可重入。
原子性来自哪里 ¶
每个方法 = 一个事务 = 若干条 `SELECT … FOR UPDATE` + 条件写。行锁把 「读状态 → 判 → 写」钉成跨进程原子:第二个进程在第一个提交前读到的 永远是旧状态,且会在行锁上等待。这与「一条带条件的 UPDATE」等价, 是契约点名认可的两种形态之一。
锁序恒为「资源行 → 动作行」:所有方法按同一顺序加锁,两个并发事务 不可能交叉持锁(交叉死锁在锁序一致时不存在)。
effect_acquire.go —— Acquire:三判合一的一个事务。
判定顺序是契约里最要紧的一行:**已完成 > 结果未知 > 执行中 > 资源被占 > 授予**。 顺序写反的代价是重复施加:把「结果未知」排到「资源被占」之后,调用方会拿到 contended(以为「晚点重试就行」),于是重试发生、重复施加随之发生。
为什么整段在一个事务里:三判合一的每一步都读同一份状态,行锁保证第二个 进程要么读到判定前的旧状态(然后在锁上等),要么读到判定后的新状态—— 不存在「两个进程同时读到没做过、资源空闲」的窗口。
effect_lease.go —— 凭据一族:Complete / Release / Abandon / Renew / MarkUncertain。
五个方法共用同一套纪律:
- 校验与写入在**同一个事务**里 ⇒ 校验失败即整体回滚,存储侧不剩半截状态;
- 资源行 → 动作行的固定锁序(见 effect.go);
- 三项联查不过即拒(ErrFenceStale / ErrNotHolder / ErrResourceMismatch), 拒绝时**不落任何一行**。
effect_reconcile.go —— 对账侧:ResolveUncertain 与 BeginReconcile。
裁决**不校验栅栏**(契约硬要求):它的依据是**下游事实**,不是持有权—— 原持有者可能已经死亡、租约早已过期,而对账恰恰要在这种情况下进行。 占用与裁决的关系:占用过期可重入(对账者死亡不永久占住);裁决成功即 清除占用(这次尝试已有结论)。
schema.go —— 连接池构造与建表。
表结构与判据的对应关系(每张表管一件事,见 doc.go):
- pgref_effect_action Key 级状态(做过没有 / 结果未知 / 执行意图 / 对账占用)
- pgref_effect_resource 资源级状态(这个资源谁占着 / 发第几代令牌)
- pgref_downstream 下游资源(已见最大令牌 + 参考业务数据,FencedSink 的落点)
- pgref_board_task 任务认领面(taskboard.Board)
- pgref_snapshot 进度快照(checkpoint.StoreCAS)
sink.go —— sideeffect.FencedSink 的 PostgreSQL 参考实现(下游栅栏适配)。
原子性要求(实现这条契约的唯一要点) ¶
「比较与写入必须在**同一次原子操作**里完成」——本实现把它落成两件事:
- **栅栏预占是一条带条件的 upsert**:INSERT … ON CONFLICT (resource) DO UPDATE SET seen_fence = EXCLUDED.seen_fence WHERE seen_fence <= presented。 影响行数为 0 ⇒ 令牌陈旧(ErrFenceStale),此刻**什么都没写**; 影响行数为 1 ⇒ 预占成功,且这行就是「比较 + 更新已见最大」本身。
- **写入体在同一事务里执行**:ctx 里带上了事务句柄(TxFromContext), 写入体的业务写入必须用 ExecWithin / QueryRowWithin 走这个事务。 写入体报错 ⇒ 整个事务回滚 ⇒ 预占与已写的业务数据一起消失—— **不得部分写**。
反面教材(为什么不能三步写) ¶
「先读已见最大 → 判断 → 再写入」三步各自独立提交时,两个持有者可以同时 读到同一个旧值、双双通过检查,然后先后写入——更旧的那次写入反而后落, 覆盖掉更新的。栅栏形同虚设。这条失败形态在 sideeffect 包内有确定性用例 (TestNonAtomicSinkIsCaught)与共享套件的并发场景双向钉住。
snapshot.go —— checkpoint.StoreCAS 的 PostgreSQL 参考实现。
契约只有两条,落库形态也是两条:
- SaveIfNewer:一条带条件的 upsert(WHERE 已持久化序号 < 本次序号), 影响行数为 0 ⇒ 判陈旧(ErrStaleSeq),**不写入**;Seq == 0 一律先拒 (它不是一次真实提交,放它落库等于给了它唯一能覆盖既有快照的写)。 并发提交同一 run 时,两个进程的条件更新在行锁上排队,只有序号更大的 那个落库——单调闸在存储侧成立,不在进程侧。
- Load:无记录返回 ErrNotFound(**不得**返回零值 + nil——那会被上游 误判成「这个运行从没跑过」从而全量重跑)。
Index ¶
- func DropSchema(ctx context.Context, pool *pgxpool.Pool, schema string) error
- func EnsureSchema(ctx context.Context, pool *pgxpool.Pool, schema string) error
- func ExecWithin(ctx context.Context, sql string, args ...any) (int64, error)
- func NewSchema(ctx context.Context, pool *pgxpool.Pool) (string, error)
- func TxFromContext(ctx context.Context) (pgx.Tx, bool)
- func WithTx(ctx context.Context, tx pgx.Tx) context.Context
- type BoardStore
- func (b *BoardStore) Claim(ctx context.Context, id taskboard.TaskID, agent taskboard.AgentID, ...) (taskboard.Task, error)
- func (b *BoardStore) Complete(ctx context.Context, id taskboard.TaskID, token string) error
- func (b *BoardStore) List(ctx context.Context) ([]taskboard.Task, error)
- func (b *BoardStore) Release(ctx context.Context, id taskboard.TaskID, token string) error
- func (b *BoardStore) Renew(ctx context.Context, id taskboard.TaskID, token string, ...) (taskboard.Task, error)
- func (b *BoardStore) Seed(tasks ...taskboard.Task)
- func (b *BoardStore) UseClock(now func() time.Time)
- type Downstream
- func (d *Downstream) Marker(ctx context.Context, resource string) (string, error)
- func (d *Downstream) SeenFence(ctx context.Context, resource string) (sideeffect.Fence, error)
- func (d *Downstream) WriteFenced(ctx context.Context, resource string, presented sideeffect.Fence, ...) error
- func (d *Downstream) WriteMarker(ctx context.Context, resource string) error
- type EffectStore
- func (s *EffectStore) Abandon(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, _ string) error
- func (s *EffectStore) Acquire(ctx context.Context, act sideeffect.Action, ttl time.Duration) (sideeffect.Outcome, sideeffect.Lease, sideeffect.Result, error)
- func (s *EffectStore) BeginReconcile(ctx context.Context, act sideeffect.Action, ttl time.Duration) (sideeffect.Reconcile, error)
- func (s *EffectStore) Complete(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, ...) (sideeffect.Result, error)
- func (s *EffectStore) MarkUncertain(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, ...) error
- func (s *EffectStore) Release(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease) error
- func (s *EffectStore) Renew(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, ...) (sideeffect.Lease, error)
- func (s *EffectStore) ResolveUncertain(ctx context.Context, act sideeffect.Action, res sideeffect.Resolution) (sideeffect.Outcome, sideeffect.Result, error)
- func (s *EffectStore) UseClock(now func() time.Time)
- type SnapStore
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DropSchema ¶
DropSchema 删除一个 schema(含其中的表)。
func EnsureSchema ¶
EnsureSchema 在 schema 里建好全部参考表(幂等,可重复执行)。
schema 由调用方先建好(或用 NewSchema 一并创建);本函数只建表。 search_path 用 SET LOCAL 在本事务内指定——否则 DDL 落到调用方连接的 缺省 schema(public)里,读侧的 search_path 指到的却是一套空 schema。
func NewSchema ¶
NewSchema 建一个随机命名的 schema,返回名字。
契约套件的 factory 每个子测试调一次,构造一份**全新**实现——在共享的 数据库里,「全新」的落法是独立 schema:表状态互不可见,t.Cleanup 时整块删除。
func TxFromContext ¶
TxFromContext 返回写入体所在的事务句柄。
写入体**必须**通过它(或 ExecWithin / QueryRowWithin)执行业务写入—— 绕开事务自己开连接写入的数据不受栅栏保护,也无法在失败时一起回滚。
Types ¶
type BoardStore ¶
type BoardStore struct {
// contains filtered or unexported fields
}
BoardStore taskboard.Board(含 RenewingBoard)的 PostgreSQL 参考实现。
func (*BoardStore) Claim ¶
func (b *BoardStore) Claim(ctx context.Context, id taskboard.TaskID, agent taskboard.AgentID, req taskboard.ClaimRequest) (taskboard.Task, error)
Claim 认领一个任务(读-判-写不可分割:一个事务完成)。
func (*BoardStore) Renew ¶
func (b *BoardStore) Renew(ctx context.Context, id taskboard.TaskID, token string, req taskboard.ClaimRequest) (taskboard.Task, error)
Renew 续期并确认持有资格(taskboard.RenewingBoard)。
校验(令牌是当前租约的 + 租约未过期)与延长在**同一个事务**里——原子; TTL 语义与 Claim 一致;续期不换令牌、不递增代次(续期不是接管)。
func (*BoardStore) Seed ¶
func (b *BoardStore) Seed(tasks ...taskboard.Task)
Seed 摆初始任务(契约里没有「创建」入口:任务拓扑是宿主的资产)。
签名与套件要求的形态一致(boardtest 经 interface{ Seed(tasks ...Task) } 取用,没有 error 返回位)。落库失败时**panic**:这是装载入口,静默失败 会让套件在一张空表上跑出误导性的结论——摆不进去就当场响。
幂等:同 id 重摆覆盖内容并清租约(套件每个子测试从全新 schema 起步, 生产宿主通常只在装载拓扑时调用它)。
func (*BoardStore) UseClock ¶
func (b *BoardStore) UseClock(now func() time.Time)
UseClock 注入时钟(boardtest 契约套件用)。
type Downstream ¶
type Downstream struct {
// contains filtered or unexported fields
}
Downstream FencedSink 的 PostgreSQL 参考实现。
func NewDownstream ¶
func NewDownstream(pool *pgxpool.Pool) *Downstream
NewDownstream 构造下游栅栏(pool 归调用方管理)。
func (*Downstream) SeenFence ¶
func (d *Downstream) SeenFence(ctx context.Context, resource string) (sideeffect.Fence, error)
SeenFence 读资源已见过的最大令牌(无记录返回 0)。
func (*Downstream) WriteFenced ¶
func (d *Downstream) WriteFenced(ctx context.Context, resource string, presented sideeffect.Fence, write func(ctx context.Context) error) error
WriteFenced 在资源上执行一次带栅栏令牌的写入。
语义(契约逐条对应):
- presented 小于已见最大 → 不执行写入,返回 ErrFenceStale(此刻零写入);
- 否则执行写入体,并把已见最大更新为 presented(**等于**也放行—— 同一代持有者的多次写入合法,如分批提交);
- 第 1 条的判定与第 2 条的写入(含令牌更新)在同一个事务里。
func (*Downstream) WriteMarker ¶
func (d *Downstream) WriteMarker(ctx context.Context, resource string) error
WriteMarker 在写入体里写一个参考业务标记(值追加 'm')。
它是参考实现的「业务写入」占位:宿主的真实业务写入走同一条路 (ExecWithin,同一事务)。供共享套件判「不得部分写」。
type EffectStore ¶
type EffectStore struct {
// contains filtered or unexported fields
}
EffectStore sideeffect.Store 的 PostgreSQL 参考实现。
一个实例绑定一个 schema(全部表在其中);并发安全。
func NewEffectStore ¶
func NewEffectStore(pool *pgxpool.Pool) *EffectStore
NewEffectStore 构造副作用闸存储。pool 归调用方管理(Close 由它负责)。
schema 必须已建好表(EnsureSchema);search_path 已被连接池指到该 schema (见 NewEffectStores),故本包的 SQL 不写 schema 前缀。
func (*EffectStore) Abandon ¶
func (s *EffectStore) Abandon(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, _ string) error
Abandon 显式放弃本次尝试:撤销执行意图,并释放(仍是自己的)资源持有。
资源闸只在仍是自己的时才放——易主了就不碰(那是新持有者的闸)。
func (*EffectStore) Acquire ¶
func (s *EffectStore) Acquire(ctx context.Context, act sideeffect.Action, ttl time.Duration) (sideeffect.Outcome, sideeffect.Lease, sideeffect.Result, error)
Acquire 一次原子判定(契约语义逐条对照 sideeffect.Store 的文档)。
授予时在同一事务里做三件事:令牌 = 资源已见最大值 + 1;持有者身份由 sideeffect.MintHolder() 在**同一事务**里铸造;执行意图(本次令牌 + 到期 时刻)随授予一并落库。三者分开写就等于没写:中间死掉 ⇒ 意图悬空或没写, 重复施加的入口重新打开。
func (*EffectStore) BeginReconcile ¶
func (s *EffectStore) BeginReconcile(ctx context.Context, act sideeffect.Action, ttl time.Duration) (sideeffect.Reconcile, error)
BeginReconcile 认领对账权:同一 Key 同一时刻至多一个对账者。
仅当该 Key 处于未知态时接受认领;占用未过期时拒绝(ErrReconcileTaken); 占用过期可重新认领(可重入)——对账者死亡不永久占住这个 Key。
func (*EffectStore) Complete ¶
func (s *EffectStore) Complete(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, res sideeffect.Result) (sideeffect.Result, error)
Complete 标记该键的动作已完成并留存结果。**先到者为准**。
幂等:同一 Key 重复 Complete 不报错,返回首次记录的结果(后到者不覆盖)。
func (*EffectStore) MarkUncertain ¶
func (s *EffectStore) MarkUncertain(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, reason string) error
MarkUncertain 把该键标记为「结果未知」(按键,与资源租约无关)。
标记写失败必须保住执行意图(契约硬要求第 5 条) ¶
本实现「校验 → 改状态」在一个事务里完成:任何一步失败都发生在**任何状态 被改之前**,整体回滚 ⇒ 意图原样留在存储里,后续 Acquire 被意图挡住。 这不是实现自由:落到数据库上时,标记与撤销意图**不得**「先撤意图、再标 未知」分两步提交——第二步失败,存储侧就什么都不剩,另一个 worker 直接重放。
标记成功后意图一并撤销(这次尝试已有结论:未知),但**闸不自动释放** ——调用方还要显式调 Release 只放资源。
func (*EffectStore) Release ¶
func (s *EffectStore) Release(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease) error
Release 释放**资源持有**,**不动**该键的执行意图。
意图没被裁决之前,这个键不得再被执行——否则「施加成功但进程死在记账 之前」会被下一次 Acquire 变成重复施加。两条出口(Abandon / Release) 不可互相替代,见 Store.Release 的契约注释。
func (*EffectStore) Renew ¶
func (s *EffectStore) Renew(ctx context.Context, act sideeffect.Action, lease sideeffect.Lease, ttl time.Duration) (sideeffect.Lease, error)
Renew 续期并**同时确认持有资格**:闸已易主或尝试已失效则拒绝续期 (假续期是重复施加的入口)。
func (*EffectStore) ResolveUncertain ¶
func (s *EffectStore) ResolveUncertain(ctx context.Context, act sideeffect.Action, res sideeffect.Resolution) (sideeffect.Outcome, sideeffect.Result, error)
ResolveUncertain 裁决「结果未知」的键。**先到者为准**。
RetrySafetyUnsafe 的动作:未生效的裁决不解除阻塞(ErrUnsafeNotRetryable), 必须 Resolution.ForceRetryable 的显式决策才放行——「不可重试」是判据, 不是注释。
func (*EffectStore) UseClock ¶
func (s *EffectStore) UseClock(now func() time.Time)
UseClock 注入时钟(sidetest 契约套件用)。
过期判定必须读它,否则过期场景恒不触发(套件的硬要求)。