流处理 Watermark 与迟到修正：窗口封口、重开和下游效果账本

订单事件按 event time 统计每小时收入。一个 Kafka 分区安静十分钟拖住全局 watermark，运维打开 idleness 后吞吐恢复；随后迟到退款到达，窗口已经输出，报表需要减掉金额，但“收入达标就发奖励”的下游动作不能再执行一遍。把 watermark 当成数据完整证明，或把每次窗口重算都当成新业务事件，都会制造静默差异。

Flink 的 watermark 表示事件时间进度，多个并行输入通常由较慢 watermark 限制；窗口允许配置 allowed lateness，并可处理或侧输出迟到元素。框架提供时间与状态原语，不知道一项业务结果能否撤销。事件驱动架构解释生产消费边界，后台任务队列处理重试；本文固定窗口何时初次封口、何时发布修订，以及下游怎样识别同一逻辑结果。

冻结 watermark 和窗口语义

streamTimePolicy.v1 至少绑定 source/partition 集合、timestamp extractor、时区、最大乱序、watermark generator、idleness 条件、window assigner、trigger、allowed lateness、晚到分类、state retention、作业 digest 和 sink correction mode。事件信封保存 event ID、event time、ingest time、source offset 与 schema generation；缺失或越界时间进入 quarantine，不能回退到处理时间后混入同一指标。

schema: streamTimePolicy.v1
job: hourly-revenue@14
timestampExtractor: order.occurred_at@2
watermark: bounded-out-of-orderness:5m
idlePartitionAfter: 15m
window: tumbling:1h:UTC
allowedLateness: 24h
correctionMode: UPSERT_REVISION

空闲标记必须带 partition generation。分区恢复时，第一条事件不能让 watermark 倒退；低于当前水位的事件按 lateness policy 处理。source 扩分区或作业升级会改变 watermark 拓扑，必须提升 policy generation。数据库 CDC提供 offset 和变更身份，但 schema、event time 与业务删除仍需本合同解释。

窗口输出是可修订对象

业务窗口身份必须跨策略稳定：logicalWindowId = H(businessStream, key, start, end) 不含 job 或 policy generation；代内投影再用 projectionId = H(logicalWindowId, policyGeneration) 区分。输出状态为 OPEN -> ON_TIME_EMITTED -> CORRECTED* -> FINALIZED。首次 watermark 越界只生成 revision 1；allowed lateness 内的新事件重算并发布更高 revision，绑定输入 offset range、aggregate digest 和前驱。唯一投影接受点是 sink 以 (projectionId, expectedPreviousRevision) CAS，canonical alias 另行绑定获批 policy，旧作业不能覆盖新代。

数据库报表可 upsert 当前 revision，同时保留 append-only correction ledger。一次性通知、积分或扣款使用 effectKey = H(effectType, logicalWindowId, businessSubject)，既不含 revision，也不含 policy generation；策略升级或历史回放因此仍指向同一业务效果。后续修正进入人工或补偿流程；允许增量调整时，每个 correction 携带 delta、前驱与独立幂等键。AI 用量计量账本展示迟到修正的账本思路。

不可逆 I/O 前先持久化 (effectKey, payloadDigest, PREPARED)，executor 以 CAS 匹配当前 canonical policy/revision 后推进 DISPATCHING，并把 effectKey 作为 provider 幂等键。响应丢失进入 UNKNOWN，只能用 provider receipt/query 对账；既不支持幂等也不能查询的 adapter 在 dispatch 后不得自动重试。Webhook 订阅正确性可交付修订事件，但 2xx 不证明下游已应用。

攻击“已经到齐”的假设

故障测试覆盖分区停顿后恢复、单机时钟跳变、同 event ID 不同 payload、source offset 回绕、watermark 参数升级后回放同一窗口、checkpoint 前后重复、provider 接受 effect 后响应丢失、迟到事件恰在 allowed-lateness 等号、side output 丢失，以及 revision 2 比 revision 1 先到 sink。断言 watermark 单调、每个输入只归属合同允许的窗口、投影 revision 不倒退、一次性 effect 跨 policy 不重复，UNKNOWN 不触发盲重试。

门禁可要求：生成式乱序轨迹与独立 oracle 一致率 100%；丢失或重复输入 offset 为 0；窗口当前值与从 append ledger 全量重算差异为 0；allowed-lateness 内修订 P99 发布低于 60 秒；超过保留期的迟到项 100% 有 dead-letter/人工去向；每个一次性 effect key 的 provider 效果最多一个。只报 watermark lag 不能证明正确。

上线先让新 policy 对历史 offset shadow 重算，不写 canonical sink；再双写 revision ledger 并比较一周，随后让 5% key 读取新投影。异常时停止新 effect dispatch，把 read alias 切回上一完整 generation；仍继续保存迟到原始事件，待修复后以新 policy 重放。回滚不能删除已发布修订，也不能把 FINALIZED 解释成“现实不再变化”。训练数据集快照适合冻结离线集合，流账本则保留随时间到达的修正事实。