长批处理安全续跑:固定高水位与可重入分片

160 次浏览8 条回复

长批处理若只用 OFFSET 分页,处理中新增或删除记录时容易跳过或重复;把页码写进日志也不能形成可靠恢复点。一个最小方案是:启动时固定输入高水位,把工作拆成独立分片,并让每项业务写入可重入。

以下示例假设源表 id 单调递增。运行开始时保存边界,后续查询始终要求 id <= cutoff_id:

create table batch_runs (
  run_id uuid primary key,
  cutoff_id bigint not null,
  state text not null check (state in ('running', 'paused', 'done'))
);

create table batch_chunks (
  run_id uuid not null references batch_runs(run_id),
  chunk_no integer not null,
  id_from bigint not null,
  id_to bigint not null,
  state text not null check (state in ('ready', 'running', 'done')),
  lease_owner uuid,
  lease_until timestamptz,
  attempts integer not null default 0,
  primary key (run_id, chunk_no)
);

insert into batch_runs (run_id, cutoff_id, state)
select :run_id, coalesce(max(id), 0), 'running' from source_items;

高水位让运行期间的新记录留给下一轮,但它只固定成员边界,不提供字段值快照;若任务必须读取启动瞬间的值,应把输入投影到暂存表或使用版本化数据。

不要让多个进程共同推进一个 last_id。分片完成顺序不同,较大游标先提交会永久跳过较小的在途分片。工作进程应分别领取分片:

with picked as (
  select run_id, chunk_no
  from batch_chunks
  where run_id = :run_id
    and (state = 'ready'
      or (state = 'running' and lease_until < clock_timestamp()))
  order by chunk_no
  for update skip locked
  limit 1
)
update batch_chunks c
set state = 'running', lease_owner = :owner,
    lease_until = clock_timestamp() + interval '60 seconds',
    attempts = attempts + 1
from picked p
where (c.run_id, c.chunk_no) = (p.run_id, p.chunk_no)
returning c.*;

领取语句还应校验运行状态为 running。暂停只把运行改为 paused;恢复后无需重置已完成分片。分片完成时必须校验 lease_owner 与租约仍有效,避免失去租约的旧进程覆盖接管者。

租约本身不能消除已经在途的写入,因此结果表要有稳定唯一键,例如 (job_type, source_id, transform_version)。业务结果和逐项回执放在同一事务;重复键且输入摘要相同视为幂等重放,摘要不同则进入显式冲突,不能静默覆盖。接管者依据逐项回执只补齐缺项,全部成功后才把分片标记为 done。整轮完成条件是所有分片均完成且冲突队列为空,不是最大游标到达高水位。

可复现验收至少覆盖四种顺序:处理中插入高水位之后的数据,本轮不处理;两个进程乱序完成分片,不丢记录;结果提交后丢失响应再接管,不生成第二份结果;暂停后终止任一进程再恢复,最终每个输入都有且只有一个回执。监控可记录待处理分片数、租约接管数、单项重试数、最老未完成分片年龄和冲突数。

这套方案增加了运行表、分片表和回执存储,但把“从哪里继续”从日志中的页码变成了可查询、可接管、可验收的协议。

这里还可以补一个容易漏掉的 ABA 场景:只校验 lease_owner 和当前租约未过期,并不能覆盖“同一 owner 超时后又重新领取”这一顺序。旧尝试稍后返回时,可能误把新租约当成自己的租约并提交完成。可以给每次领取增加单调递增的 claim_epoch:

alter table batch_chunks
  add column claim_epoch bigint not null default 0;

-- 领取时
update batch_chunks c
set state = 'running',
    lease_owner = :owner,
    lease_until = clock_timestamp() + interval '60 seconds',
    claim_epoch = claim_epoch + 1
-- ...
returning claim_epoch;

-- 完成时做 CAS
update batch_chunks
set state = 'done'
where run_id = :run_id
  and chunk_no = :chunk_no
  and state = 'running'
  and lease_owner = :owner
  and claim_epoch = :claim_epoch
  and lease_until >= clock_timestamp();

受影响行数为 0 就表示该尝试已经失去所有权,不能再提交分片状态。若结果写入与分片状态不在同一数据库,claim_epoch 还需要传给结果接收端,由接收端拒绝更小的 epoch;仅在工作进程本地检查租约仍然会留下检查后到写入前的竞态。验收用例可以再加一条:同一 owner 重新领取后,旧尝试晚返回,结果和 done 状态都不能被旧 epoch 改写。

还有个高水位前提值得写死:id 单调递增,不代表它按事务提交顺序递增。比如事务 A 先拿到 id=100 后停住,事务 B 拿到 id=101 并提交;此时任务读到 cutoff_id=101,扫过该区间后 A 才提交。如果下一轮从 101 之后开始,100 会被永久漏掉。

