fix: Admin reset key event may have reset the key sometimes (#1610)

.changeset/modern-pillows-roll.md

@@ -0,0 +1,5 @@
+---
+"@farcaster/hubble": patch
+---
+
+fix: Admin reset key event may have reset the key sometimes

apps/hubble/src/storage/db/migrations/7.clearAdminResets.test.ts

@@ -0,0 +1,61 @@
+import { performDbMigrations } from "./migrations.js";
+import { jestRocksDB } from "../jestUtils.js";
+import { MerkleTrie } from "../../../network/sync/merkleTrie.js";
+import { SyncId } from "../../../network/sync/syncId.js";
+import { Factories, HubError, OnChainEvent, SignerEventType } from "@farcaster/hub-nodejs";
+import StoreEventHandler from "../../stores/storeEventHandler.js";
+import OnChainEventStore from "../../stores/onChainEventStore.js";
+import { getOnChainEvent } from "../onChainEvent.js";
+import { ResultAsync } from "neverthrow";
+
+const db = jestRocksDB("clearAdminResets.migration.test");
+
+describe("clearAdminResets migration", () => {
+  const addEvent = async function (event: OnChainEvent, store: OnChainEventStore, trie: MerkleTrie) {
+    await expect(store.mergeOnChainEvent(event)).resolves.toBeTruthy();
+    const syncId = SyncId.fromOnChainEvent(event);
+    await trie.insert(syncId);
+    await trie.commitToDb();
+    await expectExists(event, trie, true);
+  };
+
+  const expectExists = async function (event: OnChainEvent, trie: MerkleTrie, shouldExist: boolean) {
+    const dbResult = await ResultAsync.fromPromise(
+      getOnChainEvent(db, event.type, event.fid, event.blockNumber, event.logIndex),
+      (e) => e as HubError,
+    );
+    expect(dbResult.isOk()).toBe(shouldExist);
+    const syncId = SyncId.fromOnChainEvent(event);
+    expect(await trie.exists(syncId)).toBe(shouldExist);
+  };
+
+  test("should delete admin reset signer events", async () => {
+    const syncTrie = new MerkleTrie(db);
+    const store = new OnChainEventStore(db, new StoreEventHandler(db));
+
+    const idRegistryEvent = Factories.IdRegistryOnChainEvent.build();
+    const keyRegistryEvent = Factories.SignerOnChainEvent.build();
+    const keyRegistryResetEvent = Factories.SignerOnChainEvent.build({
+      signerEventBody: Factories.SignerEventBody.build({ eventType: SignerEventType.ADMIN_RESET }),
+    });
+    const storageRegistryEvent = Factories.StorageRentOnChainEvent.build();
+
+    await addEvent(idRegistryEvent, store, syncTrie);
+    await addEvent(keyRegistryEvent, store, syncTrie);
+    await addEvent(keyRegistryResetEvent, store, syncTrie);
+    await addEvent(storageRegistryEvent, store, syncTrie);
+
+    const success = await performDbMigrations(db, 6, 7);
+    expect(success).toBe(true);
+
+    const uncachedSyncTrie = new MerkleTrie(db);
+    await uncachedSyncTrie.initialize();
+
+    // Admin reset is removed
+    await expectExists(keyRegistryResetEvent, uncachedSyncTrie, false);
+    // Rest are unaffected
+    await expectExists(idRegistryEvent, uncachedSyncTrie, true);
+    await expectExists(keyRegistryEvent, uncachedSyncTrie, true);
+    await expectExists(storageRegistryEvent, uncachedSyncTrie, true);
+  });
+});

apps/hubble/src/storage/db/migrations/7.clearAdminResets.ts

@@ -0,0 +1,54 @@
+/**
+ Remove admin resets so they can be re-added correctly (old logic picked first signer and did not check for the exact key)
+ */
+
+import { logger } from "../../../utils/logger.js";
+import RocksDB from "../rocksdb.js";
+import { OnChainEventPostfix, RootPrefix } from "../types.js";
+import { Result } from "neverthrow";
+import { HubError, isSignerOnChainEvent, OnChainEvent, SignerEventType } from "@farcaster/hub-nodejs";
+import { SyncId } from "../../../network/sync/syncId.js";
+import { MerkleTrie } from "../../../network/sync/merkleTrie.js";
+
+const log = logger.child({ component: "clearAdminResets" });
+
+export const clearAdminResets = async (db: RocksDB): Promise<boolean> => {
+  log.info({}, "Starting clearAdminResets migration");
+  const start = Date.now();
+
+  const syncTrie = new MerkleTrie(db);
+  await syncTrie.initialize();
+  let count = 0;
+
+  await db.forEachIteratorByPrefix(
+    Buffer.from([RootPrefix.OnChainEvent, OnChainEventPostfix.OnChainEvents]),
+    async (key, value) => {
+      if (!key || !value) {
+        return;
+      }
+
+      const result = Result.fromThrowable(
+        () => OnChainEvent.decode(new Uint8Array(value as Buffer)),
+        (e) => e as HubError,
+      )();
+      if (result.isOk()) {
+        const event = result.value;
+        if (isSignerOnChainEvent(event) && event.signerEventBody.eventType === SignerEventType.ADMIN_RESET) {
+          count += 1;
+          try {
+            await db.del(key);
+            await syncTrie.deleteBySyncId(SyncId.fromOnChainEvent(event));
+          } catch (e) {
+            log.error({ err: e }, `Failed to delete event ${event.type} ${event.fid} from block ${event.blockNumber}`);
+          }
+        }
+      }
+    },
+    {},
+    1 * 60 * 60 * 1000,
+  );
+  await syncTrie.commitToDb();
+  await syncTrie.stop();
+  log.info({ duration: Date.now() - start }, `Finished clearAdminResets migration. Removed ${count} events`);
+  return true;
+};

apps/hubble/src/storage/db/migrations/migrations.ts

@@ -10,6 +10,7 @@ import { oldContractEvents } from "./6.oldContractEvents.js";
 import { HubAsyncResult, HubError } from "@farcaster/hub-nodejs";
 import { RootPrefix } from "../types.js";
 import rocksdb from "../rocksdb.js";
+import { clearAdminResets } from "./7.clearAdminResets.js";
 
 type MigrationFunctionType = (db: RocksDB) => Promise<boolean>;
 const migrations = new Map<number, MigrationFunctionType>();
@@ -40,6 +41,10 @@ migrations.set(6, async (db: RocksDB) => {
   return await oldContractEvents(db);
 });
 
+migrations.set(7, async (db: RocksDB) => {
+  return await clearAdminResets(db);
+});
+
 // To Add a new migration
 // migrations.set(<next number>, async (db: RocksDB) => {
 //   <call migration script>

apps/hubble/src/storage/stores/onChainEventStore.ts

@@ -32,6 +32,7 @@ import { OnChainEventPostfix, RootPrefix } from "../db/types.js";
 import { getHubState, putHubState } from "../db/hubState.js";
 import { PageOptions } from "./types.js";
 import { logger } from "../../utils/logger.js";
+import { bytesCompare } from "@farcaster/core";
 
 const SUPPORTED_SIGNER_SCHEMES = [1];
 
@@ -187,8 +188,16 @@ class OnChainEventStore {
         OnChainEventType.EVENT_TYPE_SIGNER,
         event.fid,
       );
-      const signerAdd = signerEvents.find((value) => value.signerEventBody.eventType === SignerEventType.ADD);
+      const signerAdd = signerEvents.find(
+        (value) =>
+          value.signerEventBody.eventType === SignerEventType.ADD &&
+          bytesCompare(value.signerEventBody.key, event.signerEventBody.key) === 0,
+      );
       if (signerAdd) {
+        logger.info(
+          { fid: event.fid },
+          `Admin reset of signer from block ${existingEvent?.blockNumber || -1} with ${signerAdd.blockNumber}`,
+        );
         return txn.put(
           secondaryKey,
           makeOnChainEventPrimaryKey(signerAdd.type, signerAdd.fid, signerAdd.blockNumber, signerAdd.logIndex),

apps/hubble/src/storage/stores/onchainEventStore.test.ts

@@ -230,6 +230,10 @@ describe("OnChainEventStore", () => {
 
     test("return the signer if removal is admin reset", async () => {
       const signer = Factories.SignerOnChainEvent.build();
+      const unrelatedSigner = Factories.SignerOnChainEvent.build({
+        fid: signer.fid,
+        blockNumber: signer.blockNumber - 1,
+      });
       const signerRemoved = Factories.SignerOnChainEvent.build({
         fid: signer.fid,
         blockNumber: signer.blockNumber + 1,
@@ -247,13 +251,40 @@ describe("OnChainEventStore", () => {
         }),
       });
 
+      await set.mergeOnChainEvent(unrelatedSigner); // a different key for the same fid should not impact the reset
       await set.mergeOnChainEvent(signer);
       await set.mergeOnChainEvent(signerRemoved);
       await set.mergeOnChainEvent(adminReset);
 
       await expect(set.getActiveSigner(signer.fid, signer.signerEventBody.key)).resolves.toEqual(signer);
     });
 
+    test("does not return the signer if older admin reset is processed after removal", async () => {
+      const signer = Factories.SignerOnChainEvent.build();
+      const adminReset = Factories.SignerOnChainEvent.build({
+        fid: signer.fid,
+        blockNumber: signer.blockNumber + 1,
+        signerEventBody: Factories.SignerEventBody.build({
+          eventType: SignerEventType.ADMIN_RESET,
+          key: signer.signerEventBody.key,
+        }),
+      });
+      const signerRemoved = Factories.SignerOnChainEvent.build({
+        fid: signer.fid,
+        blockNumber: signer.blockNumber + 2,
+        signerEventBody: Factories.SignerEventBody.build({
+          eventType: SignerEventType.REMOVE,
+          key: signer.signerEventBody.key,
+        }),
+      });
+
+      await set.mergeOnChainEvent(signer);
+      await set.mergeOnChainEvent(signerRemoved);
+      await set.mergeOnChainEvent(adminReset);
+
+      await expect(set.getActiveSigner(signer.fid, signer.signerEventBody.key)).rejects.toThrow("active signer");
+    });
+
     test("does not return signer even if events are merged out of order", async () => {
       const signer = Factories.SignerOnChainEvent.build();
       const signerRemoved = Factories.SignerOnChainEvent.build({