ネットワーク遅延と部分失敗を層別に診断する

到達目標

1. DNS問い合わせからHTTP応答完了までを少なくとも六イベントの時系列として図示できる

   • DNSからHTTPまでの時刻、期限、再試行条件、未観測区間を含むトレース
   • 輸送成功とアプリケーション成功を区別する5分説明

2. 利用者期限から接続、TLS、応答、再試行へ時間を配分し、最悪時の合計を計算できる

   • DNSからHTTPまでの時刻、期限、再試行条件、未観測区間を含むトレース
   • 損失、名前解決、接続、依存遅延の仮説を反証する診断

3. パケット損失と依存サービス遅延の仮説を層別の時刻と再送観測で反証し、未知経路へ診断を移せる

   • 損失、名前解決、接続、依存遅延の仮説を反証する診断
   • 未知の多段依存で損失と依存遅延を切り分けた調査計画

能力の進行

1. recognize

   DNS、TCP、TLS、HTTPの開始・完了イベントと各層の失敗を時系列上で識別できる

   証拠: DNSからHTTPまでの時刻、期限、再試行条件、未観測区間を含むトレース

2. explain

   TCP到達、TLS認証、HTTP応答、業務処理成功が別の契約である理由を説明できる

   証拠: 輸送成功とアプリケーション成功を区別する5分説明

3. apply

   総期限から各段の予算と再試行回数を逆算し、合計が期限を超えないトレースを作成できる

   証拠: DNSからHTTPまでの時刻、期限、再試行条件、未観測区間を含むトレース

4. diagnose

   DNS時間、接続時間、再送、server timingを比較し、損失と依存遅延を反証できる

   証拠: 損失、名前解決、接続、依存遅延の仮説を反証する診断

5. lead

   冪等性、deadline伝播、retry stormを含む障害レビューを未知の依存グラフで主導できる

   証拠: 未知の多段依存で損失と依存遅延を切り分けた調査計画

なぜ重要か

「APIが遅い」という一つの観測には、名前解決、接続、暗号handshake、request送信、server処理、response転送という異なる区間が含まれる。総時間だけを見てnetwork担当またはserver担当へ責任を移すと、原因を反証できず、timeoutとretryで負荷を増幅し得る。

networkでは、相手が処理して応答だけ失われた状態と、相手へ届かなかった状態をclientから完全には区別できないことがある。特に状態を変えるPOSTを無条件にretryすると二重処理を生む。deadline、冪等key、結果照会を一つの契約として設計する必要がある。

メンタルモデル

一要求を層ではなくイベント列として観測する。各イベントに単調時計の開始・完了、残りdeadline、request IDを付ける。層の境界を越えるたびに「何が完了したと言えるか」を更新する。

timeoutは各層へ同じ値を配るのではなく、利用者の総deadlineから逆算する。再試行を許すなら、一回目のtimeout、backoff、二回目、後処理の合計が総deadline以下でなければならない。下流にも残りdeadlineを伝えないと、clientが諦めた後もserverが仕事を続ける。

動く例で考える

2000msの利用者期限へ一回のretryを収める

前提

総deadlineは2000ms。UI更新と結果保存に200ms、retry間のjitter付きbackoffに100msを確保する。状態変更操作は冪等key対応済み。

入力

一要求についてDNS、connect、TLS、time to first byte、body完了を単調時計で記録する。

操作

一回目へ800ms、backoffへ100ms、二回目へ700ms、後処理へ200ms、cancel伝播へ100msを配り、残り100ms未満では新しい試行を開始しない。

観測

localhost simulatorでcold/warmを各7回測り、制御したDNS/TLS delayを含む単調時刻を保存する。最悪時は800 + 100 + 700 + 200 + 100 = 1900msで期限内に収まる。

結論

各段へ2000msを設定すると、二試行で4000ms以上になり得る。試行ごとに残りdeadlineを再計算し、未開始区間を明示する。

次のtrace-fixture.csv相当のlocalhost simulatorは、外部network、秘密情報、curlを必要としない。DNS cacheとTLS handshakeは制御可能な待ち時間として模擬し、実protocol測定ではないことを出力へ明記する。

python3.13 - <<'PY'
import json
import re
import socket
import statistics
import threading
import time

HARNESS = "network_lab_v2"
MAX_LINE_BYTES = 256
SOCKET_DEADLINE_SECONDS = 0.25
MAX_REQUEST_ID_BYTES = 64
MAX_DELAY_MS = 1000
REQUEST_ID_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}")

