collab

package module
v0.32.0 Latest Latest
Warning

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

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

README

go-crdt/collab

collab — the wire for collaborative editing

CI Go Reference coverage license

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.

The client builds for js/wasm, so a browser tab and the server run the same code down to the merge — over either of two carriers, from one server:

carrier for browser client, gzipped
collab.WebSocket — the session's own framing anywhere, browsers included 919 KB
collab.GRPC — over gRPC native peers 4 461 KB

Everything a session carries is already bytes crdt encoded and will check on arrival, so protobuf describes fields nobody reads through it — and compiled to wasm its machinery cannot be linked away. For scale, the CRDT alone is 633 KB. Outside a browser none of that matters, which is why gRPC is still there.

From a page

The editor this was built for is TypeScript and cannot call Go, so ./wasm compiles to a binding a page uses directly. Offsets are UTF-16 code units throughout — the units a JavaScript string counts in — and an offset splitting a character is refused rather than rounded:

const session = await collab.join({ url, document: "project:default", site });

const body  = await session.text("file:main.tex");
const chat  = await session.list("chat");
const cells = await session.map("cells");

await body.insert(0, "bonjour");
await cells.set("B7", new TextEncoder().encode("42"));

await session.onChange(parts => {
  for (const part of parts) {
    if (part.kind === "text") applyEdits(part.text);   // {pos, removed, insert}
    if (part.kind === "map")  reread(part.keys);       // the keys that changed
    if (part.kind === "list") rereadWhole(part.name);  // a list says only that it moved
  }
});

Types are in wasm/collab.d.ts. A value is Uint8Array in both directions, never a string and never JSON: the CRDT does not interpret it and neither does the binding.

Keeping what people wrote

A document was saved when its last participant left, and otherwise when somebody called Flush. That is enough for a text people open and close, and not enough for what a session actually carries — the comments on it, the record of who changed what, the messages beside it. A server restarted while anybody was still connected lost everything since the document was opened, and said nothing.

store, err := collab.NewDirStore("/var/lib/loom/documents")

srv := collab.NewServer(collab.Config{
    Store:        store,
    PersistEvery: 5 * time.Second,  // bounds what a crash costs
    EvictAfter:   10 * time.Minute, // let go of what nobody is in
})
defer srv.Close(ctx)

DirStore keeps a document per file, written and renamed into place so a reader sees one whole version or the one before. The file is named after the encoding of the document name rather than the name: project:default and file:src/main.tex are not file names, and escaping them would leave the question of which characters, on which system. Documents() says what the file names cannot.

Neither runs unless it is asked for, so a server configured as before behaves as before. EvictAfter also stops a long-lived server holding every document it has ever served; reopening one costs a read from the store, not anything anybody wrote.

Using it

Server — any grpc.Server, any carrier:

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

Participant. A document holds named parts, so a caller reaches for the one it means and gets a handle that edits and publishes:

c, err := collab.Join(ctx, conn, collab.ClientConfig{Document: "notes", Site: 1})

body, _ := c.Text("file:main.tex")   // the buffer an editor binds to
chat, _ := c.List("chat")            // the messages beside it
cells, _ := c.Map("cells")           // a sheet

body.Insert(0, "hello")
chat.Append([]byte("on commence"))
cells.Set("B7", []byte("42"))
c.SetCursor(awareness.Cursor{Anchor: 5, Head: 5}, map[string]string{"name": "ada"})

for range c.Changes() {
    for _, part := range c.TakeChanges() {   // which part moved, and how
        render(part)
    }
    render(c.Peers())
}

A handle is what a caller touches rather than the replicated structure, and that is forced: editing the structure directly would produce operations nobody ever heard, so the participant would drift away from everyone else while its own screen looked right.

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.
  • 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.Flush writes 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.

collab/gitstore keeps documents in a git repository, so a document can be versioned and released the way everything else is. One commit holds both the state, which carries identities and authorship and the comments anchored to characters, and the rendered text, which is what makes the repository readable by a person. A release is a tag on a commit that already exists. It is also a federation channel: two servers sharing a repository diverge, git reports a conflict on the state file, and gitstore.Merge resolves it without anybody having to choose a side.

MultiStore writes to several stores at once and reads from all of them:

srv := collab.NewServer(collab.Config{
    Store: collab.NewMultiStore(database, repository),
})

Reading merges rather than picking, because a save that failed halfway leaves the stores holding different documents and taking the first would drop whatever only the second had — MergeSnapshots is what makes that free. Adding a store to a running server therefore backfills it. A store that cannot be read fails the load rather than serving a document quietly missing a paragraph.

Tiered is the other composition: a hot store for documents somebody is using and a cold one for documents nobody has opened in a long time.

store := collab.NewTiered(hot, archive)
srv := collab.NewServer(collab.Config{Store: store})

// On a timer, or by hand.
moved, err := store.Archive(ctx, 30*24*time.Hour)

Nothing is deleted that is not already somewhere else. Archiving reads the hot store, writes the cold one, and only then asks the hot one to release exactly what was read — so a cold store that refuses releases nothing, and a document that somebody saved while it was being copied is not released at all; it is archived on a later pass. Reading an archived document brings it back to the hot store on the way past.

A Tiered store never answers "no such document" because it could not reach the archive: nil means start a new one, and a server acts on it. An unreachable archive fails the load instead.

What counts as idle is not written for a while, which is as close to not used as a store can get — and close enough, because a server saves a document somebody is in every PersistEvery, so a busy document never looks quiet. MemoryStore and DirStore both implement Archivable; Config.EvictAfter is the same idea one level up, for the server's memory rather than the store's.

Staying in a document

Join opens one session. When it ends — a connection dropped, or the server disconnected a participant that fell behind its Backlog — it has ended.

JoinWithRetry opens another:

c, err := collab.JoinWithRetry(ctx, dial, collab.ClientConfig{
    Document: "project:paper", Site: site,
}, collab.RetryPolicy{Notify: func(s collab.LinkStatus) { /* log it */ }})

The handles taken from the client stay valid across every reconnection — a handle holds a name and looks its part up under the client's lock, which is what makes replacing the session underneath it invisible. Each attempt rejoins with what the replica already holds, so the server sends the difference rather than a snapshot, and an edit made while there was nowhere to send it is not lost and is not an error: it goes out as soon as there is a session again.

This matters more than it looks. One edit by each of P participants is P-1 messages into every other queue, so a document busier than Config.Backlog disconnects everyone in it at once — measured with 800 participants editing simultaneously, a backlog of 256 disconnected 99% of them. With Join that is a room full of documents that have stopped moving; with JoinWithRetry it is a pause.

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