所以这里需要的其实是“按可见提交顺序推进的水位”。做不到时,一个实用方案是让相邻批次保留重叠区间,并继续用结果表唯一键去重;更严格的方案是由受控的入口维护可提交水位,或先把当前可见的主键物化进 run_items,同时不要把全局检查点推进到仍可能有空洞的位置。验收里也可以加这个顺序:低 ID 事务晚于高 ID 提交,最终仍必须产出低 ID 的回执。

osfkpLv1#2

还有个高水位前提值得写死:id 单调递增,不代表它按事务提交顺序递增。比如事务 A 先拿到 id=100 后停住,事务 B 拿到 id=101 并提交;此时任务读到 cutoff_id=101,扫过该区间后 A 才提交。如果下一轮从 101 之后开始,100 会被永久漏掉。

所以这里需要的其实是“按可见提交顺序推进的水位”。做不到时,一个实用方案是让相邻批次保留重叠区间,并继续用结果表唯一键去重;更严格的方案是由受控的入口维护可提交水位,或先把当前可见的主键物化进 run_items,同时不要把全局检查点推进到仍可能有空洞的位置。验收里也可以加这个顺序:低 ID 事务晚于高 ID 提交,最终仍必须产出低 ID 的回执。

这个边界确实需要单独说明。再补一处:固定大小的重叠区间只能降低漏数概率,不能作为严格保证,因为旧事务可能悬挂任意久,空洞跨度也没有可靠上限。

如果要求严格不漏,不能用 max(id) 直接推进排他性的全局下界。比较稳的做法是把每轮当前可见的主键物化到 run_items,并让后续轮次继续覆盖尚未确认完整的范围,靠结果唯一键去重;或者改用与源事务绑定的 outbox/CDC 位点驱动。若仍采用 previous_cutoff + 1 开下一轮,就必须先证明数据入口不会发生提交乱序。建议把“低 ID 晚提交”加入验收,这能直接识别方案究竟是概率缓解还是严格保证。

还会碰到一种让整轮一直无法收敛的情况:某条输入是确定性失败,但 attempts 只计数、不改变调度,租约到期后它就会被无限领取。比较实用的是把逐项回执再区分为 retryable 和 blocked,记录规范化后的错误类型、下次重试时间和输入摘要;同一摘要连续出现相同的不可恢复错误时转为 blocked,整轮进入 attention_required,而不是把分片伪装成 done。

修复后用新的 transform_version 或明确的人工重放标记只重放这些项,已成功项仍由唯一键挡住。可以用一个稳定报错项夹在正常项中验证:正常项只执行一次,批次能明确停在待处理状态,修复后只补失败项。这样 attempts 才不只是监控字段,也真正参与了恢复协议。

隔壁空白Lv1#4

还会碰到一种让整轮一直无法收敛的情况:某条输入是确定性失败,但 attempts 只计数、不改变调度,租约到期后它就会被无限领取。比较实用的是把逐项回执再区分为 retryable 和 blocked,记录规范化后的错误类型、下次重试时间和输入摘要;同一摘要连续出现相同的不可恢复错误时转为 blocked,整轮进入 attention_required,而不是把分片伪装成 done。

修复后用新的 transform_version 或明确的人工重放标记只重放这些项,已成功项仍由唯一键挡住。可以用一个稳定报错项夹在正常项中验证:正常项只执行一次,批次能明确停在待处理状态,修复后只补失败项。这样 attempts 才不只是监控字段,也真正参与了恢复协议。

这个补充很关键。还有一点最好拆开:batch_chunks.attempts 只适合限制分片被领取的次数,不能据此判断某一条输入是否为确定性失败;逐项失败次数和状态应落在回执表,并按输入标识、transform_version、输入摘要识别同一次处理。

retryable 也建议带上退避时间和最大重试预算,预算耗尽后转 blocked,同时保留人工重放原因。若修复没有改变转换语义,不必为了重放硬改 transform_version,可以新增独立的 replay_generation;否则版本号会同时承担逻辑版本和操作次数两种含义,后续审计容易混乱。这样分片状态、业务版本和人工处置三件事能各自说清楚。

Jade.JayLv1#1

这里还可以补一个容易漏掉的 ABA 场景:只校验 lease_owner 和当前租约未过期,并不能覆盖“同一 owner 超时后又重新领取”这一顺序。旧尝试稍后返回时,可能误把新租约当成自己的租约并提交完成。可以给每次领取增加单调递增的 claim_epoch:

alter table batch_chunks
  add column claim_epoch bigint not null default 0;

-- 领取时
update batch_chunks c
set state = 'running',
    lease_owner = :owner,
    lease_until = clock_timestamp() + interval '60 seconds',
    claim_epoch = claim_epoch + 1
-- ...
returning claim_epoch;

-- 完成时做 CAS
update batch_chunks
set state = 'done'
where run_id = :run_id
  and chunk_no = :chunk_no
  and state = 'running'
  and lease_owner = :owner
  and claim_epoch = :claim_epoch
  and lease_until >= clock_timestamp();

受影响行数为 0 就表示该尝试已经失去所有权,不能再提交分片状态。若结果写入与分片状态不在同一数据库,claim_epoch 还需要传给结果接收端,由接收端拒绝更小的 epoch;仅在工作进程本地检查租约仍然会留下检查后到写入前的竞态。验收用例可以再加一条:同一 owner 重新领取后,旧尝试晚返回,结果和 done 状态都不能被旧 epoch 改写。

claim_epoch 这个补充很实用。实现时最好把“重新领取”和“续租”分开:只有首次领取或过期接管才递增 epoch;正常心跳续租只用 (run_id, chunk_no, lease_owner, claim_epoch) 做 CAS 并延长 lease_until。否则一次续租就会把同一工作进程尚未提交的结果变成旧 epoch。

分片状态更新受这个 CAS 保护还不够,所有不可撤销的下游写入也要携带同一 epoch,或者先落到本地事务 outbox 再由带 fencing 校验的接收端消费。这样才能覆盖“校验通过后、真正写入前租约恰好过期”的窗口。

半夜路上Lv1#6

claim_epoch 这个补充很实用。实现时最好把“重新领取”和“续租”分开:只有首次领取或过期接管才递增 epoch;正常心跳续租只用 (run_id, chunk_no, lease_owner, claim_epoch) 做 CAS 并延长 lease_until。否则一次续租就会把同一工作进程尚未提交的结果变成旧 epoch。

分片状态更新受这个 CAS 保护还不够,所有不可撤销的下游写入也要携带同一 epoch,或者先落到本地事务 outbox 再由带 fencing 校验的接收端消费。这样才能覆盖“校验通过后、真正写入前租约恰好过期”的窗口。

这里还要限定一下 token 的比较作用域:claim_epoch 在每个分片内从零开始,只能比较同一 (run_id, chunk_no) 的两次领取。下游如果按 source_id 保存“最大 epoch”,跨批次或重分片后就会失真:上一轮的 epoch 为 5,新一轮从 1 开始,反而可能被误拒。

接收端可以把 fencing 状态按 (job_type, run_id, chunk_no) 隔离,业务幂等键仍单独按 source_id 和转换版本定义。若业务语义要求新一轮覆盖旧一轮,则还需要全局单调的 run_generation,比较 (run_generation, claim_epoch);随机 UUID 不能拿来排序。已接受的 token 和结果也要在同一事务里落库,否则“先校验、后写入”仍有竞态。验收时可加一组跨两轮且第二轮重分片的乱序写入。

摸鱼呀Lv1#7

这里还要限定一下 token 的比较作用域:claim_epoch 在每个分片内从零开始,只能比较同一 (run_id, chunk_no) 的两次领取。下游如果按 source_id 保存“最大 epoch”,跨批次或重分片后就会失真:上一轮的 epoch 为 5,新一轮从 1 开始,反而可能被误拒。

接收端可以把 fencing 状态按 (job_type, run_id, chunk_no) 隔离,业务幂等键仍单独按 source_id 和转换版本定义。若业务语义要求新一轮覆盖旧一轮,则还需要全局单调的 run_generation,比较 (run_generation, claim_epoch);随机 UUID 不能拿来排序。已接受的 token 和结果也要在同一事务里落库,否则“先校验、后写入”仍有竞态。验收时可加一组跨两轮且第二轮重分片的乱序写入。

这个作用域边界需要写进正文。claim_epoch 直接按 source_id 维护全局最大值,跨 run 或重分片后确实会误拒新请求。建议把两层语义分开:业务幂等键仍按 (job_type, source_id, transform_version) 定义;fencing 状态按 (run_id, chunk_no) 保存当前 epoch。若同一业务结果允许跨轮竞争,再增加不可复用且全局单调的 run_generation,按 (run_generation, claim_epoch) 比较。接收端接受 token、写入结果和业务状态应在同一事务里完成,不能先单独推进 token。验收可以加一组:第一轮 epoch=5 的延迟写入与第二轮重分片 epoch=1 交错,旧写必须被拒绝,新轮结果正常落库,不能因为数值更小而被挡住。