class ProtocolError(Exception):
    def __init__(self, code):
        super().__init__(code)
        self.code = code

def receive_line(
    connection,
    timeline=None,
    deadline_seconds=SOCKET_DEADLINE_SECONDS,
):
    connection.settimeout(deadline_seconds)
    payload = bytearray()
    while True:
        try:
            part = connection.recv(min(64, MAX_LINE_BYTES + 1))
        except TimeoutError as error:
            raise ProtocolError("receive_timeout") from error
        if not part:
            raise ProtocolError("newline_required")
        newline = part.find(b"\n")
        fragment = part if newline == -1 else part[:newline]
        payload.extend(fragment)
        if len(payload) > MAX_LINE_BYTES:
            raise ProtocolError("line_too_long")
        if timeline is not None and "first_byte" not in timeline:
            timeline["first_byte"] = time.monotonic_ns()
        if newline != -1:
            if part[newline + 1:]:
                raise ProtocolError("multiple_lines")
            break
    if timeline is not None:
        timeline["body_finish"] = time.monotonic_ns()
    try:
        return payload.decode("ascii")
    except UnicodeDecodeError as error:
        raise ProtocolError("non_ascii_line") from error

def validate_request_id(request_id):
    if (
        len(request_id.encode("ascii", errors="ignore"))
        > MAX_REQUEST_ID_BYTES
        or REQUEST_ID_PATTERN.fullmatch(request_id) is None
    ):
        raise ProtocolError("invalid_request_id")
    return request_id

def parse_command(line):
    fields = line.split()
    if not fields:
        raise ProtocolError("empty_command")
    command = fields[0]
    if command == "APPLY":
        if len(fields) != 4:
            raise ProtocolError("invalid_arity")
        request_id = validate_request_id(fields[1])
        if not fields[2].isascii() or not fields[2].isdecimal():
            raise ProtocolError("invalid_delay_ms")
        delay_ms = int(fields[2])
        if not 0 <= delay_ms <= MAX_DELAY_MS:
            raise ProtocolError("invalid_delay_ms")
        if fields[3] not in {"reply", "drop"}:
            raise ProtocolError("invalid_mode")
        return command, request_id, delay_ms, fields[3]
    if command == "LOOKUP":
        if len(fields) != 2:
            raise ProtocolError("invalid_arity")
        return command, validate_request_id(fields[1])
    if command == "STOP":
        if len(fields) != 1:
            raise ProtocolError("invalid_arity")
        return (command,)
    raise ProtocolError("unknown_command")

class LocalFixture:
    def __init__(self, injected_failure_id):
        self.results = {}
        self.injected_failure_id = injected_failure_id
        self.injected_exceptions = []
        self.stopped = threading.Event()
        self.terminated = threading.Event()
        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            self.listener.setsockopt(
                socket.SOL_SOCKET,
                socket.SO_REUSEADDR,
                1,
            )
            self.listener.bind(("127.0.0.1", 0))
            self.listener.listen()
            self.listener.settimeout(0.05)
        except Exception:
            self.listener.close()
            raise
        self.port = self.listener.getsockname()[1]
        self.started = False
        # A non-daemon thread makes cleanup failures visible instead of hiding
        # a leaked server at interpreter shutdown.
        self.server = threading.Thread(
            target=self.serve,
            name="localhost-fixture",
            daemon=False,
        )

    def start(self):
        try:
            self.server.start()
            self.started = True
        except Exception:
            self.listener.close()
            raise

    def handle(self, connection, command):
        if command[0] == "APPLY":
            _, request_id, delay_ms, mode = command
            if request_id == self.injected_failure_id:
                raise RuntimeError("controlled handler failure")
            self.results.setdefault(request_id, "applied")
            time.sleep(delay_ms / 1000)
            if mode == "reply":
                connection.sendall(
                    f"RESULT {request_id} applied\n".encode("ascii")
                )
        elif command[0] == "LOOKUP":
            _, request_id = command
            value = self.results.get(request_id, "missing")
            connection.sendall(
                f"RESULT {request_id} {value}\n".encode("ascii")
            )
        else:
            self.stopped.set()
            connection.sendall(b"STOPPED\n")

    def serve(self):
        try:
            while not self.stopped.is_set():
                try:
                    connection, _ = self.listener.accept()
                except TimeoutError:
                    continue
                except OSError:
                    if self.stopped.is_set():
                        break
                    raise
                # The accepted socket receives its own deadline immediately;
                # a slow client therefore cannot retain the sole worker.
                connection.settimeout(SOCKET_DEADLINE_SECONDS)
                with connection:
                    try:
                        self.handle(
                            connection,
                            parse_command(receive_line(connection)),
                        )
                    except ProtocolError as error:
                        connection.sendall(
                            f"ERROR {error.code}\n".encode("ascii")
                        )
                    except Exception as error:
                        self.injected_exceptions.append(type(error).__name__)
                        connection.sendall(b"ERROR internal\n")
        finally:
            self.terminated.set()

    def close(self):
        self.stopped.set()
        try:
            self.listener.close()
        finally:
            if self.started:
                self.server.join(timeout=1)
        if self.started and self.server.is_alive():
            raise RuntimeError("localhost fixture did not stop")
        if self.started and not self.terminated.is_set():
            raise RuntimeError("localhost fixture termination not observed")