Two carriers, and which one to use is decided by where the code runs rather than by taste. WebSocket carries a session's own framing over a plain WebSocket; GRPC carries it over gRPC. One server serves both at once — Server.ServeWebSocket beside the registered service — and a participant on each edits the same document.

The reason there are two is measured. Everything a session carries is bytes some encoder in github.com/go-crdt/crdt produced and will check on arrival, so protobuf is describing fields nobody reads through it — and compiled to wasm its reflection and registry machinery cannot be linked away. The browser test client, gzipped, is 919 KB over the framing and 4 461 KB over gRPC, against 633 KB for the CRDT alone. Outside a browser none of that matters, and gRPC brings deadlines, interceptors and the tooling built around them.

The client builds for js/wasm either way, so a browser tab and a server run the same code down to the merge. Two browsers with no server between them carry a session over a WebRTC data channel ([DataChannel]), and two tabs of one browser over a BroadcastChannel ([JoinBroadcastChannel]) with nothing to configure at all.

A document holds named parts

What an editor holds is not one structure: the text of a file, the comments anchored into it, the record of who changed what, the messages beside it, the cells of a sheet. A document here is a github.com/go-crdt/crdt.Composite, so they travel together — one snapshot, one version, one decision about who may open it, and no instant at which the set of them disagrees.

A caller reaches for a part by name and gets a handle: Client.Text, Client.List, Client.Map. A handle edits and publishes in one step, which is why it exists rather than the replicated structure itself — a caller editing that directly would produce operations nobody ever heard, and drift away from everyone else while its own screen looked right.

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.

It is also, and less obviously, a capacity: a document where P participants edit in the same breath sends P-1 messages to each of them, so a backlog below P is a backlog that will be exceeded. Measured on one server with 800 participants each making one edit at once — a backlog of 256 disconnected 99% of them, 512 disconnected 32%, and 1024 disconnected none.

So 256 means "about two hundred and fifty people editing at the same instant", not "a queue that is usually long enough". A document with more than that wants a larger Config.Backlog, and the cost of one is a queue slot per participant rather than anything per document.

View Source 
const DefaultRetryCeiling = 30 * time.Second

DefaultRetryCeiling is the longest a link waits between attempts, when RetryPolicy.Ceiling does not say. It bounds two things at once: how much work a peer that has been down for a day is asked to do, and how stale a replica can be once that peer comes back, since nothing crosses the link until the next attempt. Half a minute is the compromise, and an operator who knows their outages can say better.

View Source 
const DefaultRetryWait = 250 * time.Millisecond

DefaultRetryWait is how long a link waits before its first attempt at coming back, when RetryPolicy.Wait does not say. It is short because most drops are brief — a process restarted, a route reconverging — and a link that is back within a second has lost nothing anybody typed.

Variables

View Source 
var ErrBroadcastClosed = errors.New("collab: broadcast channel closed")

ErrBroadcastClosed is why a BroadcastChannel carrier's [Recv] or [Send] returned: this end was closed, or the bus it spoke over was. It is the shared-bus counterpart of the error a dropped socket reports.

View Source 
var ErrChanged = errors.New("collab: the document changed since it was read")

ErrChanged reports a document that was written since it was read, so it was not released. It is not a failure: the next pass will archive the newer one.

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 ErrHostSuperseded = errors.New("collab: another tab is hosting this room")

ErrHostSuperseded is why a host's serve loop returned: another tab with a lower identifier — the one the tie-break gives priority — announced itself as host, so this tab steps down to let that one hold the room. It is the self-healing half of the election: even if two tabs ever both reach RoleHost, exactly one keeps the document and the other yields, rather than the two drifting apart as separate documents. A caller that hosts should, on this error, re-join the room (it will now find the surviving host answering) rather than treat it as a failure.

View Source 
var ErrNoDocument = errors.New("collab: a document must have a name")

ErrNoDocument reports a document with no name. The server refuses one at the door — a join must name a document — and this refuses it too rather than let it name the directory itself, which is what the empty name encodes to.

View Source 
var ErrPipeClosed = errors.New("collab: pipe closed")

ErrPipeClosed is why a Pipe carrier's [Recv] or [Send] returned: either end was closed, or the session's context was cancelled. It is the in-process counterpart of the error a dropped socket reports.

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

ErrProtocol reports a message that is not part of a session: a kind that cannot arrive at that moment — a second welcome, or a join halfway through — or bytes that are not a message at all.

View Source 
var ErrTooFarBehind = errors.New("collab: this replica holds work made against operations the document has collected")

ErrTooFarBehind reports a participant whose replica holds work made against operations the document has since collected.

It is what crdt.Composite.Collect costs, said out loud. Collecting drops operations every replica had delivered, so a replica that has them can always be caught up. A replica that has been away, has written while away, and wrote against something that is now gone, cannot: its operations name characters this document no longer holds, and applying them would strand them for ever.

Such a participant is refused rather than served, and refused rather than quietly re-seeded. Re-seeding it would work — it would get the document as it now stands — and it would throw away what that person wrote while they were away, with nothing said. Handing back an error is the only option that lets an application do the one thing that is actually right, which is to show somebody their own work and let them decide.

View Source 
var ErrTransport = errors.New("collab: transport")

ErrTransport reports a carrier that could not be opened or that failed.

Functions

func MergeSnapshots added in v0.24.0 

func MergeSnapshots(ours, theirs []byte) ([]byte, error)

MergeSnapshots combines two snapshots of the same document into one that holds everything either of them held.

It is the operation that makes a snapshot safe to keep in more than one place. Two copies of a document that were written separately have not disagreed about anything — a snapshot is a set of operations, and the union of two sets of operations is a document, which is the whole reason this project exists. Nothing here has to choose a side, and so nothing here can lose what the other side did.

Either argument may be nil, which is how a store says it has never held the document; merging with nothing gives back the other side. Merging is symmetric to the byte, because the snapshot encoding is canonical and both results hold the same operations — with one exception, which is that a side that has collected (crdt.Composite.Collect) carries a record of what it gave back, and the merge is done from that side because it is the only one that can still produce a difference. The operations are the same either way; the bytes then say which side was collected.

func Pipe added in v0.21.0 

func Pipe() (Transport, *PipeConn)

Pipe returns the two ends of an in-process session: a Transport to hand to Join, and a handle to hand to Server.ServePipe. Nothing leaves the process — messages are carried on Go channels rather than a socket — so it is how a peer that both serves a document and edits it locally binds its own editor to that document without opening a loopback connection to itself.

What it is for

