@aweftjs/sync

Commits between documents over any channel. Both ends of a link run the same code, and the link decides nothing about what it carries.

Three layers, and you take as many as you want:

The whole of a transport

A channel is four functions. Nothing in that type knows about a socket:

interface Channel {
	send(frame: Frame): void;
	receive(fn: (frame: Frame) => void): () => void;
	closed(fn: () => void): () => void;
	close(): void;
}

inProcess(), fromWebSocket(socket) and fromMessagePort(port) ship. Anything else is yours, and it is the same size: put encodeFrame(frame) on the wire, hand decodeFrame(bytes) back. recipes/sync/main.ts writes one over a TCP socket in about forty lines and runs the whole scenario over it, beside the shipped ones, to make the point that they are not special.

Delivery is always asynchronous, on every channel including the in-process one. Applying a commit from inside a watcher hands it to the receiving document's watchers after the outer watcher has already returned, so a link that delivered synchronously would work in one process and break on a socket.

Quickstart

Both ends are the same few lines. Opening the socket is yours, at either end.

import { connect, fromWebSocket } from '@aweftjs/sync';

const link = connect(fromWebSocket(socket));
link.share('board', board);                                  // this end has the document
board.tasks.push(createObject({ title: 'write it up' }));    // applies here, and goes

const link = connect(fromWebSocket(new WebSocket(url)));
const shared = link.share<Board>('board');                   // this end has nothing yet
const board = await shared.ready;                            // minted from the other end's root, then its state

One link carries as many documents as you share on it. Each is a topic named by the string you chose; the name rides on the open frame and on nothing after it.

link.share('profile', profile);
link.share('inbox', inbox);

If neither end holds the document, both hear no-document through fault and ready rejects, rather than the two waiting on each other forever.

link.close() ends everything on it. Nothing resumes: a new channel is a new link, and re-sharing on it is yours, the same way opening the socket was. If the channel ends under the link instead (the other end went away, the socket dropped), every share on it hears closed through fault, and a ready still waiting rejects with it.

Your rules go in accept

An arriving commit is handed to accept before it applies. Return the reasons to refuse it; an empty array accepts. The default accepts everything. The link has no idea what you check.

link.share('inbox', inbox, {
	accept: (commit) => commit.deltas.some((d) => d.type === 'remove')
		? [{ code: 'read-only', message: 'nothing is removed from an inbox' }]
		: [],
});

Which end may write what is a rule you write here, or a topology you arrange: one end that refuses is a server, if you want one. The package never asks who made a commit.

A refusal is reported at both ends

A commit that does not apply, because accept said no or because the applier refused it, refuses that commit and nothing else. The next commit in the same frame still applies. Both ends hear about it:

link.share('board', board, {
	refused: ({ mine, seq, reasons, commit, undo }) => {
		// mine: true when this end sent it, false when this end refused it
	},
});

The end that sent it is usually the one that can act: its write did not land over there, and undo is the commit that takes it back here. Apply it to yield, or ignore it to keep your version and let the other end yield instead. The link never chooses.

Conflicts, and who yields

Inserts never collide: two ends adding to one list at the same moment both keep their item. Replacing one slot at the same moment is different: each end applies its own write and then the other's, so the two end up swapped, and stay that way until one of them yields.

Yielding is asking the other end for its state:

shared.resync();     // this end throws its version away and moves to the other end's

The document you hold is moved, never swapped: every watcher, derived value and reference keeps working, and the change reads like any other commit. Write the rule that decides who yields once, in your refused handler or wherever your application knows best.

A state moves this end only when this end asked for one, by sharing with no document or by resync(). One that arrives unasked is refused with the fault unwanted-state at both ends, applies nothing, and ends the topic, so an end that refuses writes through accept cannot be moved past it by the other end sending its state instead (design 269).

Several networks on one document

A document can be on any number of links, in a store, and under any number of watchers at the same time, and each hears what the others land. A commit that arrives over one link is an ordinary local commit to every other link on that document, so:

• a document shared over a socket and persisted by @aweftjs/store at the same time puts every arriving commit into the store's history without either knowing about the other;

• a node holding one document at the end of two links forwards between them, which is how a chain of three converges.

The proof program runs both.

Two documents in one process

const editing = mirror(stored);
await Promise.resolve();
(editing.document as typeof stored).title = 'draft';     // reaches `stored` at the end of the tick