def open_connection(port, connector=socket.create_connection):
    return connector(("127.0.0.1", port), timeout=1)

def send_line(port, payload, *, shutdown_write=False, send=True):
    with open_connection(port) as connection:
        if send:
            connection.sendall(payload)
        if shutdown_write:
            connection.shutdown(socket.SHUT_WR)
        return receive_line(connection, deadline_seconds=1)

def rejection_code(port, payload=b"", **options):
    response = send_line(port, payload, **options)
    prefix, separator, code = response.partition(" ")
    assert prefix == "ERROR" and separator
    return code

def request(port, request_id, resolver_mode, delay_ms=0, drop=False,
            connector=socket.create_connection):
    timeline = {}
    timeline["dns_start"] = time.monotonic_ns()
    socket.getaddrinfo("localhost", port, type=socket.SOCK_STREAM)
    # These injected waits make the fixture deterministic enough to compare;
    # they do not claim to flush or observe the operating-system DNS cache.
    time.sleep(0.002 if resolver_mode == "cold" else 0.0001)
    timeline["dns_end"] = time.monotonic_ns()
    timeline["connect_start"] = time.monotonic_ns()
    with open_connection(port, connector) as connection:
        timeline["connect_end"] = time.monotonic_ns()
        timeline["tls_start"] = time.monotonic_ns()
        time.sleep(0.0002)
        timeline["tls_end"] = time.monotonic_ns()
        mode = "drop" if drop else "reply"
        connection.sendall(
            f"APPLY {request_id} {delay_ms} {mode}\n".encode("ascii")
        )
        try:
            body = receive_line(connection, timeline)
        except ProtocolError as error:
            if not drop or error.code != "newline_required":
                raise
            now = time.monotonic_ns()
            timeline.setdefault("first_byte", now)
            timeline["body_finish"] = now
            body = ""
    return {
        "request_id": request_id,
        "resolver_mode": resolver_mode,
        "monotonic_ns": timeline,
        "body": body,
        "duration_ms": (
            timeline["body_finish"] - timeline["dns_start"]
        ) / 1_000_000,
    }

