gracefully handle shutdowns

src/cli/setupServer.ts

@@ -378,9 +378,21 @@ export const setupServer = async ({
             "dumping mempool before shutdown"
         )
 
+        // mark all executors as processed
+        for (const account of senderManager.getActiveWallets()) {
+            senderManager.markWalletProcessed(account)
+        }
+
         process.exit(0)
     }
 
-    process.on("SIGINT", gracefulShutdown)
-    process.on("SIGTERM", gracefulShutdown)
+    const signals = [
+        "SIGINT",
+        "SIGTERM",
+        "unhandledRejection",
+        "uncaughtException"
+    ]
+    signals.forEach((signal) => {
+        process.on(signal, async () => await gracefulShutdown(signal))
+    })
 }

src/executor/senderManager/createMemorySenderManager.ts

@@ -53,6 +53,15 @@ export const createMemorySenderManager = ({
             )
 
             metrics.walletsAvailable.set(availableWallets.length)
+        },
+        getActiveWallets: () => {
+            // Active wallets are those that are in the total pool but not in the available pool
+            return wallets.filter(
+                (wallet) =>
+                    !availableWallets.some(
+                        (available) => available.address === wallet.address
+                    )
+            )
         }
     }
 }

src/executor/senderManager/createRedisSenderManager.ts

@@ -60,6 +60,9 @@ export const createRedisSenderManager = async ({
         entries: wallets.map((w) => w.address)
     })
 
+    // Track active wallets for this instance
+    const activeWallets = new Set<Account>()
+
     return {
         getAllWallets: () => [...wallets],
         getWallet: async () => {
@@ -79,6 +82,8 @@ export const createRedisSenderManager = async ({
                 throw new Error("wallet not found")
             }
 
+            activeWallets.add(wallet)
+
             logger.trace(
                 { executor: wallet.address },
                 "got wallet from sender manager"
@@ -91,11 +96,16 @@ export const createRedisSenderManager = async ({
             return wallet
         },
         markWalletProcessed: async (wallet: Account) => {
+            activeWallets.delete(wallet)
+
             redisQueue.push(wallet.address).then(() => {
                 redisQueue.llen().then((len) => {
                     metrics.walletsAvailable.set(len)
                 })
             })
+        },
+        getActiveWallets: () => {
+            return [...activeWallets]
         }
     }
 }

src/executor/senderManager/index.ts

@@ -26,6 +26,7 @@ export type SenderManager = {
     getAllWallets: () => Account[]
     getWallet: () => Promise<Account>
     markWalletProcessed: (wallet: Account) => Promise<void>
+    getActiveWallets: () => Account[]
 }
 
 export const getSenderManager = async ({