crdt

package
v6.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 5, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package crdt provides Conflict-free Replicated Data Types (CRDTs) built on top of the deep patch engine.

The central type is CRDT, a concurrency-safe wrapper around any value of type T. It tracks causal history using a per-field Hybrid Logical Clock (HLC) and resolves concurrent edits with Last-Write-Wins (LWW) semantics.

Basic workflow

  1. Create nodes: nodeA := crdt.NewCRDT(initial, "node-a")
  2. Edit locally: delta := nodeA.Edit(func(v *T) { v.Field = newVal })
  3. Distribute: send delta (JSON-serializable) to peers
  4. Apply remotely: nodeB.ApplyDelta(delta)

For full-state synchronization between two nodes use CRDT.Merge.

Text and sequences

Text is a convergent, ordered sequence of TextRun segments. It supports concurrent insertions and deletions across nodes and is integrated with CRDT directly — no separate registration required.

Document holds the same runs as a Text but keeps them indexed by position, so an edit costs the same whether the document holds a hundred runs or ten thousand. The two serialize identically and converge with each other; reach for a Document when the document is large or the thing being edited, and a Text when it is a small field among others.

List is the same idea for elements of any type: an ordinary slice inside a CRDT is synchronized as one value, so concurrent edits resolve by last-write-wins, whereas insertions and deletions in a List all survive.

Watching for changes

CRDT.OnChange reports each change as it is applied, carrying the operations that took effect and whether they came from a local edit, a peer's delta, or a merge. Only operations that survived conflict resolution are reported, so what arrives describes what actually happened — enough to redraw the parts of a view that moved rather than rebuilding it.

Syncing

CRDT.ApplyDelta carries a change from one replica to another, and CRDT.Merge reconciles two replicas wholesale.

A Document can also sync without exchanging whole documents: Document.StateVector says how much of each writer's output a replica holds, Document.Since returns what a peer holding that is missing, and Document.Apply integrates the result. The cost is then the size of the change rather than the size of the document.

Reclaiming history

A replica remembers when each path was written and removed so that it can recognize a stale update, and a sequence keeps deleted elements as tombstones so a concurrent insertion still has an anchor. Neither shrinks on its own. CRDT.Compact discards both, as far as a watermark the caller supplies — dropping the record is only safe for changes every replica has seen. Types that hold history of their own take part by implementing Compactable.

Collections

How a collection merges depends on how its elements are addressed:

  • Map entries are addressed by key, so concurrent writes to different keys both survive and a write to one key never disturbs another.
  • A slice whose element type carries a deep:"key" tag is addressed by that key, so concurrent edits to different elements both survive, and an element's fields merge independently. Element order is not part of the synchronized state — Diff emits nothing when a keyed slice is merely reordered — so replicas keep these slices in key order. Sort on read, or carry an explicit ordering field, when a particular order matters.
  • A slice with no key tag is synchronized as one value: concurrent edits resolve by last-write-wins, so one writer's version of the whole slice wins. Prefer a map or a keyed slice for anything edited concurrently.
  • List is a sequence that merges: concurrent insertions and deletions from different replicas all survive and every replica agrees on the order. Reach for it when position matters and several writers can edit at once — the case an ordinary slice cannot express.

Every case converges — replicas that have seen the same operations agree — but only the first, second and last merge concurrent edits rather than choosing between them.

Text and List resolve concurrency themselves by implementing Convergent; a type of your own can do the same.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Awareness 

type Awareness[T any] struct {
	// contains filtered or unexported fields
}

Awareness tracks what the other people editing alongside you are doing right now: where their cursor is, what they have selected, the name and colour to draw them in.

It is deliberately not part of the document. A cursor position is not an edit and must not be merged into the text, survive a reload, or appear in the history — and a peer that closes its laptop lid should stop being drawn rather than leave a cursor behind forever. So awareness state is held separately, last-write-wins per peer, and expires when a peer stops saying it is there.

That expiry is what makes it safe to be so simple. Because nothing here is durable, a peer being dropped is always recoverable — it reappears with its next update — and the replicas never need to agree about who left.

An Awareness is safe for concurrent use. Observers are called after the lock is released, so a callback is free to read the awareness, or to write to it.

func NewAwareness 

func NewAwareness[T any](node string, opts ...AwarenessOption[T]) *Awareness[T]

NewAwareness returns an Awareness for the given local node id, which should be the same id the node uses for its edits.

func (*Awareness[T]) Apply 

func (a *Awareness[T]) Apply(u AwarenessUpdate[T]) []PresenceChange[T]

