Opensource internals

ClickHouse Keeper Carries a Tree Digest in Every Raft Log Entry

14 min read Jay Hubbert

Raft guarantees that replicas agree on the log, not on the state machine, so clickhouse-keeper appends a digest of its whole node tree to each log entry and terminates any replica whose tree disagrees.

TL;DR: Raft replicates a log, and assumes every replica applies that log to the same state. That assumption is load-bearing and almost never true in practice — a different library version, a compiler difference, or one non-deterministic line of code and the replicas silently diverge. clickhouse-keeper closes the gap by putting a digest of its entire node tree inside the log entry itself. Every replica recomputes the digest after applying the entry and compares it with the leader’s. Mismatch is fatal, and the process aborts rather than serving divergent data.

ClickHouse ships its own ZooKeeper-compatible server, clickhouse-keeper. It speaks the ZooKeeper wire protocol, so the ordinary ClickHouse ZooKeeper client talks to it unchanged — but the consensus underneath is NuRaft, a Raft implementation in C++, not the original Java one.

That choice brings with it Raft’s central promise and Raft’s central hole.

The hole in Raft

Raft’s correctness argument goes like this:

  1. The leader appends a log entry.
  2. The leader replicates that entry to a quorum of followers.
  3. The leader commits it once a quorum has it.
  4. Every replica applies committed entries in order.

Every replica therefore has the same log. From that, Raft concludes every replica has the same state. That conclusion is only valid if applying a log entry is a deterministic function of (state, entry). In practice:

  • A std::unordered_map iterated in a different order on a different binary.
  • A floating-point operation compiled differently with different optimisation flags.
  • A time() call, a random number, or a locale-dependent comparison.
  • A different version of a library on one node.

None of these break Raft. They break the state machine. The log is still identical everywhere; the trees are not. And ZooKeeper’s contract is extremely tight — two clients connected to two different replicas must see the same tree, or the whole thing is broken in ways that are very hard to debug.

So ClickHouse added a check. And the check is the interesting part.

The idea: put the expected state in the log

The naive way to detect divergence is to compare the replicas’ trees periodically — but that is O(tree size) and requires coordination. The cleverer way is to make the leader publish its expectation as part of the data it is already replicating.

That is exactly what Keeper does. When the leader is about to append an entry to the log, it:

  1. Applies the request to its own in-memory tree (the “preprocess” step).
  2. Computes the digest of the resulting tree.
  3. Writes (zxid, digest_version, digest_value) into the log entry’s header.
