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() {
apply(c.TakeChanges()) // the edits, not the whole text
render(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
ResourceExhaustedand 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. - One replica identity per participant. Two participants sharing a site is
silent data loss rather than a conflict — both mint the same operation
identities for different characters — so the arriving session takes the
identity and the one already holding it is disconnected with
Aborted. Site zero, the server's own replica, is refused outright. - Documents outlive sessions. The last participant out writes the document;
Server.Flushwrites it without waiting. A write that fails is retried rather than forgotten.
What a binding needs
An editor cannot be handed the whole text on every keystroke somebody else
makes: that throws away the selection, the scroll position and every decoration.
Changes says something happened, TakeChanges says what — the edits, in the
order they have to be made.
c.TakeChanges() // []crdt.Change: remove this many here, put this there
c.Anchor(pos) // a handle on a character, for a comment or a stored selection
c.Position(anchor) // where it is now, or where it was if it has gone
c.AuthorRuns() // the text split by who wrote each stretch
c.InsertUTF16(pos, text) // offsets in the units a browser counts
A browser counts UTF-16 code units, and an emoji is one character and two units. A session that took the browser's offsets for runes would edit in the wrong place, silently, from the first emoji onwards — so the same operations are addressed both ways, and an offset landing inside a character is refused rather than moved.
Who may open what
Config.Authorize is asked once per session, after the join arrives and before
the document is touched, so a refused session neither reads the store nor reveals
whether the document exists:
collab.NewServer(collab.Config{
Authorize: func(ctx context.Context, document string, site crdt.SiteID) error {
return myACL.Check(userFrom(ctx), document)
},
})
It lives here rather than in a gRPC interceptor, which is where one would first look for it: an interceptor sees the method and the request metadata, and the document being joined is in neither — it arrives in the stream's first message. Authentication, being per connection rather than per document, still belongs in an interceptor; the context carries whatever it put there.
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.
collab/pgstore keeps documents in PostgreSQL, over a plain *sql.DB
and with no driver of its own, so the caller picks one:
db, _ := sql.Open("pgx", os.Getenv("DATABASE_URL"))
store, _ := pgstore.New(db)
store.Migrate(ctx)
srv := collab.NewServer(collab.Config{Store: store})
It is a module of its own, so importing collab does not drag a database driver
into anyone's build. Its tests run against a real PostgreSQL — CI fails the job
if one is missing rather than skipping it.
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
- Variables
- type Client
- func (c *Client) Anchor(pos int) (crdt.ID, error)
- func (c *Client) AuthorRuns() []crdt.AuthorRun
- func (c *Client) Changes() <-chan struct{}
- func (c *Client) Close() error
- func (c *Client) Delete(pos, length int) error
- func (c *Client) DeleteUTF16(pos, length int) error
- func (c *Client) Document() string
- func (c *Client) Done() <-chan struct{}
- func (c *Client) Err() error
- func (c *Client) Insert(pos int, text string) error
- func (c *Client) InsertUTF16(pos int, text string) error
- func (c *Client) Len() int
- func (c *Client) LenUTF16() int
- func (c *Client) Peers() []awareness.Peer
- func (c *Client) Position(anchor crdt.ID) (int, bool)
- func (c *Client) SetCursor(cursor awareness.Cursor, meta map[string]string) error
- func (c *Client) Site() crdt.SiteID
- func (c *Client) Snapshot() []byte
- func (c *Client) TakeChanges() []crdt.Change
- func (c *Client) Text() string
- func (c *Client) Version() crdt.VersionVector
- func (c *Client) Visible(anchor crdt.ID) bool
- type ClientConfig
- type Config
- type MemoryStore
- type Server
- type Store
Constants ¶
const DefaultBacklog = 256
DefaultBacklog is how many messages may be queued for one participant before the server gives up on it. See Config.
Variables ¶
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.
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 ¶
func Join(ctx context.Context, conn grpc.ClientConnInterface, cfg ClientConfig) (*Client, error)
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) Anchor ¶ added in v0.7.0
Anchor returns the identity of the character at rune offset pos, which keeps naming that character however the document moves around it. It is what a comment or a stored selection should hold; see crdt.Doc.Anchor.
func (*Client) AuthorRuns ¶ added in v0.7.0
AuthorRuns splits the visible text into stretches by who wrote them, which is what colouring a document by author needs.
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 ¶
Close ends the session. The local document is left intact, so its Client.Snapshot can resume later.
func (*Client) Delete ¶
Delete removes length runes at rune offset pos, locally and then everywhere.
func (*Client) DeleteUTF16 ¶ added in v0.8.0
DeleteUTF16 removes length code units at an offset counted in the same units.
func (*Client) Done ¶
func (c *Client) Done() <-chan struct{}
Done is closed when the session has ended, whatever the reason.
func (*Client) Err ¶
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) InsertUTF16 ¶ added in v0.8.0
InsertUTF16 adds text at an offset counted in UTF-16 code units.
func (*Client) LenUTF16 ¶ added in v0.8.0
LenUTF16 returns the length a browser would report, counting UTF-16 code units. Its companions InsertUTF16 and DeleteUTF16 take offsets in the same units, so a caller in the browser never converts by hand; see crdt.Doc.
func (*Client) Peers ¶
Peers returns the other participants and where their cursors are, ordered by site.
func (*Client) Position ¶ added in v0.7.0
Position returns where the character an anchor names sits now — or where it was, if it has been deleted. See crdt.Doc.Position.
func (*Client) SetCursor ¶
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) Snapshot ¶
Snapshot returns the document in a form ClientConfig.Resume accepts, which is how a participant keeps its place across a disconnection.
func (*Client) TakeChanges ¶ added in v0.7.0
TakeChanges returns the edits made by everyone else since it was last called, in the order a view of the text has to make them, and forgets them.
It pairs with Client.Changes: that says something happened, this says what. A view that only ever applies these holds what the document holds — see crdt.Change.
Local edits are not reported. A caller that made them already knows.
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
// Authorize, when set, decides whether a participant may open a document.
// It is asked once per session, after the join arrives and before the
// document is touched, so a refused session neither reads the store nor
// reveals whether the document exists.
//
// This belongs here rather than in a gRPC interceptor, which is where one
// would first look for it: an interceptor sees the method and the request
// metadata, and the document being joined is in neither — it arrives in the
// stream's first message. Anything deciding per document has to run after
// that message, which means here. Authentication, which is per connection
// rather than per document, still belongs in an interceptor; ctx carries
// whatever it put there.
//
// Returning a gRPC status error passes that status to the participant
// unchanged; any other error is reported as PermissionDenied.
Authorize func(ctx context.Context, document string, site crdt.SiteID) error
}
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 (*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.
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.
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.
Source Files
Directories