Apply merges an update from a peer and returns what changed.

An entry is taken only if its clock is ahead of what this node already has for that peer, which makes applying the same update twice a no-op and lets updates arrive out of order.

func (*Awareness[T]) Expire 

func (a *Awareness[T]) Expire() []PresenceChange[T]

Expire drops peers that have not been heard from within the timeout and returns what it dropped. States calls it, so most callers never need to — but a client that wants presence to fade without anyone asking can call it on a ticker.

func (*Awareness[T]) Leave 

func (a *Awareness[T]) Leave() AwarenessUpdate[T]

Leave marks this node as gone and returns the update to broadcast, so peers can drop it at once rather than waiting for it to time out.

func (*Awareness[T]) Local 

func (a *Awareness[T]) Local() (T, bool)

Local returns this node's own state.

func (*Awareness[T]) Node 

func (a *Awareness[T]) Node() string

Node returns the local node id.

func (*Awareness[T]) OnChange 

func (a *Awareness[T]) OnChange(fn func(PresenceChange[T])) (cancel func())

OnChange registers fn to be called whenever a peer joins, changes or leaves. The returned function unregisters it.

Callbacks run after the awareness lock is released, so fn may read or write the awareness. They run on the goroutine that caused the change.

func (*Awareness[T]) SetLocal 

func (a *Awareness[T]) SetLocal(state T) AwarenessUpdate[T]

SetLocal records this node's state and returns the update to broadcast.

Call it whenever the state changes, and again periodically even when it has not: an update is also the heartbeat that keeps this node from expiring on its peers.

func (*Awareness[T]) States 

func (a *Awareness[T]) States() map[string]T

States returns the state of every peer currently present, including this node. Peers that have left, or that have not been heard from within the timeout, are not included.

func (*Awareness[T]) Update 

func (a *Awareness[T]) Update() AwarenessUpdate[T]

Update returns the full state of every peer this node knows about, which is what a peer that has just connected needs in order to catch up.

type AwarenessEntry 

type AwarenessEntry[T any] struct {
	Node  string `json:"n"`
	Clock int64  `json:"c"`
	State *T     `json:"s,omitempty"`
}

AwarenessEntry is one peer's state as it travels between replicas. A nil State means the peer has left.

type AwarenessOption 

type AwarenessOption[T any] func(*Awareness[T])

AwarenessOption configures an Awareness.

func WithClock 

func WithClock[T any](now func() time.Time) AwarenessOption[T]

WithClock replaces the source of the current time. It exists so that expiry can be tested without sleeping.

func WithTTL 

func WithTTL[T any](d time.Duration) AwarenessOption[T]

WithTTL sets how long a peer is kept after it was last heard from. The default is 30 seconds, which suits a client announcing itself every few seconds.

type AwarenessUpdate 

type AwarenessUpdate[T any] struct {
	Entries []AwarenessEntry[T] `json:"e,omitempty"`
}

AwarenessUpdate is what a replica broadcasts: one or more peers' states.

func (AwarenessUpdate[T]) IsEmpty 

func (u AwarenessUpdate[T]) IsEmpty() bool

IsEmpty reports whether the update carries nothing.

type CRDT 

type CRDT[T any] struct {
	// contains filtered or unexported fields
}

CRDT represents a Conflict-free Replicated Data Type wrapper around type T.

func NewCRDT 

func NewCRDT[T any](initial T, nodeID string) *CRDT[T]

NewCRDT creates a new CRDT wrapper around a deep copy of initial, so later changes to the caller's value cannot reach inside the replica (and two replicas seeded from one value do not share state).

func (*CRDT[T]) ApplyDelta 

func (c *CRDT[T]) ApplyDelta(delta Delta[T]) bool

ApplyDelta applies a delta from a remote peer using Last-Write-Wins resolution. Returns true if any operations were accepted.

func (*CRDT[T]) Clock 

func (c *CRDT[T]) Clock() *hlc.Clock

Clock returns the internal hybrid logical clock.

func (*CRDT[T]) Compact 

func (c *CRDT[T]) Compact(before hlc.HLC) int

Compact discards bookkeeping for changes at or before before, and reports how many entries it dropped.

A replica remembers when each path was last written and when each was removed, so that it can tell a stale update from a new one. That record only grows: a path written once is remembered for good, and a long-lived replica ends up carrying more history than data — a map emptied of its five hundred keys still holds five hundred entries saying when each went.

Dropping the record is only safe for changes every replica has already seen, because what it protects against is an old update arriving late. Pass the oldest timestamp still in flight anywhere in the system — in practice the minimum, across peers, of the last delta each has acknowledged. Passing something newer risks accepting an update that should have lost, or bringing back a value that was deleted.