cpp 
// src/Coordination/KeeperServer.cpp:1289
            case nuraft::cb_func::PreAppendLogLeader:
            {
                // we are relying on the fact that request are being processed under a mutex
                // and not a RW lock
                auto & entry = *static_cast<LogEntryPtr *>(param->ctx);

                chassert(entry->get_val_type() == nuraft::app_log);
                auto next_zxid = state_machine->getNextZxid();

                auto entry_buf = entry->get_buf_ptr();

                KeeperStateMachine::ZooKeeperLogSerializationVersion serialization_version = {};
                size_t request_end_position = 0;
                auto request_for_session = state_machine->parseRequest(*entry_buf, /*final=*/false, &serialization_version, &request_end_position);
                request_for_session->zxid = next_zxid;
                auto digest_after_preprocessing = state_machine->preprocess(*request_for_session, /*lock_mutex=*/ true);
                if (!digest_after_preprocessing)
                    return nuraft::cb_func::ReturnCode::ReturnNull;

                request_for_session->digest = *digest_after_preprocessing;

and then, at the end of the entry buffer:

cpp 
// src/Coordination/KeeperServer.cpp:1336
                WriteBufferFromPointer write_buf(buffer_start, write_buffer_header_size);

                if (serialization_version < KeeperStateMachine::ZooKeeperLogSerializationVersion::WITH_TIME)
                    writeIntBinary(request_for_session->time, write_buf);

                writeIntBinary(request_for_session->zxid, write_buf);
                writeIntBinary(static_cast<uint8_t>(request_for_session->digest->version), write_buf);
                if (request_for_session->digest->version != KeeperDigestVersion::NO_DIGEST)
                    writeIntBinary(request_for_session->digest->value, write_buf);

                write_buf.finalize();

The digest is now part of the replicated log. Every follower receives it for free — no extra round trip, no extra message type, no coordinator.

Note the version byte. The digest algorithm has changed over the years (new node fields were added to it), so the version travels with the value. A follower that does not understand the version knows it cannot compare, and the check degrades to “skip”.

cpp 
// src/Coordination/KeeperCommon.h:38
enum class KeeperDigestVersion : uint8_t
{
    NO_DIGEST = 0,
    V1 = 1,
    V2 = 2, // added system nodes that modify the digest on startup so digest from V0 is invalid
    V3 = 3, // fixed bug with casting, removed duplicate czxid usage
    V4 = 4, // 0 is not a valid digest value
    V5 = 5  // added TTL fields (destroy_time and ttl) to the node digest
};

The comment on V3 — “fixed bug with casting” — is a nice reminder that this is a hash of a hash input, and any change to the input invalidates every previously computed digest.

What the digest actually is

The digest is not a Merkle tree. It is much simpler: a SipHash-64 over each node’s content, summed over the whole tree.

cpp 
// src/Coordination/KeeperCommon.cpp:153
/// When this function is updated, update KEEPER_CURRENT_DIGEST_VERSION!!
uint64_t KeeperNodeStats::calculateDigest(std::string_view path, std::string_view data) const
{
    /// Must match calculateDigest in KeeperStorage.cpp (KEEPER_CURRENT_DIGEST_VERSION).
    SipHash hash;

    hash.update(path);
    if (!data.empty())
        hash.update(data);

    hash.update(czxid);
    hash.update(mzxid);
    hash.update(getCTime());
    hash.update(mtime);
    hash.update(version);
    hash.update(cversion);
    hash.update(aversion);
    hash.update(getEphemeralOwner()); // covers EPHEMERAL and CONTAINER flags
    hash.update(getNumChildren());
    hash.update(pzxid);

    hash.update(isTTL());
    if (isTTL())
        hash.update(getTTL());

    /// TODO: Hash seq num (or replace getEphemeralOwner(), getCTime(), getTTL() above with plain ephemeral_or_seq_num_or_ttl and ctime_and_flags).

    uint64_t digest = hash.get64();

    /// 0 means no calculated digest, it's not a valid digest value.
    if (digest == 0)
        digest = 1;

    return digest;
}

Every field that distinguishes one node from another is in the hash: the path, the data, the creation and modification zxids, the version counters, the ephemeral owner, the child count, the parent zxid, and the TTL.

The TODO is honest — the sequential number is not hashed, and the comment explains why that is a known hole: getEphemeralOwner(), getCTime() and getTTL() are overloaded fields, so the same integer can mean different things for different node kinds.

And the “0 is not a valid digest value” guard is the kind of detail that saves you at 3am: 0 is used as the sentinel for “no digest”, so the real digest must never be zero.

Why a sum?

Because the digest has to be maintained incrementally. Every node’s hash can be cached, and the tree-wide digest is the sum of the cached hashes. A create adds one node’s hash; a delete subtracts it:

cpp 
// src/Coordination/KeeperMemNodesStorage.cpp:807
void KeeperMemNodesStorage::updateNodesDigest(uint64_t & current_digest, uint64_t for_zxid) const
{
    if (!keeper_context->digestEnabled())
        return;

    auto nodes_it = uncommitted_zxid_to_nodes.find(for_zxid);
    if (nodes_it == uncommitted_zxid_to_nodes.end())
        return;

    for (const auto node_it : nodes_it->second)
    {
        const auto & [path, uncommitted_node] = *node_it;
        if (uncommitted_node.node)
        {
            uncommitted_node.node->invalidateDigestCache();
            current_digest += uncommitted_node.node->getDigest(path);
        }
    }
}

and on commit:

cpp 
// src/Coordination/KeeperStorage.cpp:655
        if (!delta.path.empty())
            nodes_storage->commitDelta(delta, digest);

        if (result != Coordination::Error::ZOK)
        {
            /// Only ErrorDelta can produce error on commit.
            /// ErrorDelta should always be emitted alone (by `preprocess` on rejected request).
            /// If there could be any Deltas after an ErrorDelta, we'd have to do either commit
            /// or rollback on them here.
            chassert(std::distance(deltas.begin(), deltas.end()) == 1);

            return result;
        }
    }

    nodes_digest += digest_change;

The sum is commutative, which is exactly what you want: the order in which nodes are added does not matter, only the set. A Merkle tree would have given a stronger guarantee (it would localise which node diverged), but the sum is O(changed nodes) instead of O(tree).

The check

The check happens twice per entry: once after the preprocess (the speculative apply) and once after the commit (the real apply).

cpp 
// src/Coordination/KeeperStateMachine.cpp:1093
    if (keeper_context->digestEnabled() && request_for_session.digest)
        assertDigest(
            *request_for_session.digest,
            digest_after_preprocessing,
            *request_for_session.request,
            request_for_session.log_idx,
            request_for_session.session_id,
            false);