A browser tab holding a document for others over a data channel is also a place someone is typing. Its own editor is a participant like any other, and the honest way to say so is to Join the document it is serving. Without this that means a second, real carrier looping back to the same page — a WebRTC data channel to itself, dialled, answered and encrypted, to carry bytes that never cross a wire. Pipe is that participant with the carrier taken out: the same session, the same merge, over two channels and nothing else.

The same holds for a native server that wants a replica of a document it hosts — a headless editor, a linter, an exporter — without the cost of a second server or the round trip of a real link.

Shape

The two ends are one session and share its fate: closing either, or cancelling the context Join or Server.ServePipe was given, ends both. A [Recv] blocked when that happens returns ErrPipeClosed, and a [Send] after it does the same, so neither end can be left waiting on a peer that has gone.

A participant here edits the same document as one arriving over WebSocket or GRPC; it is a first-class participant, not a shortcut with less to it.

Types

type Archivable added in v0.26.1 

type Archivable interface {
	Store

	// Idle returns the documents this store has not been asked to write for
	// longer than d.
	//
	// Not written is the closest thing to not used that a store can know, and
	// it is close enough for the reason that matters: a server persists a
	// document somebody is in every [Config.PersistEvery], so a document with
	// anybody in it is written constantly and never looks idle. What looks idle
	// is what nobody has opened since it was last put away.
	Idle(ctx context.Context, d time.Duration) ([]string, error)

	// Release forgets a document, but only if this store still holds exactly
	// want. It reports [ErrChanged] if it holds something else, and nil if it
	// held nothing.
	//
	// The condition is the whole of it. An archiver copies a document
	// elsewhere and then asks for it to be released, and between those two
	// moments somebody may have joined the document and saved a newer one. A
	// store that released it anyway would delete a version that is nowhere
	// else. The comparison and the removal therefore happen together, under
	// whatever the store uses to keep its own writes apart.
	Release(ctx context.Context, document string, want []byte) error
}

An Archivable store can say which of its documents have gone quiet and let one go, which is what it takes to be the hot half of a Tiered.

Both methods are here rather than in Store because most stores have no business implementing them: a store that is only ever written through is complete with Load and Save, and asking every one of them for a way to forget a document would be asking for a way to lose one.

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, transport Transport, cfg ClientConfig) (*Client, error)

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

Use WebSocket unless there is a reason not to; it is what the same code compiled for a browser can afford. GRPC is there for a native peer that wants what gRPC brings with it.

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

func JoinWithRetry added in v0.28.0 

func JoinWithRetry(ctx context.Context, dial Dialer, cfg ClientConfig, policy RetryPolicy) (*Client, error)

JoinWithRetry joins a document and keeps a participant in it, opening a new session each time the one it has ends.

Why this exists

Config.Backlog says a participant that falls too far behind is disconnected and "rejoins and is caught up from its version vector". That is what the protocol supports and, until now, nothing did it: the material was here — a replica that survives the session and a join that says what it already holds — and the loop was left to whoever held the session, the same way it was for Server.Follow before Server.FollowWithRetry.

Leaving it there turned out to matter. A document busier than the backlog disconnects everyone in it in one instant, however well they are keeping up: one edit by each of P participants is P-1 messages into every other queue. Measured on one server with 800 participants editing at once, a backlog of 256 disconnected 99% of them — and an application without this loop leaves 99% of a room staring at a document that has stopped moving.

What it does that Join does not

The client it returns is the same Client, and the handles taken from it stay valid across every reconnection: a handle holds a name and looks the part up under the client's lock, which is what makes replacing the session underneath it invisible.

Each attempt rejoins with what this replica already holds, so the server sends the difference rather than a snapshot, and nothing edited while disconnected is lost — it is pushed as soon as there is somewhere to push it.

An edit made while there is no session does not fail. It is in the replica, and telling an editor that their keystroke failed when it did not is how an application starts undoing work that was never lost. The outage is reported through RetryPolicy.Notify instead, which is where an operator looks.

What it does not do

It does not resume presence: a cursor is ephemeral, and one from before an outage is a guess about where somebody was. It publishes nothing of its own on reconnection beyond the operations the server is missing.

Client.Done closes when this gives up for good — because the context ended, because Client.Close was called, or because RetryPolicy.Permanent said so — and not when a session ends.

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) 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) List added in v0.10.0 

func (c *Client) List(name string) (*List, error)

List returns a handle on the list part with this name.

func (*Client) Map added in v0.10.0 

func (c *Client) Map(name string) (*Map, error)

Map returns a handle on the map part with this name.

func (*Client) Parts added in v0.10.0 

func (c *Client) Parts() []crdt.Part

Parts returns the parts this replica holds, in the canonical order. A part that has never been written to is not among them.

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) TakeChanges added in v0.7.0 

func (c *Client) TakeChanges() []crdt.PartChange

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) Text 

func (c *Client) Text(name string) (*Text, error)

Text returns a handle on the text part with this name, which is created the first time anybody writes to it. The name is arbitrary UTF-8 and is expected to carry structure — "file:src/main.tex". An empty or invalid name is refused; see crdt.Part.