Compacting reaches the sequences inside the value too, dropping text and list entries that were deleted before the watermark. It does this to the replica's own copy rather than as an edit: no delta is produced, because what the replica represents does not change.

A replica that has compacted still converges with one that has not: merging with a peer that still remembers restores what was dropped.

func (*CRDT[T]) Edit 

func (c *CRDT[T]) Edit(fn func(*T)) Delta[T]

Edit applies fn to a copy of the current value, computes a delta, advances the local clock, and returns the delta for distribution to peers. Returns an empty Delta if the edit produces no changes.

func (*CRDT[T]) MarshalJSON 

func (c *CRDT[T]) MarshalJSON() ([]byte, error)

func (*CRDT[T]) Merge 

func (c *CRDT[T]) Merge(other *CRDT[T]) bool

Merge performs a full state-based merge with another CRDT node. For each changed field the node with the strictly newer effective timestamp (max of write clock and tombstone) wins. Text fields are always merged convergently via MergeTextRuns, bypassing LWW.

func (*CRDT[T]) NodeID 

func (c *CRDT[T]) NodeID() string

NodeID returns the unique identifier for this CRDT instance.

func (*CRDT[T]) OnChange 

func (c *CRDT[T]) OnChange(fn func(Change[T])) (cancel func())

OnChange registers fn to be called after every change to this replica, and returns a function that unregisters it.

Callbacks run synchronously on the goroutine that made the change, after the replica's lock has been released. A callback may therefore read the replica, and may edit it — an edit from inside a callback announces itself in turn, so guard against recursing forever. A slow callback holds up whoever made the change.

Nothing is serialized on your behalf: changes made from several goroutines deliver on those goroutines, so a callback can run concurrently with itself. Take a lock inside the callback, or hand changes to a single goroutine, if that matters. The Patch in each Change describes that change on its own, so they can be processed independently.

func (*CRDT[T]) Reverse 

func (c *CRDT[T]) Reverse(delta Delta[T]) Delta[T]

Reverse applies the inverse of delta to this node and returns a new Delta representing the undo operation. The returned Delta carries a fresh HLC timestamp so it is causally after the original edit and will be accepted by ApplyDelta on any peer that has already seen the original.

Calling Reverse on the returned Delta produces a redo Delta.

func (*CRDT[T]) UnmarshalJSON 

func (c *CRDT[T]) UnmarshalJSON(data []byte) error

func (*CRDT[T]) View 

func (c *CRDT[T]) View() T

View returns a deep copy of the current value.

type Change 

type Change[T any] struct {
	Patch  deep.Patch[T]
	Source ChangeSource
}

Change describes what actually happened to a replica's value.

Patch holds the operations that were applied, which for a remote delta is only the part that survived conflict resolution — an operation another replica made but this one rejected as stale does not appear. Reading it is how a consumer learns what changed without diffing snapshots: a user interface can redraw just the affected paths.

type ChangeSource 

type ChangeSource int

ChangeSource says where a change came from. 

const (
	// ChangeLocal is an edit made on this replica through [CRDT.Edit].
	ChangeLocal ChangeSource = iota
	// ChangeRemote is a delta from a peer, applied through [CRDT.ApplyDelta].
	ChangeRemote
	// ChangeMerge is the result of [CRDT.Merge] with another replica.
	ChangeMerge
)

func (ChangeSource) String 

func (s ChangeSource) String() string

type Compactable 

type Compactable interface {
	// CompactBefore returns the value with history at or before the watermark
	// discarded. What the value represents must not change.
	CompactBefore(before hlc.HLC) any
}

Compactable is implemented by values that can discard history every replica has already seen — Text and List do. CRDT.Compact calls it on the values it finds inside a replica, so a caller compacts a whole replica in one go rather than reaching into its fields.

type Convergent 

type Convergent interface {
	MergeFrom(other any) any
}

Convergent is implemented by types that resolve concurrent edits themselves rather than being replaced wholesale under last-write-wins. Text and List implement it.

A CRDT applies operations on a Convergent value unconditionally, skipping the clock filter that would discard one side of a concurrent edit, and combines two copies by calling MergeFrom rather than choosing a winner. Implement it to plug a data type of your own into that machinery: 

func (s MySet) MergeFrom(other any) any {
    o, ok := other.(MySet)
    if !ok {
        return s
    }
    return union(s, o)
}

MergeFrom must be commutative, associative and idempotent — merging in any order, any number of times, has to reach the same value — and must return a value of the receiver's own type.

