反熵とデータ修復

レプリケーションは新しい書き込みをレプリカに伝播させるが、修復はすでに発生した差異を見つけ、解消する。ノードの長時間のオフライン、ヒントレコードの期限切れ、マイグレーションの中断、メディアの破損などが原因で、通常のクエリパスではカバーされない問題が残ることがある。

読み書きパス における読み取り修復は、アクセスされたデータと読み取りに参加したレプリカのみをカバーする。反熵は、共同で責任を負うデータ範囲を能動的に比較することで、コールドデータもチェック対象に含める。それは特定のコマンドではなく、レプリカを再び一貫性のある状態に戻すための一連のメカニズムを指す。

収束に必要な条件

2つのレプリカが最終的に同じ更新セットを受け取ったと仮定する。もしそれらが同じかつ適切な競合マージルールを使用していれば、同じ状態を生成するはずである。この目標を達成するには、少なくとも「更新が最終的に到達可能であること」、「差異が検出可能であること」、「修復を実行するためのリソースが存在すること」、そして「削除情報とバージョン情報が早期に破棄されないこと」の4つが必要である。

したがって、最終的な一貫性は「放っておけば勝手に良くなる」という意味ではない。ネットワークの永続的な分断、修復が常に餓死状態になること、バージョン比較ルールが異なることなどは、収束を妨げる可能性がある。これらの前提に対する運用上の保証自体が、一貫性設計の一部なのである。

Merkle木：要約で同一範囲を除外する

データセット全体をレコードごとに交換して差異を見つけることは可能だが、転送コストが高い。Merkle木は、同じパーティション内のデータ範囲をリーフ要約にマッピングし、さらに子要約のハッシュを逐次計算する。ルート要約が同じであれば、ハッシュ衝突が無視でき、エンコーディングが一貫しているという前提の下、範囲全体をスキップできる。そうでない場合は、異なるサブツリーへとさらに降りていく。

比較を行う双方は、同じキー範囲、順序、エンコーディング、バージョンの基準を使用しなければならない。片方がツームストーンをハッシュに含め、もう片方が現在の値のみをハッシュする場合、要約の差異はレプリカの障害ではなく比較ルールに起因する可能性がある。

木構造の役割は、不要な交換を減らすことであり、すべての修復コストを O(log n) にすることではない。n 件のレコードの要約を構築するには、通常、対応するデータを少なくとも1回スキャンする必要がある。単一の局所的な差異の特定は木の高さに沿って下っていくことができるが、大量の差異がある場合は多くのノードに到達し、最終的には実際のレコードを転送する必要がある。木を永続的に維持する場合、更新にも追加のコストがかかる。

8つの固定範囲で、実際の差異を1つ特定する

以下では、双方が事前に8つの範囲を合意し、各範囲にキーでソートされたレコードを保存するものとする。レコードの形式は (key, version, value) である。モデルでは文字列、整数、nullのみを使用し、双方がJSONエンコーディングの規約を共有しており、リーフノードと内部ノードにはそれぞれタイププレフィックスが付与される。これにより、比較対象は現在のvalueだけでなく、バージョンと削除情報も含む。

from hashlib import sha256
import json

def tree(ranges):
    n = len(ranges)
    if n == 0 or n & (n - 1):
        raise ValueError("この例では範囲数が2の累乗である必要があります")
    nodes = [b""] * (2 * n)
    for i, records in enumerate(ranges):
        data = json.dumps(sorted(records, key=lambda r: r[0]),
                          ensure_ascii=True, separators=(",", ":")).encode("ascii")
        nodes[n + i] = sha256(b"leaf:" + data).digest()
    for i in range(n - 1, 0, -1):
        nodes[i] = sha256(b"node:" + nodes[2*i] + nodes[2*i+1]).digest()
    return nodes

def differences(left, right):
    if len(left) != len(right):
        raise ValueError("木の範囲分割は一致している必要があります")
    n, compared, changed = len(left) // 2, 0, []
    def visit(i):
        nonlocal compared
        compared += 1
        if left[i] == right[i]:
            return
        if i >= n:
            changed.append(i - n)
        else:
            visit(2*i)
            visit(2*i+1)
    visit(1)
    return changed, compared

A = [[(f"k{i}", 1, "old")] for i in range(8)]
B = [list(records) for records in A]
B[6] = [("k6", 2, "new")]
assert differences(tree(A), tree(B)) == ([6], 7)
assert differences(tree(A), tree(A)) == ([], 1)
C = [[(f"k{i}", 2, "new")] for i in range(8)]
assert differences(tree(A), tree(C)) == (list(range(8)), 15)
assert tree([[("k", 1, "same")]]) != tree([[("k", 2, "same")]])
assert tree([[("k", 2, None)]]) != tree([[]])  # ツームストーンはレコードなしと等しくない
assert tree([[('b', 1, 'v'), ('a', 1, 'v')]]) == tree([[('a', 1, 'v'), ('b', 1, 'v')]])