mirror is a link over an in-process pair. The same rules apply: asynchronous delivery, nothing echoed, both ends equal.

Just the commits

track(doc, on) is the whole engine underneath, for an application with its own protocol:

const tracker = track(doc, ({ commit, undo, landed }) => {
	if (!landed) send(commit);        // made here: ship it
});
tracker.receive(arrived);             // from elsewhere: applies, and is not shipped back

landed tells a commit that arrived through receive from one made here, by counting deliveries rather than holding a flag, so a watcher that writes in answer to an arriving commit produces a local commit that ships. receive must not be called from inside a watcher; every channel here queues and applies on a microtask.

asCommit(doc) says a whole document as one commit, reconcile(doc, snapshot) computes the commit that moves a document to another's state, and rootFrom(id, kind) mints an empty root that a commit about that document can reach.

Requests beside the link

A link carries commits and nothing else. When one end has to ask the other for something that is not state (a report, a search, a job started), requests(socket) puts JSON text on the same WebSocket the link's binary frames ride, and the two never meet: the socket adapter ignores text, requests ignores bytes.

import { connect, fromWebSocket, requests } from '@aweftjs/sync';

const link = connect(fromWebSocket(socket));      // commits, as binary
const asks = requests(socket);                    // requests, as text, on the same socket

asks.answer((name, args, progress) => handle(name, args, progress));   // this end answers
const result = await asks.ask('report/Daily', { day: 'mon' }, {        // and asks
	progress: (value) => bar.set(value),
	timeout: 5000,
});

Both ends may ask and both may answer. Arguments and results are what JSON.stringify carries, with undefined read as null; a value it cannot carry is refused as not-data. An answerer that throws answers with its error's reason and message, plus its reasons when it carries a list of them, and ask rejects with the same three. timeout is yours to set; none ships. When the socket closes, every ask still waiting rejects with closed. A text message that is not a request frame closes the socket, the way bytes that are not a frame end the link and close the socket under it. stop() stops asking and answering and leaves the socket alone. One requests per socket: a second throws duplicate until the first has stopped.

The frames are { id, name, args }, { id, result }, { id, progress } and { id, error }, so a client in another language writes them by hand. Design 073.

A remote error keeps its own words

Every refusal in this stack renders as reason: detail. fix. An error that came back from the other end of requests is the exception: its message is the answerer's message, unchanged, because you asked a remote question and that is the remote answer. The reason and the fix are still on the error object, and explain(error) from @aweftjs/debug shows all three.

try {
	await asking.ask('rebuild', { id });
} catch (e) {
	console.log(e.message);   // what the far end said
	console.log(e.reason);    // what to branch on
	console.log(e.fix);       // what to do about it
}

Boundaries

• Nothing about who. The link does not know who made a commit or who may write where; that is accept, and it is yours.

• No sessions, no resume. A new channel is a new link.

• No winner. Two ends that conflict stay in conflict until one yields.

• No messages on the link. A link carries commits and nothing else; an intent is state in the document, and a request is requests, beside the link and never inside it.

• No storage. That is @aweftjs/store, on the same document, beside the link.

Known limits

One live topic per name on a link. An arriving open is paired with the first share of that name that has no partner, so a second share('state') on a link that already has a live state never settles: its ready waits forever and no fault is raised. A share made after stop() waits the same way until the other end offers the name again. Share a name once per link and hand that one handle around; settling it means refusing a second share of a live name, or faulting an open for a name that is already paired.

API

Every export of @aweftjs/sync, its signature as the compiler resolves it, and its block comment.

@aweftjs/sync

Answerer

type Answerer = (name: string, args: unknown, progress: (value: unknown) => void) => unknown;

What this end does with a request addressed to it. Return the result, which may be a promise; throw to answer with an error, whose reason and reasons cross with it.

AskOptions

interface AskOptions { readonly progress?: ((value: unknown) => void) | undefined; readonly timeout?: number | undefined; }

No block comment on this export.

Channel

interface Channel { send(frame: Frame): void; receive(fn: (frame: Frame) => void): () => void; closed(fn: () => void): () => void; close(): void; }

A duplex link that carries frames in order.

A channel must deliver what it is given, in the order it was given, or close. The protocol detects loss and reordering but does not repair them: it resynchronizes instead.