type Counter 

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

Counter is a Positive-Negative Counter CRDT. Each node maintains independent increment and decrement totals; the observed value is sum(Inc) - sum(Dec).

func NewCounter 

func NewCounter(nodeID string) *Counter

NewCounter creates a new Counter for the given nodeID.

func (*Counter) Decrement 

func (c *Counter) Decrement(delta int64)

Decrement adds delta to this node's decrement total. Ignored if delta <= 0.

func (*Counter) Increment 

func (c *Counter) Increment(delta int64)

Increment adds delta to this node's increment total. Ignored if delta <= 0.

func (*Counter) Merge 

func (c *Counter) Merge(other *Counter) bool

Merge merges the state of other into this Counter. Returns true if any changes were applied.

func (*Counter) NodeID 

func (c *Counter) NodeID() string

NodeID returns the node identifier for this Counter.

func (*Counter) Value 

func (c *Counter) Value() int64

Value returns the current counter value: sum(Inc) - sum(Dec).

type DeletedRange 

type DeletedRange struct {
	ID hlc.HLC `json:"id"`
	N  int32   `json:"n"`
}

DeletedRange marks a stretch of characters as removed, without carrying the text itself: a replica being told about the deletion already has the characters, and only needs to hear that they are gone.

type Delta 

type Delta[T any] struct {
	Timestamp hlc.HLC `json:"t"`
	// contains filtered or unexported fields
}

Delta represents a set of changes with a causal timestamp. Obtain a Delta via CRDT.Edit; apply it on remote nodes via CRDT.ApplyDelta.

func (Delta[T]) MarshalJSON 

func (d Delta[T]) MarshalJSON() ([]byte, error)

func (*Delta[T]) UnmarshalJSON 

func (d *Delta[T]) UnmarshalJSON(data []byte) error

type Document 

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

Document is a collaborative text document, holding the same runs as Text but indexed for editing rather than stored as a plain slice.

Text is a value: every edit copies its runs, and finding a position walks them. That is fine for a document edited in a few places, and the cost grows with the number of runs — which grows as edits are scattered around. Document keeps the runs in a balanced tree ordered by position, so finding a position and editing there cost logarithmic rather than linear time in the number of runs, and an edit does not copy the document.

The tree is only an index. Which order the runs are in is decided the same way as in Text, by the run each one is anchored to, so the two agree on the text they describe and serialize to the same thing. A Document can be built from a Text and back again, and a replica running one converges with a replica running the other.

A Document is mutable and is not safe for concurrent use; guard it, or keep it inside a CRDT, which does.

Two documents edited independently are two replicas, and each needs its own clock. What one replica holds of another is described by a single bound per writer, which only means anything if a writer's output goes to one document: two documents drawing from one clock produce interleaved identifiers that no such bound can describe, and a sync between them would appear complete while leaving text behind. Document.Copy shares a clock deliberately, because a copy stands in for the document it came from rather than alongside it.

Which to use

Hold a Document directly for a large collaborative document — a shared editor, a long note — and exchange its runs with peers through MergeFrom. An edit costs the same whether the document holds a hundred runs or ten thousand, where a Text takes proportionally longer as the runs pile up: fifty edits to an eight-thousand-run document take microseconds against tens of milliseconds.

A Document also works inside a CRDT, and converges there, but does not bring its speed with it. CRDT.Edit copies the value and compares the copy to work out what changed, which costs time proportional to the size of the document however small the edit — the wrapper's doing, not the document's, and a Text pays it too. Put a Document in a CRDT when it sits alongside other fields that need the wrapper's conflict resolution; hold it directly when the document is the thing being edited.

func DocumentFromText 

func DocumentFromText(t Text, clock *hlc.Clock) *Document

DocumentFromText returns a document holding the runs of t.

func NewDocument 

func NewDocument(clock *hlc.Clock) *Document

NewDocument returns an empty document that draws identifiers from clock.

func (*Document) Apply 

func (d *Document) Apply(u Update)

Apply integrates an update from a peer.

func (*Document) Clock 

func (d *Document) Clock() *hlc.Clock

Clock returns the clock the document draws identifiers from.

func (*Document) Compact 

func (d *Document) Compact(before hlc.HLC)

Compact drops runs deleted at or before before; see Text.Compact for what the watermark has to be.

func (*Document) CompactBefore 

func (d *Document) CompactBefore(before hlc.HLC) any

CompactBefore implements Compactable.

func (*Document) Copy 

func (d *Document) Copy() (*Document, error)

