collab

package module
v0.1.0 Latest Latest
Warning

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

Go to latest
Published: Aug 16, 2026 License: BSD-3-Clause Imports: 10 Imported by: 0

README

collab — the wire for collaborative editing

github.com/go-crdt/collab carries a go-crdt/crdt document between the people editing it: a gRPC service, a server that hosts documents, and a client that joins one. Pure Go, CGO=0.

The service is thin on purpose. The document is a CRDT, so the server never transforms an operation and never decides an outcome — it applies what it is sent to its own replica and hands it to everyone else. Two things follow that a server-authoritative design cannot offer: a participant may edit while disconnected and reconcile on return, and the server may be restarted or replaced without anyone losing work.

Mounted on grpc-transports/websocket the same service reaches a browser, and the client here builds for js/wasm, so a browser tab and the server run the same code down to the merge.

Using it

Server — any grpc.Server, any carrier:

srv := collab.NewServer(collab.Config{Store: myStore})
collabpb.RegisterCollabServer(grpcServer, srv)

Participant:

c, err := collab.Join(ctx, conn, collab.ClientConfig{Document: "notes", Site: 1})
c.Insert(0, "hello")
c.SetCursor(awareness.Cursor{Anchor: 5, Head: 5}, map[string]string{"name": "ada"})

for range c.Changes() {
    render(c.Text(), c.Peers())
}

Coming back after a disconnection, keeping the work done offline:

c, err := collab.Join(ctx, conn, collab.ClientConfig{
    Document: "notes",
    Site:     1,
    Resume:   savedSnapshot, // from Client.Snapshot()
})

Shape of a session

One bidirectional stream per participant per document. The client opens with a Join; the server answers with a Welcome holding either the whole document or, for a participant that says what it already has, only what it missed — plus who else is present and where the server stands, so the participant can push whatever it wrote while away. After that, operations and presence flow both ways.

What it guarantees

  • Convergence, proven end to end. The acceptance test runs three replicas across two runtimes — one native, two compiled to WebAssembly and executed by Node through a real WebSocket — editing concurrently and converging on the same text.
  • Offline work is never stranded. A resuming participant pushes what the server lacks and is sent what it missed. Both directions are tested.
  • Nobody stalls the document. A participant that stops reading is disconnected with ResourceExhausted and caught up when it rejoins, rather than holding everyone up or being served state that is quietly out of date.
  • Nothing is trusted. Malformed operations, presence, version vectors and snapshots are each refused with InvalidArgument, on both sides of the wire.
  • Documents outlive sessions. The last participant out writes the document; Server.Flush writes it without waiting. A write that fails is retried rather than forgotten.

Persistence

Store is a two-method seam — Load and Save on snapshots, which are self-contained, so a document restored from one can still serve a participant that has been away. MemoryStore is the default; anything else (Postgres, object storage) implements the interface.

Status

Version 0.1. 100% statement coverage, race-clean, six-arch CI, and the WebAssembly end-to-end test running on every pull request — where a missing toolchain is a failure, not a skipped test.

License

BSD-3-Clause — see LICENSE. Copyright the go-crdt authors.

Documentation

Overview

Package collab carries a github.com/go-crdt/crdt document between the people editing it: a gRPC service, a server that hosts documents, and a client that joins one.

The service is thin on purpose. The document is a CRDT, so the server never transforms an operation and never decides an outcome — it applies what it is sent to its own replica and hands it to everyone else. Two consequences follow that a server-authoritative design cannot offer: a participant may edit while disconnected and reconcile later, and the server may be restarted or replaced without any client losing work.

Over what

Nothing here requires a particular carrier. Mounted on github.com/grpc-transports/websocket the same service reaches a browser, because that transport gives grpc-go a net.Conn a browser can open and runs unmodified under js/wasm. The client in this package builds for js/wasm too, so a browser tab and a server run the same code down to the merge.

Shape of a session

One bidirectional stream per participant per document. The client opens with a collabpb.Join; the server answers with a collabpb.Welcome holding either the whole document or, for a participant that says what it already has, only what it missed. After that, operations and presence flow both ways until either side hangs up.

Index

Constants

View Source
const DefaultBacklog = 256

DefaultBacklog is how many messages may be queued for one participant before the server gives up on it. See Config.

Variables

View Source
var ErrClosed = errors.New("collab: session closed")

ErrClosed is why a session ended when this participant closed it, and what an edit made afterwards returns.

View Source
var ErrProtocol = errors.New("collab: unexpected message from the server")

ErrProtocol reports a server that sent something a session cannot be in the middle of — a second welcome, or nothing at all.

Functions

This section is empty.

Types

type Client

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

A Client is one participant's view of a document: a replica that edits locally and is kept in step with everyone else.

It is safe for concurrent use. It builds for js/wasm, so a browser tab runs this code and the server's merge logic unchanged.

func Join

Join opens a session on conn and returns once the document has arrived, so the client is usable the moment it is returned.

The session lives until ctx is cancelled or Client.Close is called.

func (*Client) Changes

func (c *Client) Changes() <-chan struct{}

Changes receives a value whenever the document or the participants changed. It coalesces: a reader that is slow sees one wake-up, not a queue of them.

