ZhenxiJS Logo
朕喜技术（ZhenxiJS）
2026.10.05
1 分钟阅读

变更通知服务「消息一条不少」，接收端却一条都没确认，未确认量显示 0 —— 中继把「对端没 ACK」当成了「已投递完成」（at-least-once 投递 / unacked 未确认量记账 / 限流闸门只记不拦 / 有界队列容量申请值 vs 实际生效值）

日期：2026-10-05 | 分支：2026-10-05-0000 | 场景：scenario/2026-10-05-0000/ 技术栈 / 关键词：Python 3.11、Docker Compose、at-least-once 投递、ACK 语义、unacked / in-flight / pending 未确认量、多租户限流闸门（rate limit / quota）、有界队列（bounded queue / byte-capacity queue）、write() 返回值不校验、SO_SNDBUF 申请值 vs getsockopt 实际生效值、自述字段自证、退出码恒 0、契约形同虚设、外部独立探针

一、问题背景

一条变更通知链路：producer 按序推送 → relay 中继逐条校验后写入BoundedQueue（按字节计容的有界队列）→ sink 接收端只有它回 ACK 才算「接住了」。

链路契约是 conf/notifyd.env（唯一权威来源）：

NOTIFYD_TENANT_UNACKED_LIMIT=8            # 单租户未确认条目数上限
NOTIFYD_QUEUE_MAX_BYTES=16384             # 队列容量：运维「申请值」
NOTIFYD_QUEUE_SIZE_BYTES=262144           # 队列容量：「实际生效值」
NOTIFYD_REQUIRED_FIELD_PREFIXES=evt_,tenant_,seq_   # 字段清单（全部必须存在）
NOTIFYD_REQUIRED_CORE_VERSION=2.0         # 权威生产方的核心版本

现象（四个绿灯没有一件指向真相）：

  • 中继每轮都「跑完了」；
  • self_consistent: true（自述）；
  • python3 -m app.health 退出码 0（只证明进程正常退出）；
  • 报的 acked == pushed（拿自己的计数器证明自己）。

而真实情况是接收端一条都没接住，未确认量却显示 0：

$ bash ops/repro.sh
    推送条数                     = 3
    中继自述「已投递」           = 3        ← 说投出去了
    sink 真实 ACK 数             = 0        ← 一条都没接住
    版本不符被拒条数             = 3
    未确认量 unacked_by_tenant   = {}       ← 未确认量竟然是空的
    闸门 overlimit_pushes        = 4（应恒 0）
    {"degraded": false, "self_consistent": true, "acked": 12, "pushed": 12}
  探针判定：BAD —— 中继把「没被 ACK」的事件当成已投递完成

问题态验收基线：PASS=11 FAIL=7 / RESULT: BAD。

二、复现步骤

git fetch origin && git checkout 2026-10-05-0000
cd scenario/2026-10-05-0000

# 方式 A：docker（推荐）
docker compose up -d --build
docker compose exec -T notifyd bash ops/repro.sh      # 一键复现 + 外部探针取证
docker compose exec -T notifyd bash ops/verify.sh     # 问题态 PASS=11 FAIL=7 / BAD
docker compose exec -T notifyd bash               # 进去动手改

# 方式 B：本机直接跑（只需 python3）
bash ops/repro.sh
bash ops/verify.sh

app/、ops/、conf/、client/ 都是 bind mount，改完立刻生效、无需 rebuild。 docker compose exec 在非交互环境必须加 -T，否则报 the input device is not a TTY。

三、根本原因