Commit

interface Commit { readonly deltas: readonly Delta[]; readonly tag?: Tag; } (from @aweftjs/codec)

The unit that crosses every boundary.

A commit applies whole or not at all, and it carries at least one delta. Its deltas are a set, written in the canonical order of section 6.9, and a receiver replaying them one at a time would pass through states the sender never had.

tag is 4 to 32 bytes and is not a checksum of these bytes. It is a digest over the prior values of the slots this commit addresses, computed by the sender against its own state before the commit, and it answers whether the commit landed on the state the sender expected. A receiver computing a different tag treats the replicas as diverged and resynchronizes. The algorithm that fills it is still open, so nothing here computes one.

CommitsFrame

interface CommitsFrame { readonly kind: 'commits'; readonly topic: number; readonly first: number; readonly commits: readonly Commit[]; }

Commits, in the order they were made, starting at the sequence number stated.

FaultFrame

interface FaultFrame { readonly kind: 'fault'; readonly topic: number; readonly reason: string; readonly message: string; }

The link cannot carry on for this topic.

A refusal is not a fault: refusing a commit never ends anything. A fault is two documents that are not one document, or a frame about a topic nothing is open under.

The topic is the number the end being told uses, because the two cases that raise a fault are exactly the two where the sender has no number to give: nothing is open under it, or the topic is being turned away before it ever opened. Topic 0 is the link itself.

Frame

type Frame = OpenFrame | StateFrame | CommitsFrame | RefusedFrame | LeaveFrame | FaultFrame;

Everything one end of a link can say to the other.

LeaveFrame

interface LeaveFrame { readonly kind: 'leave'; readonly topic: number; }

This end is done with the topic. The link stays up for the others.

Link

interface Link { share<T extends object>(name: string, document?: T, handlers?: ShareHandlers): Shared<T>; close(): void; }

A link over one channel, carrying any number of documents.

LinkOptions

interface LinkOptions { readonly window?: number; }

What a link may be told about how it runs.

OpenFrame

interface OpenFrame { readonly kind: 'open'; readonly topic: number; readonly name: string; readonly root: RootRef | null; readonly want: boolean; }

Shares a document under a name, and gives it the number the sender's later frames use.

want asks the other end to say its whole document, which an end sets when it holds nothing for the name: a document it has just minted, or one it has chosen to give up.

PortLike

interface PortLike { postMessage(value: unknown): void; addEventListener(type: 'message' | 'messageerror', fn: (event: { data: unknown; }) => void): void; start?(): void; close(): void; }

What this package needs from a MessagePort, stated structurally so no DOM types leak in.

Refused

interface Refused { readonly mine: boolean; readonly seq: number; readonly reasons: readonly WireReason[]; readonly commit?: Commit; readonly undo?: Commit; }

One commit that did not apply, at whichever end refused it.

RefusedFrame

interface RefusedFrame { readonly kind: 'refused'; readonly topic: number; readonly seq: number; readonly reasons: readonly WireReason[]; }

One commit did not apply here, and here is why.

topic is the refusing end's number and seq is the sequence the other end assigned. A refusal refuses the commit and never the link.

RequestError

interface RequestError extends Error { readonly reason: string; readonly reasons?: readonly WireReason[]; }

What ask rejects with, and what an answerer's throw becomes at the other end.

Requests

interface Requests { ask(name: string, args?: unknown, options?: AskOptions): Promise<unknown>; answer(fn: Answerer): () => void; stop(): void; }

Requests over one socket, both directions.

RootRef

interface RootRef { readonly id: Id; readonly kind: ObservableKind; }

Where a document starts: the id of its root observable, and which kind that root is.

ShareHandlers

interface ShareHandlers { readonly accept?: ((commit: Commit) => readonly WireReason[]) | undefined; readonly refused?: ((report: Refused) => void) | undefined; readonly fault?: ((reason: string, message: string) => void) | undefined; }

What the application does about the commits crossing one topic.

Shared

interface Shared<T extends object> { readonly document: T | undefined; readonly ready: Promise<T>; resync(): void; stop(): void; }

One document shared on a link.

SocketLike

interface SocketLike { binaryType: string; readyState: number; send(data: Uint8Array | string): void; close(): void; addEventListener(type: 'open' | 'message' | 'close' | 'error', fn: (event: { data?: unknown; }) => void): void; }

