Designing a Conflict-Free Replicated Data Type (CRDT) for Collaborative Editing in Go — code snippets

snippet-1.go

package crdt

import (
	"errors"
	"sync"
)

// ID uniquely identifies an item in the CRDT space.
// Using compact numeric types minimizes heap allocation size and GC overhead.
type ID struct {
	ClientID uint64 // Globally unique ID of the client
	Counter  uint32 // Monotonically increasing clock (Lamport timestamp)
}

// Item represents a block of text in the document.
// Using a slice of bytes allows grouping contiguous insertions into a single block,
// significantly reducing pointer overhead and memory fragmentation.
type Item struct {
	ID      ID
	Origin  *ID   // ID of the item immediately to the left when this item was created
	Value   []byte
	Deleted bool
	
	// Pointers for fast in-memory document traversal
	Left  *Item
	Right *Item
}

// Doc represents the collaborative document.
type Doc struct {
	mu      sync.RWMutex
	Start   *Item
	Items   map[ID]*Item
	Clients map[uint64]uint32 // State vector (ClientID -> Max Counter seen)
}

// NewDoc initializes a Doc with a sentinel start block.
func NewDoc() *Doc {
	start := &Item{
		ID: ID{ClientID: 0, Counter: 0},
	}
	d := &Doc{
		Start:   start,
		Items:   make(map[ID]*Item),
		Clients: make(map[uint64]uint32),
	}
	d.Items[start.ID] = start
	return d
}

snippet-2.go

// Integrate inserts a remotely generated Item into the local document structure.
// It resolves concurrent conflicts deterministically without requiring a central coordinator.
func (d *Doc) Integrate(item *Item) error {
	d.mu.Lock()
	defer d.mu.Unlock()

	// Check if already integrated to prevent duplicate work
	if _, exists := d.Items[item.ID]; exists {
		return nil
	}

	// Retrieve the origin item (the left neighbor at creation time)
	var origin *Item
	if item.Origin != nil {
		var exists bool
		origin, exists = d.Items[*item.Origin]
		if !exists {
			return errors.New("missing causal dependency: origin not found")
		}
	} else {
		origin = d.Start
	}

	// Conflict resolution loop (RGA/YATA variant)
	// We scan right from the origin to find the correct placement position
	dest := origin
	for dest.Right != nil {
		next := dest.Right
		
		// Case 1: Next item has the same origin. We compare ClientIDs to enforce total order.
		if next.Origin != nil && item.Origin != nil && *next.Origin == *item.Origin {
			if next.ID.ClientID < item.ID.ClientID {
				dest = next
				continue
			} else {
				break
			}
		}
		
		// Case 2: Next item was inserted concurrently but its origin is further downstream.
		// We must skip over it to maintain causal consistency.
		if next.Origin != nil && d.isDescendantOf(next.Origin, item.Origin) {
			dest = next
			continue
		}
		
		break
	}

	// Link the new item into the double-linked list
	item.Left = dest
	item.Right = dest.Right
	if dest.Right != nil {
		dest.Right.Left = item
	}
	dest.Right = item

	// Register in the lookup map
	d.Items[item.ID] = item

	// Update our state vector
	if currentMax, ok := d.Clients[item.ID.ClientID]; !ok || item.ID.Counter > currentMax {
		d.Clients[item.ID.ClientID] = item.ID.Counter
	}

	return nil
}

// isDescendantOf checks if 'child' is causally descended from 'parent'
// by walking up the origin chain.
func (d *Doc) isDescendantOf(child *ID, parent *ID) bool {
	if child == nil {
		return false
	}
	if parent == nil {
		return true // Everything descends from the sentinel start
	}
	curr := child
	for curr != nil {
		if *curr == *parent {
			return true
		}
		item, exists := d.Items[*curr]
		if !exists {
			break
		}
		curr = item.Origin
	}
	return false
}

snippet-3.go