func (*Client) Version 

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

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].
	//
	// Size it against the number of participants who may edit at once, not
	// against how fast one of them types: one edit by each of P participants is
	// P-1 messages into every queue, so a burst on a busy document reaches the
	// limit in an instant however well everyone is keeping up. See
	// [DefaultBacklog] for what that costs, measured.
	//
	// Note what "it rejoins" asks of a caller, because it is not free.
	// [JoinWithRetry] does it: it opens a session again whenever the one it has
	// ends, rejoining with what its replica already holds so that nothing
	// edited in between is lost. A caller that uses [Join] instead gets one
	// session, and a burst on a busy document then becomes a disconnection
	// nobody recovers from.
	Backlog int

	// PersistEvery, when set, saves every document that has changed at this
	// interval, whoever is connected. Without it a document is saved when its
	// last participant leaves and when [Server.Flush] is called, so a server
	// restarted while anybody was still editing loses everything since the
	// document was opened.
	//
	// It bounds what a crash costs to this interval, which is a number an
	// operator can choose. A server that sets it must be closed with
	// [Server.Close], which stops the housekeeping and saves what is left.
	PersistEvery time.Duration

	// EvictAfter, when set, persists a document nobody has been in for this long
	// and lets go of it. Without it a long-lived server holds every document it
	// has ever served.
	//
	// A document is reloaded from the store the next time somebody joins it, so
	// evicting costs a read rather than anything anybody wrote.
	EvictAfter time.Duration

	// CollectEvery, when set, gives back what every participant of a document
	// has certainly seen: the tombstones nobody can be confused by any more.
	// See [crdt.Doc.Collect].
	//
	// It is off by default, and being off is not a failure of nerve. Collecting
	// asks for a version every replica has delivered, and a server can only
	// know one because participants tell it what they hold. If any of them has
	// gone quiet, the answer is nothing and nothing is collected — which is the
	// behaviour the literature calls for, and the reason it is acceptable is
	// that collecting affects size and never correctness:
	//
	//	When these requirements are not met, GC may block. We consider this to
	//	be acceptable, as GC does not impact correctness (only performance), and
	//	the normal operations in the object's interface remain live.
	//	  — Shapiro, Preguiça, Baquero and Zawirski, "A comprehensive study of
	//	    Convergent and Commutative Replicated Data Types", §4.1
	//
	// What it costs is stated on [ErrTooFarBehind], and it is not nothing: a
	// participant that went away, wrote while away, and wrote against something
	// this document has since collected is refused rather than served. Turn
	// this on for documents that are edited in a room, not for ones people take
	// home.
	CollectEvery time.Duration

	// OnEvictError, when set, is told about a document that could not be saved
	// as it was evicted. There is nobody left to return an error to, and the
	// document cannot be kept — a session may already have opened a fresh
	// replica of it — so this is the only place that failure can be seen.
	OnEvictError func(document string, err error)

	// OnPersistError, when set, is called for every document whose periodic
	// save failed, with the error the store gave.
	//
	// Without it a server that cannot write is silent about it. The saves go on
	// being attempted and go on failing, participants go on editing and are
	// told nothing, and the work is there until the process stops and then is
	// not. A disk that filled up, a name a filesystem will not take, a
	// credential that expired: all of them look exactly like a server that is
	// working, which is the worst way for durability to fail.
	//
	// It is called from the housekeeping goroutine, once per document per pass,
	// so an implementation that blocks delays the next pass. Counting or
	// logging is what it is for; recovery is the operator's.
	//
	// [Server.Flush] does not call it — a caller that asks for a flush is given
	// the error to handle.
	OnPersistError func(document string, err error)

	// Clock is what [Config.EvictAfter] measures with. It defaults to time.Now,
	// and exists because a caller that wants a monotonic source, or a test that
	// wants to reach an hour of idleness without waiting an hour, has nowhere
	// else to say so. It is read from more than one goroutine, so it must be
	// safe for concurrent use and must be given here rather than set afterwards.
	Clock func() time.Time

	// 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

	// AuthorizeOperations, if set, is asked about every batch a session sends,
	// and refusing ends the session.
	//
	// Authorize runs once, when somebody joins, and decides whether that site
	// may be in this document. That is the whole story for a participant: a
	// participant speaks for itself, and the site it joined as is the site its
	// operations carry.
	//
	// It is not the whole story for a link. [Server.Follow] joins as one site
	// and then relays the work of everyone on the server it follows, so what
	// arrives on a link names sites this server never authorised — thousands of
	// them, belonging to an institution rather than to a person. Inside one
	// deployment that is exactly right and there is nothing to decide. Between
	// two, it is the decision: whether this link may speak for those sites.
	//
	// from is the site the session joined as. batches carry the operations it
	// is asking to add, each naming the site that made it, so a policy can be
	// written about the relationship between the two — "this link may carry
	// operations for sites derived within lyon.ac.example" — which is what an
	// interfederation has scopes for.
	//
	// It runs after the operations have been decoded and before any of them is
	// applied, so a refused batch changes nothing. Returning a gRPC status
	// error passes that status on unchanged, as Authorize does.
	AuthorizeOperations func(ctx context.Context, document string, from crdt.SiteID, batches []crdt.PartOps) error
}

Config configures a Server.

type Dialer added in v0.24.0 

type Dialer func(ctx context.Context) (Transport, error)

A Dialer produces a route to the peer, one per attempt.

It is a function rather than a Transport because a Transport is a way to reach a peer and an attempt needs a fresh one: a gRPC client connection whose server has gone stays broken, a WebSocket that closed cannot be reopened, and a link that held on to the first one would redial nothing. What is stable across attempts is the knowledge of where the peer is and how to authenticate to it, and that is what a closure holds.

It is called on the link's own goroutine, once per attempt, with the context the link was given, so a dialler that blocks blocks the retry loop and a dialler that respects ctx is what makes cancellation prompt during a dial. Returning a transport it has already returned is allowed where the transport itself redials, as WebSocket does.

type DirStore added in v0.13.0 

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

A DirStore keeps documents as files in one directory. It is what a server wants when the documents belong to whatever else is on that disk — a project whose files are already there, backed up with them and restored with them — and it needs nothing running beside it.

A file per document, named after nothing

A document name is arbitrary UTF-8 and is expected to carry structure, so the names a real consumer uses are "project:default" and "project:ods:chapter one.ods". Those are not file names: a colon is a path separator on one system this package supports, a slash is one everywhere, and "." and ".." name something else entirely. Escaping the awkward characters would leave the question of which ones, on which system, and a name that escapes to the same file as another is two documents sharing a snapshot.

So the file is named after the encoding of the name rather than the name: base64, in the alphabet made for file names, which is total and reversible and has no awkward character in it. It is unreadable at the shell, which is what DirStore.Documents is for.

What a reader may see

A snapshot is written to a temporary file and renamed over the old one, so a reader sees the whole of one version or the whole of the one before. A crash during a save leaves the previous snapshot intact and a temporary file behind; the next NewDirStore on that directory clears those away.

func NewDirStore added in v0.13.0 

func NewDirStore(dir string) (*DirStore, error)

NewDirStore returns a store keeping documents in dir, creating it if it is not there, and clears away any temporary file a previous run left behind.

func (*DirStore) Documents added in v0.13.0 

func (s *DirStore) Documents() ([]string, error)

Documents returns the names of the documents held, which is what a caller needs to inspect a store whose file names are an encoding rather than a name.

A file whose name is not one this store wrote is skipped rather than reported: a directory shared with anything else would otherwise turn every stray file into an error nobody can act on.

func (*DirStore) Idle added in v0.26.1 

func (s *DirStore) Idle(_ context.Context, d time.Duration) ([]string, error)

Idle returns the documents whose file has not been written for longer than d, which is a DirStore's half of Archivable.

func (*DirStore) Load added in v0.13.0 

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

Load returns the snapshot for a document, or nil if there is none yet.

func (*DirStore) Release added in v0.26.1 

func (s *DirStore) Release(ctx context.Context, document string, want []byte) error

Release forgets a document if this store still holds exactly want.

The comparison and the removal are one step under the same lock as a save, so a document written while it was being archived is not deleted by the release that follows.