What this package needs from a WebSocket, stated structurally so no DOM types leak in.

A binary message is a frame and belongs to the channel; a text message is not a frame and belongs to whatever else shares the socket (requests), so send takes both.

StateFrame

interface StateFrame { readonly kind: 'state'; readonly topic: number; readonly commit?: Commit; }

The sender's whole document, said as one commit. Absent when the document is empty.

Tracked

interface Tracked { readonly commit: Commit; readonly undo: Commit; readonly landed: boolean; }

One commit this document took part in, and the commit that undoes it.

Tracker

interface Tracker { receive(commit: Commit): void; stop(): void; }

A document being tracked for replication.

WireReason

interface WireReason { readonly code: string; readonly message: string; readonly path?: readonly string[]; }

Why a commit was refused, in the form that crosses a link.

The commit itself is not carried. The end that hears a refusal is the end that sent the commit, so it already holds every delta; what it does not hold is the cause and the place.

asCommit

asCommit: (observable: unknown) => Commit | undefined

Say a whole document as one commit of add deltas.

Params

• observable: any observable in the document. The document is read from its root

Returns one commit that builds everything the document holds, or undefined when it holds nothing, because a commit carries at least one delta. The commit does not create the root: nothing addresses the root except by already having it. A receiver mints its root with the same id first, createObject(undefined, rootId) for an object root, and applies this to it. This is what an end with nothing is sent when it asks for the state, and it is also how a document built by mutation is handed to anything that wants it as one commit.

Throws not-observable when the value is not part of a document.

Example

const whole = asCommit(doc);
if (whole !== undefined) apply(replica, whole);

connect

connect: (channel: Channel, options?: LinkOptions) => Link

Share documents with the other end of a channel.

Params

• channel: the transport, already open. Both ends call this, with the same code

• options: window, how many sent commits per topic are kept for reporting a refusal

Returns the link. share puts a document on it; close ends it. A link decides nothing about the commits it carries. It numbers topics, batches what goes out, hands an arriving commit to the topic's accept, applies what is accepted and answers what is not. Which end yields in a conflict is the application's (design 054). Nothing resumes: a new channel is a new link, and re-sharing on it is the application's, the same way opening the socket was.

Example

const link = connect(fromWebSocket(socket));
const board = link.share('board', document);
await board.ready;

decodeFrame

decodeFrame: (bytes: Uint8Array<ArrayBufferLike>) => Frame

No block comment on this export.

encodeFrame

encodeFrame: (frame: Frame) => Uint8Array<ArrayBufferLike>

Turn a frame into the bytes a transport carries.

Params

• frame: the frame to write

Returns the bytes. The same frame always writes the same bytes, because every value in it goes through the one canonical encoding this stack has.

Throws whatever the encoding refuses, naming the rule broken: empty-commit, invalid-value, lone-surrogate and the rest. A frame built from a document this end holds breaks none of them.

Example

socket.send(encodeFrame({ kind: 'leave', topic: 3 }));

fromMessagePort

fromMessagePort: (port: PortLike) => Channel

A channel over a MessagePort, a Worker, or anything shaped like one.

Params

• port: the port. It is started for you if it needs starting

Returns a channel. Frames cross as bytes, which is the same encoding a socket carries, so a sandbox and a server are not two protocols.

Example

const channel = fromMessagePort(worker);

fromWebSocket

fromWebSocket: (socket: SocketLike) => Channel

A channel over a WebSocket, open or still connecting.

Params

• socket: the socket. Its binaryType is set to arraybuffer for you

Returns a channel. A frame handed over while the socket is still connecting is held and sent, in order, once it opens, so a link may share before the socket is up. A socket that is closing or closed ends the channel. Opening the socket stays yours, which is what makes reconnecting yours too: hand connect a function that makes a new one. A frame is one binary message. Frames are not compressed here: measured on a real commit stream, gzip per frame is larger than the frames themselves, because a 54 byte frame cannot pay for a gzip header. Turn on the transport's own shared-context compression instead.

Example

const link = connect(fromWebSocket(new WebSocket(url)));

inProcess

inProcess: () => [Channel, Channel]

Two channels wired to each other in one process.

