输入投递语义
ArkFlow 在崩溃后如何处理消息,取决于输入能否重新投递尚未确认的消息。本页对所有输入进行分类,并解释至少一次(at-least-once)投递契约。
背景:输出确认后再确认的规则
对每一条输入,数据源只有在下游输出确认写入之后才会被确认。这意味着在读取消息与写出消息之间发生崩溃不会静默丢弃该消息——前提是数据源能够重新投递它。
可重放与不可重放输入
可重放数据源能在崩溃后重新投递未确认的消息(通常因为它保留消息,并维护一个只在 ack 时才前进的偏移量(offset)/位置)。不可重放数据源在读取后立即丢弃消息,因此在读取之后、输出之前发生崩溃会丢失消息,除非消息先被持久化到本地。
| 输入 | 可重放? | 源侧确认 |
|---|---|---|
| Kafka | 是 | 仅在 ack 时 store_offset;enable.auto.offset.store=false |
| MQTT | 是(QoS ≥ 1) | 手动 ack(set_manual_acks(true));默认 QoS 1 |
| NATS (JetStream) | 是 | 仅在 ack 时 message.ack() |
| Pulsar | 是 | 仅在 ack 时 consumer.ack() |
| Redis (Streams) | 是(使用消费者组时) | 取决于配置 |
| SQL | 部分(增量查询) | 通过增量游标 |
| HTTP(服务端) | 否(webhook) | 无——即发即忘 |
| HTTP(客户端轮询) | 否 | 无 |
| File | 否(一次性读取流) | 无 |
| Generate | 否 | 无 |
| Modbus | 否 | 无 |
| Memory | 否 | 无 |
| WebSocket | 否 | 无 |
这对持久化意味着什么
- 可重放数据源在设计上是崩溃安全的:数据源只在 ack 时提交/确认,因此重启后会重新投递任何从未被确认的在途消息,无需本地持久化。例如 Kafka 输入设置
enable.auto.offset.store=false,使偏移量只在ack()内部前进。 - 不可重放数据源需要在输入边界有持久的 WAL(预写日志),使消息体在崩溃后幸存并可在重启时重放。通过每条流上的
durability:选项启用(参见 WAL 持久化与性能)。
至少一次契约
ArkFlow 提供至少一次投递。崩溃恢复后,在途消息可能被投递多次。输出必须容忍重复(例如 UPSERT 风格的 SQL/InfluxDB sink 天然幂等(idempotent);HTTP 与 Kafka 输出应预期可能的重复)。
精确一次投递可作为可选项启用于事务性(transactional)sink。配置了 exactly_once: true 的 Kafka 输出会把每个已确认的批次作为单个 Kafka 事务提交,因此 read_committed 下游消费者会原子地观察到它。这消除了事务内的部分写入与僵尸生产者重复;提交后边界处残留的重复仍需下游幂等来吸收。完整契约及其真实的边界参见精确一次投递。
取消安全
引擎可能放弃一个尚未完成的 Input::read——当 idle tick、barrier 或关停需要源链推进时,在途的 read future 会被丢弃并稍后重新发起。对引擎自带的通道型输入(MQTT、Pulsar、NATS、Redis、WebSocket、HTTP)而言,这种放弃不会丢失消息:每条投递只有在 codec 解码完成、ack 句柄就位之后才会进入输入的内部通道,而 read() 只在该单一认领点上等待。如果一次 read 在途中被丢弃,下一次 read() 会恰好一次地返回该投递。解码失败经由同一通道传递,以 read() 错误的形式进入引擎既有的失败路径。
简言之:取消一次 read 既不会丢失已解码的消息,也不会重复投递它——重复仍只来自上文的至少一次契约。
单节点边界
持久的 WAL 位于运行该流的节点的本地磁盘上。它防护的是进程崩溃(kill、panic、进程断电),而不是节点本身的丢失(磁盘故障、主机终止)。跨节点的高可用(复制 WAL / 共识)不在范围内。要获得 HA 级持久性,请在复制存储上运行,或在流前面放一个持久且可重放的数据源(如 Kafka)。
S3 WAL manifest 写入协调
当持久 WAL 使用对象存储(S3 兼容)后端且 parallel_put.workers > 1 时,多个 PUT 工作者可能并发地完成同一个段的上传并封存它。每次封存都会重写记录 ack 游标与已封存段列表的小 manifest.json。为防止这些并发重写互相静默覆盖——那会导致游标回退并在恢复时造成重复重放——manifest 写入通过基于 ETag 的乐观并发来协调:每次 PUT 携带它读取到的 ETag,不匹配时写入者重新读取、重新应用自己的变更并重试(最多 8 次,足以覆盖配置的工作者上限)。
这一协调要求对象存储后端支持条件 PUT(If-Match / if-none-exists)。S3 与 S3 兼容存储(以及内存测试后端)都支持。不支持条件 PUT 的后端会退化为无条件写入,这只在单写入者时安全——因此对这类后端不得配置 parallel_put.workers > 1。
对单写入者配置(parallel_put.workers = 1,默认值)而言这一切不可见:没有争用,第一次 PUT 即成功,不会发生重试。至少一次契约不变。
WAL 恢复失败(快速失败)
如果启用持久化的流在启动时无法恢复其 WAL——要么 WAL 文件不可读,要么把挂起条目重放进下游缓冲/通道失败——该流会启动失败,而不是带着可能丢失的数据静默继续(快速失败,fail-fast)。
实践中:
- redb 2.6.3 会在
Database::open时验证整棵 B-tree,因此大多数 WAL 损坏表现为StreamConfig::build返回Err——流根本不会进入运行状态。 - 如果 WAL 正常打开,但挂起条目无法转发(例如配置的缓冲拒绝写入),
Stream::run会返回Err且不读取新输入。
运维应监控流启动失败。恢复需要人工介入:要么删除配置的 durability.path 处的 WAL 文件(接受挂起条目丢失),要么从备份恢复。可以依赖 k8s/systemd 重启策略持续重试,但持续的 WAL 损坏不会自愈。
旧行为——记录失败并继续读取新输入、未确认条目实际上被丢弃——是对至少一次的静默违反,现已不复存在。