WIP: delay processing messages until sync state is initialized

packages/automerge-repo/src/Repo.ts

@@ -153,6 +153,19 @@ export class Repo extends EventEmitter<RepoEvents> {
       synchronizer.on("sync-state", ({ documentId, peerId, syncState }) => {
         storageSubsystem.saveSyncState(documentId, peerId, syncState)
       })
+
+      // todo: batch
+      synchronizer.on("missing-sync-state", async ({ documentId, peerId }) => {
+        console.log("missing-sync-state", documentId, peerId)
+
+        const syncState =
+          (await storageSubsystem.loadSyncState(documentId, peerId)) ??
+          Automerge.initSyncState()
+
+        synchronizer.intializeSyncStates([{ documentId, peerId, syncState }])
+      })
+
+      console.log("new")
     }
   }
 

packages/automerge-repo/src/network/messages.ts

@@ -127,6 +127,12 @@ export interface SyncStateMessage {
   syncState: SyncState
 }
 
+/** Notify the repo that the sync state needs to be initialized  */
+export interface MissingSyncStateMessage {
+  peerId: PeerId
+  documentId: DocumentId
+}
+
 // TYPE GUARDS
 
 export const isValidRepoMessage = (message: Message): message is RepoMessage =>

packages/automerge-repo/src/synchronizer/CollectionSynchronizer.ts

@@ -1,8 +1,9 @@
 import debug from "debug"
+import * as A from "@automerge/automerge"
 import { DocHandle } from "../DocHandle.js"
 import { stringifyAutomergeUrl } from "../AutomergeUrl.js"
 import { Repo } from "../Repo.js"
-import { RepoMessage } from "../network/messages.js"
+import { RepoMessage, SyncStateMessage } from "../network/messages.js"
 import { DocumentId, PeerId } from "../types.js"
 import { DocSynchronizer } from "./DocSynchronizer.js"
 import { Synchronizer } from "./Synchronizer.js"
@@ -50,6 +51,10 @@ export class CollectionSynchronizer extends Synchronizer {
     })
     docSynchronizer.on("message", event => this.emit("message", event))
     docSynchronizer.on("sync-state", event => this.emit("sync-state", event))
+    docSynchronizer.on("missing-sync-state", event => {
+      console.log("Missing sync state", event)
+      this.emit("missing-sync-state", event)
+    })
     return docSynchronizer
   }
 
@@ -98,15 +103,25 @@ export class CollectionSynchronizer extends Synchronizer {
   /**
    * Starts synchronizing the given document with all peers that we share it generously with.
    */
-  addDocument(documentId: DocumentId) {
+  async addDocument(documentId: DocumentId) {
     // HACK: this is a hack to prevent us from adding the same document twice
     if (this.#docSetUp[documentId]) {
       return
     }
+
+    // todo: initialize sync state
     const docSynchronizer = this.#fetchDocSynchronizer(documentId)
-    void this.#documentGenerousPeers(documentId).then(peers => {
-      docSynchronizer.beginSync(peers)
-    })
+    const peers = await this.#documentGenerousPeers(documentId)
+
+    // todo: the collection synchronizer shouldn't load sync state directly
+    for (const peerId of peers) {
+      const syncState =
+        (await this.repo.storageSubsystem?.loadSyncState(documentId, peerId)) ??
+        A.initSyncState()
+      docSynchronizer.initializeSyncState(peerId, syncState)
+    }
+
+    docSynchronizer.beginSync(peers)
   }
 
   // TODO: implement this
@@ -115,7 +130,7 @@ export class CollectionSynchronizer extends Synchronizer {
   }
 
   /** Adds a peer and maybe starts synchronizing with them */