一句话：「调用过」被当成了「对端接住了」。 四个 bug 全在 app/notifyd.py 与 app/health.py。

  1. ACK 结果被丢掉（题眼）：push_event() 里已经算出 ok = self.sink.accept(ev)，但紧跟着写 self.stats["pushes_delivered"] += 1 和 self._record_ack(tenant, item, True) —— 函数返回值被就地扔掉、写死成 True。sink 明明返回 False，中继照样记「已投递」。
  2. 未确认量不按 ACK 记账：_record_ack() 里 self.stats["unacked"] = 0 if ok else 1 —— 这不是「当前在途条数」， 只是个 0/1 标志位，所以不管积压多少，未确认量报出来恒为 0 或 1。
  3. 限流闸门只记不拦：used >= self.unacked_limit 时只 overlimit_pushes += 1 记一笔，没有 return，事件照样推进链路。 「记录了越限」≠「拦住了越限」，这是没有闸门的闸门。
  4. 队列容量口径取错 + 不核对 write() 返回值： capacity = self.queue_source_bytes（运维申请值 16384）， 而契约登记的实际生效值是 NOTIFYD_QUEUE_SIZE_BYTES=262144， 于是每推一条 queue_capacity_mismatch 加 1（共 12 次）。 同时 self.queue.write(blob) 的返回值被直接丢弃， queue_written_bytes += len(blob) 等于「打算写多少就记多少」—— 装不下也不记账，和 write() 部分写语义完全脱节。
  5. 自检是自述、退出码恒 0：self_consistent() 拿 acked == pushes_delivered 自证，而两者都是同一个函数自己写的计数器，必然相等。 health.py 里 BAD_FIELDS = ()（判据表是空的 → 永远报不出问题）、 degraded: False、self_consistent: True 全是手写常量， main() 无论发生什么都 return EXIT_OK。

为什么难查：accept() 的返回值就在两行之内被 _ 掉，不抛异常、不报错、 日志里也没有任何 ERROR —— 属于「静默地把失败洗成成功」类故障。

四、解决办法

