graceful shutdown, use multi instead of lua script

Author: arcoraven

src/db/wallets/walletNonce.ts

@@ -123,6 +123,7 @@ export const recycleNonce = async (
     });
     return;
   }
+
   const key = recycledNoncesKey(chainId, walletAddress);
   await redis.sadd(key, nonce.toString());
 };

src/index.ts

@@ -4,6 +4,15 @@ import { env } from "./utils/env";
 import { logger } from "./utils/logger";
 import "./utils/tracer";
 import { initWorker } from "./worker";
+import { CancelRecycledNoncesQueue } from "./worker/queues/cancelRecycledNoncesQueue";
+import { MigratePostgresTransactionsQueue } from "./worker/queues/migratePostgresTransactionsQueue";
+import { MineTransactionQueue } from "./worker/queues/mineTransactionQueue";
+import { NonceResyncQueue } from "./worker/queues/nonceResyncQueue";
+import { ProcessEventsLogQueue } from "./worker/queues/processEventLogsQueue";
+import { ProcessTransactionReceiptsQueue } from "./worker/queues/processTransactionReceiptsQueue";
+import { PruneTransactionsQueue } from "./worker/queues/pruneTransactionsQueue";
+import { SendTransactionQueue } from "./worker/queues/sendTransactionQueue";
+import { SendWebhookQueue } from "./worker/queues/sendWebhookQueue";
 
 const main = async () => {
   if (env.ENGINE_MODE === "server_only") {
@@ -40,3 +49,28 @@ process.on("unhandledRejection", (err) => {
     error: err,
   });
 });
+
+process.on("SIGINT", () => gracefulShutdown("SIGINT"));
+process.on("SIGTERM", () => gracefulShutdown("SIGTERM"));
+
+const gracefulShutdown = async (signal: NodeJS.Signals) => {
+  logger({
+    level: "info",
+    service: "server",
+    message: `Received ${signal}, closing server...`,
+  });
+
+  // Gracefully close workers to minimize stalled jobs.
+  // Source: https://docs.bullmq.io/guide/going-to-production#gracefully-shut-down-workers
+  await SendWebhookQueue.q.close();
+  await ProcessEventsLogQueue.q.close();
+  await ProcessTransactionReceiptsQueue.q.close();
+  await SendTransactionQueue.q.close();
+  await MineTransactionQueue.q.close();
+  await CancelRecycledNoncesQueue.q.close();
+  await PruneTransactionsQueue.q.close();
+  await MigratePostgresTransactionsQueue.q.close();
+  await NonceResyncQueue.q.close();
+
+  process.exit(0);
+};

src/utils/env.ts

@@ -62,12 +62,12 @@ export const env = createEnv({
     HTTPS_PASSPHRASE: z.string().default("thirdweb-engine"),
     TRUST_PROXY: z.boolean().default(false),
     // REDIS_MAXMEMORY should be set below the amount of memory Redis has available
-    // to avoid running out of memory. If this limit is reached, keys are evicted
-    // with a "allkeys-lru" policy (removes least recently used keys).
+    // to avoid running out of memory. If this limit is reached, Engine cannot
+    // write new keys because keys are not evicted.
     REDIS_MAXMEMORY: z.string().default("900mb"),
     // TRANSACTION_HISTORY_COUNT defines the max transaction details to keep.
-    // In testing, storing about 400k consumes 1GB memory.
-    TRANSACTION_HISTORY_COUNT: z.coerce.number().default(400_000),
+    // In testing, storing about 300k consumes 900mb memory.
+    TRANSACTION_HISTORY_COUNT: z.coerce.number().default(300_000),
     CLIENT_ANALYTICS_URL: z
       .union([UrlSchema, z.literal("")])
       .default("https://c.thirdweb.com/event"),

src/worker/tasks/cancelRecycledNoncesWorker.ts

@@ -75,12 +75,18 @@ const fromUnusedNoncesKey = (key: string) => {
 
 const getAndDeleteUnusedNonces = async (key: string) => {
   // Returns all unused nonces for this key and deletes the key.
-  const script = `
-    local key = ARGV[1]
-    local members = redis.call('SMEMBERS', key)
-    redis.call('DEL', key)
-    return members
-`;
-  const results = (await redis.eval(script, 0, key)) as string[];
-  return results.map(parseInt);
+  // Example response:
+  // [
+  //   [ null, [ '1', '2', '3', '4' ] ],
+  //   [ null, 1 ]
+  // ]
+  const multiResult = await redis.multi().smembers(key).del(key).exec();
+  if (!multiResult) {
+    throw new Error(`Error getting members of ${key}.`);
+  }
+  const [error, nonces] = multiResult[0];
+  if (error) {
+    throw new Error(`Error getting members of ${key}: ${error}`);
+  }
+  return (nonces as string[]).map((v) => parseInt(v));
 };