Sync protocol implementation (third version)

[ept]

Here is our third implementation of the sync protocol discussed in #290, superseding #324 and #332. This is joint work with @pvh, @orionz, @alexjg, and @jeffa5. We finally have an API that is simple and elegant, that encapsulates most of the complexity (simplifying network protocol implementations), and that gives us lots of flexibility for the future. In particular, it has the following properties:

• Functional (side-effect-free) API to simplify reasoning about the code and its states; protocol state for the sync with another node is encapsulated in a SyncState object that is passed into and returned from the functions
• Works with any network topology and transfer protocol, including client-server (e.g. over WebSocket), peer-to-peer (e.g. over WebRTC), and server-to-server
• Allows large sets of changes to be broken into chunks that are sent and applied incrementally (allowing progress to be indicated to the user)
• Efficient updates when nodes are already mostly in sync; the same protocol provides both real-time collaboration (small updates whenever a document changes) and also bulk re-sync after nodes have been offline for some time
• When syncing concurrently with several peers, the protocol provides a mechanism to avoid receiving the same set of changes multiple times from different peers (though this feature is not currently exposed in the API)
• Resilient to message loss, interrupted connections, and crashes that cause data loss
• Cryptographic consistency guarantee: because we check hash chains, when sync completes between two nodes, they are cryptographically guaranteed to have the same set of changes and thus be in the same state

The API consists of the following functions on the top-level Automerge object:

function initSyncState(): SyncState
function generateSyncMessage<T>(doc: Doc<T>, syncState: SyncState): [SyncState, BinarySyncMessage?]
function receiveSyncMessage<T>(doc: Doc<T>, syncState: SyncState, message: BinarySyncMessage): [Doc<T>, SyncState, Patch?]

and the following functions on Automerge.Backend:

function initSyncState(): SyncState
function generateSyncMessage(state: BackendState, syncState: SyncState): [SyncState, BinarySyncMessage?]
function receiveSyncMessage(state: BackendState, syncState: SyncState, message: BinarySyncMessage): [BackendState, SyncState, Patch?]
function encodeSyncMessage(message: SyncMessage): BinarySyncMessage
function decodeSyncMessage(bytes: BinarySyncMessage): SyncMessage
function encodeSyncState(syncState: SyncState): BinarySyncState
function decodeSyncState(bytes: BinarySyncState): SyncState

where BinarySyncMessage and BinarySyncState are type aliases for Uint8Array.

When connecting to another node for the first time, create a new SyncState with initSyncState() and then call generateSyncMessage() to get a message to send to the other node. You can have one node initiate the sync or you can have both nodes send each other sync messages concurrently — either is fine. On receiving a message, pass it to receiveSyncMessage() (which may update the document if changes were received).

The protocol may require several communication round trips. Therefore, after calling receiveSyncMessage() you should again call generateSyncMessage() to generate a response message (always passing the SyncState from the previous return value into the next function call). generateSyncMessage() will return a null message if the sync is complete and no further message needs to be sent. After the document has changed, you can call generateSyncMessage() again, and it will return a new non-null message to inform the other side about the new changes.

When a sync is complete or when the connection between two nodes is closed, the most recent SyncState should be encoded as a byte array using Automerge.Backend.encodeSyncState() and written to disk. When we later reconnect to the same remote node, reload this state using Automerge.Backend.decodeSyncState() and use it instead of calling initSyncState(). This improves efficiency by remembering how far we got in the last sync, and allows the protocol to focus just on what has changed since the last sync. Note that if you sync with multiple peers, each peer connection needs its own SyncState.

The decodeSyncMessage() function decodes a binary message into a readable JSON object, which is mostly useful for debugging. encodeSyncMessage() re-encodes such a JSON object into a binary message, which enables some advanced customisations of the protocol; most people will not need to do this.

A limitation of the current implementation is that it always sends a log of individual changes, even if a whole compressed document would be more compact. We plan to address this later (we can make this change without affecting the sync protocol API).

[dan-weaver]

is the lastSyncState an additional optimization to the bloom filter?

I’ve been wondering if say that state is discarded and your starting from an unknown sync state with non empty documents, I’m sure there’s practical benefits to using the protocol but for the sake of my understanding, will that first message be equivalent to sharing the entire document or will using the protocol still generally perform better in most cases?

[ept]

@dan-weaver Yes, it's another optimisation. If you discard it, the protocol will still work fine, but the first message of a sync will be bigger. The size of this first message grows by 1-2 bytes for each change that was made since the last recorded sync state; if you discard the last sync state, that's 1-2 bytes for every change ever made in the document. This is definitely going to be smaller than the entire log of changes; whether it's bigger or smaller than the full document depends on the compression in your particular document.

[jeffa5]

One thought that's just come to me (while adding persistence of sync_states to automerge-persistent) is that we internally call applyChanges in receiveSyncMessage and don't return that list of changes. This makes it rather fiddly to get those changes out in order to do something like persist them.

I suppose we could either have receiveSyncMessage return the list of changes to be applied, letting the user apply them and get themselves a patch or otherwise return the list of changes alongside the patch (though I think I prefer the former).

This also has the nice(?) effect of meaning that this method will no longer mutate the document (backend). It gives more focus for the mutation on where the user calls applyChanges themselves.

[pvh]

Oh yeah, this is an obvious mistake. Is getting the changes back the best way to solve this? I'm thinking about how we're leaning into RLE / columnar encoding of changes. I wonder if we ought to do something like save(lastSave) that returns any changes since your last save() call?

[ept]

You should be able to do something like this to get the changes in a message:

const docBefore = doc;
[doc, syncState] = Automerge.receiveSyncMessage(doc, syncState, message);
const changes = Automerge.getChanges(docBefore, doc);

