bitx_kv

    Git-backed KV store with CRDT sync (extension module for mizchi/bit)

    git
    kv
    sync
    Download zip
    Author
    Version
    0.48.0
    License
    Apache-2.0
    Last updated
    9 days ago
    Downloads
    40

    #GitDb - Git-based Distributed KV Store

    A distributed key-value store built on Git primitives with gossip protocol synchronization.

    #Architecture

    ┌─────────────────────────────────────────────────────────┐ │ GitDb │ ├─────────────────────────────────────────────────────────┤ │ Hierarchical KV Store │ │ - Keys: "/users/alice/profile" → Git tree path │ │ - Values: Bytes → Git blob │ ├─────────────────────────────────────────────────────────┤ │ GitFs (Copy-on-Write Layer) │ │ - Read: Git tree/blob traversal with caching │ │ - Write: In-memory working layer │ │ - Snapshot: Creates Git commits │ ├─────────────────────────────────────────────────────────┤ │ Gossip Protocol │ │ - VectorClock: Causal ordering │ │ - Announce: Broadcast HEAD state │ │ - Sync: Exchange Git objects │ │ - Merge: LWW / KeepBoth / Custom strategies │ └─────────────────────────────────────────────────────────┘

    #Features

    • Hierarchical Keys: Keys are mapped to Git tree structure (e.g., /users/alice/profile)
    • Content-Addressable: Values stored as Git blobs with SHA-1 hashing
    • Versioned: Full history via Git commits
    • Distributed: P2P sync via gossip protocol
    • Conflict Resolution: Last-Write-Wins, KeepBoth, or custom merge strategies
    • Offline-First: Work offline, sync later

    #API

    #KV Operations

    let db = GitDb::empty(NodeId::new("node1"), "/repo/.git")

    // Set/Get
    db.set("/users/alice/name", b"Alice")
    let name = db.get(fs, "/users/alice/name") // Some(b"Alice")

    // List keys
    let users = db.list(fs, "/users") // ["alice", "bob"]

    // Delete
    db.delete("/users/alice/name")

    #Versioning

    // Commit changes
    let commit_id = db.commit(fs, fs, "Add user alice", timestamp)

    // Rollback uncommitted changes
    db.rollback()

    #Gossip Sync

    // Get current state for broadcasting
    let state = db.get_gossip_state(timestamp)

    // Handle incoming gossip message
    let response = db.handle_gossip(fs, message, timestamp)

    // Select random peers for gossip
    let peers = db.select_gossip_peers(3, seed)

    // Merge with remote state
    let result = db.merge(fs, fs, their_head, their_clock, LastWriteWins, timestamp)

    #Design Decisions

    #Why Not Raft?

    Git's Merkle DAG with gossip provides eventual consistency, which is sufficient for many use cases:

    ApproachConsistencyComplexityUse Case
    RaftStrong (CP)High (leader election, log replication)Financial, config management
    Git + GossipEventualLow (push/pull + conflict resolution)Code, data, config

    This follows the same approach as Noms and Dolt.

    #Vector Clocks

    Vector clocks track causal relationships between updates:
    • Each node maintains a logical clock
    • On update: increment own clock
    • On sync: merge clocks, take max for each node
    • Enables detection of concurrent updates for conflict resolution

    #Merge Strategies

    • LastWriteWins: Take the value with the higher vector clock (or timestamp as tie-breaker)
    • KeepBoth: Create conflict markers (like Git merge conflicts)
    • Custom: User-provided resolver function

    #Intended Use Case: Cloudflare Workers

    // Durable Object export class GitDbNode { private db: GitDb; async fetch(request: Request) { const msg = await request.json() as GossipMessage; const response = this.db.handle_gossip(msg, Date.now()); return Response.json(response); } // Periodic gossip with random peers async alarm() { const peers = this.db.select_gossip_peers(3, Date.now()); for (const peer of peers) { const state = this.db.get_gossip_state(Date.now()); await fetch(peer.endpoint, { method: 'POST', body: JSON.stringify({ type: 'Announce', state }) }); } } }

    #Prior Art

    • Noms - Versioned, forkable, syncable database (Go, archived)
    • Dolt - Git for Data, SQL database with Git semantics (Go, active)
    • OrbitDB - Peer-to-peer database for IPFS (JavaScript)

    #Status

    Experimental. Part of the src/x-* experimental modules.

    GitObject

    pub(all) struct GitObject {
    id :
    ObjectId

    obj_type :
    ObjectType

    data : Bytes
    }

    Serialized git object for transfer

    GossipMessage

    pub(all) enum GossipMessage {
    Announce(GossipState)
    WantObjects(Array[
    ObjectId
    ])
    HaveObjects(Array[GitObject])
    SyncRequest(NodeId,
    ObjectId
    )
    SyncResponse(Array[GitObject],
    ObjectId
    )
    }

    Message types for gossip protocol

    GossipState

    pub(all) struct GossipState {
    node_id : NodeId
    head :
    ObjectId

    clock : VectorClock
    timestamp : Int64
    }

    State of a node in the gossip network
    pub struct Kv {
    node_id : NodeId
    tree : &
    WorkingTree

    store : &
    ObjectStore

    clock : VectorClock
    head :
    ObjectId

    sync_engine : SyncEngine
    pending_objects : Map[String, GitObject]
    snapshot_cache : Map[Int, SerializedSnapshot]
    snapshot_bytes_cache : Map[Int, Bytes]
    }

    Main Kv structure

    Kv::add_peer

    fn Kv::add_peer(self : Kv, peer : PeerInfo) -> Unit

    Add or update a peer

    Kv::apply_pending_objects

    fn Kv::apply_pending_objects(self : Kv) -> Int raise
    GitError

    Apply pending objects from sync Returns number of objects applied

    Kv::clock

    fn Kv::clock(self : Kv) -> VectorClock

    Get the current vector clock

    Kv::commit

    fn Kv::commit(self : Kv, message : String, timestamp : Int64) ->
    ObjectId
    raise
    GitError

    Commit current changes and update HEAD

    Kv::delete

    fn Kv::delete(self : Kv, key : String) -> Unit

    Delete a key

    Kv::get

    fn Kv::get(self : Kv, key : String) -> Bytes?

    Get value by hierarchical key Key format: "/path/to/key" or "path/to/key"

    Kv::get_gossip_state

    fn Kv::get_gossip_state(self : Kv, timestamp : Int64) -> GossipState

    Get current gossip state

    Kv::get_peer_cursor

    fn Kv::get_peer_cursor(self : Kv, node_id : NodeId) -> Int64?

    Get per-peer sync cursor

    Kv::get_peers

    fn Kv::get_peers(self : Kv) -> Array[PeerInfo]

    Get all peers

    Kv::handle_gossip

    fn Kv::handle_gossip(self : Kv, msg : GossipMessage, timestamp : Int64) -> GossipMessage?

    Handle incoming gossip message Returns optional response message

    Kv::has

    fn Kv::has(self : Kv, key : String) -> Bool

    Check if key exists

    Kv::has_pending_objects

    fn Kv::has_pending_objects(self : Kv) -> Bool

    Check if we have pending objects to apply

    Kv::head

    Get the current HEAD

    Kv::is_dirty

    fn Kv::is_dirty(self : Kv) -> Bool

    Check if there are uncommitted changes

    Kv::list

    fn Kv::list(self : Kv, prefix : String) -> Array[String]

    List keys under a prefix (directory)

    Kv::list_recursive

    fn Kv::list_recursive(self : Kv, prefix : String) -> Array[String]

    List all keys recursively under a prefix

    Kv::merge

    fn Kv::merge(self : Kv, their_head :
    ObjectId
    , their_clock : VectorClock, strategy : MergeStrategy, timestamp : Int64) -> MergeResult raise
    GitError

    Perform a three-way merge

    Kv::node_id

    fn Kv::node_id(self : Kv) -> NodeId

    Get the node ID

    Kv::pending_object_count

    fn Kv::pending_object_count(self : Kv) -> Int

    Get count of pending objects

    Kv::record_peer_sync_result

    fn Kv::record_peer_sync_result(self : Kv, node_id : NodeId, success : Bool, timestamp : Int64) -> Unit

    Record sync result to update retry state

    Kv::remove_peer

    fn Kv::remove_peer(self : Kv, node_id : NodeId) -> Bool

    Remove a peer

    Kv::rollback

    fn Kv::rollback(self : Kv) -> Unit

    Discard uncommitted changes

    Kv::select_gossip_peers

    fn Kv::select_gossip_peers(self : Kv, count : Int, seed : Int64, timestamp? : Int64) -> Array[PeerInfo]

    Select random peers for gossip (anti-entropy)

    Kv::serialize_snapshot

    fn Kv::serialize_snapshot(self : Kv, mode? : SnapshotMode) -> SerializedSnapshot

    Create a serialized snapshot from Kv state. Reuses cached value while the tree is clean and head/clock are unchanged.

    Kv::serialize_snapshot_bytes

    fn Kv::serialize_snapshot_bytes(self : Kv, mode? : SnapshotMode) -> Bytes

    Serialize snapshot directly to bytes. Reuses cached bytes while the tree is clean and head/clock are unchanged.

    Kv::set

    fn Kv::set(self : Kv, key : String, value : Bytes) -> Unit

    Set value by hierarchical key

    Kv::set_peer_cursor

    fn Kv::set_peer_cursor(self : Kv, node_id : NodeId, cursor : Int64) -> Unit

    Set per-peer sync cursor

    Kv::set_string

    fn Kv::set_string(self : Kv, key : String, value : String) -> Unit

    Set string value

    Kv::sync_round

    fn Kv::sync_round(self : Kv, count : Int, seed : Int64, timestamp : Int64, exchange : (PeerInfo, GossipMessage) -> Result[GossipMessage?, String]) -> SyncRoundResult

    Run one anti-entropy round against selected peers via injected transport. The exchange callback sends one message to a peer and returns an optional response.

    KvConfig

    pub(all) struct KvConfig {
    node_id : NodeId
    max_peers : Int
    gossip_interval_ms : Int
    sync_batch_size : Int
    }

    Configuration for Kv

    KvConfig::default

    fn KvConfig::default(node_id : NodeId) -> KvConfig

    MergeResult

    pub(all) enum MergeResult {
    NoOp
    FastForward(
    ObjectId
    )
    Merged(
    ObjectId
    , Array[String])
    Conflict(Array[String])
    }

    Merge result

    MergeStrategy

    pub(all) enum MergeStrategy {
    LastWriteWins
    KeepBoth
    Custom((Bytes, Bytes) -> Bytes)
    }

    Merge strategy for conflicts

    NodeId

    pub(all) struct NodeId {
    id : String
    }

    Node identifier in the gossip network
    impl Eq for NodeId
    impl Hash for NodeId
    impl Show for NodeId

    NodeId::new

    fn NodeId::new(id : String) -> NodeId

    PeerInfo

    pub(all) struct PeerInfo {
    node_id : NodeId
    last_seen : Int64
    head :
    ObjectId

    endpoint : String
    }

    Peer information

    PeerSyncState

    pub(all) struct PeerSyncState {
    peer : PeerInfo
    cursor : Int64?
    last_sync_at : Int64?
    last_attempt_at : Int64?
    retry_count : Int
    next_retry_at : Int64
    }

    SerializationStats

    pub(all) struct SerializationStats {
    total_bytes : Int
    object_count : Int
    commit_count : Int
    tree_count : Int
    blob_count : Int
    blob_bytes : Int
    }

    Statistics about serialized data

    SerializedObject

    pub(all) struct SerializedObject {
    id : Bytes
    obj_type : Int
    data : Bytes
    }

    Serialized git object

    SerializedSnapshot

    pub(all) struct SerializedSnapshot {
    version : Int
    node_id : String
    head : Bytes
    clock : Array[(String, Int64)]
    objects : Array[SerializedObject]
    }

    Serialized snapshot for cold storage

    SerializedSnapshot::byte_size

    fn SerializedSnapshot::byte_size(self : SerializedSnapshot) -> Int

    Calculate total serialized size

    SerializedSnapshot::from_bytes

    fn SerializedSnapshot::from_bytes(data : Bytes) -> SerializedSnapshot

    Deserialize snapshot from bytes - optimized with slice operations

    SerializedSnapshot::stats

    SerializedSnapshot::to_bytes

    fn SerializedSnapshot::to_bytes(self : SerializedSnapshot) -> Bytes

    Serialize snapshot to bytes (simple binary format) Format: 4 bytes: version (little-endian) 4 bytes: node_id length N bytes: node_id 20 bytes: head 4 bytes: clock entry count For each clock entry: 4 bytes: key length N bytes: key 8 bytes: value (little-endian) 4 bytes: object count For each object: 20 bytes: id 1 byte: type 4 bytes: data length N bytes: data

    SnapshotMode

    pub(all) enum SnapshotMode {
    FullHistory
    HeadState
    TreeAndBlobOnly
    }

    Controls how many git objects are included in a snapshot.

    SyncEngine

    pub(all) struct SyncEngine {
    config : SyncEngineConfig
    peers : Map[String, PeerSyncState]
    }

    SyncEngine::get_cursor

    fn SyncEngine::get_cursor(self : SyncEngine, node_id : NodeId) -> Int64?

    SyncEngine::get_peers

    fn SyncEngine::get_peers(self : SyncEngine) -> Array[PeerInfo]

    SyncEngine::new

    fn SyncEngine::new(config? : SyncEngineConfig) -> SyncEngine

    SyncEngine::record_sync_result

    fn SyncEngine::record_sync_result(self : SyncEngine, node_id : NodeId, success : Bool, timestamp : Int64) -> Unit

    SyncEngine::remove_peer

    fn SyncEngine::remove_peer(self : SyncEngine, node_id : NodeId) -> Bool

    SyncEngine::select_peers

    fn SyncEngine::select_peers(self : SyncEngine, count : Int, seed : Int64, timestamp : Int64) -> Array[PeerInfo]

    SyncEngine::set_cursor

    fn SyncEngine::set_cursor(self : SyncEngine, node_id : NodeId, cursor : Int64) -> Unit

    SyncEngine::upsert_peer

    fn SyncEngine::upsert_peer(self : SyncEngine, peer : PeerInfo) -> Unit

    SyncEngineConfig

    pub(all) struct SyncEngineConfig {
    anti_entropy_interval_ms : Int64
    retry_base_ms : Int64
    retry_max_ms : Int64
    }

    Configuration for sync peer selection and retry

    SyncEngineConfig::default

    SyncResult

    pub(all) struct SyncResult {
    objects_sent : Int
    objects_received : Int
    conflicts : Array[String]
    new_head :
    ObjectId

    }

    Result of a sync operation

    SyncRoundResult

    pub(all) struct SyncRoundResult {
    selected : Int
    attempted : Int
    succeeded : Int
    failed : Int
    messages_sent : Int
    messages_received : Int
    }

    Result of one anti-entropy sync round

    VectorClock

    pub(all) struct VectorClock {
    clocks : Map[String, Int64]
    }

    Vector clock for causal ordering

    VectorClock::compare

    fn VectorClock::compare(self : VectorClock, other : VectorClock) -> Int

    Compare vector clocks: -1 = before, 0 = concurrent, 1 = after

    VectorClock::increment

    fn VectorClock::increment(self : VectorClock, node : NodeId) -> VectorClock

    VectorClock::merge

    fn VectorClock::merge(self : VectorClock, other : VectorClock) -> VectorClock

    VectorClock::new