// SplitBlock splits an Item at the specified offset, returning the newly created right-hand Item.
// This is critical for Run-Length Encoded (RLE) sequence CRDTs, allowing us to represent
// long contiguous runs of text in a single struct and split them dynamically only when edits occur.
func (d *Doc) SplitBlock(targetID ID, offset int) (*Item, error) {
	d.mu.Lock()
	defer d.mu.Unlock()

	target, exists := d.Items[targetID]
	if !exists {
		return nil, errors.New("target block not found")
	}

	if offset <= 0 || offset >= len(target.Value) {
		return nil, errors.New("invalid split offset")
	}

	// Create the new right item
	rightID := ID{
		ClientID: targetID.ClientID,
		Counter:  targetID.Counter + uint32(offset),
	}

	rightItem := &Item{
		ID:      rightID,
		Origin:  &ID{ClientID: targetID.ClientID, Counter: targetID.Counter + uint32(offset-1)},
		Value:   target.Value[offset:],
		Deleted: target.Deleted,
		Left:    target,
		Right:   target.Right,
	}

	// Update the left (target) item's value by reslicing.
	// This avoids creating a new underlying array allocation.
	target.Value = target.Value[:offset]

	// Update double-linked list pointers
	if target.Right != nil {
		target.Right.Left = rightItem
	}
	target.Right = rightItem

	// Register the new right item in our lookup map
	d.Items[rightID] = rightItem

	return rightItem, nil
}

snippet-4.go

// StateVector maps ClientID to the highest Counter integrated for that client.
type StateVector map[uint64]uint32

// GetStateVector returns a copy of the current document state vector.
func (d *Doc) GetStateVector() StateVector {
	d.mu.RLock()
	defer d.mu.RUnlock()

	sv := make(StateVector, len(d.Clients))
	for clientID, clock := range d.Clients {
		sv[clientID] = clock
	}
	return sv
}

// ComputeDelta returns a slice of Items that the remote peer is missing
// based on their provided StateVector.
func (d *Doc) ComputeDelta(remoteSV StateVector) []*Item {
	d.mu.RLock()
	defer d.mu.RUnlock()

	var delta []*Item

	// Traverse the document in logical order from start to end
	curr := d.Start
	for curr != nil {
		// Skip sentinel
		if curr.ID.ClientID == 0 && curr.ID.Counter == 0 {
			curr = curr.Right
			continue
		}

		remoteClock, exists := remoteSV[curr.ID.ClientID]
		// Since items can be split or represent blocks, we calculate the max counter of this block
		maxCounterOfBlock := curr.ID.Counter + uint32(len(curr.Value)) - 1
		
		// If the remote peer has not seen this counter range, we append it to the delta
		if !exists || maxCounterOfBlock > remoteClock {
			delta = append(delta, &Item{
				ID:      curr.ID,
				Origin:  curr.Origin,
				Value:   curr.Value,
				Deleted: curr.Deleted,
			})
		}
		curr = curr.Right
	}

	return delta
}

snippet-5.go

package crdt

import (
	"context"
	"sync"
)

type DocumentUpdate struct {
	Items []*Item
}

// UpdatePool recycles DocumentUpdate structs to minimize GC pressure during rapid typing.
var UpdatePool = sync.Pool{
	New: func() interface{} {
		return &DocumentUpdate{
			Items: make([]*Item, 0, 16),
		}
	},
}

type DocActor struct {
	doc        *Doc
	inbox      chan *DocumentUpdate
	broadcast  chan *DocumentUpdate
	clients    map[chan *DocumentUpdate]struct{}
	clientsMu  sync.Mutex
}

func NewDocActor(doc *Doc) *DocActor {
	return &DocActor{
		doc:       doc,
		inbox:     make(chan *DocumentUpdate, 1024),
		broadcast: make(chan *DocumentUpdate, 256),
		clients:   make(map[chan *DocumentUpdate]struct{}),
	}
}

// Start runs the actor loop in a background goroutine.
// All modifications to the document occur sequentially on this single thread,
// eliminating lock contention and data races.
func (a *DocActor) Start(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case update := <-a.inbox:
			var integrated []*Item
			for _, item := range update.Items {
				err := a.doc.Integrate(item)
				if err == nil {
					integrated = append(integrated, item)
				}
			}

			// Broadcast integrated updates to all other clients
			if len(integrated) > 0 {
				broadcastUpdate := UpdatePool.Get().(*DocumentUpdate)
				broadcastUpdate.Items = append(broadcastUpdate.Items[:0], integrated...)
				
				a.clientsMu.Lock()
				for clientChan := range a.clients {
					select {
					case clientChan <- broadcastUpdate:
					default:
						// Slow client connection; drop or handle backpressure
					}
				}
				a.clientsMu.Unlock()
			}

			// Clear slice and return update struct to the pool
			update.Items = update.Items[:0]
			UpdatePool.Put(update)
		}
	}
}