Returns a pair. What one sends, the other hears, on a microtask. The frames are handed over as they are, with no encoding. Measured, that is 0.011 us against 4.29 us to encode and decode one, and nothing between the two ever leaves the heap. Treat a frame as read-only on both sides, which every type here already says.

Example

const [a, b] = inProcess();
const left = connect(a);
const right = connect(b);

mirror

mirror: (source: object, target?: object | undefined) => { readonly document: object; stop(): void; }

Keep a second document in step with a first, both ways.

Params

• source: any observable in the document to copy from. Its state is the state both start at

• target: the document to keep in step. Minted from the source's root when left out

Returns the document being kept in step, and the function that stops it. This is a link over an in-process pair, with connect at both ends and nothing else: the same code a link over a socket runs. It is asynchronous for the same reason every link is, so the two documents are in step at the end of the tick, not at the end of the statement. The target starts by asking for the source's state, so a target holding something else is moved to the source. After that the two are equal ends and a change on either crosses.

Throws not-observable when the source is not part of a document.

Example

const editing = mirror(stored);
await Promise.resolve();
(editing.document as { title: string }).title = 'draft';  // reaches `stored`

reconcile

reconcile: (observable: unknown, target: Snapshot) => Commit | undefined

The commit that turns this document into what a snapshot says.

Params

• observable: any observable in the document to move. It is read and not written

• target: the state to reach, as snapshot returns it

Returns one commit, or undefined when the document already says the same thing. This is what makes a resynchronization keep the document. Rebuilding from a snapshot would hand back a different tree, and every watcher, every derived value and every piece of interface holding the old one would be pointing at a document nothing writes to any more. Applying a difference leaves identity alone and reads to a watcher like any other change. The two documents must share a root id: a document is only ever moved to another state of itself.

Throws not-observable when the value is not part of a document, and root-mismatch when the two do not share a root id.

Example

const fix = reconcile(mine, snapshot(theirs));
if (fix !== undefined) apply(mine, fix);

requests

requests: (socket: SocketLike) => Requests

Requests over a socket, beside a link.

Params

• socket: the same socket fromWebSocket takes, open or still connecting. Text messages are this channel's; binary ones are the link's and are ignored here

Returns this end's ask, answer and stop. A request is one text message, { id, name, args }; its answer is { id, result }, each progress report { id, progress }, and a failure { id, error }. A text message that is not one of those closes the socket, as bytes that are not a frame end the link. A request that arrives with no answerer registered is answered missing. One channel per socket: a second requests on the same socket throws duplicate until the first has stopped.

Throws duplicate when this socket already has a request channel.

Example

const asks = requests(socket);
asks.answer((name, args, progress) => handle(name, args, progress));
const result = await asks.ask('report/Daily', { day: 'mon' });

rootFrom

rootFrom: (id: Id, kind: ObservableKind) => object

Make an empty document root of a stated kind and id.

Params

• id: the root's id. A replica shares its source's, or commits about it are unreachable

• kind: which of the three kinds the root is

Returns the root observable, holding nothing. Apply asCommit of the source to fill it.

Throws kind-conflict for a kind that is not object, array or map.

Example

const doc = rootFrom(frame.root.id, frame.root.kind);

track

track: (document: unknown, on: (event: Tracked) => void) => Tracker

Watch a document for replication.

Params

• document: any observable in the document. The whole document is tracked, from its root

• on: called once per commit, in the order the document saw them

Returns a tracker. receive is the only way to apply a commit to a tracked document, and calling apply around it would make an arriving commit look locally made. receive must not be called from inside a watcher. Delivery is deferred, so a commit applied there reaches its watchers after the outer one returned, and the ordering this rests on is gone. Queue it and apply on a microtask, which is what every channel here does.

Throws not-observable when the value is not part of a document.

Example

const tracker = track(doc, ({ commit, landed }) => {
  if (!landed) channel.send({ kind: 'commits', topic, first: next++, commits: [commit] });
});

Refusals

Every error this package raises carries a reason to switch on and a fix that says what to do. These are its reasons, from errors.txt.

Recipes

The programs in the stack's gate that use this package, each a job someone would have.

• recipes/two-clients: Two people editing one document at once, including what happens when they write the same slot and who yields

• recipes/optimistic-write: A write that applies locally before the server sees it, is refused, and is rolled back

• recipes/sync: the package's own recipe