[jeffa5]

Besides the potential performance enhancement of calling sync with an old version I think my problem, more specifically is this:

Before applying changes I'd want to be able to persist them to ensure I can rebuild a document on failure, calling applyChanges internally means I don't get this opportunity.

Thus I'd propose to change the API for receiveSyncMessage to something like this:

function receiveSyncMessage(state: BackendState, syncState: SyncState, message: BinarySyncMessage): [BackendState, SyncState, BinaryChange[]]

[ept]

@jeffa5 Interesting idea. Moving applyChanges out of receiveSyncMessage would have the advantage of making the functions more orthogonal, and we could even remove the BackendState from the return value, since the backend state would not change if applyChanges is not called.

The problem is that the advanceHeads() computation relies on knowing the local heads before and after a received batch of changes is applied. If applyChanges is moved out of receiveSyncMessage, the advanceHeads() computation would need to be moved to generateSyncMessage instead, but then we have the problem that in between the calls to receiveSyncMessage and generateSyncMessage the user may do more than just apply the changes received in a particular message (e.g. they may also apply changes from other sources), and then the differences in heads would no longer accurately reflect the changes we received from a particular peer. This would lead to sharedHeads being wrong, which in turn would lead to difficult-to-debug sync issues.

I think it would be okay if we could guarantee that the API is always used like this:

[syncState, changes] = Backend.receiveSyncMessage(backend, syncState, message);
[backend, patch] = Backend.applyChanges(backend, changes);
[syncState, response] = Backend.generateSyncMessage(backend, syncState);

i.e. the applyChanges with the changes from the message is the only thing that happens to the backend between receiveSyncMessage and generateSyncMessage, and no other changes are made during this time. But it would be easy to accidentally misuse this API, which makes me worried.

Is there some other way we can integrate persistence? Maybe you could allow a callback to be registered that is called by applyChanges whenever new changes arrive?

[jeffa5]

Ah yes, that does make things more complicated. I agree that we should keep the API hard to misuse.

In that case I would think a callback as you describe would be a good fit. I'd be happy to leave that for a later PR in order to get this through.

As for integrating persistence, I like the strategy you describe in https://github.com/automerge/automerge/issues/331 so wouldn't need any changes related to the save functionality etc.

[ept]

I think this is done; any follow-up work can happen on separate PRs. The Rust counterpart to this PR is in automerge/automerge-rs#97 — thanks @jeffa5 for porting it so quickly.

[HerbCaudill]

I've been writing some tests in order to understand the new sync protocol. It works fine for me with two peers, but with three I can easily get into an endless loop.

it.only('syncs divergent changes', () => {
  let doc = A.from({ wrens: 1, goldfinches: 12 })

  const alice = new ConnectedDoc('alice', doc)
  const bob = new ConnectedDoc('bob', doc)
  const charlie = new ConnectedDoc('charlie', doc)

  connect(alice, bob)
  connect(bob, charlie)
  connect(alice, charlie)

  alice.disconnect()
  bob.disconnect()
  charlie.disconnect()

  // each one makes a change
  alice.change(s => (s.wrens = 42))
  bob.change(s => (s.goldfinches = 0))
  charlie.change(s => (s.cassowaries = 13))

  // while they're disconnected, they have divergent docs
  assert.notDeepStrictEqual(alice.doc, bob.doc)
  assert.notDeepStrictEqual(bob.doc, charlie.doc)
  assert.notDeepStrictEqual(alice.doc, charlie.doc)

  alice.connect()
  bob.connect()
  charlie.connect()

  // after connecting, their docs converge
  assert.deepStrictEqual(alice.doc, bob.doc)
  assert.deepStrictEqual(bob.doc, charlie.doc)
  assert.deepStrictEqual(alice.doc, charlie.doc)
})

(From https://github.com/herbcaudill/automerge/blob/more-sync-tests/test/sync_integration_test.js#L97-L130
-- see that file for the ConnectedDoc class and other helpers.)

In this case, I end up with these two changes over and over until the stack overflows.

sending update bob->charlie { goldfinches: 0, wrens: 42, cassowaries: 13 }
sending update charlie->bob { goldfinches: 0, wrens: 1, cassowaries: 13 }
sending update bob->charlie { goldfinches: 0, wrens: 42, cassowaries: 13 }
sending update charlie->bob { goldfinches: 0, wrens: 1, cassowaries: 13 }
sending update bob->charlie { goldfinches: 0, wrens: 42, cassowaries: 13 }
sending update charlie->bob { goldfinches: 0, wrens: 1, cassowaries: 13 }
...

I haven't really dug into this - I might be doing something silly - but was wondering if my problem has anything to do with this teaser in the description above:

When syncing concurrently with several peers, the protocol provides a mechanism to avoid receiving the same set of changes multiple times from different peers (though this feature is not currently exposed in the API)

[skokenes]

@HerbCaudill I don't have an answer for you but I think its worth pointing out something interesting about your syncing here. I have tested syncing more than 2 peers with no issue, but always in a graph with no loops, ie something like:

Your test above has a loop in it like so:

I wonder if that has something to do with it?

[HerbCaudill]

No, even if I just have two connections (alice<->bob and bob<->charlie, or alice<->bob and alice<->charlie) it gets into an infinite loop.

And a cyclic connection graph should work - @ept promised! 😅

[pvh]

@HerbCaudill slideware is not normative :)

That said, the protocol is designed to support N-peer sync but our tests have been mostly focused on two-peer sync. Would you be able to submit a failing PR? In sync_test.js there's a function (sync) that could be pretty easily generalized to a third peer.

[ept]

It ought to work with cycles in the network topology, so this sounds like a bug! I will look at #352.