fix: last block bugs with events providers (#1243)

* fix: last block bugs with events providers * Remove usages of name registry worker * Better defaults for OP constants * Fix tests * Add changeset
farcasterxyz · Aug 14, 2023 · 1fcfd49 · 1fcfd49
1 parent 637002e
commit 1fcfd49
Show file tree

Hide file tree

Showing 7 changed files with 50 additions and 42 deletions.
diff --git a/.changeset/lucky-insects-work.md b/.changeset/lucky-insects-work.md
@@ -0,0 +1,7 @@
+---
+"@farcaster/hub-web": patch
+"@farcaster/core": patch
+"@farcaster/hubble": patch
+---
+
+fix: fetching l2 events was not respecting --l2-first-block
diff --git a/apps/hubble/src/eth/fnameRegistryEventsProvider.test.ts b/apps/hubble/src/eth/fnameRegistryEventsProvider.test.ts
@@ -146,7 +146,7 @@ describe("fnameRegistryEventsProvider", () => {
     });
 
     it("fetches events from where it left off", async () => {
-      await hub.putHubState({ lastFnameProof: transferEvents[0]?.id ?? 0, lastEthBlock: 0 });
+      await hub.putHubState({ lastFnameProof: transferEvents[0]?.id ?? 0, lastEthBlock: 0, lastL2Block: 0 });
       mockFnameRegistryClient.setMinimumSince(transferEvents[0]?.id ?? 0);
       await provider.start();
       expect(await getUserNameProof(db, utf8ToBytes("test2"))).toBeTruthy();

diff --git a/apps/hubble/src/eth/l2EventsProvider.ts b/apps/hubble/src/eth/l2EventsProvider.ts
@@ -33,9 +33,9 @@ export class OptimismConstants {
   public static StorageRegistryAddress = "0x000000fC0a4Fccee0b30E360773F7888D1bD9FAA" as const;
   public static KeyRegistryAddress = "0x000000fc6548800fc8265d8eb7061d88cefb87c2" as const;
   public static IdRegistryAddress = "0x000000fc99489b8cd629291d97dbca62b81173c4" as const;
-  public static FirstBlock = 12500000;
+  public static FirstBlock = 108222360; // ~Aug 14 2023 8pm UTC, approx 1.5 weeks before planned migration
   public static ChunkSize = 1000;
-  public static ChainId = 420; // OP Goerli
+  public static ChainId = 10; // OP mainnet
 }
 
 const RENT_EXPIRY_IN_SECONDS = 365 * 24 * 60 * 60; // One year
@@ -502,8 +502,8 @@ export class L2EventsProvider {
     let lastSyncedBlock = this._firstBlock;
 
     const hubState = await this._hub.getHubState();
-    if (hubState.isOk() && hubState.value.lastEthBlock) {
-      lastSyncedBlock = hubState.value.lastEthBlock;
+    if (hubState.isOk() && hubState.value.lastL2Block) {
+      lastSyncedBlock = hubState.value.lastL2Block;
     }
 
     if (this._resyncEvents) {
@@ -639,8 +639,13 @@ export class L2EventsProvider {
 
     // Update the last synced block if all the historical events have been synced
     if (this._isHistoricalSyncDone) {
-      const hubState = HubState.create({ lastEthBlock: blockNumber });
-      await this._hub.putHubState(hubState);
+      const hubState = await this._hub.getHubState();
+      if (hubState.isOk()) {
+        hubState.value.lastL2Block = blockNumber;
+        await this._hub.putHubState(hubState.value);
+      } else {
+        log.error({ errCode: hubState.error.errCode }, `failed to get hub state: ${hubState.error.message}`);
+      }
     }
 
     this._blockTimestampsCache.clear(); // Clear the cache periodically to avoid unbounded growth

diff --git a/apps/hubble/src/hubble.ts b/apps/hubble/src/hubble.ts
@@ -7,7 +7,6 @@ import {
   IdRegistryEvent,
   Message,
   NameRegistryEvent,
-  UpdateNameRegistryEventExpiryJobPayload,
   HubAsyncResult,
   HubError,
   bytesToHexString,
@@ -38,10 +37,6 @@ import { RootPrefix } from "./storage/db/types.js";
 import Engine from "./storage/engine/index.js";
 import { PruneEventsJobScheduler } from "./storage/jobs/pruneEventsJob.js";
 import { PruneMessagesJobScheduler } from "./storage/jobs/pruneMessagesJob.js";
-import {
-  UpdateNameRegistryEventExpiryJobQueue,
-  UpdateNameRegistryEventExpiryJobWorker,
-} from "./storage/jobs/updateNameRegistryEventExpiryJob.js";
 import { sleep } from "./utils/crypto.js";
 import {
   idRegistryEventToLog,
@@ -281,9 +276,6 @@ export class Hub implements HubInterface {
   private checkIncomingPortsJobScheduler: CheckIncomingPortsJobScheduler;
   private updateNetworkConfigJobScheduler: UpdateNetworkConfigJobScheduler;
 
-  private updateNameRegistryEventExpiryJobQueue: UpdateNameRegistryEventExpiryJobQueue;
-  private updateNameRegistryEventExpiryJobWorker?: UpdateNameRegistryEventExpiryJobWorker;
-
   engine: Engine;
   ethRegistryProvider?: EthEventsProvider;
   fNameRegistryEventsProvider?: FNameRegistryEventsProvider;
@@ -414,9 +406,6 @@ export class Hub implements HubInterface {
     );
     this.adminServer = new AdminServer(this, this.rocksDB, this.engine, this.syncEngine, options.rpcAuth);
 
-    // Setup job queues
-    this.updateNameRegistryEventExpiryJobQueue = new UpdateNameRegistryEventExpiryJobQueue(this.rocksDB);
-
     // Setup job schedulers/workers
     this.pruneMessagesJobScheduler = new PruneMessagesJobScheduler(this.engine);
     this.periodSyncJobScheduler = new PeriodicSyncJobScheduler(this, this.syncEngine);
@@ -431,14 +420,6 @@ export class Hub implements HubInterface {
       this.testDataJobScheduler = new PeriodicTestDataJobScheduler(this.rpcServer, options.testUsers as TestUser[]);
     }
 
-    if (this.ethRegistryProvider) {
-      this.updateNameRegistryEventExpiryJobWorker = new UpdateNameRegistryEventExpiryJobWorker(
-        this.updateNameRegistryEventExpiryJobQueue,
-        this.rocksDB,
-        this.ethRegistryProvider,
-      );
-    }
-
     // Allowed peers can be undefined, which means permissionless connections
     this.allowedPeerIds = this.options.allowedPeers;
     // Denied peers by default is an empty list
@@ -588,10 +569,6 @@ export class Hub implements HubInterface {
     // Start the sync engine
     await this.syncEngine.initialize(this.options.rebuildSyncTrie ?? false);
 
-    if (this.updateNameRegistryEventExpiryJobWorker) {
-      this.updateNameRegistryEventExpiryJobWorker.start();
-    }
-
     let bootstrapAddrs = this.options.bootstrapAddrs ?? [];
     // Add mainnet bootstrap addresses if none are provided
     if (bootstrapAddrs.length === 0 && this.options.network === FarcasterNetwork.MAINNET) {
@@ -716,17 +693,12 @@ export class Hub implements HubInterface {
     // Stop admin, gossip and sync engine
     await Promise.all([this.adminServer.stop(), this.gossipNode.stop(), this.syncEngine.stop()]);
 
-    if (this.updateNameRegistryEventExpiryJobWorker) {
-      this.updateNameRegistryEventExpiryJobWorker.stop();
-    }
-
     // Stop cron tasks
     this.pruneMessagesJobScheduler.stop();
     this.periodSyncJobScheduler.stop();
     this.pruneEventsJobScheduler.stop();
     this.checkFarcasterVersionJobScheduler.stop();
     this.testDataJobScheduler?.stop();
-    this.updateNameRegistryEventExpiryJobWorker?.stop();
     this.validateOrRevokeMessagesJobScheduler.stop();
     this.gossipContactInfoJobScheduler.stop();
     this.checkIncomingPortsJobScheduler.stop();
@@ -1131,11 +1103,6 @@ export class Hub implements HubInterface {
       },
     );
 
-    if (!event.expiry) {
-      const payload = UpdateNameRegistryEventExpiryJobPayload.create({ fname: event.fname });
-      await this.updateNameRegistryEventExpiryJobQueue.enqueueJob(payload);
-    }
-
     return mergeResult;
   }
 

diff --git a/packages/core/src/protobufs/generated/hub_state.ts b/packages/core/src/protobufs/generated/hub_state.ts
@@ -5,10 +5,11 @@ import _m0 from "protobufjs/minimal";
 export interface HubState {
   lastEthBlock: number;
   lastFnameProof: number;
+  lastL2Block: number;
 }
 
 function createBaseHubState(): HubState {
-  return { lastEthBlock: 0, lastFnameProof: 0 };
+  return { lastEthBlock: 0, lastFnameProof: 0, lastL2Block: 0 };
 }
 
 export const HubState = {
@@ -19,6 +20,9 @@ export const HubState = {
     if (message.lastFnameProof !== 0) {
       writer.uint32(16).uint64(message.lastFnameProof);
     }
+    if (message.lastL2Block !== 0) {
+      writer.uint32(24).uint64(message.lastL2Block);
+    }
     return writer;
   },
 
@@ -43,6 +47,13 @@ export const HubState = {
 
           message.lastFnameProof = longToNumber(reader.uint64() as Long);
           continue;
+        case 3:
+          if (tag != 24) {
+            break;
+          }
+
+          message.lastL2Block = longToNumber(reader.uint64() as Long);
+          continue;
       }
       if ((tag & 7) == 4 || tag == 0) {
         break;
@@ -56,13 +67,15 @@ export const HubState = {
     return {
       lastEthBlock: isSet(object.lastEthBlock) ? Number(object.lastEthBlock) : 0,
       lastFnameProof: isSet(object.lastFnameProof) ? Number(object.lastFnameProof) : 0,
+      lastL2Block: isSet(object.lastL2Block) ? Number(object.lastL2Block) : 0,
     };
   },
 
   toJSON(message: HubState): unknown {
     const obj: any = {};
     message.lastEthBlock !== undefined && (obj.lastEthBlock = Math.round(message.lastEthBlock));
     message.lastFnameProof !== undefined && (obj.lastFnameProof = Math.round(message.lastFnameProof));
+    message.lastL2Block !== undefined && (obj.lastL2Block = Math.round(message.lastL2Block));
     return obj;
   },
 
@@ -74,6 +87,7 @@ export const HubState = {
     const message = createBaseHubState();
     message.lastEthBlock = object.lastEthBlock ?? 0;
     message.lastFnameProof = object.lastFnameProof ?? 0;
+    message.lastL2Block = object.lastL2Block ?? 0;
     return message;
   },
 };

diff --git a/packages/hub-web/src/generated/request_response.ts b/packages/hub-web/src/generated/request_response.ts
@@ -106,6 +106,7 @@ export interface SyncStatusRequest {
 export interface SyncStatusResponse {
   isSyncing: boolean;
   syncStatus: SyncStatus[];
+  engineStarted: boolean;
 }
 
 export interface SyncStatus {
@@ -770,7 +771,7 @@ export const SyncStatusRequest = {
 };
 
 function createBaseSyncStatusResponse(): SyncStatusResponse {
-  return { isSyncing: false, syncStatus: [] };
+  return { isSyncing: false, syncStatus: [], engineStarted: false };
 }
 
 export const SyncStatusResponse = {
@@ -781,6 +782,9 @@ export const SyncStatusResponse = {
     for (const v of message.syncStatus) {
       SyncStatus.encode(v!, writer.uint32(18).fork()).ldelim();
     }
+    if (message.engineStarted === true) {
+      writer.uint32(24).bool(message.engineStarted);
+    }
     return writer;
   },
 
@@ -805,6 +809,13 @@ export const SyncStatusResponse = {
 
           message.syncStatus.push(SyncStatus.decode(reader, reader.uint32()));
           continue;
+        case 3:
+          if (tag != 24) {
+            break;
+          }
+
+          message.engineStarted = reader.bool();
+          continue;
       }
       if ((tag & 7) == 4 || tag == 0) {
         break;
@@ -818,6 +829,7 @@ export const SyncStatusResponse = {
     return {
       isSyncing: isSet(object.isSyncing) ? Boolean(object.isSyncing) : false,
       syncStatus: Array.isArray(object?.syncStatus) ? object.syncStatus.map((e: any) => SyncStatus.fromJSON(e)) : [],
+      engineStarted: isSet(object.engineStarted) ? Boolean(object.engineStarted) : false,
     };
   },
 
@@ -829,6 +841,7 @@ export const SyncStatusResponse = {
     } else {
       obj.syncStatus = [];
     }
+    message.engineStarted !== undefined && (obj.engineStarted = message.engineStarted);
     return obj;
   },
 
@@ -840,6 +853,7 @@ export const SyncStatusResponse = {
     const message = createBaseSyncStatusResponse();
     message.isSyncing = object.isSyncing ?? false;
     message.syncStatus = object.syncStatus?.map((e) => SyncStatus.fromPartial(e)) || [];
+    message.engineStarted = object.engineStarted ?? false;
     return message;
   },
 };

diff --git a/protobufs/schemas/hub_state.proto b/protobufs/schemas/hub_state.proto
@@ -3,4 +3,5 @@ syntax = "proto3";
 message HubState {
   uint32 last_eth_block = 1;
   uint64 last_fname_proof = 2;
+  uint64 last_l2_block = 3;
 }