reorder queue logic

Bossett · Sep 5, 2024 · ddb3656 · ddb3656
1 parent b54d958
commit ddb3656
Showing 1 changed file with 10 additions and 7 deletions.
diff --git a/src/util/subscription.ts b/src/util/subscription.ts
@@ -17,6 +17,8 @@ import { Mutex } from 'async-mutex'
 
 const includedRecords = new Set(['app.bsky.feed.post'])
 let runningEvents = 0
+const mutex = new Mutex()
+const delay = (ms: number) => new Promise((res) => setTimeout(res, ms))
 
 export abstract class FirehoseSubscriptionBase {
   public sub: Subscription<RepoEvent>
@@ -35,9 +37,6 @@ export abstract class FirehoseSubscriptionBase {
   abstract handleEvent(evt: RepoEvent): Promise<void>
 
   async run(subscriptionReconnectDelay: number) {
-    const delay = (ms: number) => new Promise((res) => setTimeout(res, ms))
-    const mutex = new Mutex()
-
     await mutex.runExclusive(() => (runningEvents = 0))
 
     try {
@@ -57,11 +56,15 @@ export abstract class FirehoseSubscriptionBase {
                 runningEvents++
               })
 
-              this.handleEvent(evt).finally(async () => {
-                await mutex.runExclusive(async () => {
-                  if (runningEvents > 0) runningEvents--
+              this.handleEvent(evt)
+                .catch((err) => {
+                  console.log(`err in handleEvent ${err}`)
+                })
+                .finally(async () => {
+                  await mutex.runExclusive(async () => {
+                    if (runningEvents > 0) runningEvents--
+                  })
                 })
-              })
 
               // no longer awaiting this
             }