安全采集、回放与线上到离线闭环：练习答案

参考实现

完整实现：

def clean(event):
    if event.get("code") not in {"ok","timeout","denied"}:raise ValueError("code")
    ms=event.get("duration_ms")
    if type(ms) is not int or ms<0:raise ValueError("duration")
    out={"code":event["code"],"duration_ms":ms}
    if "retry_count" in event:
        n=event["retry_count"]
        if type(n) is not int or not 0<=n<=3:raise ValueError("retry")
        out["retry_count"]=n
    return out
assert clean({"code":"ok","duration_ms":1,"retry_count":2,"secret":"x"})=={"code":"ok","duration_ms":1,"retry_count":2}
try:clean({"code":"ok","duration_ms":1,"retry_count":True})
except ValueError:print("bool rejected")
else:raise AssertionError("bool accepted")

精确类型检查保护统计语义，避免布尔值混入次数分布。

返回正文

双超时回放答案

将正文实际采集块保存为 replay.py；导入会先跑正文成功对照，本答案再记录新的双失败轨迹。

from replay import task,scrub
import json
from tempfile import TemporaryDirectory
from pathlib import Path
calls=0;writes=[]
def unavailable(name,args):
    global calls
    calls+=1
    raise TimeoutError('never record this detail')
with TemporaryDirectory() as folder:
    path=Path(folder)/'failed.jsonl'
    def emit(e):
        with path.open('a') as f:f.write(json.dumps(scrub(e))+'\n')
    assert task(unavailable,emit) is None
    saved=[json.loads(x) for x in path.read_text().splitlines()]
    assert saved[-1]['status']=='failed' and calls==2
    errors=iter(e for e in saved if e['event']=='error')
    def replay(name,args):
        if name!='lookup':raise PermissionError('write forbidden')
        assert next(errors)['error']=='timeout'
        raise TimeoutError('recorded timeout')
    actual=[]
    assert task(replay,actual.append) is None and actual[-1]['status']=='failed'
    assert calls==2 and writes==[]
    assert [e['event'] for e in saved]==[e['event'] for e in actual]
print('both attempts fail in recording and replay without live or write calls')

回放只读取旧事件，因此两次 failed 证明控制流被复现；它没有请求修复后的上游，不能说明上游当前可用。真实可用性需要另一次有标记的只读集成检查。