Hotfix pubsub subscribe

packages/cli/src/__tests__/ceramic-daemon.test.ts

@@ -28,6 +28,7 @@ const waitChange = (doc: EventEmitter, count = 1): Promise<void> => {
 }
 const port = 7777
 const apiUrl = 'http://localhost:' + port
+const topic = '/ceramic'
 
 /**
  * Create an IPFS instance
@@ -64,6 +65,11 @@ describe('Ceramic interop: core <> http-client', () => {
                 }, Bootstrap: []
             }
         })
+        if (!ipfs.pubsub) {
+            ipfs.pubsub = {}
+        }
+        ipfs.pubsub.subscribe = jest.fn()
+        ipfs.pubsub.ls = jest.fn(() => Promise.resolve([topic]))
     })
 
     afterAll(async () => {
@@ -76,7 +82,7 @@ describe('Ceramic interop: core <> http-client', () => {
         // performed yet by the time the test checks.  To eliminate this race condition we should set
         // anchorOnRequest to false in the config for the InMemoryAnchorService and anchor manually
         // throughout the tests.
-        core = await Ceramic.create(ipfs)
+        core = await Ceramic.create(ipfs, {pubsubTopic: topic})
 
         const doctypeHandler = new TileDoctypeHandler()
         doctypeHandler.verifyJWS = (): Promise<void> => { return }

packages/core/src/__tests__/dispatcher.test.ts

@@ -4,6 +4,7 @@ import Document from "../document"
 import { TileDoctype } from "@ceramicnetwork/doctype-tile"
 import DocID from "@ceramicnetwork/docid";
 
+const TOPIC = '/ceramic'
 const FAKE_CID = new CID('bafybeig6xv5nwphfmvcnektpnojts33jqcuam7bmye2pb54adnrtccjlsu')
 const FAKE_CID2 = new CID('bafybeig6xv5nwphfmvcnektpnojts44jqcuam7bmye2pb54adnrtccjlsu')
 const FAKE_DOC_ID = "kjzl6cwe1jw147dvq16zluojmraqvwdmbh61dx9e0c59i344lcrsgqfohexp60s"
@@ -12,6 +13,9 @@ const ipfs = {
   pubsub: {
     subscribe: jest.fn(),
     unsubscribe: jest.fn(),
+    ls: jest.fn(() => new Promise((resolve) => {
+        resolve([TOPIC])
+    })),
     publish: jest.fn()
   },
   dag: {
@@ -23,7 +27,6 @@ const ipfs = {
   },
   id: (): any => ({ id: 'ipfsid' })
 }
-const TOPIC = '/ceramic'
 
 class TileDoctypeMock extends TileDoctype {
   get doctype() {
@@ -50,9 +53,13 @@ describe('Dispatcher', () => {
     await dispatcher.init()
   })
 
+  afterEach(async () => {
+    await dispatcher.close()
+  })
+
   it('is constructed correctly', async () => {
     expect(dispatcher._documents).toEqual({})
-    expect(ipfs.pubsub.subscribe).toHaveBeenCalledWith(TOPIC, expect.anything())
+    expect(ipfs.pubsub.subscribe).toHaveBeenCalledWith(TOPIC, expect.anything(), expect.anything())
   })
 
   it('closes correctly', async () => {
@@ -62,8 +69,16 @@ describe('Dispatcher', () => {
   })
 
   it('makes registration correctly', async () => {
-    const doc = new Document(DocID.fromString(FAKE_DOC_ID), dispatcher, null)
-    doc._doctype = new TileDoctypeMock()
+    const doc = new Document(
+      DocID.fromString(FAKE_DOC_ID),
+      dispatcher,
+      null,
+      false,
+      {},
+      null,
+      null
+    )
+    doc['_doctype'] = new TileDoctypeMock(null, {})
     await dispatcher.register(doc)
 
     const publishArgs = ipfs.pubsub.publish.mock.calls[0]
@@ -94,8 +109,16 @@ describe('Dispatcher', () => {
   })
 
   it('handle message correctly', async () => {
-    const doc = new Document(DocID.fromString(FAKE_DOC_ID), dispatcher, null)
-    doc._doctype = new TileDoctypeMock()
+    const doc = new Document(
+      DocID.fromString(FAKE_DOC_ID),
+      dispatcher,
+      null,
+      false,
+      {},
+      null,
+      null
+    )
+    doc['_doctype'] = new TileDoctypeMock(null, {})
     await dispatcher.register(doc)
 
     // Store the query ID sent when the doc is registered so we can use it as the response ID later

packages/core/src/dispatcher.ts

@@ -12,6 +12,8 @@ import DocID from "@ceramicnetwork/docid";
 
 const IPFS_GET_TIMEOUT = 30000 // 30 seconds
 const IPFS_MAX_RECORD_SIZE = 256000 // 256 KB
+const IPFS_RESUBSCRIBE_INTERVAL_DELAY = 1000 * 60 // 1 minute
+const TESTING = process.env.NODE_ENV == 'test'
 
 /**
  * Ceramic Pub/Sub message type.
@@ -45,6 +47,8 @@ export default class Dispatcher extends EventEmitter {
 
   private logger: Logger
   private _isRunning = true
+  private _isSubscribed = false
+  private _resubscribeInterval: any
 
   constructor (public _ipfs: IpfsApi, public topic: string) {
     super()
@@ -58,8 +62,70 @@ export default class Dispatcher extends EventEmitter {
    */
   async init(): Promise<void> {
     this._peerId = this._peerId || (await this._ipfs.id()).id
-    await this._ipfs.pubsub.subscribe(this.topic, this.handleMessage.bind(this))
-    this._log({ peer: this._peerId, event: 'subscribed', topic: this.topic })
+    await this._subscribe()
+    !TESTING && this._resubscribeOnDisconnect()
+  }
+
+  /**
+   * Subscribes IPFS pubsub to `this.topic` and logs an `subscribe` event.
+   */
+  async _subscribe(): Promise<void> {
+    if (this._isSubscribed) {
+      console.log(`Pubsub subscribe skipped. Already subscribed to topic ${this.topic}`)
+    } else {
+      try {
+        await this._ipfs.pubsub.subscribe(
+          this.topic,
+          this.handleMessage.bind(this),
+          {timeout: !TESTING && IPFS_GET_TIMEOUT}
+        )
+
+        const { isSubscribed, error } = await this._confirmIsSubscribed()
+        if (isSubscribed) {
+          this._log({peer: this._peerId, event: 'subscribed', topic: this.topic })
+        } else if (error) {
+          throw error
+        } else {
+          throw new Error(`Pubsub subscribe failed for topic ${this.topic}`)
+        }
+      } catch (error) {
+        console.error(error)
+      }
+    }
+  }
+
+  /**
+   * Periodically checks that IPFS pubsub is subscribed to `this.topic` and
+   * attempts to subscribe if not.
+   */
+  _resubscribeOnDisconnect(): void {
+    this._resubscribeInterval = setInterval(async () => {
+      const { isSubscribed } = await this._confirmIsSubscribed()
+      !isSubscribed && await this._subscribe()
+    }, IPFS_RESUBSCRIBE_INTERVAL_DELAY)
+  }
+
+  /**
+   * Sets `this._iSubscribed` if the `this.topic` is listed as an IPFS pubsub
+   * subscription.
+   * @returns A tuple of `this._isSubscribed` and an error if one was caught.
+   */
+  async _confirmIsSubscribed(): Promise<{isSubscribed: boolean, error: Error | null}> {
+    let isSubscribed = true
+    let error = null
+
+    try {
+      const subscriptions = await this._ipfs.pubsub.ls({ timeout: IPFS_GET_TIMEOUT })
+      if (!subscriptions.includes(this.topic)) {
+        isSubscribed = false
+      }
+    } catch (_error) {
+      isSubscribed = false
+      error = _error
+    }
+    this._isSubscribed = isSubscribed
+
+    return { isSubscribed: this._isSubscribed, error }
   }
 
   /**
@@ -319,6 +385,8 @@ export default class Dispatcher extends EventEmitter {
   async close(): Promise<void> {
     this._isRunning = false
 
+    clearInterval(this._resubscribeInterval)
+
     await Promise.all(Object.values(this._documents).map(async (doc) => await doc.close()))
 
     await this._ipfs.pubsub.unsubscribe(this.topic)

packages/core/src/document.ts

@@ -746,7 +746,7 @@ class Document extends EventEmitter {
 
     await this._applyQueue.onEmpty()
 
-    this._context.anchorService.removeAllListeners(this.id.toString())
+    this._context.anchorService && this._context.anchorService.removeAllListeners(this.id.toString())
     await Utils.awaitCondition(() => this._isProcessing, () => false, 500)
   }