support start() before peers connect

lib/index.js

@@ -67,6 +67,7 @@ function initSync(peer, config) {
   assertSHSEExists(peer)
   const debug = makeDebug(`ppppp:sync`)
   const algo = new Algorithm(peer)
+  let started = false
 
   const streams = /** @type {Array<SyncStream>} */ ([])
 
@@ -92,7 +93,8 @@ function initSync(peer, config) {
     assertSHSEExists(peer)
     if (rpc.shse.pubkey === peer.shse.pubkey) return // connecting to myself
     if (!iamClient) return
-    const local = toPull.duplex(createStream(rpc.shse.pubkey, true))
+    const stream = createStream(rpc.shse.pubkey, true)
+    const local = toPull.duplex(stream)
 
     let abort = /** @type {CallableFunction | null} */ (null)
     const remote = rpc.sync.connect((networkError) => {
@@ -108,6 +110,7 @@ function initSync(peer, config) {
       }
     })
     abort = pull(local, remote, local)
+    if (started) stream.initiate()
   }
   peer.on('rpc:connect', onSyncRPCConnect)
 
@@ -119,6 +122,8 @@ function initSync(peer, config) {
   }
 
   function start() {
+    if (started) return
+    started = true
     for (const stream of streams) {
       stream.initiate()
     }

lib/stream.js

@@ -105,6 +105,13 @@ class SyncStream extends Pipeable {
       this.#requested.add(goal.id)
     }
     this.resume()
+
+    this.#goals.watch((/** @type {any} */ goal) => {
+      if (!this.#requested.has(goal.id)) {
+        this.#requested.add(goal.id)
+        this.resume()
+      }
+    })
   }
 
   #canSend() {

test/account-sync.test.js

@@ -51,13 +51,16 @@ test('sync an account tangle', async (t) => {
     "bob doesn't have alice's account tangle"
   )
 
-  bob.goals.set(aliceID, 'all')
-  alice.goals.set(aliceID, 'all')
 
+  // start() on purpose before connect, to test whether this also works
+  bob.sync.start()
   const remoteAlice = await p(bob.connect)(alice.getAddress())
   assert('bob connected to alice')
 
-  bob.sync.start()
+  // Set goals on purpose after connect, to test whether this also works
+  bob.goals.set(aliceID, 'all')
+  alice.goals.set(aliceID, 'all')
+
   await p(setTimeout)(1000)
   assert('sync!')