"""Deterministic teaching models; no network, no production measurements.""" import argparse import math def slo(): total, bad, target = 12_000_000, 8_400, 0.999 allowed = round(total * (1 - target)) remaining = allowed - bad burn = (240 / 20_000) / (1 - target) assert allowed == 12_000 and remaining == 3_600 assert math.isclose(burn, 12) small_tenant_error = 0.99 * 0.0001 + 0.01 * 0.08 assert math.isclose(small_tenant_error, 0.000899) print(dict(allowed=allowed, remaining=remaining, burn=round(burn, 6), global_error=small_tenant_error)) def latency(): assert 20 + 30 + 180 + 30 + 40 == 300 remaining = 300 - 35 - 70 assert remaining == 195 print(dict(all_three=0.99 ** 3, all_hundred=0.99 ** 100, downstream_budget_ms=remaining)) def availability(): result = 0.9999 * 0.9995 * 0.9995 assert math.isclose(result, 0.998900349975) print(dict(independent_series=result, exposed_quotes=3000 * 0.4 * 3)) def gate_decision(bad, total, baseline_bad, baseline_total): if total <= 0 or baseline_total <= 0: return "INSUFFICIENT_EVIDENCE" if not (0 <= bad <= total and 0 <= baseline_bad <= baseline_total): raise ValueError("bad counts must be subsets of total counts") error, baseline = bad / total, baseline_bad / baseline_total return "PASS" if error <= 0.001 and error - baseline <= 0.002 else "STOP" def gate(): decision = gate_decision(50, 10_000, 80, 100_000) assert decision == "STOP" assert gate_decision(0, 0, 0, 100) == "INSUFFICIENT_EVIDENCE" assert gate_decision(0, 1000, 0, 1000) == "PASS" print(dict(decision=decision, absolute_error_difference=0.005 - 0.0008)) def decision(): """U69：场景输入缺项时输出区间与三态，不按节点数均分。""" rate, window_s, limit = 3_000, 3, 5_000 exposure = {share: int(rate * share * window_s) for share in (0.4, 0.7)} assert exposure == {0.4: 3_600, 0.7: 6_300} interval = (0, rate * window_s) verdict = { share: ("PASS" if value <= limit else "STOP") for share, value in exposure.items() } assert interval == (0, 9_000) and interval[1] > limit print(dict(exposure=exposure, unknown_share_interval=interval, unknown_share_verdict="INSUFFICIENT_EVIDENCE", verdicts=verdict)) def windows(): """U69：多窗口告警的最小事件数三态，低流量不按比例寻呼。""" target, min_events = 0.999, 1_000 tolerance = 1 - target def judge(total, bad): if total <= 0 or total < min_events: return "INSUFFICIENT_EVIDENCE" burn = (bad / total) / tolerance return "STOP" if burn >= 1 else "PASS" assert judge(20_000, 240) == "STOP" and abs((240 / 20_000) / tolerance - 12) < 1e-9 assert judge(200, 1) == "INSUFFICIENT_EVIDENCE" assert judge(0, 0) == "INSUFFICIENT_EVIDENCE" print(dict(burn_12=12.0, at_20k=judge(20_000, 240), at_200=judge(200, 1), at_zero=judge(0, 0), min_events=min_events)) def retry(): """U69：剩余预算传递与冗余放大，重试不能重新获得完整预算。 时刻与时长分开：t20 发出、150ms 窗口在 t170 超时；t170 是时刻， 剩余预算按时刻计算，不能把 170 当毫秒时长。""" budget_ms, reserve_ms, first_sent_ms, first_timeout_ms = 300, 30, 20, 150 first_fails_at = first_sent_ms + first_timeout_ms remaining = budget_ms - first_fails_at - reserve_ms amplification = {share: round(1 + share, 4) for share in (0.05, 0.60)} assert first_fails_at == 170 and remaining == 100 assert amplification == {0.05: 1.05, 0.6: 1.6} verdict = "OVER_BUDGET" if first_timeout_ms > remaining else "WITHIN_BUDGET" print(dict(remaining_ms=remaining, amplification=amplification, retry_same_150ms=verdict, wrong_if_timestamp_read_as_duration=budget_ms - first_fails_at, worst_three_layers=3 ** 3)) def durability(): """U69：五状态乘三类故障的确认点矩阵（分类结果，不注入真实故障）。""" states = ("BEFORE_COMMIT", "AFTER_COMMIT_BEFORE_RESPONSE", "BEFORE_REPLICA_RECEIVE", "BEFORE_REPLICA_APPLY", "AFTER_NOTIFY") faults = ("PROCESS_CRASH", "HOST_LOSS", "REGION_LOSS") # 每格：客户端是否已知、可用恢复动作。 matrix = { "BEFORE_COMMIT": {"PROCESS_CRASH": "SAFE_RETRY", "HOST_LOSS": "SAFE_RETRY", "REGION_LOSS": "SAFE_RETRY"}, "AFTER_COMMIT_BEFORE_RESPONSE": {"PROCESS_CRASH": "QUERY_FIRST", "HOST_LOSS": "QUERY_FIRST", "REGION_LOSS": "QUERY_FIRST"}, "BEFORE_REPLICA_RECEIVE": {"PROCESS_CRASH": "QUERY_FIRST", "HOST_LOSS": "QUERY_FIRST", "REGION_LOSS": "DUPLICATE_RISK"}, "BEFORE_REPLICA_APPLY": {"PROCESS_CRASH": "QUERY_FIRST", "HOST_LOSS": "DUPLICATE_RISK", "REGION_LOSS": "DUPLICATE_RISK"}, "AFTER_NOTIFY": {"PROCESS_CRASH": "DUPLICATE_RISK", "HOST_LOSS": "DUPLICATE_RISK", "REGION_LOSS": "DUPLICATE_RISK"}, } assert all(matrix[state][fault] for state in states for fault in faults) upper = 5 * 1_500 assert upper == 7_500 print(dict(matrix={state: dict(matrix[state]) for state in states}, replication_lag_bound=upper, note="bound assumes uniform writes and similar transaction sizes")) def fitness(): """U69：适应度检查三态判定；低样本输出证据不足。""" min_samples = 1_000 def judge(bad, total, baseline_bad, baseline_total, error_limit=0.001, delta_limit=0.002): if total < min_samples or baseline_total <= 0: return "INSUFFICIENT_EVIDENCE" error, baseline = bad / total, baseline_bad / baseline_total return "PASS" if error <= error_limit and error - baseline <= delta_limit else "STOP" assert judge(1, 100, 0, 100) == "INSUFFICIENT_EVIDENCE" assert judge(50, 10_000, 80, 100_000) == "STOP" assert judge(0, 20_000, 0, 20_000) == "PASS" print(dict(low_sample=judge(1, 100, 0, 100), canary=judge(50, 10_000, 80, 100_000), clean=judge(0, 20_000, 0, 20_000), min_samples=min_samples)) def recovery(): """U69：RTO 分段、备份恢复算术与段级证据标注。""" segments = [("discover", 30, "监控覆盖假设"), ("isolate", 20, "自动仲裁假设"), ("promote", 40, "演练证据"), ("route", 30, "估算"), ("warmup", 60, "抽样校验估算")] rto = sum(seconds for _, seconds, _ in segments) target = 60 tidb = 1024 ** 4 minutes = tidb / (200 * 1024 ** 2) / 60 assert rto == 180 and abs(minutes - 87.4) < 0.05 print(dict(rto_s=rto, target_s=target, gap_s=rto - target, backup_minutes=round(minutes, 1), evidence={name: source for name, _, source in segments})) if __name__ == "__main__": checks = dict(slo=slo, latency=latency, availability=availability, gate=gate, decision=decision, windows=windows, retry=retry, durability=durability, fitness=fitness, recovery=recovery) parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("case", choices=[*checks, "all"]) case = parser.parse_args().case for name in checks if case == "all" else [case]: checks[name]()