Repository: ponder-sh/ponder
Pull request #1504 (2025-02-11 16:20:44)

fix: memory usage with multichain ordering


Patch:
diff --git a/.changeset/olive-trainers-joke.md b/.changeset/olive-trainers-joke.md
new file mode 100644
index 00000000..2c8925d2
--- /dev/null
+++ b/.changeset/olive-trainers-joke.md
@@ -0,0 +1,5 @@
+---
+"ponder": patch
+---
+
+Fixed a memory leak for apps with `ordering: "multichain"` introduced in v0.9.3.
diff --git a/packages/core/src/sync/index.ts b/packages/core/src/sync/index.ts
index fbfc8100..532bcfe9 100644
--- a/packages/core/src/sync/index.ts
+++ b/packages/core/src/sync/index.ts
@@ -326,6 +326,22 @@ export const createSync = async (params: {
           ({ filter }) => filter.chainId === network.chainId,
         );
 
+        async function* decodeEventGenerator(
+          eventGenerator: AsyncGenerator<{
+            events: RawEvent[];
+            checkpoint: string;
+          }>,
+        ) {
+          for await (const { events, checkpoint } of eventGenerator) {
+            const decodedEvents = decodeEvents(params.common, sources, events);
+            params.common.logger.debug({
+              service: "app",
+              msg: `Decoded ${decodedEvents.length} '${network.name}' events`,
+            });
+            yield { events: decodedEvents, checkpoint };
+          }
+        }
+
         const localSyncGenerator = getLocalSyncGenerator({
           common: params.common,
           network,
@@ -350,18 +366,10 @@ export const createSync = async (params: {
           ),
         });
 
-        async function* decodeEventGenerator() {
-          for await (const { events, checkpoint } of localEventGenerator) {
-            const decodedEvents = decodeEvents(params.common, sources, events);
-            params.common.logger.debug({
-              service: "app",
-              msg: `Decoded ${decodedEvents.length} '${network.name}' events`,
-            });
-            yield { events: decodedEvents, checkpoint };
-          }
-        }
-
-        return bufferAsyncGenerator(decodeEventGenerator(), 1);
+        return bufferAsyncGenerator(
+          decodeEventGenerator(localEventGenerator),
+          1,
+        );
       },
     );
 
diff --git a/packages/core/src/utils/generators.ts b/packages/core/src/utils/generators.ts
index 4958296d..eafba706 100644
--- a/packages/core/src/utils/generators.ts
+++ b/packages/core/src/utils/generators.ts
@@ -9,25 +9,21 @@ import { promiseWithResolvers } from "@/utils/promiseWithResolvers.js";
 export async function* mergeAsyncGenerators<T>(
   generators: AsyncGenerator<T>[],
 ): AsyncGenerator<T> {
-  const results: T[] = [];
-  let count = generators.length;
-  let pwr = promiseWithResolvers<void>();
+  const promises = generators.map((gen) => gen.next());
 
-  generators.map(async (generator) => {
-    for await (const result of generator) {
-      results.push(result);
-      pwr.resolve();
-    }
-    count--;
-    pwr.resolve();
-  });
+  while (promises.length > 0) {
+    const wrappedPromises = promises.map((promise, index) =>
+      promise.then((result) => ({ index, result })),
+    );
+
+    const { result, index } = await Promise.race(wrappedPromises);
 
-  while (count > 0 || results.length > 0) {
-    if (results.length > 0) {
-      yield results.shift()!;
+    if (result.done) {
+      generators.splice(index, 1);
+      promises.splice(index, 1);
     } else {
-      await pwr.promise;
-      pwr = promiseWithResolvers<void>();
+      yield result.value;
+      promises[index] = generators[index]!.next();
     }
   }
 }


Test patch:
diff --git a/packages/core/src/utils/generators.test.ts b/packages/core/src/utils/generators.test.ts
index 05516c3f..4375683f 100644
--- a/packages/core/src/utils/generators.test.ts
+++ b/packages/core/src/utils/generators.test.ts
@@ -67,16 +67,14 @@ test("mergeAsyncGenerators() yields all results", async () => {
 
   p1.resolve(1);
   p2.resolve(2);
-  await new Promise((res) => setTimeout(res));
   p3.resolve(3);
   p4.resolve(4);
-  await new Promise((res) => setTimeout(res));
 
   lock.resolve();
 
   await new Promise((res) => setTimeout(res));
 
-  expect(results).toStrictEqual([1, 2, 3, 4]);
+  expect(results).toStrictEqual([1, 3, 2, 4]);
 });
 
 test("bufferAsyncGenerator() prefetches results", async () => {