func (*DirStore) Save added in v0.13.0 

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

Save records the snapshot, replacing any previous one.

It writes a temporary file, flushes it, and renames it over the old one, so that a reader sees one whole version or the other and never half of either. On every system this package supports, a rename within a directory replaces the destination in one step.

type GRPCServer added in v0.18.0 

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

GRPCServer presents a Server as the generated gRPC service.

It exists because the Server itself no longer does. The session logic speaks the wire format in wire.go — four small types, hand-written — and that is what let it stop depending on the generated protobuf code. The reason is a measurement: compiling the server for the browser with protobuf attached takes the WebAssembly binding from 5.3 MB to 19.3, because gRPC and protobuf come with it. A browser holding a document for a colleague on another continent cannot pay that, and it is the same reason wire.go exists at all.

So gRPC is a binding rather than a foundation: this type converts, and the document logic never sees a protobuf message.

func GRPCService added in v0.18.0 

func GRPCService(s *Server) *GRPCServer

GRPC presents a Server over gRPC. Register the result with collabpb.RegisterCollabServer on any grpc.Server.

func (*GRPCServer) Session added in v0.18.0 

func (g *GRPCServer) Session(stream collabpb.Collab_SessionServer) error

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

type LinkStatus added in v0.24.0 

type LinkStatus struct {
	// Up is true for the one report that says the link is established and the
	// local replica has been caught up. The rest of the fields are zero.
	Up bool

	// Err is why the attempt ended, for a report that is not Up.
	Err error

	// Attempt counts the consecutive failures since the link was last up, so
	// the first report of an outage carries 1. It is what tells a second
	// failure from a hundredth.
	Attempt int

	// RetryIn is how long the link will wait before trying again — the jitter
	// already applied, so it is the real interval and not the policy's idea of
	// it.
	RetryIn time.Duration

	// DownFor is how long this outage has lasted: zero on the first report,
	// growing with each one. It is the number a health check thresholds on,
	// because "down for six seconds" and "down for six hours" want different
	// people woken up.
	DownFor time.Duration
}

A LinkStatus is what a reconnecting link tells RetryPolicy.Notify: whether it is up, and if it is not, why, since when and for how much longer.

It answers the two questions an operator has about a link that is not working — is it down, and how long has it been down — without their having to keep the state themselves, because the obvious mistake is to report only the failure and leave "still failing" indistinguishable from "failed once".

type List added in v0.10.0 

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

A List is a handle on one list part — comments, a change log, the messages beside a document.

func (*List) Append added in v0.10.0 

func (l *List) Append(values ...[]byte) error

Append adds values after the last one, which is what a chat or a log does.

func (*List) Delete added in v0.10.0 

func (l *List) Delete(pos, count int) error

Delete removes count values from index pos.

func (*List) Get added in v0.10.0 

func (l *List) Get(pos int) ([]byte, error)

Get returns a copy of the value at index pos.

func (*List) Insert added in v0.10.0 

func (l *List) Insert(pos int, values ...[]byte) error

Insert adds values at index pos, locally and then everywhere.

func (*List) Len added in v0.10.0 

func (l *List) Len() int

Len returns how many values are present.

func (*List) Name added in v0.10.0 

func (l *List) Name() string

Name returns the part's name.

func (*List) Part added in v0.10.0 

func (l *List) Part() crdt.Part

Part names this handle's part.

func (*List) Values added in v0.10.0 

func (l *List) Values() [][]byte

Values returns copies of every value present, in order. It is what a view of a list reads when it is told the list changed; see crdt.PartChange.

type Map added in v0.10.0 

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

A Map is a handle on one map part, such as the cells of a sheet.

func (*Map) Delete added in v0.10.0 

func (m *Map) Delete(key string) error

Delete removes key, locally and then everywhere. It writes a tombstone whether or not this replica holds the key; see crdt.Map.

func (*Map) Edit added in v0.29.0 

func (m *Map) Edit(fn func(*crdt.Map) ([]crdt.MapOp, error)) error

Edit runs fn against the map part itself and sends whatever operations it produced to everyone else.

It is what a structured type built on a map is driven through. A github.com/go-crdt/crdt/structured.Sequence, a Tree or a RecordMap is a binding over a crdt.Map rather than a thing of its own, and each of its methods changes the map here and hands back the operations that change: 

err := m.Edit(func(mp *crdt.Map) ([]crdt.MapOp, error) {
    _, ops, err := structured.SequenceOf(mp).Insert(after, value)
    return ops, err
})

The lock is held for the whole of fn, so what it reads and what it writes are one moment. fn must not call back into the client.

func (*Map) Get added in v0.10.0 

func (m *Map) Get(key string) ([]byte, bool)

Get returns a copy of the value at key, and whether the key is present. It is what a view reads for each key a crdt.PartChange names.

func (*Map) Keys added in v0.10.0 

func (m *Map) Keys() []string

Keys returns the keys present, ascending.

func (*Map) Len added in v0.10.0 

func (m *Map) Len() int

Len returns how many keys are present, not counting deleted ones.

func (*Map) Name added in v0.10.0 

func (m *Map) Name() string

Name returns the part's name.

func (*Map) Part added in v0.10.0 

func (m *Map) Part() crdt.Part

Part names this handle's part.

func (*Map) Read added in v0.29.0 

func (m *Map) Read(fn func(*crdt.Map))

Read runs fn against the map part for reading. It sends nothing, and anything fn writes to the map would be this replica's alone — so do not: use Map.Edit.

The lock is held for the whole of fn, so a view built from several reads sees one moment rather than a moment per read.

func (*Map) Set added in v0.10.0 

func (m *Map) Set(key string, value []byte) error

Set stores value at key, locally and then everywhere.

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) Idle added in v0.26.1 

func (s *MemoryStore) Idle(_ context.Context, d time.Duration) ([]string, error)

Idle returns the documents this store has not been asked to write for longer than d, which is a MemoryStore's half of Archivable.

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) Release added in v0.26.1 

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

Release forgets a document if this store still holds exactly want.

The comparison and the removal are one step under the same lock, so a save that lands while a document is being archived cannot be deleted by the release that follows it.

func (*MemoryStore) Save 

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

Save records a copy of the snapshot.

type MultiStore added in v0.24.0 

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

A MultiStore keeps every document in several stores at once.

What it is for

The stores in this project answer different questions. A database answers "what is the document now", quickly, which is what a server restarting needs. A git repository answers "what did it say last Tuesday, and who wrote this sentence", which is what a person needs. Neither answers the other's question, and an operator who wants both has until now had to choose.

