バイナリプロトコル設計

バイナリプロトコル設計では、まず受信側がバイト列から明確で有限かつ検証可能なメッセージを取得する方法を定義する必要があります。コンパクトなエンコーディングはその一部に過ぎず、フィールドの意味、重複処理、未知の値、長さの上限、状態遷移、およびエラー動作を規定することも必要です。テキストにも厳格なフレーミングが存在し得ますし、バイナリの方が特定のテキストよりも大きくなることもあり、形式だけで品質を推測することはできません。

フレーミング：長さプレフィックスにも完全な契約が必要

TCP はアプリケーションの書き込み境界を保持しません。2回の send が1回の read で発生することがあり、1回の send が複数の read に分割されることもあります。UDP はデータグラム境界を保持しますが、それでもアプリケーションメッセージが完全かどうか、断片化や複数レコードを許可するかどうか、受信バッファによる切り捨てが発生しないかどうかを判断する必要があります。

HTTP ヘッダーの空行はヘッダーの終了のみを示し、レスポンスボディ全体の長さを独立して決定するものではありません。SMTP DATA も補完的なドットエスケープに依存します。これらの局所的なメカニズムを、他の状態を必要としない汎用フレーミングとして扱うことはできません。

長さフィールドは信頼できるメモリ割り当て指示ではありません。受信側はまず固定ヘッダーと長さの上限を検証し、その後逐步的にデータを受信すべきです。長接続の場合、未完了メッセージの数、総バッファサイズ、組立時間、展開後のサイズを制限する必要があります。単一フレームを 1 MiB に制限しても、1万の接続がそれぞれ半分のフレームを保持し続けることを防げません。

導出可能な教育用プロトコル

以下に定義するのは、パーサーの境界を説明するためのみの小規模プロトコルであり、既存の標準ではありません。これは信頼性の高い順序付きバイトストリーム上で動作し、12 バイトの固定ヘッダーと TLV ボディを使用します。空のボディは許可され、すべてのマルチバイト整数はビッグエンディアンです。

TLV type の最上位ビットはクリティカルフィールドを示し、下位 7 ビットがフィールド ID です。この例では、既知の ID 1 は厳密な UTF-8 名、ID 2 は 4 バイトの uint32 数量です。既知の ID の重複は許可されません。他のオプションフィールドはスキップされ、未知のクリティカルフィールドは拒否されます。既知のフィールドの欠如は構文レベルでは許可されますが、ビジネス操作では必須として要求される場合があります。

この「クリティカルビット」は、この文書のプロトコルが能動的に定義したルールであり、すべての TLV フォーマットに共通するものではありません。読者は長さがわかればオプション値をスキップできることを理解できますが、それによってセマンティクスを理解できるわけではなく、破棄しても正しく転送できることが保証されるわけではありません。

増分パーサー：待機、失敗、正常な完了を分離する

以下のパーサーはボディの長さを制限し、フレームごとに TLV を検証します。呼び出し元から提供されるチャンクを受け取り、上位層が有界な読み込みで供給することを要求し、接続レベルの予算とタイムアウトを管理します。この例は完全なサーバーではありません。

import struct
HEADER = struct.Struct('>2sBBII')
MAX_BODY = 4096

def parse_tlvs(body):
    pos, values = 0, {}
    while pos < len(body):
        if len(body) - pos < 3:
            raise ValueError('truncated TLV header')
        kind, size = struct.unpack_from('>BH', body, pos)
        pos += 3
        if size > len(body) - pos:
            raise ValueError('TLV exceeds body')
        value = body[pos:pos + size]
        pos += size
        field_id = kind & 0x7f
        if field_id not in (1, 2):
            if kind & 0x80:
                raise ValueError('unknown critical field')
            continue
        if field_id in values:
            raise ValueError('duplicate known field')
        if field_id == 1:
            values[1] = value.decode('utf-8', errors='strict')
        else:
            if size != 4:
                raise ValueError('quantity must be uint32')
            values[2] = struct.unpack('>I', value)[0]
    return values

class Decoder:
    def __init__(self):
        self.buffer = bytearray()

    def feed(self, chunk):
        self.buffer.extend(chunk)
        messages = []
        while len(self.buffer) >= HEADER.size:
            magic, version, flags, size, request_id = HEADER.unpack_from(self.buffer)
            if magic != b'KB' or version != 1 or flags != 0 or size > MAX_BODY:
                raise ValueError('invalid header')
            end = HEADER.size + size
            if len(self.buffer) < end:
                break
            fields = parse_tlvs(bytes(self.buffer[HEADER.size:end]))
            del self.buffer[:end]
            messages.append((request_id, fields))
        return messages

    def eof(self):
        if self.buffer:
            raise ValueError('truncated frame')