差異が1つの場合、7組の要約が比較される：ルートで1回、各層で2つの子を比較し、合計3層。範囲0–3と4–5の同一サブツリーは展開されず、最終的に範囲6のみが特定される。8つの範囲すべてが異なる場合、すべての15ノードを比較する必要がある。「木の高さは3しかない」からといって、すべての差異が3回の比較で済むわけではない。

カウントは、⁠すでに構築された2つの木間の要約比較のみを含み、各範囲のスキャン、ソート、ハッシュ、および差異レコードの転送は含まない。単一のリーフ範囲に百万件のレコードが含まれている可能性もあり、リーフに到達した後でもさらに比較を行う必要がある。範囲の境界は同じメタデータによって固定されなければならない。両側のレコード数が異なるからといって、各自が数で均等に分割してはならない。そうすると、同じ位置にあるリーフが同じキーのセットではなくなる。

差異が増えた場合、木はいくつの範囲をスキップできるか？

要約が同じであれば、対応するサブツリーをスキップできる。要約が異なる場合は、さらに比較を行う理由を示すだけで、どちらが更新されたかは教えてくれない。以下では、実際のSHA-256を使用して2つの木を計算する。リーフを1つ変更すると比較パスが見え、すべてのリーフを変更するとすべての15ノードに到達する。短いハッシュは表示用であり、比較には完全な要約が使用され、カウントも構築済みの木間の比較のみを統計する。

差異の発見から修復方法の決定へ

ある範囲で、Aが key=X, version=7 を持ち、Bが version=5 を持っているとする。バージョンルールが7が5をカバーすることを証明できる場合のみ、7を新しい状態として扱える。2つのベクトルバージョンが並列に発生した場合、要約は差異があることだけを伝え、アプリケーションに勝者を選ばせることはできない。

修復は、通常の書き込みと同じマージルールを使用する必要がある。ビジネスロジックが並列のsiblingを保持する場合、修復は必要なバージョンを伝播させる必要がある。LWW（Last-Writer-Wins）を採用する場合、そのタイムスタンプソートと並列値の破棄というセマンティクスを受け入れる必要がある。詳細は競合解決を参照。

Gossipはメンバーシップやスケジューリング情報を伝播できるが、反熵はデータの差異を担当する。これらを組み合わせることはできるが、すべてのMerkle修復がgossipによって駆動されるわけではない。理想的なメッセージ拡散のラウンド数を、スキャン、転送、マージを含むデータベース修復の所要時間に直接当てはめることはできない。

削除もレプリケートする理由：ツームストーン

削除がローカルでレコードを削除するだけの場合、古い値をまだ保持しているレプリカに対して「このレコードは削除された」と伝えることができない。ツームストーンは削除の事実を保持し、マージルールが古い値を上書きすべきであることを判定できるようにする。

A、Bがすでに削除を記録しており、Cはオフラインのまま古い値を保持していると考える。A、BがCが復帰する前に唯一の削除証拠をクリアした場合、後の修復でCの古い値が再び伝播される可能性がある。これが「削除の復活」問題である。

ツームストーンの保持期間、最大オフライン時間、修復のカバレッジサイクル、およびノードの再参加戦略は連携する必要がある。単に古いパラメータを大きく設定すれば万事解決するわけではない。関連するレプリカが削除を受領したことを証明するか、再参加時に強制再構築を行う必要がある。Cassandraの修復とツームストーンの制約についてはRepairを参照。

ツームストーンはバージョン付きの事実である

簡略化のため、同じキーの厳密に増加するバージョンのみを考慮する。バージョン5は古い値、バージョン6は削除とする。None はそのレプリカにレコードが何も存在しないことを示し、(6, None) こそがバージョン6の削除の事実を表す。

def merge(a, b):
    candidates = [record for record in (a, b) if record is not None]
    return max(candidates, key=lambda record: record[0]) if candidates else None

old = (5, "value")
deleted = (6, None)
assert merge(deleted, old) == deleted
assert merge(None, old) == old  # 削除証拠を早期に失うと、古い値が再び勝者になる
assert merge(deleted, (7, "recreated")) == (7, "recreated")