The alternative already in use is worse than choosing: writing to one store from the server and to the other from somewhere else — a browser, a sync job — which is two sources of truth held together by whichever of them happens to run last.

Reading merges rather than picking

[Load] reads every store and merges what they return. The obvious design is to read the first store that has the document and stop, and it is wrong here for a reason particular to this problem: a save that failed halfway leaves the stores holding different documents, and reading only the first would quietly drop whatever only the second had. Merging is the only answer that loses nothing, and a CRDT is what makes it available.

It has a consequence worth having on purpose: adding a store to a running server backfills it. The new store returns nothing, the merge is the other store's document unchanged, and the next save writes it across.

The cost is that opening a document reads every store instead of one, and merges when more than one has content. That is paid once per document, when it is opened, and not per edit.

A store that cannot be read makes the document unavailable

If any store fails to read, [Load] fails. It does not fall back to the stores that answered, because what came back would be a document that is missing whatever the unreadable store alone held — and the next save would then write that shortened document over the store that was merely unreachable. Serving a document that is quietly missing a paragraph is worse than serving none: an error stops at one document and an operator can fix it, while silent loss is discovered by the person who wrote the paragraph.

Writing tries every store, and fails if any refused

[Save] writes to all of them even after one has failed, so that a store being down does not stop the others from being written, and then reports every failure together. It returns an error if any store refused, because a caller that gets nil back has to be able to believe the document is durable in all of them.

They are written one after another rather than at the same time because there are two of them, not two hundred.

func NewMultiStore added in v0.24.0 

func NewMultiStore(stores ...Store) *MultiStore

NewMultiStore returns a store that writes to all of the given stores and reads from all of them.

It panics if given none: a store that silently keeps nothing would look like it was working, and there is no configuration in which that is what somebody meant. One is allowed, and behaves as that store does.

func (*MultiStore) Load added in v0.24.0 

func (m *MultiStore) Load(ctx context.Context, document string) ([]byte, error)

Load returns the merge of what every store holds, or nil if none of them has the document yet.

func (*MultiStore) Save added in v0.24.0 

func (m *MultiStore) Save(ctx context.Context, document string, snapshot []byte) error

Save writes the snapshot to every store, and reports every store that refused it.

type PipeConn added in v0.21.0 

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

PipeConn is the server end of a Pipe, the side Server.ServePipe serves. It is an opaque handle: the session is spoken through it, not by the caller.

type RetryPolicy added in v0.24.0 

type RetryPolicy struct {
	// Wait is how long to wait before the first attempt after a link drops.
	// Each further failure doubles it, up to Ceiling. Defaults to
	// [DefaultRetryWait]; it may not be negative, and may not exceed Ceiling.
	Wait time.Duration

	// Ceiling is the longest the link will ever wait between attempts.
	// Defaults to [DefaultRetryCeiling]; it may not be negative.
	Ceiling time.Duration

	// Permanent, when set, is asked about the error that ended each attempt and
	// stops the link by returning true — the error it was asked about is then
	// what [Server.FollowWithRetry] returns.
	//
	// It exists because this package cannot answer the question honestly for
	// everybody. The errors that are genuinely permanent are refused before the
	// loop is ever entered, and everything that can then end an attempt came
	// off a network or off a peer, where "permanent" is a judgement about a
	// deployment and not about an error value. A peer refusing the link is the
	// case that decides the shape: it is a policy decision, policy is edited
	// and credentials are rotated, so a link that gave up on it would need
	// somebody to notice and restart a process — while a link that keeps asking
	// costs one attempt per Ceiling, which is nothing. So the default is to
	// retry it, and an operator who disagrees says so here, with
	// [errors.Is] over whatever their peer returns.
	Permanent func(error) bool

	// Notify, when set, is told every time the link changes state: down, with
	// why and for how long, and up again. See [LinkStatus].
	//
	// A library has no business choosing where that goes. Writing to stdout
	// would put a federation link's troubles into the middle of whatever the
	// process's own output is, in a format nobody asked for, and a link that
	// says nothing at all is a link nobody can operate — an outage would be
	// visible only as a replica that had quietly stopped converging. So it is
	// handed over instead, and the operator's logger, metric or health check
	// decides.
	//
	// It is called on the link's own goroutine, in order, and blocking in it
	// blocks the link — including the moment it is trying to come back. It must
	// not call back into the link.
	Notify func(LinkStatus)
}

A RetryPolicy is how a link comes back, which Server.Follow says is the operator's to decide. This is where they decide it.

The zero value is a working policy: DefaultRetryWait doubling to DefaultRetryCeiling, jittered, retrying everything, telling nobody.

type Role added in v0.26.0 

type Role int

A Role is which side of the protocol a tab took, decided by [electRole]. 

const (
	// RoleClient is a tab that joins a document another tab is holding.
	RoleClient Role = iota + 1
	// RoleHost is the tab holding the document and answering the others.
	RoleHost
)

type Server 

type Server struct {
	// 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) Close added in v0.13.0 

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

Close stops the housekeeping Config.PersistEvery and Config.EvictAfter ask for, and saves everything that has changed. It does not end the sessions in progress: those belong to whatever is serving them, and stopping that is the caller's to do first.

Calling it twice is harmless. A server that asked for neither still has one, so a caller need not know which kind it configured.

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) Follow added in v0.17.0 

func (s *Server) Follow(ctx context.Context, peer Transport, document string, as crdt.SiteID) error

Follow makes this server a participant in another server's copy of a document, so that the two converge.

What it is for

Not capacity. One server holds a document for a thousand participants at about three kilobytes and two and a half microseconds each, flat, which is twelve percent of a core for a document five people are typing in — see BenchmarkFanOut. A second server earns its place for two other reasons: a participant far from the first pays the round trip on every keystroke echo, and a site that goes down takes its documents with it until somebody brings them back.

Both are answered by a replica near each participant rather than by splitting a document across servers, which is what this is. It is also what the CRDT is for: two replicas that have seen the same operations hold the same document, in any order, with no agreement about the order and nothing to coordinate on the write path. There is no leader here and no consensus, which is why it works between datacentres without paying a round trip per edit.

A participant. The server being followed cannot tell the difference and does not need to: a link joins its document, is sent what it is missing, and is broadcast to like anybody else. Everything the local document learns is sent out, and everything that arrives is applied and broadcast onwards — to everyone except the link it arrived on, which is the loop prevention the subscriber machinery already had.

That prevention is not enough on its own, and the missing half is in applyOperations: two servers that follow each other would otherwise pass an operation back and forth forever, each applying it harmlessly and telling the other again. Operations that do not advance the version are not passed on.

Per document