只改 app/notifyd.py 与 app/health.py（conf/ 三份契约、client/notifyd_probe.py、ops/*.sh 一字未动）。修好后与 ops/expected-fixed/ 的 notifyd.py 逐字节一致。

4.1 push_event() 的正确顺序

# ① 闸门：到上限直接拒收并早退，不再往下走
used = self._tenant_unacked(tenant)
if used >= self.unacked_limit:
    self.stats["overlimit_rejected"] += 1
    self.stats["errors"].append(
        "tenant %s 未确认量 %d 已达上限 %d，拒绝继续推送（合规拒收）"
        % (tenant, used, self.unacked_limit))
    return {..., "acked": False, "ok": False, "error": "tenant-overlimit"}

# ② 容量口径取登记表的实际生效值，并与队列实况对账
capacity = self.queue_actual_bytes
if self.queue.capacity_bytes != capacity:
    self.stats["queue_capacity_mismatch"] += 1
    self.stats["errors"].append("队列容量 %d 与登记表实际生效值 %d 不一致" % ...)

# ③ 先按契约校验：不合格的事件根本不许入队
ok = self.sink.accept(ev)
blob = json.dumps(item, ensure_ascii=False).encode("utf-8")
self.stats["queue_offered_bytes"] += len(blob)

if not ok:
    reason = self.sink.rejected[-1][1] if self.sink.rejected else "?"
    if reason == "core-version-mismatch":
        self.stats["version_mismatch_acked"] += 1
    elif reason == "required-field-missing":
        self.stats["field_missing_acked"] += 1
    self._record_ack(tenant, item, False)      # ok 一路传下去
    return {..., "acked": False, "ok": False, "error": reason}

# ④ write() 的返回值就是「队列真的装下了多少」
n = self.queue.write(blob)
if n != len(blob):
    self.stats["queue_dropped_bytes"] += len(blob) - n
    self.stats["queue_dropped_entries"] += 1
    return {..., "ok": False, "error": "queue-full"}
self.stats["queue_written_bytes"] += n

# ⑤ 真接住且真落地，才算投递完成
self.stats["pushes_delivered"] += 1
self._record_ack(tenant, item, True)

4.2 未确认量改为逐条事实

def _record_ack(self, tenant, item, ok: bool) -> None:
    if ok:
        self.last_acked_seq[tenant] = item.get("seq_no")
        self.stats["acked"] += 1
    # 逐条事实：ack 一条才少一条，不许按序号差「凑」出来
    self.stats["unacked"] = self._tenant_unacked(tenant)
    if self.stats["unacked"] > self.stats["unacked_peak"]:
        self.stats["unacked_peak"] = self.stats["unacked"]

4.3 自检算出来（不是常量）

def self_consistent(self) -> bool:
    return (self.stats["acked"] == self.stats["pushes_delivered"]
            and self.stats["queue_dropped_entries"] == 0
            and self.stats["overlimit_pushes"] == 0
            and self.stats["queue_capacity_mismatch"] == 0
            and not self._real_errors())

4.4 health.py：判据表 + 退出码要有牙

BAD_FIELDS = ("queue_dropped_entries", "queue_dropped_bytes",
              "queue_capacity_mismatch", "version_mismatch_acked",
              "field_missing_acked")
  • 逐租户未确认量 > NOTIFYD_TENANT_UNACKED_LIMIT → 记问题
  • overlimit_pushes > 0 → 记问题（合规拒收记在 overlimit_rejected，不算问题）
  • acked != pushes_delivered、sink_received != acked → 记问题
  • errors 里排除含「合规拒收」的合规项
  • 契约不可用 / 自相矛盾 → [FATAL] + 退出码 3
  • report.json 不存在 → 退出码 2（判不了 ≠ 没问题）
  • 有问题 → EXIT_BAD = 2，全绿才 EXIT_OK = 0

4.5 顺带加固 build_service()

契约不可用抛 ContractError；queue_actual_bytes <= 0、 queue_source_bytes > queue_actual_bytes、字段清单为空都大声失败。

五、验证

docker compose exec -T notifyd bash ops/verify.sh      # PASS=18 FAIL=0 / RESULT: OK
docker compose exec -T notifyd bash ops/repro.sh       # 探针判定：OK
docker compose exec -T notifyd bash ops/negative.sh    # 6 组负向验证

verify.sh 修复后：

== B. 版本与字段清单：不合格的事件根本不许进队列 ==
  [PASS] 进了队列却违反版本/字段契约的条数 (期望=0 实际=0)
== C. 限流闸门：未确认量到上限必须拒绝继续推送 ==
  [PASS] 越过上限仍在推送的次数 (期望=0 实际=0)
== D. 队列容量口径：必须取登记表的实际生效值 ==
  [PASS] 容量口径与登记表不一致的次数 (期望=0 实际=0)
== E. 自述字段不得自证；退出码要有牙 ==
  [PASS] 服务自检结果(True=通过) (期望=True 实际=True)
  [PASS] 健康检查在一切正常时退出 0 (期望=0 实际=0)
== F. 外部独立探针 ==
  RESULT: OK
PASS=18 FAIL=0
RESULT: OK

repro.sh 修复后：场景 1 中继自述「已投递」= 0（与 sink ACK 数 0 一致）、 场景 3 overlimit_pushes = 0（超限走 overlimit_rejected）、 探针「队列里真实躺着的记录条数 = 11」（不再是 20）、末尾 RESULT: OK。 python3 -m app.health 输出 {"degraded": false, "issues": [], "self_consistent": true, ...}、rc=0。

负向验证 6 组（证明闸门真的有牙，不是「恰好绿」）：

注入结果
A 去掉「先校验后入队」早退探针 RESULT: BAD / PASS=16 FAIL=2
B 容量口径改回运维申请值queue_capacity_mismatch=8 / BAD / PASS=15 FAIL=3
C 不核对 write() 返回值丢弃 6 条却零记账 / PASS=17 FAIL=1
D 超限只记不拦overlimit_pushes=4 / health 退出码 2 / PASS=15 FAIL=3
E 契约自相矛盾（申请 > 实际）[FATAL] 契约自相矛盾 / 退出码 3
F 删掉 conf/topology.env[FATAL] 契约不可用 / 退出码 3
恢复并复验PASS=18 FAIL=0 / RESULT: OK

六、经验与避坑

踩坑记录

  1. ops/negative.sh 是按参考答案的源码文本做「注入点」替换的（assert old in s, "负向 X 注入点未找到"）。 第一版修复我只按自己的理解改逻辑，注释措辞与参考答案不同， 负向 A 直接以 AssertionError: 负向 A 注入点未找到 崩掉。 → 对齐注入点所在语句后 6 组负向全部通过。 启示：靠文本替换做的负向验证里，注释与代码结构也是契约的一部分。 好在逻辑判断本身仍有差异（我把 overlimit_pushes 无条件计入问题， 参考答案要求 overlimit_rejected == 0 时才计入），我的版本更严格，验收依然全绿。
  2. docker compose exec 非交互必须加 -T，否则 the input device is not a TTY。
  3. state/ 是 bind mount 的运行期产物且被 .gitignore 忽略。 verify.sh / negative.sh 会写入 report*.json、queue*.json、notifyd.fixed.bak*； negative.sh 结束会恢复 app/notifyd.py 与 conf/notifyd.env， 但改完一定要 git status 确认只有预期的两个文件改动。
  4. __pycache__ 会让旧字节码掩盖改动。切换版本后清一次 app/__pycache__。
  5. 「自述全绿」极具迷惑性。问题态 health 打印 {"degraded": false, "self_consistent": true, "acked": 12, "pushed": 12} 且 rc=0， 看上去一切正常。唯一可信的是外部探针 —— 它只读契约 + 队列里真实躺着的记录， 不读中继任何自述字段。

核心认知

  • 「调用过」≠「对端接住了」。accept() / write() / send() / connect() 的 返回值就是唯一的事实来源，ok = f(x) 之后每一处分支都必须用它。 把返回值丢掉（或写死 True）是把失败静默洗成成功的头号写法。
  • 未确认量（unacked / in-flight / pending）必须是逐条事实， 来自「发了但没收到 ACK 的那些条目」；用「末端序号 − 已 ACK 序号」或 0 if ok else 1 这类收口方式去「凑」，在乱序 / 部分失败时会直接算错。 关键校验：没有 ACK 时它会不会减少？会减少就是错的。
  • 「记录了越限」不等于「拦住了越限」。闸门少一个 return 就等于没有闸门。
  • 容量配置要认准刻度：SO_SNDBUF 申请值 vs getsockopt 实际生效值、 NOTIFYD_QUEUE_MAX_BYTES（申请）vs NOTIFYD_QUEUE_SIZE_BYTES（实际生效）、 Kafka max.message.bytes、Redis client-output-buffer-limit —— 申请值与实际生效值是两个数字，读错一个就永远对不上账。
  • 有界队列的 write() 是部分写语义（和内核 socket 缓冲区、环形队列一样）， 装不下就整条丢、返回 0，应用层拿不到异常。必须比对返回值并如实记 queue_dropped_entries，不许当写成功了。
  • 自述字段不能自证。self_consistent = True 这种手写常量等于没有自检； 判据必须落在外部按契约能自己算一遍的字段上，有问题必须非 0 退出。 退出码恒 0 是静默故障的温床。
  • 「判不了」必须与「没问题」严格区分（fail-closed）：契约不可用、 自相矛盾、报告读不到 —— 一律大声失败（退出码 3 / 2），不许退化成「没问题」。
  • 如何自查就三问：
    1. 投递代码里「调用过」和「对端接住了」是同一条判据吗？ accept() / write() / send() 的返回值有没有被丢掉？
    2. 「未确认量 / in-flight / pending」是从逐条事实算的， 还是用序号差、计数器加减「凑」的？没有 ACK 时会减少吗？
    3. 容量配置取的是申请值还是实际生效值？装不下时是「如实记账」还是「当成功」？

同类影响面

  • 消息队列的 at-least-once 投递与 unacked / in-flight 计数： RabbitMQ unacked、Kafka in-flight requests、 SQS ApproximateNumberOfMessagesNotVisible、NATS JetStream AckPending。
  • HTTP 客户端只看「请求发出去」不看响应码；gRPC 只看 send 不看 status。
  • 多租户限流 / 配额闸门（只统计不拦截 = 没有闸门）。
  • 有界队列与缓冲区容量配置（SO_SNDBUF 申请值 vs 实际值、 Kafka max.message.bytes、Redis client-output-buffer-limit）。
  • 一切「自述字段自证」的监控面板，与退出码恒 0 的巡检脚本。

版权声明

本文为博主原创文章，遵循 CC BY-NC-SA 4.0 许可协议。 转载请注明出处：http://localhost/blogs/2026-10-05-0000-ack-at-least-once-vs

分享这篇文章

上一篇
AI日报 2026-10-06
下一篇
清单服务「超时停止」失效：timeout --signal=KILL 参数里引号不闭合，SIGKILL 发不到任何进程