write() 返回值不校验导致排空「假成功」：报告 events_drained=1200 全绿、退出码 0，下游只落盘 128 字节、1198 条事件静默丢失
日期：2026-10-03 ｜ 分支：
2026-10-03-0000｜ 第99题 技术栈：Python 3.11 / Docker Compose / 事件排空（drain）流水线 /write()部分写 / 下游接收缓冲 / 背压（backpressure）/ NDJSON 分帧 / 契约配置自洽校验 / 健康检查与退出码
关键词：write() 返回值不校验、部分写（partial write）、半截记录（truncated record）、
排空假成功、events_drained=1200 但下游只收到 128 字节、events_lost=0 撒谎、
退出码恒 0、看板全绿但数据丢失、硬编码 65536/8400、下游接收上限 SINK_WRITE_CAP_BYTES、
契约未采纳、DRAIN_BATCH_EVENTS、按字节切批 vs 按条数切批、NDJSON 静默丢数据、
环境变量被配置文件压掉、负向验证没牙。
一、问题背景
一条「事件总线 → 下游存储」的排空链路：
- 生成器每轮产生 1200 条事件，按 NDJSON（一行一条）写进缓冲
state/buffer/round-<id>.buf。 - 排空器（drainer）把缓冲里的事件按周期分组，逐批
write()给下游。 - 下游（sink）是部分写语义：每次最多接收
SINK_WRITE_CAP_BYTES字节， 超出的部分直接丢弃，并把实收字节数作为write()的返回值。
现象非常反直觉 —— 所有自述指标全绿：
[generator] 本轮事件 1200 条 -> state/buffer/round-....buf
[supervisor] 报告 = {"periods": [1779999960000], "events_drained": 1200,
"events_lost": 0, "ok": true} ← 自述：全都排空了
[supervisor] round OK
[supervisor] 下游已收 = 66213 字节 ← 自述：按"应当写入"的字节数算
exit code = 0 ← 退出码也是 0
但磁盘取证打脸：
缓冲事件数 = 1200
下游落盘字节 = 128 ← 事实
消费者完整记录 = 2
消费者半截记录 = 2 ← 还有读一半就断的
落盘尾 64 字节 = b'{"body":"","seq":1187,...}\n{"body'
1198 条事件凭空消失，而且没有任何告警。
二、复现步骤
git clone <repo> && cd <repo>
git checkout 2026-10-03-0000
cd scenario/2026-10-03-0000
docker compose up -d --build
docker compose exec -T drain bash ops/verify.sh
问题态输出：
== A. 下游不能出现半截记录 ==
[FAIL] 下游半截记录数 (期望=0 实际=2)
== B. 事件不能丢：下游可见完整事件数 == 本轮生成数 ==
[FAIL] 下游可见完整事件数 (期望=1200 实际=2)
PASS=4 FAIL=2
RESULT: BAD
最小验证「write() 到底收下了多少」：
docker compose exec -T drain bash -c 'cd /srv && python3 -c "
import sys;sys.path.insert(0,\"app\")
import sink
s=sink.Sink()
print(\"下游单次接收上限 =\", s.cap_bytes, \"字节\")
n=s.write(b\"A\"*8400)
print(\"write(8400 字节) 返回 =\", n, \"-> 丢弃\", 8400-n, \"字节\")
"'
下游单次接收上限 = 64 字节
write(8400 字节) 返回 = 64 -> 丢弃 8336 字节
三、根本原因
五个缺陷叠加，核心是第一个。
1（核心）write() 的返回值被直接扔掉
def _emit(report, out, period, batch, nbytes):
payload = b"".join(raw for _ev, raw in batch)
out.write(payload) # ← 返回值没接！
report["events_drained"] += len(batch) # ← 按"我认为发出去的"计数
「我调用过 write()」和「下游收下了」被当成了同一件事。
下游 write() 返回 64，排空器却记 1200 条全部排空 —— 报告从此与事实无关。
这是 write() 部分写（partial write）语义的经典误用。
POSIX 的 write() 本来就不保证写完全部字节；这里下游还主动丢超限部分。
2 批量上限硬编码，从不参考下游能力
DEFAULT_CHUNK = 8400 # 每批事件数
DEFAULT_BATCH_BYTES = 65536 # 单批字节数
下游每次只接 64 字节，而第一批就按 66 KB 发 —— 第一批就注定了 99.9% 丢失。
3 契约存在但从未被采纳
conf/sink.env 明确登记了下游能力：
SINK_WRITE_CAP_BYTES=64 # 下游单次接收上限
SINK_READ_ATOMIC_BYTES=64
但 conf.load_drain_params() 只读 gateway.env，从不读 sink.env；
而 gateway.env 里连 DRAIN_BATCH_EVENTS 都没有。配置写了没人看。
4 退出码恒 0 —— 故障无声
print("[drain] round OK")
return 0 # 不管上面发生了什么
supervisor 的判据 ok = (events_drained == 1200 and events_lost == 0 and not stopped)
全部依赖排空器自己的计数器 —— 一个自述字段自己证明自己，看板必然全绿。
5 消费端把「窗口边界」当成「记录边界」
for w in windows:
for line in w.split(b"\n"): # ← 窗口边界劈开的记录就"读不完整"
读取端一次拉 64 字节，却把每个窗口单独 split。
窗口边界落在记录中间时那条记录就成了「半截记录」。
这是读取端自身的缺陷，不是下游没收到，但它让取证结果非常难解读。
四、解决办法
1 校验 write() 返回值，不一致就停发并大声失败
def _emit(report, out, period, batch, nbytes):
payload = b"".join(raw for _ev, raw in batch)
n = out.write(payload)
if n != len(payload):
# 「我调用过 write()」≠「下游收下了」
report["events_lost"] += len(batch)
report["batches"].append({
"period": period, "events": 0, "bytes": len(payload),
"status": "backpressure",
"error": "下游只接收 %d/%d 字节" % (n, len(payload)),
})
return True # 通知调用方：立刻停发
report["events_drained"] += len(batch)
report["batches"].append({
"period": period, "events": len(batch), "bytes": n, "status": "ok",
})
return False
2 批量上限自适应收敛到下游能力，绝不硬编码
biggest = max((len(raw) for _ev, raw in events), default=0)
# 契约自洽闸门：单条事件必须装得进下游一次接收窗口
if biggest > out.cap_bytes:
raise ContractError(
"契约不自洽：单条事件 %d 字节 > 下游单次接收上限 %d 字节，任何切批都会丢数据"
% (biggest, out.cap_bytes))
# 批量上限 = min(契约值, 下游一次接收窗口能装下的条数)
per_batch = max(1, out.cap_bytes // biggest)
if chunk > per_batch:
chunk = per_batch
report["batch_events_clamped"] = chunk
切批改成按字节收敛（条数只是上限）：
for ev, raw in group:
if cur and (cur_bytes + len(raw) > out.cap_bytes or len(cur) >= chunk):
stopped = _emit(report, out, period, cur, cur_bytes)
if stopped:
break
cur, cur_bytes = [], 0
cur.append((ev, raw))
cur_bytes += len(raw)
3 补齐契约并让环境变量优先
conf/gateway.env 补上：
DRAIN_BATCH_EVENTS=8400
app/conf.py 让环境变量能覆盖文件，否则故障没法用 env 注入复现：
ENV_OVERRIDABLE = ("DRAIN_BATCH_EVENTS", "DRAIN_MAX_EVENTS", "DRAIN_POLL_MS",
"SINK_WRITE_CAP_BYTES", "SINK_READ_ATOMIC_BYTES")
def load_drain_params():
params = _parse_env(os.path.join(CONF_DIR, "gateway.env"))
params.update(_parse_env(os.path.join(CONF_DIR, "sink.env")))
for k in ENV_OVERRIDABLE:
if os.environ.get(k):
params[k] = os.environ[k]
return params
4 退出码反映真实结果
if rep.get("stopped") or rep["events_lost"]:
print("[drain] FATAL: 排空未完成，下游背压", file=sys.stderr)
return 4
print("[drain] round OK")
return 0
5 读取端跨窗口拼接
rest = b""
out, bad = [], 0
for w in windows:
rest += w
while b"\n" in rest:
line, rest = rest.split(b"\n", 1)
line = line.strip()
if not line:
continue
try:
out.append(json.loads(line))
except Exception:
bad += 1
if len(rest) > MAX_LINE_BYTES: # 4096：超过就判定"永远读不完整"
bad += 1
rest = b""
if rest.strip():
bad += 1
五、验证
正向：验收 6/6
docker compose exec -T drain bash ops/verify.sh
[PASS] 下游半截记录数 (期望=0 实际=0)
[PASS] 下游可见完整事件数 (期望=1200 实际=1200)
[PASS] 报告 events_lost (期望=0 实际=0)
[PASS] health rc（本轮已排空） (期望=0 实际=0)
[PASS] 契约自洽（单条事件 <= 下游单次接收上限） (期望=1 实际=1)
[PASS] 外部探针 rc (期望=0 实际=0)
PASS=6 FAIL=0
RESULT: OK
磁盘取证对比（前 → 后）：
下游落盘字节 128 → 66213
消费者完整记录 2 → 1200
消费者半截记录 2 → 0
负向 A：批量超下游上限 → 必须停发 + 非 0 退出
cd scenario/2026-10-03-0000
cat > app/_neg.py <<'PY'
import importlib, os, sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import drainer, supervisor
RealSink = importlib.import_module("sink").Sink
class BigCapSink:
def __init__(self, path=None, cap_bytes=None):
self.cap_bytes = 65536 # 谎报的大窗口
self.real = RealSink(path=path, cap_bytes=64)
def write(self, data):
return self.real.write(data) # 真实下游只收 64 字节
drainer.sinkmod.Sink = BigCapSink
print("[neg] supervisor rc =", supervisor.main())
PY
docker compose exec -T drain bash -c 'cd /srv && python3 app/_neg.py; echo "rc=$?"'
rm -f app/_neg.py
[supervisor] FATAL: 本轮未排空
[supervisor] 报告 = {"events_drained": 0, "events_lost": 1149, "ok": false}
[neg] supervisor rc = 2
关键：没有谎报成功，events_lost 如实记 1149，退出码非 0。
负向 B：契约自相矛盾 → 启动即 FATAL
docker compose exec -T drain bash -c 'cd /srv && SINK_WRITE_CAP_BYTES=16 python3 app/supervisor.py; echo "rc=$?"'
[supervisor] FATAL: 契约不自洽: 单条事件 57 字节 > 下游单次接收上限 16 字节，任何切批都会丢数据
rc=3
六、经验与避坑
1 「我调用过 write()」永远不等于「对方收下了」
任何 I/O 原语都要看返回值：write() / send() / fwrite() 都可能部分写，
read() 也可能短读。必须写循环补齐，或显式处理不足量：
while data:
n = write(fd, data)
if n <= 0:
raise IOError("下游拒绝接收")
data = data[n:]
如果对端是「超限即丢」的语义（本题的下游、某些消息中间件的 max.message.bytes），
必须把返回值与输入长度比对，不一致就立刻停发并如实记录丢失量。
绝不能 report["sent"] += len(batch) 就完事。
2 自述字段不能自己证明自己
events_drained、ok: true、round OK 都是排空器自己的计数器。
判据必须来自可外部独立复核的事实：下游落盘字节数、下游可见的完整记录数、
独立消费者探针。
一个实用判据：让外部探针（消费者视角）成为唯一的裁判。
本题的 client/consumer_probe.py 就是这么设计的 —— 它不看排空器的报告，只看下游落盘。
3 退出码恒 0 是「静默故障」的温床
如果 main() 无论发生什么都 return 0，那么 CI、看板、告警全部失灵。
退出码必须区分故障类别（本题：2=未排空、3=契约不自洽、4=背压中断），
让上游能据此分流处理。
4 批量/缓冲区大小必须来自契约，且要与对端能力自洽
硬编码的 65536 / 8400 看着像「保守的大值」，实则是自相矛盾的配置。
正确做法是启动时校验：
单条记录字节数 <= 对端单次接收上限 # 否则任何切批都丢数据
每批字节数 <= 对端单次接收上限 # 按字节切，不是按条数切
按条数切批是错的。记录定长时两者等价，一旦变长（本例 body 可 padding），
条数相同字节数可以差几十倍。按字节切批 + 条数作为上限，才是安全写法。
5 负向验证前，先确认注入通道真的通了（本题最大的坑）
我第一版修复后，跑「批量超上限」的负向验证，结果全绿、rc=0， 看上去像「闸门没牙」。实际原因是：
os.environ 里 DRAIN_BATCH_EVENTS = 8400 ← 我注入的
conf 读到的 DRAIN_BATCH_EVENTS = 1 ← conf/gateway.env 覆盖了它
因为 conf.py 用 params.update(读到的文件)，完全不看 os.environ。
我的 env 压根没到达被测代码。
排查手法：同时打印「注入值」与「被测代码实际读到的值」：
docker compose exec -T drain bash -c 'cd /srv && DRAIN_BATCH_EVENTS=8400 python3 -c "
import os,sys;sys.path.insert(0,\"app\");import conf
print(\"os.environ =\", os.environ.get(\"DRAIN_BATCH_EVENTS\"))
print(\"conf 读到 =\", conf.load_drain_params().get(\"DRAIN_BATCH_EVENTS\"))
"'
两者不一致 → 是注入通道问题，不是闸门问题。
通用教训：负向验证失败时，先质疑「我的注入生效了吗」，
再怀疑「闸门坏了」。配置加载器不认环境变量，是容器化部署里极常见的坑
（尤其 ENTRYPOINT 里 source 了文件之后）。
6 别把「读取端窗口」当成「记录边界」
分帧（framing）类 bug 有个典型伪装：数据其实完整落盘了，是读取端自己读断了。
本题的 downstream.py 就是这样 —— 每个 64 字节窗口单独 split(b"\n")，
窗口边界劈开记录就报「半截记录」。
定位方法：先看原始字节，再看解析结果。
docker compose exec -T drain bash -c 'cd /srv && python3 -c "
d=open(\"state/sink/downstream.log\",\"rb\").read()
print(\"字节 =\", len(d), \" 换行 =\", d.count(b\"\\n\"))
print(repr(d[:140]))
"'
字节数和换行数都对、能正常 json.loads 逐行读 → 是读取端的问题；
字节数就少了 → 才是写入端的问题。
7 参考答案也可能跑不过验收 —— 不要盲信 expected-fixed/
本题仓库自带的 ops/expected-fixed/ 实测 PASS=3 FAIL=3 / RESULT: BAD：
supervisor rc=2
[supervisor] 报告 = {"events_drained": 0, "events_lost": 1200, "ok": false}
[FAIL] 下游可见完整事件数 (期望=1200 实际=1)
PASS=3 FAIL=3
原因：它 group[idx:idx+chunk] 只按条数切批，不参考 out.cap_bytes；
一旦 DRAIN_BATCH_EVENTS 缺省成 8400，第一批 1200 条 ≈ 66 KB 直接超限，
闸门正确报警（退出码 2）—— 但这不符合题目「修好后应 RESULT: OK」的要求。
它把「契约完整」当成了前提：只有手工补上 DRAIN_BATCH_EVENTS=1 才可能过。
教训：expected-fixed/ 是「作者当时的解法」，不是「唯一正确解」。
验收脚本 ops/verify.sh 才是最终裁判。我加了「按下游能力自适应收敛」这道闸门，
所以 DRAIN_BATCH_EVENTS=8400 也能自动收敛到 1，契约缺省/写错都不会丢数据 —— 更健壮。
8 ops/problem-state/ 与 ops/expected-fixed/ 可能被搞混
进仓库时 app/*.py 与 ops/problem-state/*.py 的 md5 完全一致，
但 ops/problem-state/ 里 conf.py、health.py、events.py、sink.py、supervisor.py
其实是修复版内容（与 expected-fixed/ 完全相同），只有 drainer.py、downstream.py 是问题态。
规避：别信目录名，直接看特征值 ——
grep -c "DEFAULT_BATCH_BYTES = 65536" app/drainer.py，
1 = 问题态，0 = 修复态。
9 rootless Docker：/tmp 下的 bind mount 会静默变成空目录
最终验收时把仓库 clone 到 /tmp/final-check 再 docker compose up，
容器里 /srv/ops/、/srv/app/ 全是空的，bash ops/verify.sh 报
bash: ops/verify.sh: No such file or directory。
docker inspect 看挂载声明完全正常，但容器内：
docker compose exec -T drain bash -c 'mount | grep /srv/ops'
# overlay on /srv/ops type overlay (...) ← 不是 bind，是个空 overlay 目录
原因：rootless Docker 的 shell /tmp 与 daemon 的 /tmp 不是同一个，
bind mount /tmp/... 时 daemon 找不到源路径，不报错、直接创建空目录顶上。
规避：复现目录放工作区（$PWD、/workspace）下，别放 /tmp。
注意本机 df -h /tmp / 会显示两者是同一个挂载点，看起来完全正常 —— 别被它骗了。
10 运维小贴士
docker compose exec在非 TTY 环境（CI）要加-T，否则报the input device is not a TTY。ops/verify.sh在RESULT: BAD时exit 1，日志可能被 CI 截断； 取证时用| tail -n或单独跑子项，别让退出码吃掉输出。- 调试
drainer时注意supervisor.reset_state()会rmtree(DATA_DIR)， 先手工造缓冲再跑supervisor.py，缓冲会被清掉。想单独调试就直接调drainer.drain()。
一句话总结：
write() 的返回值就是「下游到底收下了多少」的唯一事实来源；
不看它，报告里的 events_drained 和退出码 0 就只是自我感觉良好，
真正的事实只能去下游落盘里数。