and:

cpp 
// src/Coordination/KeeperStateMachine.cpp:1260
                if (keeper_context->digestEnabled() && request_for_session->digest)
                    assertDigest(
                        *request_for_session->digest,
                        storage->getNodesDigest(true, /*lock_transaction_mutex=*/true),
                        *request_for_session->request,
                        request_for_session->log_idx,
                        request_for_session->session_id,
                        true);

The preprocess check catches a divergence before the entry is committed — the follower refuses to advance at all. The commit check catches a divergence introduced by the commit path itself.

And the failure mode is deliberately violent:

cpp 
// src/Coordination/KeeperStateMachine.cpp:764
void assertDigest(
    const KeeperDigest & expected,
    const KeeperDigest & actual,
    const Coordination::ZooKeeperRequest & request,
    uint64_t log_idx,
    uint64_t session_id,
    bool committing)
{
    if (!KeeperStorage::checkDigest(expected, actual))
    {
        LOG_FATAL(
            getLogger("KeeperStateMachine"),
            "Digest for nodes is not matching after {} request of type '{}' at log index {} for session {}.\nExpected digest - {}, actual digest - {} "
            "(digest {}). Keeper will terminate to avoid inconsistencies.\nExtra information about the request:\n{}",
            committing ? "committing" : "preprocessing",
            request.getOpNum(),
            log_idx,
            session_id,
            expected.value,
            actual.value,
            expected.version,
            request.toString());
        std::terminate();
    }
}

std::terminate(), not an exception, not a degraded mode. The reasoning is that a Keeper that has diverged is dangerous: it will happily answer get requests with a tree that is different from its peers’, and the resulting client-side confusion is far worse than an outage. So the process dies, loudly, with the offending request printed, and a human comes and looks at it.

The digest is off by default in the server config:

cpp 
// src/Coordination/KeeperContext.cpp:111
    digest_enabled = config.getBool("keeper_server.digest_enabled", false);
    digest_enabled_on_commit = config.getBool("keeper_server.digest_enabled_on_commit", false);

digest_enabled_on_commit is a separate, more expensive mode: it recomputes the digest from scratch on every commit rather than maintaining it incrementally, catching bugs in the incremental maintenance itself.

Snapshots

The digest is the divergence detector; snapshots are how a replica catches up without replaying the entire log.

