Merge pull request #134 from ssbc/preferred-live-crash

Be more forgiving in getPreferredEpoch stream

Author: Powersource

lib/epochs.js

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
 const { promisify: p } = require('util')
-const { fromMessageSigil } = require('ssb-uri2')
+const { fromMessageSigil, toMessageSigil } = require('ssb-uri2')
 const pull = require('pull-stream')
 const pullDefer = require('pull-defer')
 const pullFlatMerge = require('pull-flat-merge')
@@ -12,7 +12,15 @@ const OverwriteFields = require('@tangle/overwrite-fields')
 const clarify = require('clarify-error')
 const Butt64 = require('butt64')
 const isCanonicalBase64 = require('is-canonical-base64')
-const { where, and, type, live, toPullStream } = require('ssb-db2/operators')
+const {
+  where,
+  and,
+  type,
+  live,
+  key,
+  isDecrypted,
+  toPullStream,
+} = require('ssb-db2/operators')
 const {
   validator: {
     group: {
@@ -28,6 +36,7 @@ const isSubsetOf = require('set.prototype.issubsetof')
 
 const { groupRecp } = require('./operators')
 const getTangleUpdates = require('./tangles/get-tangle-updates')
+const pullMany = require('pull-many')
 
 const msgPattern = toPattern(new Butt64('ssb:message/[a-zA-Z0-9-]+/', null, 32))
 const feedPattern = toPattern(new Butt64('ssb:feed/[a-zA-Z0-9-]+/', null, 32))
@@ -268,14 +277,16 @@ function Epochs(ssb) {
     // then skip all the preferrentEpochs until we get up to the current
     // This is important for listMembers to not send confusing results
     getPreferredEpoch(groupId, (err, preferredEpoch) => {
-      if (err) return deferredSource.abort(clarify(err, 'failed to get initial preferred epoch'))
-
+      // if we're live and this fails we don't really mind, just go straight to live
       if (!live) {
-        deferredSource.resolve(pull.once(preferredEpoch))
+        if (err) deferredSource.abort(clarify(err, 'failed to get initial preferred epoch'))
+        else deferredSource.resolve(pull.once(preferredEpoch))
+
         return
       }
 
-      var sync = false
+      // if we couldn't get current preferred, we'll just go live
+      var sync = !!err
       const source = pull(
         epochsReduce.stream(groupId, { getters: allGetters, live }),
         pull.asyncMap(buildPreferredEpoch),
@@ -509,24 +520,58 @@ function epochNodeStream(ssb, groupId, opts = {}) {
 
   return deferredSource
 }
-function getGroupInit(ssb, groupId, cb) {
-  ssb.box2.getGroupInfo(groupId, (err, info) => {
-    // prettier-ignore
-    if (err) return cb(clarify(err, 'Failed to get group info for ' + groupId))
-    if (!info) return cb(new Error('Unknown group'))
 
-    // Fetch the tangle root
-    ssb.db.get(info.root, (err, rootVal) => {
-      // prettier-ignore
-      if (err) return cb(clarify(err, 'Failed to load group root with id ' + info.root))
+function getRootVal(ssb, msgId, cb) {
+  pull(
+    pullMany([
+      ssb.db.query(
+        where(and(isDecrypted('box2'), key(toMessageSigil(msgId)))),
+        live({ old: true }),
+        toPullStream()
+      ),
+      pull(
+        ssb.db.reindexed(),
+        pull.filter((msg) => fromMessageSigil(msg.key) === msgId)
+      ),
+    ]),
+    pull.take(1),
+    pull.drain(
+      (msg) => cb(null, msg.value),
+      (err) => {
+        if (err) cb(Error('Failed getting root msg async', { cause: err }))
+      }
+    )
+  )
+}
 
-      if (!isInitRoot(rootVal))
+function getGroupInit(ssb, groupId, cb) {
+  pull(
+    ssb.box2.getGroupInfoUpdates(groupId),
+    pull.take(1),
+    pull.drain(
+      (info) => {
+        if (!info) return cb(new Error('Unknown group'))
+
+        // Fetch the tangle root
+        // This is based on a live stream since sometimes the group info comes in very quick, before the root msg has had time to get put into the db
+        // and sometimes it might take ages (we haven't gotten that feed yet)
+        getRootVal(ssb, info.root, (err, rootVal) => {
+          // prettier-ignore
+          if (err) return cb(clarify(err, 'Failed to load group root with id ' + info.root))
+
+          if (!isInitRoot(rootVal))
+            // prettier-ignore
+            return cb(clarify(new Error(isInitRoot.string), 'Malformed group/init root message'))
+
+          cb(null, { key: info.root, value: rootVal })
+        })
+      },
+      (err) => {
         // prettier-ignore
-        return cb(clarify(new Error(isInitRoot.string), 'Malformed group/init root message'))
-
-      cb(null, { key: info.root, value: rootVal })
-    })
-  })
+        if (err) return cb(clarify(err, 'Failed to get group info for ' + groupId))
+      }
+    )
+  )
 }
 
 /* HELPERS */

test/lib/epochs.test.js

@@ -26,7 +26,7 @@ test('lib/epochs (getEpochs, getMembers)', async (t) => {
   async function sync(label) {
     return run(`(sync ${label})`, replicate(peers), { isTest: false })
   }
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   const [aliceId, bobId, oscarId] = await getRootIds(peers)
   await run(
@@ -153,7 +153,7 @@ test('lib/epochs (getMissingMembers)', async (t) => {
       { isTest: false }
     )
   }
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   await run(
     'start tribes',
@@ -260,7 +260,7 @@ test('lib/epochs (getPreferredEpoch - 4.4. same membership)', async (t) => {
     Server({ name: 'bob' }),
     Server({ name: 'oscar' }),
   ]
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   const [alice, bob, oscar] = peers
   const [bobId, oscarId] = await getRootIds([bob, oscar])
@@ -381,7 +381,7 @@ test('lib/epochs (getPreferredEpoch - 4.5. subset membership)', async (t) => {
     Server({ name: 'carol' }),
     Server({ name: 'oscar' }),
   ]
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   const [alice, bob, carol, oscar] = peers
   const [bobId, carolId, oscarId] = await getRootIds([bob, carol, oscar])