Copy returns a document holding the same text, independent of this one.

It shares the tree rather than duplicating it, which is safe because no operation ever changes a node: editing builds new nodes along the path it touches and leaves the rest alone, so this document goes on describing what it describes now however much the copy is edited afterwards. A copy is therefore a few words of memory rather than a walk of the whole document, whatever its size.

That is what a CRDT needs. Every edit it makes takes a copy of the value first, so that it has something to compare the result against and can say what changed; copying the document in full made that cost grow with the document rather than with the edit.

This is the method deep.Clone looks for, so it applies wherever a document is copied.

func (*Document) Delete 

func (d *Document) Delete(pos, count int)

Delete removes count visible characters starting at pos. The runs stay as tombstones, which is what a concurrent insertion next to them anchors to.

func (*Document) Diff 

func (d *Document) Diff(other *Document) (*DocumentPatch, error)

Diff reports the change from d to other.

The engine calls this instead of comparing the two documents structurally, which would walk both indexes to describe a change the document can state directly. Working out what other holds that d does not is the same question Document.Since answers for a peer.

func (*Document) Insert 

func (d *Document) Insert(pos int, value string)

Insert places value at pos, counted in visible runes.

func (*Document) Len 

func (d *Document) Len() int

Len returns the number of visible characters, in runes.

func (*Document) MarshalJSON 

func (d *Document) MarshalJSON() ([]byte, error)

MarshalJSON writes the document as its runs, the same shape a Text takes.

func (*Document) Merge 

func (d *Document) Merge(other *Document)

Merge combines other into d.

func (*Document) MergeFrom 

func (d *Document) MergeFrom(other any) any

MergeFrom implements Convergent. Merging combines the two sets of runs and rebuilds the index: an edit has to be fast because it happens on every keystroke, whereas a merge happens once per exchange with a peer and can afford to walk the document. Reusing the ordering MergeTextRuns already implements also means the two types cannot disagree about it.

func (*Document) Patch 

func (d *Document) Patch(p deep.Patch[Document], logger *slog.Logger) error

Patch applies p to d, merging rather than overwriting.

func (*Document) Since 

func (d *Document) Since(sv StateVector) Update

Since returns what a replica holding sv is missing.

A run the peer holds entirely is left out; one it holds part of is trimmed to the part it does not. That is what makes syncing cost the size of the change rather than the size of the document.

func (*Document) StateVector 

func (d *Document) StateVector() StateVector

StateVector returns what this document holds, to hand to a peer so it can work out what to send.

It is maintained as the document changes rather than derived on demand, so asking costs one entry per writer rather than a walk of every run.

func (*Document) String 

func (d *Document) String() string

String returns the visible text.

func (*Document) Text 

func (d *Document) Text() Text

Text returns the document's runs in order, in the form Text uses. This is the document's whole state: it serializes, merges and compacts as a Text.

func (*Document) UnmarshalJSON 

func (d *Document) UnmarshalJSON(data []byte) error

UnmarshalJSON reads runs written by either a Document or a Text.

type DocumentPatch 

type DocumentPatch struct {
	Update Update
}

DocumentPatch is the change from one document to another, expressed as the part the first is missing rather than as the whole of the second.

func (*DocumentPatch) Apply 

func (p *DocumentPatch) Apply(d **Document)

Apply merges the change into the document.

func (*DocumentPatch) FlatOperation 

func (p *DocumentPatch) FlatOperation() (old, new any)

FlatOperation describes the change for a patch's flat operation form. Saying what is missing rather than what the document became is what keeps a delta the size of the edit: a document held inside a CRDT would otherwise put its whole contents into every delta it produced.

type LWW 

type LWW[T any] struct {
	Value     T       `json:"v"`
	Timestamp hlc.HLC `json:"t"`
}

LWW represents a Last-Write-Wins register for type T. Embed LWW fields in a struct to track per-field causality. Use Set to update the value; it accepts the write only if ts is strictly newer.

func (*LWW[T]) Set 

func (l *LWW[T]) Set(v T, ts hlc.HLC) bool

Set updates the register's value and timestamp if ts is after the current timestamp. Returns true if the update was accepted.

type List 

type List[T any] []ListEntry[T]

List is a convergent sequence of T.

An ordinary slice inside a CRDT is synchronized as a single value, so concurrent edits resolve by last-write-wins and one writer's version of the whole slice wins. A List merges instead: concurrent insertions and deletions from different replicas all survive, and every replica converges on the same order.

