raft-snapshot-compaction-20260729

Answer

Below is the complete implementation of Raft log compaction with InstallSnapshot RPC.

Core Data Structures

The implementation introduces a Snapshot type with metadata (LastIncludedIndex, LastIncludedTerm, Config) and raw state-machine data. For transmission, snapshots are split into SnapshotChunk objects with offset tracking and CRC-32 verification:

// SnapshotMeta holds metadata about a snapshot.
type SnapshotMeta struct {
    LastIncludedIndex uint64
    LastIncludedTerm  uint64
    Config            []byte  // cluster configuration at snapshot time
    Size              int
}

// InstallSnapshotArgs for the RPC with chunking support.
type InstallSnapshotArgs struct {
    Term              uint64
    LeaderID          int
    LastIncludedIndex uint64
    LastIncludedTerm  uint64
    Config            []byte   // membership config at snapshot boundary
    Offset            uint64   // byte offset of this chunk
    Data              []byte   // raw bytes of this chunk
    Done              bool     // true iff this is the last chunk
    Crc32             uint32   // CRC-32 IEEE of Data
    BoundCRC          uint32   // rolling CRC of all data up to end of this chunk
    SnapshotSize      uint64   // total snapshot size (only on Done=true)
}

Leader Side: Detecting Behind Followers

On every replication attempt, the leader checks if a follower's nextIndex has fallen behind the snapshot boundary:

func (rn *RaftNode) sendAppendEntriesToPeer(peer int, entriesHint int) {
    prevNextIdx := rn.nextIndex[peer]
    snapLastIdx := rn.lastSnapshotMeta.LastIncludedIndex

    // Follower too far behind → send InstallSnapshot instead
    if prevNextIdx <= snapLastIdx && rn.lastSnapshot != nil {
        rn.sendInstallSnapshotToPeer(peer)
        return
    }
    // ... normal AppendEntries path ...
}

Leader Side: Chunking with CRC Verification

The leader creates the snapshot, then splits it into fixed-size chunks (512 KiB each). Each chunk carries: - Offset for assembly order - Crc32 per-chunk checksum - BoundCRC rolling CRC of all data from offset 0 through this chunk

func (rn *RaftNode) chunkSnapshot(snap *Snapshot) []InstallSnapshotArgs {
    data := snap.Data
    totalSize := uint64(len(data))
    var chunks []InstallSnapshotArgs
    offset := uint64(0)
    var boundCRC uint32

    for offset < totalSize {
        end := offset + SnapshotChunkSize
        if end > totalSize { end = totalSize }
        chunkData := data[offset:end]
        chunkCRC := crc32.ChecksumIEEE(chunkData)
        boundCRC = crc32.Update(boundCRC, crc32.IEEETable, chunkData)

        chunks = append(chunks, InstallSnapshotArgs{
            Term: rn.currentTerm, Offset: offset, Data: chunkData,
            Done: end >= totalSize, Crc32: chunkCRC,
            BoundCRC: boundCRC, SnapshotSize: totalSize,
            LastIncludedIndex: snap.Meta.LastIncludedIndex,
            LastIncludedTerm:  snap.Meta.LastIncludedTerm,
            Config:            snap.Meta.Config,
        })
        offset = end
    }
    return chunks
}

Follower Side: Atomic Snapshot Application

The follower accumulates chunks, verifies CRCs at each step, and applies the snapshot atomically on receipt of the final chunk:

func (rn *RaftNode) handleInstallSnapshot(args InstallSnapshotArgs) InstallSnapshotResult {
    // Step 1: Term check and leader recognition
    if args.Term < rn.currentTerm { return result }

    // Step 2: Initialize pending state on offset 0
    if args.Offset == 0 {
        rn.pendingSnapshot = make([]byte, 0, args.SnapshotSize)
        rn.pendingBoundCRC = 0
    }

    // Step 3: Verify per-chunk CRC
    chunkCRC := crc32.ChecksumIEEE(args.Data)
    if chunkCRC != args.Crc32 {
        rn.pendingSnapshot = nil  // abort
        return result
    }

    // Step 4: Accumulate data and verify rolling CRC
    rn.pendingSnapshot = append(rn.pendingSnapshot, args.Data...)
    rn.pendingBoundCRC = crc32.Update(rn.pendingBoundCRC, crc32.IEEETable, args.Data)
    if rn.pendingBoundCRC != args.BoundCRC {
        rn.pendingSnapshot = nil  // abort
        return result
    }

    // Step 5: When Done, apply the snapshot atomically
    if args.Done {
        rn.applySnapshot(rn.pendingSnapshot, meta)
        // This restores state machine, truncates log,
        // updates commitIndex/lastApplied, and installs cluster config
    }
}

The applySnapshot method atomically: 1. Restores the state machine from the snapshot data 2. Installs the cluster configuration from the snapshot 3. Truncates the log to the snapshot boundary 4. Advances commitIndex and lastApplied

Membership-Change Integration

Every snapshot includes the cluster configuration at its boundary. Membership-change entries (EntryMembership) are regular log entries that update the configuration when applied. Snapshots capture the latest committed config, ensuring new followers joining via snapshot get the correct membership.