feat(store-sync, store-indexer): order logs by logIndex (#2037)

latticexyz · Dec 7, 2023 · 85b9461 · 85b9461
1 parent 017e57a
commit 85b9461
Show file tree

Hide file tree

Showing 11 changed files with 49 additions and 32 deletions.
diff --git a/.changeset/fluffy-days-carry.md b/.changeset/fluffy-days-carry.md
@@ -0,0 +1,6 @@
+---
+"@latticexyz/store-indexer": major
+"@latticexyz/store-sync": major
+---
+
+The postgres indexer is now storing the `logIndex` of the last update of a record to be able to return the snapshot logs in the order they were emitted onchain.
diff --git a/packages/store-indexer/bin/postgres-indexer.ts b/packages/store-indexer/bin/postgres-indexer.ts
@@ -59,8 +59,8 @@ try {
     // TODO: move this to `.findFirst` after upgrading drizzle or `rows[0]` after enabling `noUncheckedIndexedAccess: true`
     .then((rows) => rows.find(() => true));
 
-  if (chainState?.lastUpdatedBlockNumber != null) {
-    startBlock = chainState.lastUpdatedBlockNumber + 1n;
+  if (chainState?.blockNumber != null) {
+    startBlock = chainState.blockNumber + 1n;
     console.log("resuming from block number", startBlock);
   }
 } catch (error) {

diff --git a/packages/store-indexer/src/postgres/common.ts b/packages/store-indexer/src/postgres/common.ts
@@ -7,7 +7,8 @@ export type RecordData = {
   staticData: Hex | null;
   encodedLengths: Hex | null;
   dynamicData: Hex | null;
-  lastUpdatedBlockNumber: string;
+  recordBlockNumber: string;
+  logIndex: number;
 };
 
 export type RecordMetadata = {

diff --git a/packages/store-indexer/src/postgres/deprecated/getLogs.ts b/packages/store-indexer/src/postgres/deprecated/getLogs.ts
@@ -55,29 +55,26 @@ export async function getLogs(
     // Get the first record in a way that returns a possible `undefined`
     // TODO: move this to `.findFirst` after upgrading drizzle or `rows[0]` after enabling `noUncheckedIndexedAccess: true`
     .then((rows) => rows.find(() => true));
-  const indexerBlockNumber = chainState?.lastUpdatedBlockNumber ?? 0n;
+  const indexerBlockNumber = chainState?.blockNumber ?? 0n;
   benchmark("query chainState");
 
   const records = await database
     .select()
     .from(tables.recordsTable)
     .where(or(...conditions))
     .orderBy(
-      asc(tables.recordsTable.lastUpdatedBlockNumber)
+      asc(tables.recordsTable.blockNumber)
       // TODO: add logIndex (https://github.com/latticexyz/mud/issues/1979)
     );
   benchmark("query records");
 
-  const blockNumber = records.reduce(
-    (max, record) => bigIntMax(max, record.lastUpdatedBlockNumber ?? 0n),
-    indexerBlockNumber
-  );
+  const blockNumber = records.reduce((max, record) => bigIntMax(max, record.blockNumber ?? 0n), indexerBlockNumber);
   benchmark("find block number");
 
   const logs = records
     // TODO: add this to the query, assuming we can optimize with an index
     .filter((record) => !record.isDeleted)
-    .map((record) => recordToLog({ ...record, lastUpdatedBlockNumber: record.lastUpdatedBlockNumber.toString() }));
+    .map(recordToLog);
   benchmark("map records to logs");
 
   return { blockNumber, logs };

diff --git a/packages/store-indexer/src/postgres/queryLogs.ts b/packages/store-indexer/src/postgres/queryLogs.ts
@@ -46,7 +46,7 @@ export function queryLogs(sql: Sql, opts: z.infer<typeof input>): PendingQuery<R
         SELECT
           version AS "indexerVersion",
           chain_id AS "chainId",
-          last_updated_block_number AS "chainBlockNumber"
+          block_number AS "chainBlockNumber"
         FROM ${sql(`${schemaName}.config`)}
         LIMIT 1
       ),
@@ -58,10 +58,11 @@ export function queryLogs(sql: Sql, opts: z.infer<typeof input>): PendingQuery<R
           '0x' || encode(static_data, 'hex') AS "staticData",
           '0x' || encode(encoded_lengths, 'hex') AS "encodedLengths",
           '0x' || encode(dynamic_data, 'hex') AS "dynamicData",
-          last_updated_block_number AS "recordBlockNumber"
+          block_number AS "recordBlockNumber",
+          log_index AS "logIndex"
         FROM ${sql(`${schemaName}.records`)}
         ${where}
-        ORDER BY last_updated_block_number ASC
+        ORDER BY block_number, log_index ASC
       )
     SELECT
       (SELECT COUNT(*) FROM records) AS "totalRows",

diff --git a/packages/store-indexer/src/postgres/recordToLog.ts b/packages/store-indexer/src/postgres/recordToLog.ts
@@ -2,7 +2,9 @@ import { StorageAdapterLog } from "@latticexyz/store-sync";
 import { decodeDynamicField } from "@latticexyz/protocol-parser";
 import { RecordData } from "./common";
 
-export function recordToLog(record: RecordData): StorageAdapterLog & { eventName: "Store_SetRecord" } {
+export function recordToLog(
+  record: Omit<RecordData, "recordBlockNumber">
+): StorageAdapterLog & { eventName: "Store_SetRecord" } {
   return {
     address: record.address,
     eventName: "Store_SetRecord",

diff --git a/packages/store-sync/src/postgres-decoded/createStorageAdapter.test.ts b/packages/store-sync/src/postgres-decoded/createStorageAdapter.test.ts
@@ -50,9 +50,9 @@ describe("createStorageAdapter", async () => {
     expect(await db.select().from(storageAdapter.tables.configTable)).toMatchInlineSnapshot(`
       [
         {
+          "blockNumber": 12n,
           "chainId": 31337,
-          "lastUpdatedBlockNumber": 12n,
-          "version": "0.0.2",
+          "version": "0.0.4",
         },
       ]
     `);
@@ -71,13 +71,14 @@ describe("createStorageAdapter", async () => {
       [
         {
           "address": "0x6E9474e9c83676B9A71133FF96Db43E7AA0a4342",
+          "blockNumber": 12n,
           "dynamicData": "0x000001a400000045",
           "encodedLengths": "0x0000000000000000000000000000000000000000000000000800000000000008",
           "isDeleted": false,
           "key0": null,
           "key1": null,
           "keyBytes": "0x",
-          "lastUpdatedBlockNumber": 12n,
+          "logIndex": 1,
           "staticData": null,
           "tableId": "0x746200000000000000000000000000004e756d6265724c697374000000000000",
         },

diff --git a/packages/store-sync/src/postgres/createStorageAdapter.test.ts b/packages/store-sync/src/postgres/createStorageAdapter.test.ts
@@ -48,9 +48,9 @@ describe("createStorageAdapter", async () => {
     expect(await db.select().from(storageAdapter.tables.configTable)).toMatchInlineSnapshot(`
       [
         {
+          "blockNumber": 12n,
           "chainId": 31337,
-          "lastUpdatedBlockNumber": 12n,
-          "version": "0.0.2",
+          "version": "0.0.4",
         },
       ]
     `);
@@ -69,13 +69,14 @@ describe("createStorageAdapter", async () => {
       [
         {
           "address": "0x6E9474e9c83676B9A71133FF96Db43E7AA0a4342",
+          "blockNumber": 12n,
           "dynamicData": "0x000001a400000045",
           "encodedLengths": "0x0000000000000000000000000000000000000000000000000800000000000008",
           "isDeleted": false,
           "key0": null,
           "key1": null,
           "keyBytes": "0x",
-          "lastUpdatedBlockNumber": 12n,
+          "logIndex": 1,
           "staticData": null,
           "tableId": "0x746200000000000000000000000000004e756d6265724c697374000000000000",
         },

diff --git a/packages/store-sync/src/postgres/createStorageAdapter.ts b/packages/store-sync/src/postgres/createStorageAdapter.ts
@@ -54,7 +54,8 @@ export async function createStorageAdapter<TConfig extends StoreConfig = StoreCo
               staticData: log.args.staticData,
               encodedLengths: log.args.encodedLengths,
               dynamicData: log.args.dynamicData,
-              lastUpdatedBlockNumber: blockNumber,
+              blockNumber,
+              logIndex: log.logIndex ?? 0,
               isDeleted: false,
             })
             .onConflictDoUpdate({
@@ -63,7 +64,8 @@ export async function createStorageAdapter<TConfig extends StoreConfig = StoreCo
                 staticData: log.args.staticData,
                 encodedLengths: log.args.encodedLengths,
                 dynamicData: log.args.dynamicData,
-                lastUpdatedBlockNumber: blockNumber,
+                blockNumber,
+                logIndex: log.logIndex ?? 0,
                 isDeleted: false,
               },
             })
@@ -105,14 +107,16 @@ export async function createStorageAdapter<TConfig extends StoreConfig = StoreCo
               key0: log.args.keyTuple[0],
               key1: log.args.keyTuple[1],
               staticData: newStaticData,
-              lastUpdatedBlockNumber: blockNumber,
+              blockNumber,
+              logIndex: log.logIndex ?? 0,
               isDeleted: false,
             })
             .onConflictDoUpdate({
               target: [tables.recordsTable.address, tables.recordsTable.tableId, tables.recordsTable.keyBytes],
               set: {
                 staticData: newStaticData,
-                lastUpdatedBlockNumber: blockNumber,
+                blockNumber,
+                logIndex: log.logIndex,
                 isDeleted: false,
               },
             })
@@ -155,15 +159,17 @@ export async function createStorageAdapter<TConfig extends StoreConfig = StoreCo
               key1: log.args.keyTuple[1],
               encodedLengths: log.args.encodedLengths,
               dynamicData: newDynamicData,
-              lastUpdatedBlockNumber: blockNumber,
+              blockNumber,
+              logIndex: log.logIndex ?? 0,
               isDeleted: false,
             })
             .onConflictDoUpdate({
               target: [tables.recordsTable.address, tables.recordsTable.tableId, tables.recordsTable.keyBytes],
               set: {
                 encodedLengths: log.args.encodedLengths,
                 dynamicData: newDynamicData,
-                lastUpdatedBlockNumber: blockNumber,
+                blockNumber: blockNumber,
+                logIndex: log.logIndex ?? 0,
                 isDeleted: false,
               },
             })
@@ -181,7 +187,8 @@ export async function createStorageAdapter<TConfig extends StoreConfig = StoreCo
               staticData: null,
               encodedLengths: null,
               dynamicData: null,
-              lastUpdatedBlockNumber: blockNumber,
+              blockNumber,
+              logIndex: log.logIndex ?? 0,
               isDeleted: true,
             })
             .where(
@@ -200,12 +207,12 @@ export async function createStorageAdapter<TConfig extends StoreConfig = StoreCo
         .values({
           version,
           chainId,
-          lastUpdatedBlockNumber: blockNumber,
+          blockNumber,
         })
         .onConflictDoUpdate({
           target: [tables.configTable.chainId],
           set: {
-            lastUpdatedBlockNumber: blockNumber,
+            blockNumber,
           },
         })
         .execute();

diff --git a/packages/store-sync/src/postgres/tables.ts b/packages/store-sync/src/postgres/tables.ts
@@ -10,7 +10,7 @@ const schemaName = transformSchemaName("mud");
 const configTable = pgSchema(schemaName).table("config", {
   version: varchar("version").notNull(),
   chainId: asNumber("chain_id", "bigint").notNull().primaryKey(),
-  lastUpdatedBlockNumber: asBigInt("last_updated_block_number", "numeric").notNull(),
+  blockNumber: asBigInt("block_number", "numeric").notNull(),
 });
 
 const recordsTable = pgSchema(schemaName).table(
@@ -28,7 +28,8 @@ const recordsTable = pgSchema(schemaName).table(
     encodedLengths: asHex("encoded_lengths"),
     dynamicData: asHex("dynamic_data"),
     isDeleted: boolean("is_deleted"),
-    lastUpdatedBlockNumber: asBigInt("last_updated_block_number", "numeric").notNull(),
+    blockNumber: asBigInt("block_number", "numeric").notNull(),
+    logIndex: asNumber("log_index", "numeric").notNull(),
   },
   (table) => ({
     pk: primaryKey(table.address, table.tableId, table.keyBytes),

diff --git a/packages/store-sync/src/postgres/version.ts b/packages/store-sync/src/postgres/version.ts
@@ -1 +1 @@
-export const version = "0.0.2";
+export const version = "0.0.4";
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		export const version = "0.0.2";
		export const version = "0.0.4";