func (*Client) Close

func (c *Client) Close() error

Close ends the session. The local document is left intact, so its Client.Snapshot can resume later.

func (*Client) Delete

func (c *Client) Delete(pos, length int) error

Delete removes length runes at rune offset pos, locally and then everywhere.

func (*Client) Document

func (c *Client) Document() string

Document returns the name of the document joined.

func (*Client) Done

func (c *Client) Done() <-chan struct{}

Done is closed when the session has ended, whatever the reason.

func (*Client) Err

func (c *Client) Err() error

Err returns why the session ended, or nil while it is still running. Once Client.Done is closed it is never nil: a session that was closed deliberately reports ErrClosed rather than the transport's cancellation.

func (*Client) Insert

func (c *Client) Insert(pos int, text string) error

Insert adds text at rune offset pos, locally and then everywhere.

func (*Client) Len

func (c *Client) Len() int

Len returns the number of characters, counted in runes.

func (*Client) Peers

func (c *Client) Peers() []awareness.Peer

Peers returns the other participants and where their cursors are, ordered by site.

func (*Client) SetCursor

func (c *Client) SetCursor(cursor awareness.Cursor, meta map[string]string) error

SetCursor publishes where this participant is. meta carries whatever the editor wants shown — a display name, a colour — and is not interpreted here.

Cursor positions are ephemeral and are never persisted.

func (*Client) Site

func (c *Client) Site() crdt.SiteID

Site returns this participant's replica identity.

func (*Client) Snapshot

func (c *Client) Snapshot() []byte

Snapshot returns the document in a form ClientConfig.Resume accepts, which is how a participant keeps its place across a disconnection.

func (*Client) Text

func (c *Client) Text() string

Text returns the document as it stands here.

func (*Client) Version

func (c *Client) Version() crdt.VersionVector

Version returns what this participant holds, for ClientConfig.Resume or for diagnostics.

type ClientConfig

type ClientConfig struct {
	// Document names the document to join. It is created if it does not exist.
	Document string

	// Site is this participant's replica identity, and must differ from every
	// other participant's in the document. See [crdt.DeriveSiteID].
	Site crdt.SiteID

	// Resume is a snapshot from an earlier session, obtained from
	// [Client.Snapshot]. When set, the participant keeps the work it did while
	// disconnected and is sent only what it missed, rather than the whole
	// document.
	Resume []byte
}

ClientConfig describes a participant joining a document.

type Config

type Config struct {
	// Store keeps documents between sessions. Defaults to a [MemoryStore].
	Store Store

	// Backlog is how many messages may be queued for one participant.
	// A participant that falls further behind than this is disconnected with
	// ResourceExhausted rather than served stale state or allowed to stall
	// everyone else; it rejoins and is caught up from its version vector.
	// Defaults to [DefaultBacklog].
	Backlog int
}

Config configures a Server.

type MemoryStore

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

MemoryStore keeps documents in memory. It is the default, it is what the tests use, and it is enough for a single process that does not need to survive a restart. Anything else — Postgres, object storage — implements Store.

func NewMemoryStore

func NewMemoryStore() *MemoryStore

NewMemoryStore returns an empty store.

func (*MemoryStore) Documents

func (s *MemoryStore) Documents() []string

Documents returns the names of the documents held, which is what a caller needs to inspect or migrate a store.

func (*MemoryStore) Load

func (s *MemoryStore) Load(_ context.Context, document string) ([]byte, error)

Load returns a copy of the stored snapshot, or nil if the document is new.

func (*MemoryStore) Save

func (s *MemoryStore) Save(_ context.Context, document string, snapshot []byte) error

Save records a copy of the snapshot.

type Server

type Server struct {
	collabpb.UnimplementedCollabServer
	// contains filtered or unexported fields
}

A Server hosts documents. Register it with collabpb.RegisterCollabServer on any grpc.Server — over github.com/grpc-transports/websocket for browsers, over plain TCP for anything else.

Documents stay in memory once opened, so a long-lived server holds every document it has served. Call Server.Flush to persist them.

func NewServer

func NewServer(cfg Config) *Server

NewServer returns a server ready to register.

func (*Server) Flush

func (s *Server) Flush(ctx context.Context) error

Flush persists every document that has changed since it was last written. A server that wants durability without waiting for participants to leave calls this on a timer, or before shutting down.

func (*Server) Session

func (s *Server) Session(stream collabpb.Collab_SessionServer) error

Session is the service method: one bidirectional stream, one participant, one document.

type Store

type Store interface {
	// Load returns the snapshot for a document, or nil if there is none yet.
	// Returning nil is how a store says "new document", and is not an error.
	Load(ctx context.Context, document string) ([]byte, error)

	// Save records the current snapshot, replacing any previous one.
	Save(ctx context.Context, document string, snapshot []byte) error
}

A Store keeps documents between sessions. It holds snapshots, which are self-contained: a document restored from one can still serve a participant that has been away, because the snapshot carries the whole history.

Implementations must be safe for concurrent use.

Directories

Path Synopsis
pgstore module
Command wasmtest is the browser half of the end-to-end proof.
Command wasmtest is the browser half of the end-to-end proof.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL