DOCSEARCHbyAlgolia
Submit Docs

go-ds-crdt

repository·master·Indexed 19 days ago

https://github.com/ipfs/go-ds-crdt

A distributed key-value store implementation using Merkle-CRDTs that satisfies the go-datastore and Batching interfaces. It provides decentralized state synchronization with automatic replication, conflict resolution via a delta-CRDT Add-Wins Observed-Removed set, and resilience against network partitions and corrupted messages. The library requires a go-datastore implementation for permanent storage, a Broadcaster for network updates, and a DAG Syncer implementing the ipld.DAGService interface.

Tokens
8.7K
Snippets
43
Records
49
Agent score
67%

What's inside go-ds-crdt

  1. What is go-ds-crdt?

    master

    go-ds-crdt is a distributed key-value store implementation based on Merkle-CRDTs. It implements the Datastore and Batching interfaces from the go-datastore project.

    Key characteristics include:

    • Automatic Replication: Every key-value pair written to a node is automatically replicated to all other nodes in the network.
    • Decentralized Membership: Nodes can join or leave the network at will without prior notification. The system does not require knowledge of the total number of replicas.
    • Resilience: The system handles dropped, reordered, corrupted, or duplicated network messages, as well as network partitions (which resolve once connectivity is restored).
    • Conflict Resolution: It uses a delta-CRDT Add-Wins Observed-Removed set. The value for a key is determined by the highest priority, where priority is defined by the height of the Merkle-CRDT node where the key was introduced.

  2. Run the GlobalDB CLI

    master

    The GlobalDB CLI can be started with specific options to define the data directory or to run in daemon mode.

    # Standard interactive mode with a custom data directory
./globaldb -datadir /path/to/data

# Daemon mode for continuous operation
./globaldb -daemon -datadir /path/to/data

  3. Requirements for using go-ds-crdt

    master

    To use go-ds-crdt, you must provide or configure three main components:

    1. Permanent Storage: A thread-safe go-datastore implementation. The project recommends using the Pebble implementation.
    2. Broadcaster: A component to broadcast and receive updates between replicas. If using libp2p, you can use libp2p PubSub with the provided PubsubBroadcaster.
    3. DAG Syncer: A component implementing the ipld.DAGService interface to publish and retrieve Merkle DAGs across the network. An example of a compatible service is IPFS-Lite.

  5. Initialize a Merkle-CRDT Datastore

    master

    To create a new replicated key-value store, use the crdt.New function. This requires providing an underlying ds.Datastore for persistence, a ds.Key for the namespace, an ipld.DAGService for managing IPLD nodes, and a Broadcaster for network communication.

    Important Lifecycle Notes:

    • Syncing: If using an asynchronous underlying datastore, you must call Sync() regularly. To ensure all prefixes are synced, it is recommended to call Sync("/") with an empty prefix.
    • Closing: Always call Close() on the crdt.Datastore before closing the underlying persistent store to ensure a clean shutdown and prevent the need for a repair loop on the next startup.
    // Example initialization (requires implementation of Broadcaster and DAGService)
datastore, err := crdt.New(
    underlyingStore, // ds.Datastore
    namespace,        // ds.Key
    dagService,       // ipld.DAGService
    broadcaster,     // crdt.Broadcaster
    nil,             // opts (nil uses DefaultOptions)
)

  6. Example interaction with GlobalDB

    master

    This example demonstrates starting the CLI with a data directory and performing basic key-value operations and peer connection.

    # Start the CLI
./globaldb -datadir /path/to/data

# Interactive session
> put exampleKey exampleValue
> get exampleKey
[exampleKey] -> exampleValue
> list
[exampleKey] -> exampleValue
> connect /ip4/192.168.1.3/tcp/33123/p2p/12D3KooWEkgRTTXGsmFLBembMHxVPDcidJyqFcrqbm9iBE1xhdXq

  7. Configure the Merkle-CRDT Datastore with Options

    master

    Use crdt.DefaultOptions() to get a sensible configuration, or provide a custom crdt.Options struct to crdt.New.

    Key Configuration Fields

    • RebroadcastInterval: How often the system re-publishes its latest heads to new replicas. Default: 1m.
    • PutHook: A function triggered when an element is successfully added (or updated) to the datastore as the prevalent value.
    • DeleteHook: A function triggered when a version of an element is successfully removed. Note: due to concurrent updates, an element might still exist after a DeleteHook is triggered; use Has() to verify presence.
    • NumWorkers: Number of workers for retrieving and merging deltas while walking DAGs. Default: 5.
    • DAGSyncerTimeout: How long to wait for a DAGSyncer to receive a delta. Default: 5m.
    • MaxBatchDeltaSize: Automatically commits batches if the delta size exceeds this value (Default: 1MiB). This prevents DAG nodes from becoming too large for network transfer.
    • RepairInterval: Frequency of walking the full DAG to recover from dirty states. Default: 1h.
    • MultiHeadProcessing: If true, allows multiple new heads to be processed in parallel, increasing throughput but potentially increasing branching.
    • BroadcastBatchDelay: Batches new Head broadcasts to reduce network traffic. A value of 0 disables batching (Default: 0).
    opts := crdt.DefaultOptions()
opts.PutHook = func(k ds.Key, v []byte) {
    // Handle successful put
}
opts.NumWorkers = 10
// Use opts in crdt.New(...)

  8. GlobalDB CLI Commands

    master

    Once the CLI is running, you can use the following interactive commands to manage the distributed CRDT-based database:

    list                 List all items in the store.
get <key>            Retrieve the value for a specified key.
put <key> <value>    Store a value with a specified key.
connect <multiaddr>  Connect to a peer using its multiaddress.
debug <on/off/peers/subs> Enable/disable debug logging, list connected peers, or show pubsub subscribers.
exit                 Quit the CLI.

  10. Handle errors in Merkle-CRDT broadcasting

    master

    When interacting with the Broadcaster, you may encounter crdt.ErrNoMoreBroadcast. This error indicates that the receiving process should abort because no new blocks will be broadcasted.

    // Example error handling
val, err := broadcaster.Next(ctx)
if err == crdt.ErrNoMoreBroadcast {
    // Stop listening
    return
}

  11. Replace a DAG head with a new CID

    master

    Use the Replace method to update a head. This is useful when a new state has been computed and you want to move the pointer from an old head to a new head. If the old and new heads belong to different DAGNames, a warning is logged, but the replacement proceeds.

    If the underlying datastore implements ds.Batching, Replace will use a batch to ensure the deletion of the old head and the addition of the new head are atomic.

    err := heads.Replace(ctx, oldHead, newHead)
if err != nil {
	// handle error
}

  12. Initialize a PubSubBroadcaster

    master

    Use NewPubSubBroadcaster to create a broadcaster that propagates CRDT updates across a network using libp2p-pubsub.

    Important Lifecycle Requirements:

    1. Topic Validation: Register any topic validators before calling NewPubSubBroadcaster.
    2. Shutdown Order: You must cancel the context passed to the broadcaster to shut it down before closing the crdt.Datastore. Failing to do so may cause the application to hang.
    3. Context Cancellation: The broadcaster manages its own cleanup (cancelling subscriptions and closing the topic) automatically when the provided context is cancelled.
    // Example initialization
// ctx: context used for lifecycle management
// psub: an initialized *pubsub.PubSub instance
// topicName: the string name of the pubsub topic

broadcaster, err := crdt.NewPubSubBroadcaster(ctx, psub, "my-crdt-topic")
if err != nil {
    return err
}