Elements are placed relative to their neighbours rather than by index, and concurrent insertions at the same position are ordered by ID, so replicas agree regardless of the order updates arrive in. Use it for collaborative ordered data — a task list being reordered by several people, a document outline, a playlist.

The zero List is empty and ready to use. Like Text, a List is a value: Insert and Delete return a new List rather than mutating the receiver.

func MergeLists 

func MergeLists[T any](a, b List[T]) List[T]

MergeLists combines two Lists into one containing every element either side has seen, in the order both sides agree on. A deletion on either side wins, since a tombstone records an element that was removed rather than never seen.

func NewList 

func NewList[T any](clock *hlc.Clock, values ...T) List[T]

NewList returns a List holding values, in order, with IDs drawn from clock.

func (List[T]) At 

func (l List[T]) At(pos int) (T, bool)

At returns the live element at pos, and whether pos is in range.

func (List[T]) Compact 

func (l List[T]) Compact(before hlc.HLC) List[T]

Compact drops entries deleted at or before before, returning the rest.

A deletion leaves the entry behind as a tombstone, because a replica that has not heard about it may still insert next to what was deleted and needs something to anchor to. Dropping one is only safe once every replica has seen the deletion, so before must be older than anything still in flight; see CRDT.Compact, which takes the same watermark.

An entry that something is still anchored to is kept whatever its age: removing it would leave whatever follows it without a place, and re-anchoring would change this replica's order without changing anyone else's.

func (List[T]) CompactBefore 

func (l List[T]) CompactBefore(before hlc.HLC) any

CompactBefore implements Compactable.

func (List[T]) Delete 

func (l List[T]) Delete(pos, count int) List[T]

Delete removes count elements starting at pos, counted over live elements. Removed entries are kept as tombstones so concurrent insertions that named them still have an anchor.

func (List[T]) Diff 

func (l List[T]) Diff(other List[T]) deep.Patch[List[T]]

Diff reports the change from l to other as a single whole-value operation: the receiving side merges rather than overwrites, so the operation carries the sequence and lets MergeLists work out the result.

func (List[T]) Insert 

func (l List[T]) Insert(pos int, value T, clock *hlc.Clock) List[T]

Insert places value at pos, counted over live elements. A pos of 0 inserts at the head; a pos at or past the end appends.

func (List[T]) Items 

func (l List[T]) Items() []T

Items returns the live elements in order.

func (List[T]) Len 

func (l List[T]) Len() int

Len returns the number of live elements.

func (List[T]) MergeFrom 

func (l List[T]) MergeFrom(other any) any

MergeFrom implements Convergent, so a List merges with a peer's copy instead of one replacing the other under last-write-wins.

func (*List[T]) Patch 

func (l *List[T]) Patch(p deep.Patch[List[T]], logger *slog.Logger) error

Patch applies p to l, merging rather than overwriting.

type ListEntry 

type ListEntry[T any] struct {
	// ID identifies this element for all time. It is assigned once, by the
	// replica that inserted the element, and never reused. The deep:"key" tag
	// makes a List nested inside a struct diff and merge entry by entry rather
	// than as one opaque value.
	ID hlc.HLC `deep:"key" json:"id"`
	// Prev is the ID of the element this one was inserted after — the zero
	// value for an insertion at the head. Position is expressed relative to a
	// neighbour rather than as an index, so it survives concurrent edits that
	// shift indices around.
	Prev hlc.HLC `json:"p,omitempty"`
	// Deleted marks a removed element. The entry stays as a tombstone so that a
	// concurrent insertion that named it as Prev still has somewhere to attach.
	Deleted bool `json:"d,omitempty"`
	Value   T    `json:"v"`
}

ListEntry is one element of a List together with the metadata that places it in the sequence.

type Map 