def tlv(kind, value):
    return struct.pack('>BH', kind, len(value)) + value

def frame(body, request_id=7):
    return HEADER.pack(b'KB', 1, 0, len(body), request_id) + body

body = tlv(1, '温度'.encode()) + tlv(2, struct.pack('>I', 300))
wire = frame(body)
assert len(body) == 16 and len(wire) == 28
expected = [(7, {1: '温度', 2: 300})]
# 空の読み込みから完全な読み込みまで、すべてのバイナリ読み込みの境界を列挙する。
for split in range(len(wire) + 1):
    decoder = Decoder()
    assert decoder.feed(wire[:split]) + decoder.feed(wire[split:]) == expected
    decoder.eof()
# 2つのフレームが結合されていても、フレームごとに消費する必要がある。
assert Decoder().feed(wire + wire) == expected + expected
assert Decoder().feed(frame(body + tlv(3, b'optional'))) == expected

bad = [
    frame(body + tlv(0x83, b'critical')),
    frame(body + tlv(1, b'duplicate')),
    frame(tlv(1, b'\xff')),
    frame(tlv(2, b'\x01')),
    frame(b'\x01\x00\x04x'),
    HEADER.pack(b'KB', 1, 0, MAX_BODY + 1, 7),
    wire[:-1],
]
for candidate in bad:
    decoder = Decoder()
    try:
        decoder.feed(candidate)
        decoder.eof()
    except ValueError:
        pass
    else:
        raise AssertionError('invalid input accepted')
print('28 B メッセージ：すべてのバイナリ境界、フレーム結合、オプションフィールド、7種類の拒否が通過')

解析エラーが発生した後、この例では呼び出し元にストリームのデコードを中止するよう要求します。次の KB がフレームヘッダーではないかと推測し続けるのではなく、このバイト列がペイロード内にも合法に出現する可能性があるためです。ノイズからの再同期が本当に必要な場合は、同期、完全性、および回復ルールを別途設計する必要があります。

本番環境の実装では、バイト単位のバッファ移動のコスト、例外発生前に解析済みのメッセージの配信方法、メモリ所有権、および非同期キャンセルを考慮する必要があります。ゼロコピーでもライフサイクルの制約から逃れることはできません。小さなビューを保持していると、大きなバッファ全体が解放できなくなる可能性があります。

ヘッダーの半分を受け取った場合、待機すべきかエラーとするか？

11バイト目で EOF に切り替えると、待機は切り捨てになります。17バイト目で最初のフレームは完全に完成しており、2番目のフレームがまだ到着していなくても配信できます。さらに宣言された長さが上限を超えるの場合、12バイト目で拒否すべきです。4097バイト待ってエラーを確定させるべきではありません。

整数エンコーディング：Protobuf varint と QUIC varint は異なる

Protobuf varint は下位グループが先に現れ、uint64 で最大 10 バイトです。QUIC は長さマーカーの後にネットワークバイトオーダーでエンコーディングされ、最大 62 ビットです。QUIC の一般的な整数エンコーディングでは、最短表現でない某些の表現が許可されていますが、frame type などの特定のフィールドには最短エンコーディングの要件があり、「すべての varint が最短でなければならない」と一般化することはできません。QUIC §16

def uvarint(n):
    if not 0 <= n < 1 << 64:
        raise ValueError('not uint64')
    result = bytearray()
    while n >= 128:
        result.append((n & 127) | 128)
        n >>= 7
    result.append(n)
    return bytes(result)

def quic_integer(n):
    for length, bits, prefix in [(1, 6, 0), (2, 14, 1), (4, 30, 2), (8, 62, 3)]:
        if 0 <= n < 1 << bits:
            return (n | (prefix << (8 * length - 2))).to_bytes(length, 'big')
    raise ValueError('not uint62')

