in-memory realtime sync (#1115)

* start in-memory realtime sync * fix realtime log * notes * no queue for onRealtimeSyncEvent * pass factory information to buildEvents * cleanup * backpressure log * remove reorged events * on conflict do nothing * fix reorg * filter out reverted call traces * fix finalize interval * fix reorg handling * filter call traces with function selector * cleanup * fix test * nits * chore: changeset * cleanup
ponder-sh · Oct 10, 2024 · 1a7d1f8 · 1a7d1f8
1 parent 6d45c52
commit 1a7d1f8
Show file tree

Hide file tree

Showing 16 changed files with 1,143 additions and 432 deletions.
diff --git a/.changeset/large-crabs-listen.md b/.changeset/large-crabs-listen.md
@@ -0,0 +1,5 @@
+---
+"@ponder/core": patch
+---
+
+Improved performance of realtime event processing.
diff --git a/packages/core/src/_test/utils.ts b/packages/core/src/_test/utils.ts
@@ -239,7 +239,7 @@ export const getRawRPCData = async () => {
       block: blocks[0],
       transactions: [],
       transactionReceipts: [],
-      traces: [
+      callTraces: [
         {
           action: {
             from: ALICE,
@@ -293,7 +293,7 @@ export const getRawRPCData = async () => {
       transactionReceipts: transactionReceipts.filter(
         (tr) => tr?.blockNumber === blocks[1]?.number,
       ),
-      traces: [
+      callTraces: [
         {
           action: {
             callType: "call",
@@ -363,7 +363,7 @@ export const getRawRPCData = async () => {
       transactionReceipts: transactionReceipts.filter(
         (tr) => tr?.blockNumber === blocks[2]?.number,
       ),
-      traces: [
+      callTraces: [
         {
           action: {
             callType: "call",
@@ -403,7 +403,7 @@ export const getRawRPCData = async () => {
       transactionReceipts: transactionReceipts.filter(
         (tr) => tr?.blockNumber === blocks[3]?.number,
       ),
-      traces: [
+      callTraces: [
         {
           action: {
             callType: "call",
@@ -441,43 +441,43 @@ export const getRawRPCData = async () => {
       block: blocks[4]!,
       transactions: [],
       transactionReceipts: [],
-      traces: [],
+      callTraces: [],
     },
   } as unknown as {
     block1: {
       logs: [];
       block: SyncBlock;
       transactions: [];
       transactionReceipts: [];
-      traces: [SyncCreateTrace, SyncCreateTrace];
+      callTraces: [SyncCreateTrace, SyncCreateTrace];
     };
     block2: {
       logs: [SyncLog, SyncLog];
       block: SyncBlock;
       transactions: [SyncTransaction, SyncTransaction];
       transactionReceipts: [SyncTransactionReceipt, SyncTransactionReceipt];
-      traces: [SyncCallTrace, SyncCallTrace];
+      callTraces: [SyncCallTrace, SyncCallTrace];
     };
     block3: {
       logs: [SyncLog];
       block: SyncBlock;
       transactions: [SyncTransaction];
       transactionReceipts: [SyncTransactionReceipt];
-      traces: [SyncCallTrace];
+      callTraces: [SyncCallTrace];
     };
     block4: {
       logs: [SyncLog];
       block: SyncBlock;
       transactions: [SyncTransaction];
       transactionReceipts: [SyncTransactionReceipt];
-      traces: [SyncCallTrace];
+      callTraces: [SyncCallTrace];
     };
     block5: {
       logs: [];
       block: SyncBlock;
       transactions: [];
       transactionReceipts: [];
-      traces: [];
+      callTraces: [];
     };
   };
 };
@@ -592,7 +592,7 @@ export const getEventsTrace = async (
 
   return [
     {
-      trace: rpcData.block3.traces[0],
+      trace: rpcData.block3.callTraces[0],
       block: rpcData.block3.block,
       transaction: rpcData.block3.transactions[0]!,
       transactionReceipt: rpcData.block3.transactionReceipts[0]!,

diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
@@ -234,7 +234,7 @@ export async function run({
 
     indexingService.updateIndexingStore({ indexingStore, schema });
 
-    sync.startRealtime();
+    await sync.startRealtime();
 
     await metadataStore.setStatus(sync.getStatus());
 

diff --git a/packages/core/src/build/configAndIndexingFunctions.test.ts b/packages/core/src/build/configAndIndexingFunctions.test.ts
@@ -3,6 +3,7 @@ import type { Options } from "@/common/options.js";
 import type { CallTraceFilter, LogFactory, LogFilter } from "@/sync/source.js";
 import {
   http,
+  type Address,
   getEventSelector,
   getFunctionSelector,
   parseAbiItem,
@@ -399,8 +400,7 @@ test("buildConfigAndIndexingFunctions() validates address empty string", async (
       a: {
         network: "mainnet",
         abi: [event0],
-        // @ts-expect-error
-        address: "",
+        address: "" as Address,
       },
     },
   }) as Config;
@@ -426,8 +426,8 @@ test("buildConfigAndIndexingFunctions() validates address prefix", async () => {
       a: {
         network: "mainnet",
         abi: [event0],
-        // @ts-expect-error
-        address: "0b0000000000000000000000000000000000000001",
+
+        address: "0b0000000000000000000000000000000000000001" as Address,
       },
     },
   }) as Config;

diff --git a/packages/core/src/sync-historical/index.test.ts b/packages/core/src/sync-historical/index.test.ts
@@ -25,9 +25,9 @@ const getRequestQueue = async (requestQueue: RequestQueue) => {
     request: (request: any) => {
       if (request.method === "trace_filter") {
         let traces = [
-          ...rpcData.block2.traces,
-          ...rpcData.block3.traces,
-          ...rpcData.block4.traces,
+          ...rpcData.block2.callTraces,
+          ...rpcData.block3.callTraces,
+          ...rpcData.block4.callTraces,
         ];
 
         if (request.params[0].fromBlock !== undefined) {

diff --git a/packages/core/src/sync-realtime/filter.test.ts b/packages/core/src/sync-realtime/filter.test.ts
@@ -70,21 +70,21 @@ test("isCallTraceFilterMatched", async (context) => {
   let isMatched = isCallTraceFilterMatched({
     filter: context.sources[3].filter,
     block: rpcData.block3.block,
-    callTrace: rpcData.block3.traces[0],
+    callTrace: rpcData.block3.callTraces[0],
   });
   expect(isMatched).toBe(true);
 
   isMatched = isCallTraceFilterMatched({
     filter: context.sources[2].filter,
     block: rpcData.block3.block,
-    callTrace: rpcData.block3.traces[0],
+    callTrace: rpcData.block3.callTraces[0],
   });
   expect(isMatched).toBe(true);
 
   isMatched = isCallTraceFilterMatched({
     filter: context.sources[3].filter,
     block: rpcData.block2.block,
-    callTrace: rpcData.block2.traces[0],
+    callTrace: rpcData.block2.callTraces[0],
   });
   expect(isMatched).toBe(false);
 });