-  addPeer(peerId: PeerId) {
+  async addPeer(peerId: PeerId) {
     log(`adding ${peerId} & synchronizing with them`)
 
     if (this.#peers.has(peerId)) {
@@ -125,9 +140,18 @@ export class CollectionSynchronizer extends Synchronizer {
     this.#peers.add(peerId)
     for (const docSynchronizer of Object.values(this.#docSynchronizers)) {
       const { documentId } = docSynchronizer
-      void this.repo.sharePolicy(peerId, documentId).then(okToShare => {
-        if (okToShare) docSynchronizer.beginSync([peerId])
-      })
+      const okToShare = await this.repo.sharePolicy(peerId, documentId)
+      if (okToShare) {
+        // todo: the collection synchronizer shouldn't load sync state directly
+        const syncState =
+          (await this.repo.storageSubsystem?.loadSyncState(
+            documentId,
+            peerId
+          )) ?? A.initSyncState()
+
+        docSynchronizer.initializeSyncState(peerId, syncState)
+        docSynchronizer.beginSync([peerId])
+      }
     }
   }
 
@@ -140,4 +164,10 @@ export class CollectionSynchronizer extends Synchronizer {
       docSynchronizer.endSync(peerId)
     }
   }
+
+  intializeSyncStates(syncStateMessages: SyncStateMessage[]) {
+    for (const { documentId, peerId, syncState } of syncStateMessages) {
+      this.#docSynchronizers[documentId].initializeSyncState(peerId, syncState)
+    }
+  }
 }

packages/automerge-repo/src/synchronizer/DocSynchronizer.ts

@@ -126,27 +126,15 @@ export class DocSynchronizer extends Synchronizer {
   }
 
   #getSyncState(peerId: PeerId) {
-    if (!this.#peers.includes(peerId)) {
-      this.#log("adding a new peer", peerId)
-      this.#peers.push(peerId)
+    const syncState = this.#syncStates[peerId]
+    if (!syncState) {
+      throw new Error(`sync state for peer ${peerId} hasn't been initalized`)
     }
 
-    // when a peer is added, we don't know if it has the document or not
-    if (!(peerId in this.#peerDocumentStatuses)) {
-      this.#peerDocumentStatuses[peerId] = "unknown"
-    }
-
-    const prevSyncState = this.#syncStates[peerId]
-    if (prevSyncState) {
-      return prevSyncState
-    }
-
-    return A.initSyncState()
+    return syncState
   }
 
   #setSyncState(peerId: PeerId, syncState: A.SyncState) {
-    // TODO: we only need to do this on reconnect
-
     const previousSyncState = this.#syncStates[peerId]
 
     this.#syncStates[peerId] = syncState
@@ -300,8 +288,17 @@ export class DocSynchronizer extends Synchronizer {
     if (message.documentId !== this.#handle.documentId)
       throw new Error(`channelId doesn't match documentId`)
 
-    // We need to block receiving the syncMessages until we've checked local storage
-    if (!this.#handle.inState([READY, REQUESTING, UNAVAILABLE])) {
+    const syncState = this.#syncStates[message.senderId]
+
+    if (!syncState) {
+      this.emit("missing-sync-state", {
+        peerId: message.senderId,
+        documentId: this.documentId,
+      })
+    }
+
+    // We need to block receiving the syncMessages until we've loeaded the document and the sync state for the peer
+    if (!this.#isDocumentLoaded() || !syncState) {
       this.#pendingSyncMessages.push({ message, received: new Date() })
       return
     }
@@ -310,6 +307,21 @@ export class DocSynchronizer extends Synchronizer {
     this.#processSyncMessage(message, new Date())
   }
 
+  /** called by collection synchronizer to load sync state initially */
+  initializeSyncState(peerId: PeerId, syncState: A.SyncState) {
+    this.#syncStates[peerId] = syncState
+
+    console.log("initialized sycn state", this.documentId, peerId, syncState)
+
+    if (this.#isDocumentLoaded()) {
+      this.#processAllPendingSyncMessages()
+    }
+  }
+
+  #isDocumentLoaded() {
+    return !this.#handle.inState([READY, REQUESTING, UNAVAILABLE])
+  }
+
   #processSyncMessage(message: SyncMessage | RequestMessage, received: Date) {
     if (isRequestMessage(message)) {
       this.#peerDocumentStatuses[message.senderId] = "wants"
@@ -373,10 +385,13 @@ export class DocSynchronizer extends Synchronizer {
   }
 
   #processAllPendingSyncMessages() {
-    for (const message of this.#pendingSyncMessages) {
-      this.#processSyncMessage(message.message, message.received)
-    }
+    this.#pendingSyncMessages = this.#pendingSyncMessages.filter(pending => {
+      if (!this.#syncStates[pending.message.senderId]) {
+        return true
+      }
 
-    this.#pendingSyncMessages = []
+      this.#processSyncMessage(pending.message, pending.received)
+      return false
+    })
   }
 }

packages/automerge-repo/src/synchronizer/Synchronizer.ts

@@ -3,6 +3,7 @@ import {
   MessageContents,
   RepoMessage,
   SyncStateMessage,
+  MissingSyncStateMessage,
 } from "../network/messages.js"
 
 export abstract class Synchronizer extends EventEmitter<SynchronizerEvents> {
@@ -12,4 +13,5 @@ export abstract class Synchronizer extends EventEmitter<SynchronizerEvents> {
 export interface SynchronizerEvents {
   message: (arg: MessageContents) => void
   "sync-state": (arg: SyncStateMessage) => void
+  "missing-sync-state": (arg: MissingSyncStateMessage) => void
 }