A link follows one document. The alternative — a link that mirrors a whole store — is simpler to operate and replicates documents nobody is looking at, which between continents is bandwidth spent on nothing. Idle documents are evicted here already, and a link is what keeps one alive, so the set of documents a server replicates is the set somebody is using.

What this does not do

It does not reconnect. A link that drops stays dropped, and the error is returned to whoever called Follow, because the policy for coming back — immediately, with a backoff, never — belongs to the operator and not to a library. Server.FollowWithRetry does not overturn that: it is one such policy, written down and opted into by an operator whose answer is "with a backoff", and Follow behaves exactly as it did for everyone else. It does not discover peers. It does not replicate presence: cursors are ephemeral and a link that carried them would have to decide what a cursor in another datacentre means when the link is a second behind.

func (*Server) FollowWithRetry added in v0.24.0 

func (s *Server) FollowWithRetry(ctx context.Context, dial Dialer, document string, as crdt.SiteID, policy RetryPolicy) error

FollowWithRetry follows document on the peer dial reaches, exactly as Server.Follow does, and re-establishes the link when it drops — waiting longer after each failure, never longer than the policy's ceiling, and never the same interval as anybody else.

It returns when ctx is cancelled, returning ctx's error, or when RetryPolicy.Permanent says an attempt's failure is not worth another, returning that failure. It returns immediately, without dialling, if the call itself is wrong: no dialler, no document name, a link claiming the server's own replica, or a policy that waits for a negative time or longer than its own ceiling.

Why this is here rather than in every caller

Server.Follow's reasoning stands: the policy belongs to the operator. What does not follow from it is leaving everyone to write the loop, because it is the same loop every time and it is usually written twice wrong. Without jitter, every link in a datacentre that lost the same peer waits the same interval and returns together, so the peer coming back up meets the whole fleet at once and goes down again — the retry itself becomes the outage. Without a ceiling, doubling either arrives somewhere absurd or, more often, is bounded by an attempt counter that gives up, and a federation link that gives up is a replica that stops converging while the process it lives in carries on looking healthy.

So the loop is written here, once, and nothing acquires it by accident. Server.Follow is unchanged and remains what a caller with a different answer builds on.

What is retried, and what is not

Everything the loop can see, and that is a deliberate line rather than a shrug. The two failures that are genuinely permanent — a document with no name, and a link claiming [serverSite] — are decided once, before the loop is entered, so they are returned to the caller rather than re-asked forever. After that, an attempt can only end because a dialler could not reach the peer, because a carrier broke, because the peer refused the link or spoke something unexpected, or because ctx ended. The first two are what a link between datacentres does on a normal day. The third is a deployment's business, not this package's, which is what RetryPolicy.Permanent is for. The last ends the loop rather than being retried.

Jitter

The delay is drawn uniformly from the half-open band between half the current interval and the whole of it, rather than from zero to it: full jitter decorrelates best but can draw a delay near zero many times running, which is the hammering this exists to avoid, while a floor of half the interval bounds the attempt rate and still leaves no two links in step.

It is not a knob. An operator has something to say about how long to wait and how stale they will tolerate, and nothing to say about a jitter fraction — but offered the field they would be able to set it to zero, which is the single mistake this whole function exists to prevent.

func (*Server) ServePipe added in v0.21.0 

func (s *Server) ServePipe(ctx context.Context, sc *PipeConn) error

ServePipe runs one session over the server end of a Pipe, with this server holding the document. It returns when the session ends — because the client end closed, because ctx was cancelled, or because the session itself failed.

It is the in-process counterpart of Server.ServeWebSocket and, in a browser, [Server.ServeDataChannel]: there is no request to upgrade and no origin to check, because there is no boundary to cross. Give it the PipeConn that Pipe returned beside the Transport the local editor joined over.

func (*Server) ServeWebSocket added in v0.11.0 

func (s *Server) ServeWebSocket(origins ...string) http.Handler

ServeWebSocket returns an http.Handler that runs sessions over WebSockets — the carrier a browser can afford, and the one WebSocket dials.

Mount it where the browser will reach it. Everything else is the same server: the same documents, the same store, the same Config.Authorize, and a participant here edits the same document as one arriving over gRPC.

origins, when not empty, are the Origin header values allowed to open a session, which is the check that stops another site's page from opening one with the visitor's cookies. An empty list allows only same-origin requests.

func (*Server) Stable added in v0.32.0 

func (s *Server) Stable(name string) (crdt.CompositeVersion, bool)

What every participant has certainly seen.

A replica may drop a tombstone once every replica has delivered the deletion that made it — see crdt.Doc.Collect, which asks for such a version and cannot compute one. A server can: it is the thing every operation passes through, and participants tell it what they have applied.

This is the telling. A participant sends its version after it applies what the server sent; the server keeps the last one from each, and the meet of them — the element-wise minimum — is what everybody here has. Nothing depends on an acknowledgement arriving: one that is late or lost holds the answer back, and holding it back is the safe direction.

What it is not

It is the meet over the participants **connected now**. A replica that is offline holding work of its own is not in it and cannot be: the server has never heard of what it did. So this is not yet a version anything may be collected against — deciding that a replica is gone is a policy, and it is not one a version vector can make. What this gives is the measurement that policy would have to be worth making: whether, in a room that is actually being used, the meet advances at all.

Stable returns the version every participant of the named document has acknowledged, and false if the document is not open or nobody has said anything yet.

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.

type Text added in v0.10.0 

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

A Text is a handle on one text part: the buffer of a file, and what an editor binds to.

func (*Text) Anchor added in v0.10.0 

func (t *Text) Anchor(pos int) (crdt.ID, error)

Anchor returns the identity of the character at rune offset pos, which keeps naming that character however the text moves around it. It is what a comment or a stored selection should hold; see crdt.Doc.Anchor.

func (*Text) AnchorUTF16 added in v0.12.0 

func (t *Text) AnchorUTF16(pos int) (crdt.ID, error)

AnchorUTF16 is Text.Anchor with pos counted in UTF-16 code units, the units a page counts in. An offset falling between the two units of one character is refused rather than rounded; see crdt.ErrSurrogateBoundary.

func (*Text) AuthorRuns added in v0.10.0 

func (t *Text) AuthorRuns() []crdt.AuthorRun

AuthorRuns splits the visible text into stretches by who wrote them, which is what colouring a document by author needs.

func (*Text) AuthorRunsUTF16 added in v0.12.0 

func (t *Text) AuthorRunsUTF16() []crdt.AuthorRun

AuthorRunsUTF16 is Text.AuthorRuns with every offset and length counted in UTF-16 code units, so that a page can colour the string it holds without converting anything by hand.