assert uvarint(300).hex() == 'ac02'
assert quic_integer(300).hex() == '412c'
assert len(uvarint(127)) == 1 and len(uvarint(128)) == 2
assert len(quic_integer(63)) == 1 and len(quic_integer(64)) == 2
assert len(uvarint(2**64 - 1)) == 10
assert len(quic_integer(2**62 - 1)) == 8
# ZigZag は、絶対値の小さい符号付き数を、小さい符号なし数にマッピングする。
zigzag = lambda n: 2*n if n >= 0 else -2*n-1
assert [zigzag(n) for n in [0, -1, 1, -2]] == [0, 1, 2, 3]
assert uvarint(zigzag(-2)) == b'\x03'
print('2種類の varint、境界、ZigZag マッピングが通過')

ZigZag は値のマッピングであり、その後 varint エンコーディングが行われます。Protobuf の sint32/sint64 はこれを使用します。int32/int64 の負数は別のルールを採用しており、負数は 10 バイトになる可能性があります。ワイヤータイプがどちらも 0 だからといって、互いに交換可能と宣言してはいけません。

同じ 300 なのに、なぜ 2 つの varint は全く異なるのか？

まず 63/64 付近で QUIC が 2 バイトに増える様子を観察し、次に 127/128 付近で Protobuf の変化を観察します。これらは同じ「最上位ビットが長さを示す」というルールではありません。300 に移動し、ac02 と 412c のビットグループ順序を照らし合わせます。

Protobuf：wire type が境界を与え、スキーマがセマンティクスを与える

syntax = "proto3";
message Person {
  string name = 1;
  uint32 age = 2;
}

name="Li"、age=300 のエンコーディングの 1 つは 0a 02 4c 69 10 ac 02 です。フィールド 1 のタグは (1 << 3) | 2 = 0a で、その後長さが 2、UTF-8 が続きます。フィールド 2 のタグは 10 で、その後 varint ac 02 が続きます。フィールド名はワイヤー上に現れず、年齢の単位や制約もタグ自体によって解釈されません。

したがって、Protobuf はすべてのフィールドが独立した長さを持つ統一された 3 部構成の TLV を明示的に携带しているわけではありません。fixed32/fixed64 などはリトルエンディアンを使用し、TCP/IP ヘッダーで一般的なビッグエンディアンとは異なります。可変長整数は、独自のビットグループルールに従って理解する必要があります。Protobuf Encoding

未知フィールドの保持、フィールドの存在、および JSON への変換動作は、具体的なランタイムとインターフェースに依存します。バイナリメッセージをオブジェクトとして読み取り、既知のフィールドのみを手動でコピーすると、本来保持可能だった未知の内容が失われる可能性があります。ワイヤー互換性があるからといって、すべての中間処理が无损であるわけではありません。Proto3 未知フィールド

バイトオーダー、正規化、完全性はそれぞれが決定する

新しいプロトコルでは、フィールドの幅、符号、バイトオーダー、浮動小数点表現、テキストエンコーディング、単位を明文で規定すべきであり、「CPU のネイティブ形式を選べばよい」というわけではありません。C struct をそのまま送信すると、パディング、アラインメント、ABI の差異が持ち込まれます。CPU が未アラインアクセスを許可していても、それが相手の言語やハードウェアでも同様に許可されるわけではありません。

署名、ハッシュ、キャッシュキーについては、生バイト列にバインドするのか、正規化されたセマンティクスにバインドするのかを定義する必要があります。2つのオブジェクトが等価であっても、それらのシリアライズされたバイト列が一意であるとは限りません。一意な表現が必要な場合は、フィールドの順序、重複キー、浮動小数点 NaN、数字の最短エンコーディングなどのルールを明確にする必要があります。CBOR の決定論的エンコーディングは、これらの要件を明示的に書き出した例であり、「すべての CBOR が本来的に正規化されている」わけではありません。CBOR 決定論的エンコーディング

CRC やチェックサムは、適用可能な故障モデルの下での予期せぬ損傷を検出するために主に使用され、メッセージが信頼できる相手から来たことを証明するものではありません。認証には MAC、AEAD、または署名と鍵の信頼が必要です。圧縮についても展開後の上限を規定する必要があり、元のフレームが小さくても、展開後のリソースオーバーヘッドが小さいとは限りません。

デザインが完了したかどうかの基準は、失敗入力に対しても確定的な動作があるかどうか

切り捨て、結合、境界長、未知のクリティカルフィールド、重複値、およびクロス言語ベクトルを優先してテストし、その後ファジングテストとリソース予算テストを行います。フォールトトレランスは、明文で定義された相互運用動作であり、各実装が異なる入力を推測して受け入れるものではありません。Maintaining Robust Protocols