crdt

Documentation

Overview

Package crdt implements CRDT-based distributed replication for the ProofGraph. Since the ProofGraph is append-only (no updates or deletes), a G-Set (Grow-Only Set) CRDT is sufficient. Nodes are content-addressed by their SHA-256 hash, making merge a simple set union with no conflict resolution needed.

Index

• Constants
• type DistributedStore
  • func NewDistributedStore() *DistributedStore
  • func NewDistributedStoreWithGSet(gset *GSet) *DistributedStore
  • func (d *DistributedStore) GSet() *GSet
  • func (d *DistributedStore) GetChain(_ context.Context, nodeID string) ([]*proofgraph.Node, error)
  • func (d *DistributedStore) GetNode(_ context.Context, id string) (*proofgraph.Node, error)
  • func (d *DistributedStore) GetNodesByType(_ context.Context, kind proofgraph.NodeType, fromLamport, toLamport uint64) ([]*proofgraph.Node, error)
  • func (d *DistributedStore) GetRange(_ context.Context, fromLamport, toLamport uint64) ([]*proofgraph.Node, error)
  • func (d *DistributedStore) StoreNode(_ context.Context, node *proofgraph.Node) error
• type GSet
  • func NewGSet() *GSet
  • func (g *GSet) Add(node *proofgraph.Node) error
  • func (g *GSet) All() []*proofgraph.Node
  • func (g *GSet) Contains(nodeHash string) bool
  • func (g *GSet) Delta(otherHashes map[string]bool) []*proofgraph.Node
  • func (g *GSet) Get(nodeHash string) (*proofgraph.Node, bool)
  • func (g *GSet) Hashes() map[string]bool
  • func (g *GSet) Len() int
  • func (g *GSet) Merge(other *GSet) []*proofgraph.Node
• type MemoryTransport
  • func NewMemoryTransport(nodeID string) *MemoryTransport
  • func (t *MemoryTransport) Connect(peer *MemoryTransport)
  • func (t *MemoryTransport) Disconnect(peer *MemoryTransport)
  • func (t *MemoryTransport) Peers() []string
  • func (t *MemoryTransport) Receive(ctx context.Context) (*SyncMessage, error)
  • func (t *MemoryTransport) Send(ctx context.Context, peerID string, msg *SyncMessage) error
• type SyncMessage
• type SyncMessageType
• type SyncProtocol
  • func NewSyncProtocol(nodeID string, gset *GSet, transport Transport) *SyncProtocol
  • func (s *SyncProtocol) HandleMessage(ctx context.Context, msg *SyncMessage) (*SyncMessage, error)
  • func (s *SyncProtocol) Start(ctx context.Context) error
  • func (s *SyncProtocol) Stop()
  • func (s *SyncProtocol) WithClock(clock func() time.Time) *SyncProtocol
  • func (s *SyncProtocol) WithInterval(d time.Duration) *SyncProtocol
• type Transport

Constants

const DefaultSyncInterval = 5 * time.Second

DefaultSyncInterval is the default gossip tick interval.

Types

type DistributedStore

type DistributedStore struct {
	// contains filtered or unexported fields
}

DistributedStore wraps a GSet and implements the proofgraph.Store interface. It provides a distributed, eventually-consistent store for ProofGraph nodes.

func NewDistributedStore

func NewDistributedStore() *DistributedStore

NewDistributedStore creates a new DistributedStore backed by a GSet.

func NewDistributedStoreWithGSet

func NewDistributedStoreWithGSet(gset *GSet) *DistributedStore

NewDistributedStoreWithGSet creates a DistributedStore sharing an existing GSet. This allows the store and sync protocol to operate on the same data.

func (*DistributedStore) GSet

func (d *DistributedStore) GSet() *GSet

GSet returns the underlying GSet for use with the sync protocol.

func (*DistributedStore) GetChain

func (d *DistributedStore) GetChain(_ context.Context, nodeID string) ([]*proofgraph.Node, error)

GetChain retrieves the chain of nodes from a given node ID back to genesis via a breadth-first traversal of parent links.

func (*DistributedStore) GetNode

func (d *DistributedStore) GetNode(_ context.Context, id string) (*proofgraph.Node, error)

GetNode retrieves a node by ID (hash).

func (*DistributedStore) GetNodesByType

func (d *DistributedStore) GetNodesByType(_ context.Context, kind proofgraph.NodeType, fromLamport, toLamport uint64) ([]*proofgraph.Node, error)

GetNodesByType retrieves all nodes of a given type within a Lamport range.

func (*DistributedStore) GetRange

func (d *DistributedStore) GetRange(_ context.Context, fromLamport, toLamport uint64) ([]*proofgraph.Node, error)

GetRange retrieves nodes in a Lamport clock range.

func (*DistributedStore) StoreNode

func (d *DistributedStore) StoreNode(_ context.Context, node *proofgraph.Node) error

StoreNode persists a single node into the GSet.

type GSet

type GSet struct {
	// contains filtered or unexported fields
}

GSet implements a Grow-Only Set CRDT for ProofGraph nodes. Nodes can only be added, never removed or updated. This provides eventual consistency across replicas without coordination.

func NewGSet