func (*Text) Delete added in v0.10.0 

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

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

func (*Text) DeleteUTF16 added in v0.10.0 

func (t *Text) DeleteUTF16(pos, length int) error

DeleteUTF16 removes length code units at an offset counted in the same units.

func (*Text) Insert added in v0.10.0 

func (t *Text) Insert(pos int, text string) error

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

func (*Text) InsertUTF16 added in v0.10.0 

func (t *Text) InsertUTF16(pos int, text string) error

InsertUTF16 adds text at an offset counted in UTF-16 code units.

func (*Text) Len added in v0.10.0 

func (t *Text) Len() int

Len returns the number of characters, counted in runes.

func (*Text) LenUTF16 added in v0.10.0 

func (t *Text) LenUTF16() int

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 (*Text) Name added in v0.10.0 

func (t *Text) Name() string

Name returns the part's name.

func (*Text) Part added in v0.10.0 

func (t *Text) Part() crdt.Part

Part names this handle's part, which is what a crdt.PartChange from Client.TakeChanges carries.

func (*Text) Position added in v0.10.0 

func (t *Text) Position(anchor crdt.ID) (int, bool)

Position returns where the character an anchor names sits now — or where it was, if it has been deleted. See crdt.Doc.Position.

func (*Text) PositionUTF16 added in v0.12.0 

func (t *Text) PositionUTF16(anchor crdt.ID) (pos int, ok bool)

PositionUTF16 is Text.Position with the offset reported in UTF-16 code units. ok is false for an anchor this replica has never seen, exactly as it is there — which is not the same question as whether the character is still in the text; that one is Text.Visible.

func (*Text) String added in v0.10.0 

func (t *Text) String() string

String returns the text as it stands here.

func (*Text) Visible added in v0.10.0 

func (t *Text) Visible(anchor crdt.ID) bool

Visible reports whether the character an anchor names is still in the text.

type Tiered added in v0.26.1 

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

A Tiered store keeps documents somebody is using in one store and documents nobody has opened for a long time in another.

What it is for

A server holds every document it has ever served until it is told to let go — see Config.EvictAfter — and a store holds every document it has ever been given, with nothing to let go of it at all. A year of a busy service is a directory of documents that were opened once, and there has been no way to move them anywhere cheaper without deleting them.

Nothing is deleted that is not already somewhere else

Archiving is three steps in one order: read the hot store, write the cold one, then ask the hot one to release exactly what was read. Every way it can fail leaves the document somewhere. A cold store that will not take it releases nothing. A document that was written between the read and the release is not released, because Archivable.Release compares before it removes; it is archived on a later pass, when it has gone quiet again.

An archived document is not a missing one

Store.Load returning nil means "there is no such document, start a new one", and a server acts on it: it opens an empty document and, at its next save, writes it over whatever was there. So a Tiered store never answers nil for a document the cold store has, and never answers nil because it could not reach the cold store — it fails instead. A document that is unreachable is not a document that does not exist, and confusing the two is how a store loses what it was given.

Reading an archived document brings it back: it is written to the hot store on the way past, so the next read does not go looking again and the next save has somewhere to land.

func NewTiered added in v0.26.1 

func NewTiered(hot Archivable, cold Store) *Tiered

NewTiered returns a store that writes to hot and falls back to cold.

It panics if either is nil, which is a mistake in the call rather than a state anything could recover from.

func (*Tiered) Archive added in v0.26.1 

func (t *Tiered) Archive(ctx context.Context, idleFor time.Duration) (int, error)

Archive moves every document the hot store has not been asked to write for idleFor into the cold store, and returns how many it moved.

It is a method rather than a timer of its own because a server already has housekeeping running on a schedule an operator chose, and a store that starts goroutines is a store that has to be closed.

A document that could not be archived does not stop the ones after it: the errors are returned together, and what was moved is reported whatever else happened.

func (*Tiered) Load added in v0.26.1 

func (t *Tiered) Load(ctx context.Context, document string) ([]byte, error)

Load returns the document from the hot store, or from the cold one, bringing it back to the hot store on the way.

func (*Tiered) Save added in v0.26.1 

func (t *Tiered) Save(ctx context.Context, document string, snapshot []byte) error

Save writes to the hot store, which is where a document being edited belongs.

type Transport added in v0.11.0 

type Transport interface {
	// contains filtered or unexported methods
}

A Transport is how a participant reaches a server. WebSocket works anywhere, a browser included; GRPC works outside one.

There are two because of what they cost where they run. Outside a browser a carrier costs nothing anybody notices, and gRPC brings deadlines, interceptors and everything already built around them. Inside one it is paid for on every load: protobuf alone is six times the size of the whole CRDT compiled to wasm — see wire.go for the measurements — so the browser gets a framing of four message kinds over a plain WebSocket instead.

Both carry the same session, byte for byte in the fields that matter, because every field in these messages is something github.com/go-crdt/crdt encoded and will check on arrival. A participant on one and a participant on the other can edit the same document.

func GRPC added in v0.11.0 

func GRPC(conn grpc.ClientConnInterface) Transport

GRPC returns a transport that opens sessions on a gRPC connection.

It is deliberately not the default. Everything it carries is bytes some encoder in github.com/go-crdt/crdt produced, so protobuf is describing fields nobody reads through it — and compiled for a browser it costs six times the CRDT itself. Outside a browser that does not matter, and gRPC brings deadlines, interceptors and the tooling built around them, which is reason enough to keep it. See Transport and WebSocket.

func WebSocket added in v0.11.0 

func WebSocket(url string, opts ...WebSocketOption) Transport

WebSocket returns a transport that opens sessions at url, which is "ws://" or "wss://" and the path the server's handler is mounted at.

This is the transport a browser uses, and the one to reach for by default: it is what the same code compiled to wasm can afford. See Transport.

type WebSocketOption added in v0.11.0 

type WebSocketOption func(*wsTransport)

A WebSocketOption configures WebSocket.

func WithHTTPHeader added in v0.11.0 

func WithHTTPHeader(h http.Header) WebSocketOption

WithHTTPHeader sends these headers with the opening handshake, which is where a cookie or a bearer token goes when the participant is not a browser.

It does not exist for a browser, because a page cannot put a header on a WebSocket handshake — and does not need to, since the browser sends the cookies for that origin itself. Code meant to run in both places should let the cookie do the work; see Config.Authorize.

Source Files

View all Source files

Directories

Path Synopsis
Command browsertest is the real-browser half of the WebRTC proof.
 Command browsertest is the real-browser half of the WebRTC proof.
gitstore module
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
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.