fix: handle firehose subscription error reconnect (Close bluesky-soci…

…al#44) (bluesky-social#46)

.env.example

@@ -19,4 +19,7 @@ FEEDGEN_HOSTNAME="example.com"
 FEEDGEN_PUBLISHER_DID="did:plc:abcde...."
 
 # Only use this if you want a service did different from did:web
-# FEEDGEN_SERVICE_DID="did:plc:abcde..."
+# FEEDGEN_SERVICE_DID="did:plc:abcde..."
+
+# Delay between reconnect attempts to the firehose subscription endpoint (in milliseconds)
+FEEDGEN_SUBSCRIPTION_RECONNECT_DELAY=3000

src/config.ts

@@ -15,4 +15,5 @@ export type Config = {
   subscriptionEndpoint: string
   serviceDid: string
   publisherDid: string
+  subscriptionReconnectDelay: number
 }

src/index.ts

@@ -15,6 +15,8 @@ const run = async () => {
       'wss://bsky.social',
     publisherDid:
       maybeStr(process.env.FEEDGEN_PUBLISHER_DID) ?? 'did:example:alice',
+    subscriptionReconnectDelay:
+      maybeInt(process.env.FEEDGEN_SUBSCRIPTION_RECONNECT_DELAY) ?? 3000,
     hostname,
     serviceDid,
   })

src/server.ts

@@ -63,7 +63,7 @@ export class FeedGenerator {
 
   async start(): Promise<http.Server> {
     await migrateToLatest(this.db)
-    this.firehose.run()
+    this.firehose.run(this.cfg.subscriptionReconnectDelay)
     this.server = this.app.listen(this.cfg.port, this.cfg.listenhost)
     await events.once(this.server, 'listening')
     return this.server

src/util/subscription.ts

@@ -36,17 +36,22 @@ export abstract class FirehoseSubscriptionBase {
 
   abstract handleEvent(evt: RepoEvent): Promise<void>
 
-  async run() {
-    for await (const evt of this.sub) {
-      try {
-        await this.handleEvent(evt)
-      } catch (err) {
-        console.error('repo subscription could not handle message', err)
-      }
-      // update stored cursor every 20 events or so
-      if (isCommit(evt) && evt.seq % 20 === 0) {
-        await this.updateCursor(evt.seq)
+  async run(subscriptionReconnectDelay: number) {
+    try {
+      for await (const evt of this.sub) {
+        try {
+          await this.handleEvent(evt)
+        } catch (err) {
+          console.error('repo subscription could not handle message', err)
+        }
+        // update stored cursor every 20 events or so
+        if (isCommit(evt) && evt.seq % 20 === 0) {
+          await this.updateCursor(evt.seq)
+        }
       }
+    } catch (err) {
+      console.error('repo subscription errored', err)
+      setTimeout(() => this.run(subscriptionReconnectDelay), subscriptionReconnectDelay)
     }
   }