def main():
    fixture = LocalFixture(injected_failure_id="inject-error")
    baseline = {"cold": [], "warm": []}
    protocol_rejections = {}
    connection_failure = "not_injected"
    real_connector_calls = [0]
    real_connector_calls_at_refusal = None
    injector_calls = [0]
    ambiguous = None
    lookup_result = None
    delayed = None
    baseline_median_ms = None
    try:
        fixture.start()
        for mode in baseline:
            for repeat in range(7):
                baseline[mode].append(
                    request(fixture.port, f"{mode}-{repeat}", mode)
                )
        baseline_values = [
            sample["duration_ms"]
            for samples in baseline.values()
            for sample in samples
        ]
        baseline_median_ms = statistics.median(baseline_values)

        delayed = request(
            fixture.port,
            "delay-fixture",
            "warm",
            delay_ms=25,
        )
        protocol_rejections = {
            "unterminated": rejection_code(
                fixture.port,
                b"LOOKUP never-finished",
                shutdown_write=True,
            ),
            "oversized": rejection_code(
                fixture.port,
                b"X" * (MAX_LINE_BYTES + 1) + b"\n",
            ),
            "timeout": rejection_code(fixture.port, send=False),
            "invalid_id": rejection_code(
                fixture.port,
                b"LOOKUP bad/id\n",
            ),
            "excessive_delay": rejection_code(
                fixture.port,
                b"APPLY too-slow 1001 reply\n",
            ),
            "invalid_arity": rejection_code(
                fixture.port,
                b"APPLY missing-mode 0\n",
            ),
            "invalid_mode": rejection_code(
                fixture.port,
                b"APPLY wrong-mode 0 maybe\n",
            ),
        }
        assert send_line(
            fixture.port,
            b"APPLY inject-error 0 reply\n",
        ) == "ERROR internal"

        # The fault injector raises before the real connector can create a
        # socket. The counters prove the refusal did not call the real path.
        def real_connector(address, timeout):
            real_connector_calls[0] += 1
            return socket.create_connection(address, timeout=timeout)

        def refused_connector(address, timeout):
            del address, timeout
            injector_calls[0] += 1
            raise ConnectionRefusedError(
                "controlled refusal before socket creation"
            )

        try:
            request(
                fixture.port,
                "refused-fixture",
                "warm",
                connector=refused_connector,
            )
        except ConnectionRefusedError:
            connection_failure = "injected_before_socket_creation"
        real_connector_calls_at_refusal = real_connector_calls[0]
        assert real_connector_calls_at_refusal == 0
        assert injector_calls[0] == 1

        ambiguous = request(
            fixture.port,
            "order-001",
            "warm",
            drop=True,
            connector=real_connector,
        )
        lookup_result = send_line(
            fixture.port,
            b"LOOKUP order-001\n",
        ).split()[-1]
        assert send_line(fixture.port, b"STOP\n") == "STOPPED"
    finally:
        fixture.close()

    budget = {
        "deadline_ms": 2000,
        "first_attempt_ms": 800,
        "backoff_ms": 100,
        "second_attempt_ms": 700,
        "post_processing_ms": 200,
        "cancel_propagation_ms": 100,
    }
    budget["worst_case_ms"] = sum(
        value
        for name, value in budget.items()
        if name != "deadline_ms"
    )
    assert budget["worst_case_ms"] <= budget["deadline_ms"]
    assert ambiguous is not None
    assert delayed is not None
    assert baseline_median_ms is not None

    return {
        "harness": HARNESS,
        "fixture": "trace-fixture.csv",
        "bind_address": "127.0.0.1",
        "external_network_used": False,
        "tls_simulated": True,
        "dns_cache_simulated": True,
        "baseline": baseline,
        "baseline_median_ms": baseline_median_ms,
        "budget": budget,
        "retry_stop_conditions": [
            "remaining deadline is below 100ms",
            "non-idempotent APPLY has an unknown result",
        ],
        "faults": {
            "dependency_delay_ms": delayed["duration_ms"],
            "connection_failure": connection_failure,
            "real_connector_calls": real_connector_calls_at_refusal,
            "injector_calls": injector_calls[0],
        },
        "non_idempotent": {
            "request_id": "order-001",
            "first_response": ambiguous["body"] or "response-lost",
            "automatic_retry": "stopped",
            "lookup_result": lookup_result,
        },
        "protocol_security": {
            "max_line_bytes": MAX_LINE_BYTES,
            "socket_deadline_ms": int(
                SOCKET_DEADLINE_SECONDS * 1000
            ),
            "request_id_ascii_allowlist": REQUEST_ID_PATTERN.pattern,
            "max_request_id_bytes": MAX_REQUEST_ID_BYTES,
            "max_delay_ms": MAX_DELAY_MS,
            "rejections": protocol_rejections,
        },
        "cleanup": {
            "injected_exception": (
                "observed"
                if fixture.injected_exceptions == ["RuntimeError"]
                else "not_observed"
            ),
            "server_stopped": fixture.terminated.is_set(),
            "thread_alive": fixture.server.is_alive(),
            "listener_closed": fixture.listener.fileno() == -1,
        },
    }

if __name__ == "__main__":
    print(json.dumps(main(), ensure_ascii=False, indent=2))
PY

cold/warmは各7回だが、このfixtureのcoldは注入delay、warmは短い注入delayであり、OS resolver cacheのflushを証明しない。TLS区間も暗号処理のsimulationである。実環境へ移すときだけ管理下の検証先とcurl等を用い、同じrequest IDでclient/serverを相関する。

同じrequest IDでclientとserverを相関する。clientのconnect区間と再送が増え、serverにrequestが届いていなければ損失説を支持する。connectは平常でserver内部のDB spanだけが伸びれば依存遅延説を支持する。どちらも観測できなければ原因確定ではなく観測不足である。

ラボ成果物: 「DNSからHTTPまでの時系列トレースとタイムアウト予算」には、coldとwarmの両経路、単位、単調時計、request ID、最悪時の足し算、retry停止条件を含める。

トレードオフと失敗モード

もっともらしい誤診と反証

