Revert getEvents performance improvements

packages/core/src/bin/utils/run.ts

@@ -124,8 +124,13 @@ export async function run({
     initialCheckpoint,
   });
 
-  const handleEvents = async (events: Event[], toCheckpoint: Checkpoint) => {
-    indexingService.updateTotalSeconds(toCheckpoint);
+  const handleEvents = async (
+    events: Event[],
+    lastEventCheckpoint: Checkpoint | undefined,
+  ) => {
+    if (lastEventCheckpoint !== undefined) {
+      indexingService.updateLastEventCheckpoint(lastEventCheckpoint);
+    }
 
     if (events.length === 0) return { status: "success" } as const;
 
@@ -150,14 +155,20 @@ export async function run({
     worker: async (event: RealtimeEvent) => {
       switch (event.type) {
         case "newEvents": {
+          const lastEventCheckpoint = await syncStore.getLastEventCheckpoint({
+            sources: sources,
+            fromCheckpoint: event.fromCheckpoint,
+            toCheckpoint: event.toCheckpoint,
+          });
           for await (const rawEvents of syncStore.getEvents({
             sources,
             fromCheckpoint: event.fromCheckpoint,
             toCheckpoint: event.toCheckpoint,
+            limit: 1_000,
           })) {
             const result = await handleEvents(
               decodeEvents(syncService, rawEvents),
-              event.toCheckpoint,
+              lastEventCheckpoint,
             );
             if (result.status === "error") onReloadableError(result.error);
           }
@@ -229,14 +240,21 @@ export async function run({
       fromCheckpoint,
       toCheckpoint,
     } of syncService.getHistoricalCheckpoint()) {
+      const lastEventCheckpoint = await syncStore.getLastEventCheckpoint({
+        sources: sources,
+        fromCheckpoint,
+        toCheckpoint,
+      });
+
       for await (const rawEvents of syncStore.getEvents({
         sources: sources,
         fromCheckpoint,
         toCheckpoint,
+        limit: 1_000,
       })) {
         const result = await handleEvents(
           decodeEvents(syncService, rawEvents),
-          toCheckpoint,
+          lastEventCheckpoint,
         );
 
         if (result.status === "killed") {
@@ -250,15 +268,6 @@ export async function run({
 
     await historicalStore.flush({ isFullFlush: true });
 
-    // set completed indexing metrics
-    common.metrics.ponder_indexing_completed_seconds.set(
-      syncService.checkpoint.blockTimestamp -
-        syncService.startCheckpoint.blockTimestamp,
-    );
-    common.metrics.ponder_indexing_completed_timestamp.set(
-      syncService.checkpoint.blockTimestamp,
-    );
-
     // Become healthy
     common.logger.info({
       service: "indexing",

packages/core/src/common/options.ts

@@ -35,7 +35,6 @@ export type Options = {
   indexingCacheBytes: number;
 
   syncMaxIntervals: number;
-  syncEventsQuerySize: number;
 };
 
 export const buildOptions = ({ cliOptions }: { cliOptions: CliOptions }) => {
@@ -107,6 +106,5 @@ export const buildOptions = ({ cliOptions }: { cliOptions: CliOptions }) => {
     indexingCacheBytes: os.freemem() / 3,
 
     syncMaxIntervals: 50_000,
-    syncEventsQuerySize: 10_000,
   } satisfies Options;
 };

packages/core/src/indexing/index.ts

@@ -5,7 +5,7 @@ import {
   processEvents,
   processSetupEvents,
   updateIndexingStore,
-  updateTotalSeconds,
+  updateLastEventCheckpoint,
 } from "./service.js";
 import type { Context, Service } from "./service.js";
 
@@ -15,7 +15,7 @@ const methods = {
   processEvents,
   processSetupEvents,
   updateIndexingStore,
-  updateTotalSeconds,
+  updateLastEventCheckpoint,
 };
 
 export const createIndexingService = extend(create, methods);

packages/core/src/indexing/service.ts

@@ -66,7 +66,8 @@ export type Service = {
   eventCount: {
     [eventName: string]: { [networkName: string]: number };
   };
-  startCheckpoint: Checkpoint;
+  firstEventCheckpoint: Checkpoint | undefined;
+  lastEventCheckpoint: Checkpoint | undefined;
 
   /**
    * Reduce memory usage by reserving space for objects ahead of time
@@ -181,7 +182,8 @@ export const create = ({
     indexingStore,
     isKilled: false,
     eventCount,
-    startCheckpoint: syncService.startCheckpoint,
+    firstEventCheckpoint: undefined,
+    lastEventCheckpoint: undefined,
     currentEvent: {
       contextState,
       context: {
@@ -272,6 +274,21 @@ export const processEvents = async (
   | { status: "success" }
   | { status: "killed" }
 > => {
+  // set first event checkpoint
+  if (events.length > 0 && indexingService.firstEventCheckpoint === undefined) {
+    indexingService.firstEventCheckpoint = decodeCheckpoint(
+      events[0].encodedCheckpoint,
+    );
+
+    // set total seconds
+    if (indexingService.lastEventCheckpoint !== undefined) {
+      indexingService.common.metrics.ponder_indexing_total_seconds.set(
+        indexingService.lastEventCheckpoint.blockTimestamp -
+          indexingService.firstEventCheckpoint.blockTimestamp,
+      );
+    }
+  }
+
   const eventCounts: { [eventName: string]: number } = {};
 
   for (let i = 0; i < events.length; i++) {
@@ -377,26 +394,30 @@ export const processEvents = async (
       ).blockTimestamp;
 
       indexingService.common.metrics.ponder_indexing_completed_seconds.set(
-        eventTimestamp - indexingService.startCheckpoint.blockTimestamp,
+        eventTimestamp - indexingService.firstEventCheckpoint!.blockTimestamp,
       );
       indexingService.common.metrics.ponder_indexing_completed_timestamp.set(
         eventTimestamp,
       );
 
-      // Note: allows for terminal and logs to be updated
+      // Note(kyle) this is only needed for sqlite
       await new Promise(setImmediate);
     }
   }
 
   // set completed seconds
-  if (events.length > 0) {
+  if (
+    events.length > 0 &&
+    indexingService.firstEventCheckpoint !== undefined &&
+    indexingService.lastEventCheckpoint !== undefined
+  ) {
     const lastEventInBatchTimestamp = decodeCheckpoint(
       events[events.length - 1].encodedCheckpoint,
     ).blockTimestamp;
 
     indexingService.common.metrics.ponder_indexing_completed_seconds.set(
       lastEventInBatchTimestamp -
-        indexingService.startCheckpoint.blockTimestamp,
+        indexingService.firstEventCheckpoint.blockTimestamp,
     );
     indexingService.common.metrics.ponder_indexing_completed_timestamp.set(
       lastEventInBatchTimestamp,
@@ -430,14 +451,18 @@ export const kill = (indexingService: Service) => {
   indexingService.isKilled = true;
 };
 
-export const updateTotalSeconds = (
+export const updateLastEventCheckpoint = (
   indexingService: Service,
-  endCheckpoint: Checkpoint,
+  lastEventCheckpoint: Checkpoint,
 ) => {
-  indexingService.common.metrics.ponder_indexing_total_seconds.set(
-    endCheckpoint.blockTimestamp -
-      indexingService.startCheckpoint.blockTimestamp,
-  );
+  indexingService.lastEventCheckpoint = lastEventCheckpoint;
+
+  if (indexingService.firstEventCheckpoint !== undefined) {
+    indexingService.common.metrics.ponder_indexing_total_seconds.set(
+      indexingService.lastEventCheckpoint.blockTimestamp -
+        indexingService.firstEventCheckpoint.blockTimestamp,
+    );
+  }
 };
 
 const updateCompletedEvents = (indexingService: Service) => {

packages/core/src/sync-historical/service.test.ts

@@ -428,6 +428,7 @@ test("start() adds log filter events to sync store", async (context) => {
     sources: [sources[0]],
     fromCheckpoint: zeroCheckpoint,
     toCheckpoint: maxCheckpoint,
+    limit: 100,
   });
   const events = await drainAsyncGenerator(ag);
 
@@ -471,6 +472,7 @@ test("start() adds factory events to sync store", async (context) => {
     sources: [sources[1]],
     fromCheckpoint: zeroCheckpoint,
     toCheckpoint: maxCheckpoint,
+    limit: 100,
   });
   const events = await drainAsyncGenerator(ag);
 
@@ -501,6 +503,7 @@ test("start() adds block filter events to sync store", async (context) => {
     sources: [sources[4]],
     fromCheckpoint: zeroCheckpoint,
     toCheckpoint: maxCheckpoint,
+    limit: 100,
   });
   const events = await drainAsyncGenerator(ag);
 
@@ -542,6 +545,7 @@ test("start() adds trace filter events to sync store", async (context) => {
     sources: [sources[3]],
     fromCheckpoint: zeroCheckpoint,
     toCheckpoint: maxCheckpoint,
+    limit: 100,
   });
   const events = await drainAsyncGenerator(ag);
 
@@ -580,6 +584,7 @@ test("start() adds factory trace filter events to sync store", async (context) =
     sources: [sources[2]],
     fromCheckpoint: zeroCheckpoint,
     toCheckpoint: maxCheckpoint,
+    limit: 100,
   });
   const events = await drainAsyncGenerator(ag);