type Map[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Map is a distributed LWW key-value map CRDT built on top of CRDT.

Concurrent writes to the same key are resolved by Last-Write-Wins: the write with the strictly higher HLC timestamp wins. Deletions remove the key from the map and record a tombstone timestamp, so a delete with a newer timestamp wins over an older set, and a set with a newer timestamp wins over an older delete.

func NewMap 

func NewMap[K comparable, V any](nodeID string) *Map[K, V]

NewMap returns an empty Map CRDT bound to the given node ID.

func (*Map[K, V]) Contains 

func (m *Map[K, V]) Contains(key K) bool

Contains reports whether key exists in the map.

func (*Map[K, V]) Delete 

func (m *Map[K, V]) Delete(key K)

Delete removes key from the map. It is a no-op if the key does not exist.

func (*Map[K, V]) Get 

func (m *Map[K, V]) Get(key K) (V, bool)

Get returns the value for key and true if the key exists. It returns the zero value and false otherwise.

func (*Map[K, V]) Keys 

func (m *Map[K, V]) Keys() []K

Keys returns a slice of all live keys. The order is non-deterministic.

func (*Map[K, V]) Len 

func (m *Map[K, V]) Len() int

Len returns the number of entries in the map.

func (*Map[K, V]) Merge 

func (m *Map[K, V]) Merge(other *Map[K, V]) bool

Merge performs a full state-based LWW merge with another Map node. Returns true if the local state changed.

func (*Map[K, V]) NodeID 

func (m *Map[K, V]) NodeID() string

NodeID returns the unique identifier for this Map instance.

func (*Map[K, V]) Set 

func (m *Map[K, V]) Set(key K, value V)

Set sets key to value.

type PresenceChange 

type PresenceChange[T any] struct {
	Node  string
	Kind  PresenceKind
	State T // the peer's new state; the zero value when it left
}

PresenceChange describes one peer's arrival, update or departure.

type PresenceKind 

type PresenceKind int

PresenceKind is what happened to a peer. 

const (
	// PresenceJoined is a peer heard from for the first time.
	PresenceJoined PresenceKind = iota
	// PresenceUpdated is a peer whose state changed.
	PresenceUpdated
	// PresenceLeft is a peer that said goodbye or fell silent past the
	// timeout.
	PresenceLeft
)

func (PresenceKind) String 

func (k PresenceKind) String() string

type Set 

type Set[T comparable] struct {
	// contains filtered or unexported fields
}

Set is an Add-Wins Observed-Remove Set (OR-Set) CRDT built on top of CRDT.

Each Add creates a uniquely-tagged entry using the node's HLC. Remove only tombstones entries that exist at call time; a concurrent Add from another node produces a different tag, so after Merge the element is still present (add wins over remove).

func NewSet 

func NewSet[T comparable](nodeID string) *Set[T]

NewSet returns an empty Set CRDT bound to the given node ID.

func (*Set[T]) Add 

func (s *Set[T]) Add(elem T)

Add appends a new uniquely-tagged entry for elem. The tag is the current HLC timestamp serialised as a string map key.

Two clock ticks occur per Add: one to mint the entry's tag and a second inside the underlying Edit for the resulting Delta's Timestamp. Both values are monotonic per the HLC mutex, so the extra tick is harmless; keeping the tag-mint outside Edit means the tag is fixed before the inner closure runs, which keeps the data-flow easy to follow.

func (*Set[T]) Contains 

func (s *Set[T]) Contains(elem T) bool

Contains reports whether elem has at least one live (non-deleted) entry.

func (*Set[T]) Items 

func (s *Set[T]) Items() []T

Items returns a deduplicated slice of all live elements.

func (*Set[T]) Len 

func (s *Set[T]) Len() int

Len returns the number of distinct live elements.

Cost is O(n) in the number of entries (live + tombstoned) because OR-Set duplicates require dedup; the prior implementation built a full slice via Items just to take its length, which this avoids.

func (*Set[T]) Merge 

func (s *Set[T]) Merge(other *Set[T]) bool

Merge performs a full state-based OR-Set merge with another Set node. Returns true if the local state changed.

func (*Set[T]) NodeID 

func (s *Set[T]) NodeID() string

NodeID returns the unique identifier for this Set instance.

func (*Set[T]) Remove 

func (s *Set[T]) Remove(elem T)

Remove marks all non-deleted entries whose Elem equals elem as deleted. Only entries visible at call time are tombstoned; concurrent adds on other nodes create entries with different tags that this Remove never sees.

type StateVector 

type StateVector map[string]int32

StateVector says how much of each origin's output a replica already holds.

Every character a replica writes is identified by the node that wrote it and a counter that only goes up, so "everything node A wrote up to 500" describes a replica's knowledge of A completely, however that text was later split, merged or moved. A state vector is one such bound per origin, which is enough for a peer to work out exactly what to send back — see Document.Since.

It is small: one entry per node that has ever written, not per character.

func (StateVector) Includes 

func (sv StateVector) Includes(run TextRun) bool

Includes reports whether the state vector accounts for every character of the run.

func (StateVector) MarshalBinary 

func (sv StateVector) MarshalBinary() ([]byte, error)

MarshalBinary encodes a state vector: one entry per node, with the counter each has been seen up to.

func (*StateVector) UnmarshalBinary 

func (sv *StateVector) UnmarshalBinary(data []byte) error

UnmarshalBinary decodes a state vector encoded by MarshalBinary.

type Text 

type Text []TextRun

Text represents a CRDT-friendly text structure using runs.

A Text held by this package is always stored in document order. Every operation here either preserves that order or restores it: Insert splices a run into the position the ordering would give it, Delete only flips flags, and MergeTextRuns — the one place runs arrive from elsewhere — derives the order from scratch. Operations can therefore read the runs directly instead of rebuilding the ordering tree, which is what keeps editing from costing more as a document grows.

func MergeTextRuns 

func MergeTextRuns(a, b Text) Text

MergeTextRuns merges two Text states into a single convergent state.

func (Text) Compact 

func (t Text) Compact(before hlc.HLC) Text

Compact drops deleted runs that were removed at or before before, returning the remaining text.

Deleting text does not reclaim it: the run stays as a tombstone, because a replica that has not heard about the deletion may still insert next to what was deleted, and the tombstone is what that insertion attaches to. A document edited for long enough is mostly tombstones — two hundred typed-then-deleted phrases leave two hundred runs behind nine visible characters.

Dropping one is only safe once every replica has seen the deletion, so before must be a timestamp older than anything still in flight anywhere in the system; see CRDT.Compact, which takes the same watermark for a replica's own bookkeeping.

A tombstone that something is still anchored to is kept regardless of its age. Removing it would leave the runs anchored to it without a place, and re-anchoring them would change how this replica orders the document without changing how any other replica orders it — the two would then disagree. Keeping it costs a run and preserves the order exactly, and it becomes collectable once whatever anchored to it is itself deleted and compacted.

func (Text) CompactBefore 

func (t Text) CompactBefore(before hlc.HLC) any

CompactBefore implements Compactable.

func (Text) Delete 

func (t Text) Delete(pos, length int) Text

Delete removes length characters starting at pos.

func (Text) Diff 

func (t Text) Diff(other Text) deep.Patch[Text]

Diff compares t with other and returns a Patch.

func (Text) Insert 

func (t Text) Insert(pos int, value string, clock *hlc.Clock) Text

Insert inserts a string at the given character position.

func (Text) Len 

func (t Text) Len() int

Len returns the number of visible characters, counted in runes — the same unit Insert and Delete take positions in.

func (Text) MergeFrom 

func (t Text) MergeFrom(other any) any

MergeFrom implements Convergent, so a Text merges with a peer's copy instead of one replacing the other under last-write-wins.

func (*Text) Patch 

func (t *Text) Patch(p deep.Patch[Text], logger *slog.Logger) error

Patch applies p to t.

func (Text) String 

func (t Text) String() string

String returns the full text content, skipping deleted runs.

type TextRun 

type TextRun struct {
	ID    hlc.HLC `deep:"key" json:"id"`
	Value string  `json:"v"`
	Prev  hlc.HLC `json:"p,omitempty"`
	// N is the number of runes in Value. Positions and identifiers are counted
	// in runes, so nearly every operation needs this number, and counting it
	// means walking the string — which for a document held as one long run
	// means walking the whole document, once per operation. Carrying the count
	// makes those operations depend on the number of runs instead of the length
	// of the text. A run without one (an older document, or a literal built by
	// hand) is counted on demand.
	N       int32 `json:"n,omitempty"`
	Deleted bool  `json:"d,omitempty"`
}

TextRun represents a contiguous run of characters with a unique starting ID.

type Update 

type Update struct {
	Runs    Text           `json:"r,omitempty"`
	Deleted []DeletedRange `json:"d,omitempty"`
}

Update is what one replica sends another: the text the other does not have, and which characters have been deleted.

Runs carries only what is missing, which after the first exchange is usually just what has been typed since. Deleted carries every deletion the sender knows about rather than only recent ones, because a deletion is not itself timestamped — it is a flag on text that already exists. The set stays small, since it holds one entry per deleted stretch rather than per character, and Document.Compact discards the ones every replica has seen.

func (Update) IsEmpty 

func (u Update) IsEmpty() bool

IsEmpty reports whether the update carries nothing.

func (Update) MarshalBinary 

func (u Update) MarshalBinary() ([]byte, error)

MarshalBinary encodes the update in the compact format described above.

func (*Update) UnmarshalBinary 

func (u *Update) UnmarshalBinary(data []byte) error

UnmarshalBinary decodes an update encoded by MarshalBinary.

Source Files

View all Source files

Directories

Path Synopsis
Package hlc implements a Hybrid Logical Clock (HLC) for distributed causality tracking.
 Package hlc implements a Hybrid Logical Clock (HLC) for distributed causality tracking.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.