1. 誤診: 「HTTP p99悪化はpacket loss」。反証: TCP再送、connect時間、複数地点の同時刻を確認し、serverの依存spanと比較する。再送が平常でDB区間だけ増えていれば損失説を退ける。
2. 誤診: 「DNS時間が0msなのでDNSに問題はない」。反証: warm cacheとcold queryを分け、TTL、負のcache、複数addressを確認する。測定がresolver callを含んでいない可能性も検証する。

retryは成功率を上げる可能性と、障害中の負荷を増やす可能性を同時に持つ。全clientが同時に再試行しないようjitterを入れ、総試行数を制限し、server側のcapacityと期限を共有する。TLS 0-RTTを使う場合はreplay可能性に適した操作かを確認する。

知識チェック

1. TCP ACKを受けたら注文は確定済みか。分からない。byteの輸送確認とapplicationの状態遷移は別である。
2. HTTP 200なら利用者操作は成功か。契約次第である。body内の業務結果、後続処理、client保存まで別の失敗点がある。
3. DNS cacheは常に遅延を減らすか。通常はqueryを省けるが、stale addressや負のcacheにより回復を遅らせることがある。
4. timeoutを長くすれば成功率は上がるか。待てば成功する原因には効くが、queueとresource占有を増やし、利用者deadlineを破り得る。

5分teach-back

八イベントの時系列を使い、各段で保証されたことと未確定なことを説明する。2000ms予算の足し算と、POST timeout後に直ちにretryしない理由を一分ずつ含める。

未知へのtransfer

パケット損失と依存遅延を区別する診断を、CDN、API gateway、二つの内部service、databaseからなる未知経路で行う。観測点が欠ける区間を明示し、責任を断定する前に追加する時刻とrequest IDを設計する。

出典と次の学習

RFC 1035はDNS、RFC 9293はTCP、RFC 6298はTCP再送timeout、RFC 8446はTLS 1.3、RFC 9110はHTTP semanticsの正準として使う。各層の仕様を横断して要約し、長い引用や原図の転載はしていない。正確なURLはlesson metadataのsourcesに集約した。

次の学習では、要求とdomain modelへ進み、networkの結果不明状態をapplication契約として表現する。1日後は時系列、7日後は予算計算、30日後は実trace、90日後はretryによる障害増幅を再レビューする。

実践ラボ

名前解決から応答完了までを期限へ収める

提出成果物: DNSからHTTPまでの時系列トレースとタイムアウト予算

1. cold DNSとwarm DNSで、名前解決、接続、TLS、最初のbyte、本文完了の単調時計時刻を記録する
2. 利用者期限2000msから各段の上限、後処理余白、キャンセル伝播時間を配分する
3. 一回の再試行を含む最悪時合計を計算し、期限を超える場合は試行予算を修正する
4. 制御可能な遅延依存と接続失敗を別々に注入し、時系列上の差を観測する
5. 非冪等操作では再試行を止める条件と、結果不明を照会するための識別子を記録する

説明して理解を確かめる

5分で、DNS、TCP、TLS、HTTPの役割、TCP成功が業務成功を保証しない理由、exactly-onceを安易に主張できない理由を説明する。

アセスメント

1. 問い: HTTPのp99が800msへ悪化した。『パケット損失だ』という診断と『依存DBが遅い』という診断をどう区別するか。

   期待する証拠: TCP再送または接続時間、server-side区間、依存span、同時刻比較による反証

2. 問い: POSTの応答がtimeoutした。即時retryが常に安全でない理由と必要な契約を示す。

   期待する証拠: 処理済みだが応答喪失の状態、冪等key、結果照会、期限とbackoffを含む説明

別問題へ転用する

パケット損失と依存遅延を区別する診断

復習スケジュール

1. 1日後

   総期限2000msへ一回の再試行を入れるなら、各試行へ何ms配分するか

2. 7日後

   TCP成功から業務成功までに残る結果不明状態は何か

3. 30日後

   パケット損失説を反証するclient側とserver側の観測を一つずつ示す

4. 90日後

   総期限2000msへ一回の再試行を入れるなら、各試行へ何ms配分するか

評価ルーブリック

出典

以下の外部資料は利用者が選択したときだけ開きます。

• RFC 1035: Domain Names - Implementation and Specification (standard)
• RFC 9293: Transmission Control Protocol (TCP) (standard)
• RFC 6298: Computing TCP's Retransmission Timer (standard)
• RFC 8446: The Transport Layer Security (TLS) Protocol Version 1.3 (standard)
• RFC 9110: HTTP Semantics (standard)