3番目のアサートも重要である。ツームストーンは「今後このキーを永遠に禁止する」わけではない。ビジネスロジックで再作成が許可されている場合、新しい有効なバージョンが削除を上書きできる。阻止されるのは古い値の復活のみである。実際のシステムで同じバージョンでも異なるコンテンツを許可する場合、バージョン番号のみを比較するこの関数をそのまま適用することはできず、決定論的な競合ルールを定義する必要がある。

安全なクリーンアップには証拠のカバレッジが必要である。関連する古いレプリカが削除を受領したか、除外されて再参加時に完全に再構築される必要がある。オンラインの2つのレプリカの要約が同じだからといって、長期オフラインだった3番目のレプリカが安全であるとは限らない。したがって、修復完了の記録にはレプリカセットと範囲を含める必要があり、クラスター全体の「最近の成功時刻」のみを保存してはならない。

全量、増分、およびノード再構築

これらの名称は製品によって実装が異なる。Cassandraの repair と rebuild は互いに代替不可であり、増分修復が一般的な変更ログの読み取りに等しいわけではない。操作範囲とバージョンの制約については、デプロイ版本文書を参照すること。

継続的な書き込み下で要約比較を行う場合、一貫したデータビューまたはバージョンのタイムスタンプが必要である。そうでないと、ルートの要約を比較した直後にデータが変化し、繰り返しスキャンすることでアクティブな書き込みを修復不可能な差異として誤認する可能性がある。スナップショット、段階的プロトコル、または再検証により、今回のカバレッジの境界を定義する。

強一貫性レプリケーションでも回復と整合性チェックが必要

コンセンサスプロトコルは、自身の故障モデルの下でログの順序とコミットの安全性を維持し、遅延したフォロワーはログの追跡またはスナップショットのインストールを行うことができる。しかし、静かなディスク破損、誤ったソフトウェアの適用、バックアップの破損などは、プロトコルの仮定を超える可能性がある。

したがって、「Raftがあればデータ検証は不要」とは書けない。ログの追跡、状態スナップショット、チェックサム、およびレプリカ間の整合性チェックは、異なるレベルの問題を解決する。修復の成功には、ネットワーク転送の成功だけでなく、論理状態の確認も含まれる必要がある。

修復予算の見積もり

1回のラウンドで600 GBの差異データを転送する必要があり、修復に50 MB/sの継続的な帯域幅を割り当てたとする。転送のみでも理論上約12000秒、つまり3.3時間かかり、要約の構築、ディスク読み取り、マージは含まれていない。2時間以内に完了させる必要がある場合、現在の予算は明らかに不足している。

修復中に10 MB/sの速度で補完が必要な差異が発生し続けると、純粋な解消速度は40 MB/sとなる。すべての量を10進数のGB、MBで計算する：

from fractions import Fraction
backlog = 600 * 10**9
repair_rate, new_difference_rate = 50 * 10**6, 10 * 10**6
assert Fraction(backlog, repair_rate) == 12000
assert Fraction(backlog, repair_rate - new_difference_rate) == 15000
hours = Fraction(15000, 3600)
assert hours == Fraction(25, 6)  # 約4.17時間
required_rate = Fraction(backlog, 2 * 3600) + new_difference_rate
assert required_rate == Fraction(280_000_000, 3)  # 約93.33 MB/s

したがって、2時間でバックログを解消するには、約93.33 MB/sの有効な修復スループットが必要であり、追加のI/Oボトルネックは含まれていない。差異生成速度が修復速度に達するかそれを超えると、この定常状態モデルには有限の追従時間が存在しない。ここで言う「新規差異」は、ターゲットレプリカにまだ到達しておらず、確かに修復が必要なバイトであり、全体のビジネス書き込みスループットをそのまま代用することはできない。更新のカバレッジや圧縮により、実際に転送されるバイト数が変化する可能性がある。

このような見積もりは、並列度、実行ウィンドウ、およびリソースの余裕を逆算するために使用される。修復のレートリミットが低すぎると、新規差異に永遠についていけない。高すぎると、オンラインの読み書きとディスクとネットワークを競合させることになる。スキャン、ストリーミング、マージ、未カバー範囲を個別に監視する必要があり、「タスクが開始された」ことだけを見ていてはならない。

完全な検証では、以下の点に回答する必要がある。：どのトークン/キー範囲がいつ比較されたか、どのレプリカが参加したか、スキップまたは失敗があったか、ツームストーンの保持期限までにカバーできるか、修復後に要約またはバージョンのチェックが通過したか。

参考

• Dynamo：Merkle木に基づくレプリカ同期
• Cassandra：Repair
• Cassandra：Read Repair
• Cassandra：Hints

キーワード：anti-entropy、Merkle tree、収束、tombstone、削除の復活、full repair、incremental repair、rebuild