func NewGSet() *GSet

NewGSet creates an empty G-Set.

func (*GSet) Add

func (g *GSet) Add(node *proofgraph.Node) error

Add inserts a node into the set. Idempotent -- adding an existing node is a no-op. Returns an error if the node's hash does not match its computed hash.

func (*GSet) All

func (g *GSet) All() []*proofgraph.Node

All returns all nodes as a snapshot.

func (*GSet) Contains

func (g *GSet) Contains(nodeHash string) bool

Contains checks if a node hash is in the set.

func (*GSet) Delta

func (g *GSet) Delta(otherHashes map[string]bool) []*proofgraph.Node

Delta returns nodes in this set that are NOT in the other set (identified by hash). Used for efficient sync: send only what the peer does not have.

func (*GSet) Get

func (g *GSet) Get(nodeHash string) (*proofgraph.Node, bool)

Get retrieves a node by hash.

func (*GSet) Hashes

func (g *GSet) Hashes() map[string]bool

Hashes returns all node hashes as a snapshot.

func (*GSet) Len

func (g *GSet) Len() int

Len returns the number of nodes.

func (*GSet) Merge

func (g *GSet) Merge(other *GSet) []*proofgraph.Node

Merge combines another GSet into this one (union). Returns nodes that were new. Thread-safe on both sets.

type MemoryTransport

type MemoryTransport struct {
	// contains filtered or unexported fields
}

MemoryTransport is an in-process transport for testing multi-node sync. Each MemoryTransport instance represents one node in the cluster.

func NewMemoryTransport

func NewMemoryTransport(nodeID string) *MemoryTransport

NewMemoryTransport creates a new in-memory transport with the given node ID.

func (*MemoryTransport) Connect

func (t *MemoryTransport) Connect(peer *MemoryTransport)

Connect establishes a bidirectional link between this transport and a peer.

func (*MemoryTransport) Disconnect

func (t *MemoryTransport) Disconnect(peer *MemoryTransport)

Disconnect removes the bidirectional link between this transport and a peer. Useful for simulating network partitions.

func (*MemoryTransport) Peers

func (t *MemoryTransport) Peers() []string

Peers returns the IDs of all connected peers.

func (*MemoryTransport) Receive

func (t *MemoryTransport) Receive(ctx context.Context) (*SyncMessage, error)

Receive blocks until a message arrives or the context is cancelled.

func (*MemoryTransport) Send

func (t *MemoryTransport) Send(ctx context.Context, peerID string, msg *SyncMessage) error

Send delivers a message to a specific peer's inbox.

type SyncMessage

type SyncMessage struct {
	SenderID    string             `json:"sender_id"`
	MessageType SyncMessageType    `json:"message_type"`
	Hashes      []string           `json:"hashes,omitempty"` // for DIGEST and PULL messages
	Nodes       []*proofgraph.Node `json:"nodes,omitempty"`  // for PUSH messages
	Timestamp   time.Time          `json:"timestamp"`
}

SyncMessage is exchanged between nodes during gossip.

type SyncMessageType

type SyncMessageType string

SyncMessageType identifies the kind of gossip message.

const (
	// SyncDigest is sent to advertise which nodes the sender has (hash list).
	SyncDigest SyncMessageType = "DIGEST"
	// SyncPush carries nodes the receiver is missing.
	SyncPush SyncMessageType = "PUSH"
	// SyncPull requests nodes the sender is missing (hash list).
	SyncPull SyncMessageType = "PULL"
)

type SyncProtocol

type SyncProtocol struct {
	// contains filtered or unexported fields
}

SyncProtocol manages gossip-based ProofGraph synchronization.

func NewSyncProtocol

func NewSyncProtocol(nodeID string, gset *GSet, transport Transport) *SyncProtocol

NewSyncProtocol creates a new gossip sync protocol.

func (*SyncProtocol) HandleMessage

func (s *SyncProtocol) HandleMessage(ctx context.Context, msg *SyncMessage) (*SyncMessage, error)

HandleMessage processes an incoming sync message and returns a response (or nil).

func (*SyncProtocol) Start

func (s *SyncProtocol) Start(ctx context.Context) error

Start begins the gossip loop. It runs two goroutines: 1. A ticker that periodically sends digests to a random peer. 2. A receiver that processes incoming messages. Start blocks until ctx is cancelled or Stop is called.

func (*SyncProtocol) Stop

func (s *SyncProtocol) Stop()

Stop gracefully shuts down the gossip loop.

func (*SyncProtocol) WithClock

func (s *SyncProtocol) WithClock(clock func() time.Time) *SyncProtocol

WithClock overrides the timestamp source (useful for testing).

func (*SyncProtocol) WithInterval

func (s *SyncProtocol) WithInterval(d time.Duration) *SyncProtocol

WithInterval overrides the default gossip interval.

type Transport

type Transport interface {
	// Send delivers a message to a specific peer.
	Send(ctx context.Context, peerID string, msg *SyncMessage) error
	// Receive blocks until a message arrives or the context is cancelled.
	Receive(ctx context.Context) (*SyncMessage, error)
	// Peers returns the IDs of all known peers.
	Peers() []string
}

Transport is the network layer for gossip messages.