fix: stabilize watch fromBlock CI

src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.test.ts

@@ -1,5 +1,5 @@
 import * as BunTest from 'bun:test';
-import { type Address, getAddress } from 'viem';
+import { type Address, type PublicClient, getAddress } from 'viem';
 import {
   type AnnouncementLog,
   ERC5564AnnouncerAbi,
@@ -12,7 +12,7 @@ import setupTestStealthKeys from '../../helpers/test/setupTestStealthKeys';
 import setupTestWallet from '../../helpers/test/setupTestWallet';
 import type { SuperWalletClient } from '../../helpers/types';
 import type { StealthActions } from '../../stealthClient/types';
-import {
+import watchAnnouncementsForUser, {
   createWatchedAnnouncementsQueue,
   processWatchedAnnouncementsBatch,
   startWatchHeartbeat
@@ -110,6 +110,54 @@ const announce = async ({
   };
 };
 
+const getNextBlockNumber = async (walletClient: SuperWalletClient) =>
+  (await walletClient.getBlockNumber({ cacheTime: 0 })) + 1n;
+
+describe('watchAnnouncementsForUser setup failures', () => {
+  test('does not start heartbeat when polling setup fails', async () => {
+    const setupError = new Error('initial block number failed');
+    const reportedErrors: string[] = [];
+    let getBlockNumberCalls = 0;
+    const schemeId = VALID_SCHEME_ID.SCHEME_ID_1;
+    const schemeIdBigInt = BigInt(schemeId);
+    const { spendingPublicKey, viewingPrivateKey } =
+      setupTestStealthKeys(schemeId);
+
+    await expect(
+      watchAnnouncementsForUser({
+        ERC5564Address: ERC5564_CONTRACT_ADDRESS as Address,
+        args: {
+          schemeId: schemeIdBigInt,
+          caller: '0x00000000000000000000000000000000000000AA'
+        },
+        clientParams: {
+          publicClient: {
+            getBlockNumber: async () => {
+              getBlockNumberCalls += 1;
+              throw setupError;
+            }
+          } as unknown as PublicClient
+        },
+        handleLogsForUser: () => {},
+        onError: error => {
+          reportedErrors.push(error.message);
+        },
+        onHeartbeat: () => {},
+        pollOptions: {
+          pollingInterval: 10
+        },
+        spendingPublicKey,
+        viewingPrivateKey
+      })
+    ).rejects.toThrow('initial block number failed');
+
+    await sleep(35);
+
+    expect(getBlockNumberCalls).toEqual(1);
+    expect(reportedErrors).toHaveLength(0);
+  });
+});
+
 describe('watchAnnouncementsForUser', () => {
   let stealthClient: StealthActions;
   let walletClient: SuperWalletClient;
@@ -131,7 +179,7 @@ describe('watchAnnouncementsForUser', () => {
 
   test('awaits async handlers and still delivers relevant announcements', async () => {
     const watchedAnnouncements: AnnouncementLog[] = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
     const unwatch = await stealthClient.watchAnnouncementsForUser({
       ERC5564Address,
       args: {
@@ -186,7 +234,7 @@ describe('watchAnnouncementsForUser', () => {
   test('reports rejected handlers once and keeps watching later announcements', async () => {
     const handledAnnouncements: AnnouncementLog[] = [];
     const handlerErrors: string[] = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
     let handlerCalls = 0;
 
     const unwatch = await stealthClient.watchAnnouncementsForUser({
@@ -272,7 +320,7 @@ describe('watchAnnouncementsForUser', () => {
     const loggedErrors: Error[] = [];
     const handledAnnouncements: AnnouncementLog[] = [];
     let handlerCalls = 0;
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
 
     console.error = ((value: unknown) => {
       if (value instanceof Error) {
@@ -360,7 +408,7 @@ describe('watchAnnouncementsForUser', () => {
     const originalConsoleError = console.error;
     const loggedErrors: Error[] = [];
     const handledAnnouncements: AnnouncementLog[] = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
     let handlerCalls = 0;
 
     console.error = ((value: unknown) => {
@@ -452,7 +500,7 @@ describe('watchAnnouncementsForUser', () => {
   test('recovers when fallback console logging throws', async () => {
     const originalConsoleError = console.error;
     const handledAnnouncements: AnnouncementLog[] = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
     let consoleErrorCalls = 0;
     let handlerCalls = 0;
 
@@ -613,7 +661,7 @@ describe('watchAnnouncementsForUser', () => {
       rawLogCount: number;
       relevantLogCount: number;
     }> = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
 
     const unwatch = await stealthClient.watchAnnouncementsForUser({
       ERC5564Address,
@@ -697,7 +745,7 @@ describe('watchAnnouncementsForUser', () => {
     let activeHandlers = 0;
     let handlerCalls = 0;
     let overlapDetected = false;
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
 
     const unwatch = await stealthClient.watchAnnouncementsForUser({
       ERC5564Address,
@@ -791,7 +839,7 @@ describe('watchAnnouncementsForUser', () => {
     const firstHandlerStarted = createDeferred();
     const firstBatchFinished = createDeferred();
     const handledAnnouncements: AnnouncementLog[] = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
     let sawLaterBatch = false;
     let sawFirstBatch = false;
 
@@ -879,7 +927,7 @@ describe('watchAnnouncementsForUser', () => {
 
   test('does not emit announcements that do not apply to the user', async () => {
     const watchedAnnouncements: AnnouncementLog[] = [];
-    const fromBlock = (await walletClient.getBlockNumber()) + 1n;
+    const fromBlock = await getNextBlockNumber(walletClient);
     const unwatch = await stealthClient.watchAnnouncementsForUser({
       ERC5564Address,
       args: {

src/lib/actions/watchAnnouncementsForUser/watchAnnouncementsForUser.ts

@@ -57,6 +57,7 @@ async function reportWatchError({
 }
 
 const DEFAULT_WATCH_HEARTBEAT_INTERVAL_MS = 4_000;
+const DEFAULT_WATCH_POLLING_INTERVAL_MS = 4_000;
 
 type WatchedAnnouncementsQueue = {
   enqueue: (task: () => Promise<void>) => void;
@@ -96,6 +97,20 @@ type WatchedAnnouncementEventLog = Omit<
   >;
 };
 
+const getInitialPreviousBlockNumber = async ({
+  fromBlock,
+  publicClient
+}: {
+  fromBlock?: WatchAnnouncementsForUserParams['fromBlock'];
+  publicClient: ReturnType<typeof handleViemPublicClient>;
+}): Promise<bigint> => {
+  if (typeof fromBlock === 'bigint') {
+    return fromBlock - 1n;
+  }
+
+  return publicClient.getBlockNumber({ cacheTime: 0 });
+};
+
 const getObservedBlockFromLogs = (
   logs: WatchedAnnouncementEventLog[]
 ): bigint | undefined =>
@@ -245,6 +260,136 @@ export function startWatchHeartbeat<T = void>({
   };
 }
 
+export async function startPollingAnnouncementEvents<T = void>({
+  args,
+  ERC5564Address,
+  excludeList,
+  fromBlock,
+  handleLogsForUser,
+  includeList,
+  onError,
+  pollOptions,
+  publicClient,
+  spendingPublicKey,
+  viewingPrivateKey,
+  watchedAnnouncementsQueue
+}: {
+  args: WatchAnnouncementsForUserParams<T>['args'];
+  ERC5564Address: WatchAnnouncementsForUserParams<T>['ERC5564Address'];
+  excludeList?: WatchAnnouncementsForUserParams<T>['excludeList'];
+  fromBlock?: WatchAnnouncementsForUserParams<T>['fromBlock'];
+  handleLogsForUser: WatchAnnouncementsForUserParams<T>['handleLogsForUser'];
+  includeList?: WatchAnnouncementsForUserParams<T>['includeList'];
+  onError?: WatchAnnouncementsForUserParams<T>['onError'];
+  pollOptions?: WatchAnnouncementsForUserParams<T>['pollOptions'];
+  publicClient: ReturnType<typeof handleViemPublicClient>;
+  spendingPublicKey: WatchAnnouncementsForUserParams<T>['spendingPublicKey'];
+  viewingPrivateKey: WatchAnnouncementsForUserParams<T>['viewingPrivateKey'];
+  watchedAnnouncementsQueue: WatchedAnnouncementsQueue;
+}): Promise<WatchAnnouncementsForUserReturnType> {
+  let active = true;
+  let timeoutId: ReturnType<typeof setTimeout> | undefined;
+  let previousBlockNumber = await getInitialPreviousBlockNumber({
+    fromBlock,
+    publicClient
+  });
+  const pollingInterval =
+    pollOptions?.pollingInterval ??
+    publicClient.pollingInterval ??
+    DEFAULT_WATCH_POLLING_INTERVAL_MS;
+  const enqueueLogs = (logs: WatchedAnnouncementEventLog[]) => {
+    if (logs.length === 0) {
+      return;
+    }
+
+    const batches =
+      pollOptions?.batch === false ? logs.map(log => [log]) : [logs];
+
+    for (const batch of batches) {
+      watchedAnnouncementsQueue.enqueue(async () => {
+        await processWatchedAnnouncementsBatch({
+          logs: batch,
+          spendingPublicKey,
+          viewingPrivateKey,
+          publicClient,
+          excludeList,
+          includeList,
+          fromBlock,
+          handleLogsForUser,
+          onError
+        });
+      });
+    }
+  };
+
+  const pollAnnouncements = async () => {
+    try {
+      const blockNumber = await publicClient.getBlockNumber({ cacheTime: 0 });
+      if (!active) {
+        return;
+      }
+
+      const nextFromBlock = previousBlockNumber + 1n;
+
+      if (nextFromBlock > blockNumber) {
+        return;
+      }
+
+      const logs = (await publicClient.getContractEvents({
+        address: ERC5564Address,
+        abi: ERC5564AnnouncerAbi,
+        eventName: 'Announcement',
+        args,
+        fromBlock: nextFromBlock,
+        toBlock: blockNumber,
+        strict: true
+      })) as WatchedAnnouncementEventLog[];
+
+      if (!active) {
+        return;
+      }
+
+      previousBlockNumber = blockNumber;
+      enqueueLogs(logs);
+    } catch (error) {
+      await reportWatchError({
+        error,
+        fallbackMessage:
+          'watchAnnouncementsForUser failed while polling announcement events',
+        onError
+      });
+    }
+  };
+
+  const scheduleNextPoll = () => {
+    timeoutId = setTimeout(() => {
+      void runPollLoop();
+    }, pollingInterval);
+  };
+
+  const runPollLoop = async () => {
+    if (!active) {
+      return;
+    }
+
+    await pollAnnouncements();
+
+    if (active) {
+      scheduleNextPoll();
+    }
+  };
+
+  void runPollLoop();
+
+  return () => {
+    active = false;
+
+    if (timeoutId !== undefined) {
+      clearTimeout(timeoutId);
+    }
+  };
+}
+
 /**
  * Watches for announcement events relevant to the user.
  *
@@ -285,6 +430,22 @@ async function watchAnnouncementsForUser<T = void>({
       });
     }
   });
+  // Poll with getContractEvents directly so logs cannot slip through while an RPC filter is still being created.
+  const unwatchContractEvent = await startPollingAnnouncementEvents({
+    args,
+    ERC5564Address,
+    excludeList,
+    fromBlock,
+    handleLogsForUser,
+    includeList,
+    onError,
+    pollOptions,
+    publicClient,
+    spendingPublicKey,
+    viewingPrivateKey,
+    watchedAnnouncementsQueue
+  });
+
   const stopHeartbeat = startWatchHeartbeat({
     fromBlock,
     onError,
@@ -293,31 +454,6 @@ async function watchAnnouncementsForUser<T = void>({
     publicClient
   });
 
-  const unwatchContractEvent = publicClient.watchContractEvent({
-    address: ERC5564Address,
-    abi: ERC5564AnnouncerAbi,
-    eventName: 'Announcement',
-    args,
-    ...(fromBlock === 'latest' ? {} : { fromBlock }),
-    onLogs: logs => {
-      watchedAnnouncementsQueue.enqueue(async () => {
-        await processWatchedAnnouncementsBatch({
-          logs,
-          spendingPublicKey,
-          viewingPrivateKey,
-          publicClient,
-          excludeList,
-          includeList,
-          fromBlock,
-          handleLogsForUser,
-          onError
-        });
-      });
-    },
-    strict: true,
-    ...pollOptions
-  });
-
   return () => {
     stopHeartbeat();
     unwatchContractEvent();