A Keeper snapshot is a serialization of the entire tree — nodes, data, ACLs, sessions — plus the metadata needed to resume:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:243
void KeeperStorageSnapshot::serialize(const KeeperStorageSnapshot & snapshot, WriteBuffer & out, KeeperContextPtr keeper_context)
{
    writeBinary(static_cast<uint8_t>(snapshot.version), out);
    serializeSnapshotMetadata(snapshot.snapshot_meta, out);

    if (snapshot.version >= SnapshotVersion::V5)
    {
        writeBinary(snapshot.zxid, out);
        if (keeper_context->digestEnabled())
        {
            writeBinary(static_cast<uint8_t>(KEEPER_CURRENT_DIGEST_VERSION), out);
            writeBinary(snapshot.nodes_digest, out);
        }
        else
            writeBinary(static_cast<uint8_t>(KeeperDigestVersion::NO_DIGEST), out);
    }

The snapshot carries the digest too (since V5), so a replica that loads a snapshot can check that the snapshot’s tree matches what the sender claimed.

Determinism

Snapshots must be byte-identical across replicas, or two replicas that snapshot at the same log index would produce different files and the transfer machinery would have no way to compare them. Two comments in the serializer say so explicitly:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:262
    /// Better to sort before serialization, otherwise snapshots can be different on different replicas
    std::vector<std::pair<ACLId, Coordination::ACLs>> sorted_acl_map(snapshot.acl_map.begin(), snapshot.acl_map.end());
    ::sort(sorted_acl_map.begin(), sorted_acl_map.end());

and, for sessions:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:306
    /// Session must be saved in a sorted order,
    /// otherwise snapshots will be different
    std::vector<std::pair<int64_t, int64_t>> sorted_session_and_timeout(
        snapshot.session_and_timeout.begin(), snapshot.session_and_timeout.end());
    ::sort(sorted_session_and_timeout.begin(), sorted_session_and_timeout.end());

The node tree itself is written in path order (the storage iterates it sorted), so that part is deterministic for free.

The format is versioned, with a version per feature:

cpp 
// src/Coordination/KeeperSnapshotManager.h:34
enum SnapshotVersion : uint8_t
{
    V0 = 0,
    V1 = 1, /// with ACL map
    V2 = 2, /// with 64 bit buffer header
    V3 = 3, /// compress snapshots with ZSTD codec
    V4 = 4, /// add Node size to snapshots
    V5 = 5, /// add ZXID and digest to snapshots
    V6 = 6, /// remove is_sequential, per node size, data length
    V7 = 7, /// acl_id narrowed from uint64_t to uint32_t, seq_num widened from int32_t to int64_t
    V8 = 8, /// add destroy_time and ttl for TTL nodes
    V9 = 9, /// container nodes (ephemeralOwner = INT64_MIN sentinel)
};

static constexpr auto MAX_SUPPORTED_SNAPSHOT_VERSION = SnapshotVersion::V9;

A newer Keeper writes V9; an older one that cannot understand V9 will refuse rather than misparse:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:1281
    uint8_t version = 0;
    readBinary(version, *in);
    if (version > static_cast<uint8_t>(MAX_SUPPORTED_SNAPSHOT_VERSION))
        throw Exception(ErrorCodes::UNKNOWN_FORMAT_VERSION, "Unsupported snapshot version {}", version);

Compression

Snapshots are compressed. Two formats are supported — the legacy custom LZ4 frame and Zstandard — and the reader detects which one it has by sniffing the magic number:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:759
bool KeeperSnapshotManager::isZstdCompressed(nuraft::ptr<nuraft::buffer> buffer)
{
    static constexpr unsigned char ZSTD_COMPRESSED_MAGIC[4] = {0x28, 0xB5, 0x2F, 0xFD};

    ReadBufferFromNuraftBuffer reader(buffer);
    unsigned char magic_from_buffer[4]{};
    reader.readStrict(reinterpret_cast<char *>(&magic_from_buffer), sizeof(magic_from_buffer));
    buffer->pos(0);
    return memcmp(magic_from_buffer, ZSTD_COMPRESSED_MAGIC, 4) == 0;
}

0x28 0xB5 0x2F 0xFD is the Zstandard frame magic. This matters because the setting compress_snapshots_with_zstd_format (default true) can be flipped on a live cluster, and old snapshots must still be readable.

On-disk layout

Snapshot files are named after the log index they cover, plus a random component so that two concurrent writes of the same index cannot collide:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:70
    std::string getSnapshotFileName(uint64_t up_to_log_idx, bool compress_zstd, uint64_t recovery_generation = 0)
    {
        /// Unique-from-birth name avoids collisions between concurrent same-index writes.
        auto base = recovery_generation
            ? fmt::format("snapshot_{}_recovered_{}_{:016x}.bin", up_to_log_idx, recovery_generation, thread_local_rng())
            : fmt::format("snapshot_{}_{:016x}.bin", up_to_log_idx, thread_local_rng());
        if (compress_zstd)
            base += ".zstd";
        return base;
    }

A partially-written snapshot is detected by a tmp_ marker file. Both the marker and the real file are removed if the marker is found on startup:

cpp 
// src/Coordination/KeeperSnapshotManager.cpp:435
        std::vector<std::string> snapshot_files;
        for (auto it = disk->iterateDirectory(""); it->isValid(); it->next())
        {
            if (it->name().starts_with(tmp_keeper_file_prefix))
            {
                incomplete_files.emplace(it->name().substr(tmp_keeper_file_prefix.size()), it->path());
                continue;
            }

            if (it->name().starts_with("snapshot_") && !clean_incomplete_file(it->path()))
                snapshot_files.push_back(it->path());
        }

The number of snapshots kept on disk is snapshots_to_keep, default 3.

Snapshot identity

Raft’s snapshot installation is a delicate operation: the follower must replace its entire state with the leader’s, and the leader must know which snapshot it is installing. The identity of a snapshot is (last_log_idx, last_log_term):

cpp 
// src/Coordination/KeeperStateMachine.cpp:101
/// Snapshot identity is (last_log_idx, last_log_term); Raft log matching guarantees one per committed prefix.
static bool sameSnapshotIdentity(const nuraft::snapshot & lhs, const nuraft::snapshot & rhs)

The term matters: two snapshots at the same log index but different terms are different snapshots (the leader may have overwritten its own log entry). NuRaft’s own snapshot object only holds Raft metadata, so ClickHouse clones it to keep the richer metadata separately:

cpp 
// src/Coordination/KeeperStateMachine.cpp:94
/// nuraft::snapshot holds only Raft metadata (last_log_idx, last_log_term, size, cluster_config).
static nuraft::ptr<nuraft::snapshot> cloneSnapshotMeta(nuraft::snapshot & s)

and the apply_snapshot path fails closed if the snapshot it is asked to apply is not the one it was told to expect:

cpp 
// src/Coordination/KeeperStateMachine.cpp:1334
            if (s.get_last_log_idx() < pending_snapshot_to_apply->get_last_log_idx())
            {
                /// A newer install's save displaced this pending context (NuRaft drops its lock
                /// between save and apply). Skip-as-success is sound only when local commits
                /// already cover the snapshot; otherwise fail closed — NuRaft exits the process
                /// and `init` recovers from the displacing install, which is fully saved on disk.
                const auto last_committed = keeper_context->lastCommittedIndex();
                if (last_committed >= s.get_last_log_idx())

The log itself

Underneath all of this is the changelog: Keeper’s own Raft log storage, separate from the tree.

cpp 
// src/Coordination/KeeperLogStore.cpp:60
uint64_t KeeperLogStore::append(nuraft::ptr<nuraft::log_entry> & entry)
{
    ProfiledExclusiveLock lock(changelog_lock, ProfileEvents::KeeperChangelogLockWaitMicroseconds);
    uint64_t idx = changelog.getNextEntryIndex();
    changelog.appendEntry(idx, entry);
    return idx;
}

The changelog is compressed and rotated:

cpp 
// src/Coordination/Changelog.h:193
struct LogFileSettings
{
    bool force_sync = true;
    bool compress_logs = true;
    uint64_t rotate_interval = 100000;
    uint64_t max_size = 0;
    uint64_t overallocate_size = 0;
    uint64_t latest_logs_cache_size_threshold = 0;
    uint64_t latest_logs_cache_entry_count_threshold = 0;
    /// 0 = automatically use the number of CPU cores (resolved by Changelog's constructor).
    uint64_t startup_read_max_streams = 0;
    uint64_t startup_read_buffer_size = 8 * 1024 * 1024;
};

rotate_interval = 100000 entries per file, force_sync = true so an acknowledged write survives power loss, and up to 8 MiB of read buffer per startup read stream.

The log is compacted once a snapshot covers it — compact(last_log_index) drops everything below the snapshot point — but a configurable number of entries is kept:

cpp 
// src/Coordination/CoordinationSettings.cpp:41
    DECLARE(UInt64, reserved_log_items, 100000, "How many log items to store (don't remove during compaction)", 0) \

Trade-offs

What the design buys:

  • Divergence is detected, not discovered later. The check runs on every entry, so a divergence is caught at the moment it happens, with the offending request printed.
  • Zero extra round trips. The digest rides along in the log entry that was being replicated anyway.
  • Deterministic snapshots. Sorted ACLs and sessions mean two replicas that snapshot at the same index produce byte-identical files.
  • Fail-stop rather than fail-silent. A diverged Keeper kills itself.

What it costs:

  • The digest is off by default. keeper_server.digest_enabled defaults to false. You have to opt in.
  • A sum, not a Merkle tree. The digest tells you that the trees differ, not where. Debugging a divergence means bisecting the log.
  • The digest is not cryptographically strong. SipHash-64 is a non-cryptographic hash; it is fast and good against accidental divergence, which is the threat model, but it is not a defence against a malicious replica.
  • The sequential number is not hashed. The TODO in calculateDigest documents this: a node’s seq_num is excluded, so two trees that differ only in sequential numbering would hash the same.
  • Terminating is drastic. A false positive (a bug in the digest maintenance, or a version mismatch during a rolling upgrade) kills a production node. That is why the version byte and the “no digest” value exist, and why the digest is opt-in.
  • Snapshots are O(tree). A million-node tree means a million-node snapshot, written to disk and potentially shipped over the network to a lagging follower.

Wrapping up

The one thing to remember: Keeper’s divergence check is a digest embedded in the replicated log. Raft replicates the log; by putting the expected state inside the log, the expected state gets replicated too, for free.

The part people get wrong: the digest is not on by default. If you run a multi-node Keeper and want to know that your replicas agree, you have to set keeper_server.digest_enabled = true yourself. And when you do, a mismatch kills the process — that is the design, not a bug.

What to go do about it: turn on keeper_server.digest_enabled on a staging Keeper cluster and watch KeeperCommits in system.events. If the cluster is healthy you will see commits proceed; if a node’s binary differs from its peers’, the first divergent entry will abort it, and the LOG_FATAL message will name the request, the log index, and both digests.

ClickhouseKeeperRaftZookeeperInternals