fix(archiver): filter tagged log queries by block number (#21388)

Resolves the referenceBlock hash to a block number in the AztecNode and
passes it down as upToBlockNumber so the LogStore stops returning logs
from blocks beyond the client's sync point. Also adds an ordering check
on log insertion to guard against out-of-order appends.

Fixes F-417

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: spalladino

yarn-project/archiver/src/errors.ts

@@ -74,6 +74,22 @@ export class BlockAlreadyCheckpointedError extends Error {
   }
 }
 
+/** Thrown when logs are added for a tag whose last stored log has a higher block number than the new log. */
+export class OutOfOrderLogInsertionError extends Error {
+  constructor(
+    public readonly logType: 'private' | 'public',
+    public readonly tag: string,
+    public readonly lastBlockNumber: number,
+    public readonly newBlockNumber: number,
+  ) {
+    super(
+      `Out-of-order ${logType} log insertion for tag ${tag}: ` +
+        `last existing log is from block ${lastBlockNumber} but new log is from block ${newBlockNumber}`,
+    );
+    this.name = 'OutOfOrderLogInsertionError';
+  }
+}
+
 /** Thrown when a proposed block conflicts with an already checkpointed block (different content). */
 export class CannotOverwriteCheckpointedBlockError extends Error {
   constructor(

yarn-project/archiver/src/modules/data_source_base.ts

@@ -165,16 +165,21 @@ export abstract class ArchiverDataSourceBase
     return (await this.store.getPendingChainValidationStatus()) ?? { valid: true };
   }
 
-  public getPrivateLogsByTags(tags: SiloedTag[], page?: number): Promise<TxScopedL2Log[][]> {
-    return this.store.getPrivateLogsByTags(tags, page);
+  public getPrivateLogsByTags(
+    tags: SiloedTag[],
+    page?: number,
+    upToBlockNumber?: BlockNumber,
+  ): Promise<TxScopedL2Log[][]> {
+    return this.store.getPrivateLogsByTags(tags, page, upToBlockNumber);
   }
 
   public getPublicLogsByTagsFromContract(
     contractAddress: AztecAddress,
     tags: Tag[],
     page?: number,
+    upToBlockNumber?: BlockNumber,
   ): Promise<TxScopedL2Log[][]> {
-    return this.store.getPublicLogsByTagsFromContract(contractAddress, tags, page);
+    return this.store.getPublicLogsByTagsFromContract(contractAddress, tags, page, upToBlockNumber);
   }
 
   public getPublicLogs(filter: LogFilter): Promise<GetPublicLogsResponse> {

yarn-project/archiver/src/store/kv_archiver_store.test.ts

@@ -49,6 +49,7 @@ import {
   CannotOverwriteCheckpointedBlockError,
   CheckpointNumberNotSequentialError,
   InitialCheckpointNumberNotSequentialError,
+  OutOfOrderLogInsertionError,
 } from '../errors.js';
 import { MessageStoreError } from '../store/message_store.js';
 import type { InboxMessage } from '../structs/inbox_message.js';
@@ -2329,6 +2330,32 @@ describe('KVArchiverDataStore', () => {
       ]);
     });
 
+    it('throws on out-of-order private log insertion', async () => {
+      const sharedTag = makePrivateLogTag(99, 0, 0);
+
+      // Create blocks 4 and 5 with the same shared tag
+      const prevArchive1 = logsCheckpoints[logsCheckpoints.length - 1].checkpoint.blocks[0].archive;
+      const checkpoint4 = await makeCheckpointWithLogs(numBlocksForLogs + 1, {
+        previousArchive: prevArchive1,
+        numTxsPerBlock,
+        privateLogs: { numLogsPerTx: numPrivateLogsPerTx },
+      });
+      checkpoint4.checkpoint.blocks[0].body.txEffects[0].privateLogs[0] = makePrivateLog(sharedTag);
+
+      const prevArchive2 = checkpoint4.checkpoint.blocks[0].archive;
+      const checkpoint5 = await makeCheckpointWithLogs(numBlocksForLogs + 2, {
+        previousArchive: prevArchive2,
+        numTxsPerBlock,
+        privateLogs: { numLogsPerTx: numPrivateLogsPerTx },
+      });
+      checkpoint5.checkpoint.blocks[0].body.txEffects[0].privateLogs[0] = makePrivateLog(sharedTag);
+
+      // Store block 5's logs first (higher block number), then try to store block 4's logs
+      // (lower block number) — this should fail.
+      await store.addLogs([checkpoint5.checkpoint.blocks[0]]);
+      await expect(store.addLogs([checkpoint4.checkpoint.blocks[0]])).rejects.toThrow(OutOfOrderLogInsertionError);
+    });
+
     it('is possible to request logs for non-existing tags and determine their position', async () => {
       const tags = [makePrivateLogTag(99, 88, 77), makePrivateLogTag(1, 1, 1)];
 
@@ -2347,6 +2374,48 @@ describe('KVArchiverDataStore', () => {
       ]);
     });
 
+    it('filters logs up to specified block number', async () => {
+      // Tags are unique per block, so create a shared tag across blocks by adding logs with the same tag
+      const sharedTag = makePrivateLogTag(1, 2, 1);
+
+      // Add extra blocks with logs sharing the same tag
+      for (let blockNum = numBlocksForLogs + 1; blockNum <= numBlocksForLogs + 2; blockNum++) {
+        const previousArchive = logsCheckpoints[logsCheckpoints.length - 1].checkpoint.blocks[0].archive;
+        const newCheckpoint = await makeCheckpointWithLogs(blockNum, {
+          previousArchive,
+          numTxsPerBlock,
+          privateLogs: { numLogsPerTx: numPrivateLogsPerTx },
+        });
+        const newLog = newCheckpoint.checkpoint.blocks[0].body.txEffects[1].privateLogs[1];
+        newLog.fields[0] = sharedTag.value;
+        newCheckpoint.checkpoint.blocks[0].body.txEffects[1].privateLogs[1] = newLog;
+        await store.addCheckpoints([newCheckpoint]);
+        await store.addLogs([newCheckpoint.checkpoint.blocks[0]]);
+        logsCheckpoints.push(newCheckpoint);
+      }
+
+      // Without filter, should return logs from block 1 and the extra blocks
+      const allLogs = await store.getPrivateLogsByTags([sharedTag]);
+      expect(allLogs[0].some(log => log.blockNumber > numBlocksForLogs)).toBe(true);
+
+      // With upToBlockNumber=numBlocksForLogs, should only return the original log from block 1
+      const filteredLogs = await store.getPrivateLogsByTags([sharedTag], 0, BlockNumber(numBlocksForLogs));
+      expect(filteredLogs[0].length).toBeGreaterThan(0);
+      for (const log of filteredLogs[0]) {
+        expect(log.blockNumber).toBeLessThanOrEqual(numBlocksForLogs);
+      }
+      expect(filteredLogs[0].length).toBeLessThan(allLogs[0].length);
+    });
+
+    it('returns all logs when upToBlockNumber is not set', async () => {
+      const tag = makePrivateLogTag(1, 2, 1);
+
+      const logsWithoutFilter = await store.getPrivateLogsByTags([tag]);
+      const logsWithUndefined = await store.getPrivateLogsByTags([tag], 0, undefined);
+
+      expect(logsWithoutFilter).toEqual(logsWithUndefined);
+    });
+
     describe('pagination', () => {
       const paginationTag = makePrivateLogTag(1, 2, 1);
 
@@ -2368,6 +2437,20 @@ describe('KVArchiverDataStore', () => {
         }
       });
 
+      it('pagination works correctly with upToBlockNumber', async () => {
+        // With a low upToBlockNumber, the filtered set should be smaller than MAX_LOGS_PER_TAG
+        const filteredPage0 = await store.getPrivateLogsByTags([paginationTag], 0, BlockNumber(5));
+        for (const log of filteredPage0[0]) {
+          expect(log.blockNumber).toBeLessThanOrEqual(5);
+        }
+
+        // Page 1 with the same filter should only contain remaining filtered logs
+        const filteredPage1 = await store.getPrivateLogsByTags([paginationTag], 1, BlockNumber(5));
+        for (const log of filteredPage1[0]) {
+          expect(log.blockNumber).toBeLessThanOrEqual(5);
+        }
+      });
+
       it('returns first page of logs when page=0', async () => {
         const logsByTags = await store.getPrivateLogsByTags([paginationTag], 0);
 
@@ -2535,6 +2618,32 @@ describe('KVArchiverDataStore', () => {
       ]);
     });
 
+    it('throws on out-of-order public log insertion', async () => {
+      const sharedTag = makePublicLogTag(99, 0, 0);
+
+      // Create blocks 4 and 5 with the same shared tag
+      const prevArchive1 = logsCheckpoints[logsCheckpoints.length - 1].checkpoint.blocks[0].archive;
+      const checkpoint4 = await makeCheckpointWithLogs(numBlocksForLogs + 1, {
+        previousArchive: prevArchive1,
+        numTxsPerBlock,
+        publicLogs: { numLogsPerTx: numPublicLogsPerTx, contractAddress },
+      });
+      checkpoint4.checkpoint.blocks[0].body.txEffects[0].publicLogs[0] = makePublicLog(sharedTag, contractAddress);
+
+      const prevArchive2 = checkpoint4.checkpoint.blocks[0].archive;
+      const checkpoint5 = await makeCheckpointWithLogs(numBlocksForLogs + 2, {
+        previousArchive: prevArchive2,
+        numTxsPerBlock,
+        publicLogs: { numLogsPerTx: numPublicLogsPerTx, contractAddress },
+      });
+      checkpoint5.checkpoint.blocks[0].body.txEffects[0].publicLogs[0] = makePublicLog(sharedTag, contractAddress);
+
+      // Store block 5's logs first (higher block number), then try to store block 4's logs
+      // (lower block number) — this should fail.
+      await store.addLogs([checkpoint5.checkpoint.blocks[0]]);
+      await expect(store.addLogs([checkpoint4.checkpoint.blocks[0]])).rejects.toThrow(OutOfOrderLogInsertionError);
+    });
+
     it('is possible to request logs for non-existing tags and determine their position', async () => {
       const tags = [makePublicLogTag(99, 88, 77), makePublicLogTag(1, 1, 0)];
 
@@ -2553,6 +2662,52 @@ describe('KVArchiverDataStore', () => {
       ]);
     });
 
+    it('filters logs up to specified block number', async () => {
+      const sharedTag = makePublicLogTag(1, 2, 1);
+
+      // Add extra blocks with logs sharing the same tag
+      for (let blockNum = numBlocksForLogs + 1; blockNum <= numBlocksForLogs + 2; blockNum++) {
+        const previousArchive = logsCheckpoints[logsCheckpoints.length - 1].checkpoint.blocks[0].archive;
+        const newCheckpoint = await makeCheckpointWithLogs(blockNum, {
+          previousArchive,
+          numTxsPerBlock,
+          publicLogs: { numLogsPerTx: numPublicLogsPerTx, contractAddress },
+        });
+        const newLog = newCheckpoint.checkpoint.blocks[0].body.txEffects[1].publicLogs[1];
+        newLog.fields[0] = sharedTag.value;
+        newCheckpoint.checkpoint.blocks[0].body.txEffects[1].publicLogs[1] = newLog;
+        await store.addCheckpoints([newCheckpoint]);
+        await store.addLogs([newCheckpoint.checkpoint.blocks[0]]);
+        logsCheckpoints.push(newCheckpoint);
+      }
+
+      // Without filter, should return logs from block 1 and the extra blocks
+      const allLogs = await store.getPublicLogsByTagsFromContract(contractAddress, [sharedTag]);
+      expect(allLogs[0].some(log => log.blockNumber > numBlocksForLogs)).toBe(true);
+
+      // With upToBlockNumber=numBlocksForLogs, should only return the original log from block 1
+      const filteredLogs = await store.getPublicLogsByTagsFromContract(
+        contractAddress,
+        [sharedTag],
+        0,
+        BlockNumber(numBlocksForLogs),
+      );
+      expect(filteredLogs[0].length).toBeGreaterThan(0);
+      for (const log of filteredLogs[0]) {
+        expect(log.blockNumber).toBeLessThanOrEqual(numBlocksForLogs);
+      }
+      expect(filteredLogs[0].length).toBeLessThan(allLogs[0].length);
+    });
+
+    it('returns all logs when upToBlockNumber is not set', async () => {
+      const tag = makePublicLogTag(1, 2, 1);
+
+      const logsWithoutFilter = await store.getPublicLogsByTagsFromContract(contractAddress, [tag]);
+      const logsWithUndefined = await store.getPublicLogsByTagsFromContract(contractAddress, [tag], 0, undefined);
+
+      expect(logsWithoutFilter).toEqual(logsWithUndefined);
+    });
+
     describe('pagination', () => {
       const paginationTag = makePublicLogTag(1, 2, 1);
 
@@ -2574,6 +2729,28 @@ describe('KVArchiverDataStore', () => {
         }
       });
 
+      it('pagination works correctly with upToBlockNumber', async () => {
+        const filteredPage0 = await store.getPublicLogsByTagsFromContract(
+          contractAddress,
+          [paginationTag],
+          0,
+          BlockNumber(5),
+        );
+        for (const log of filteredPage0[0]) {
+          expect(log.blockNumber).toBeLessThanOrEqual(5);
+        }
+
+        const filteredPage1 = await store.getPublicLogsByTagsFromContract(
+          contractAddress,
+          [paginationTag],
+          1,
+          BlockNumber(5),
+        );
+        for (const log of filteredPage1[0]) {
+          expect(log.blockNumber).toBeLessThanOrEqual(5);
+        }
+      });
+
       it('returns first page of logs when page=0', async () => {
         const logsByTags = await store.getPublicLogsByTagsFromContract(contractAddress, [paginationTag], 0);
 

yarn-project/archiver/src/store/kv_archiver_store.ts

@@ -470,10 +470,11 @@ export class KVArchiverDataStore implements ContractDataSource {
    * array implies no logs match that tag.
    * @param tags - The tags to search for.
    * @param page - The page number (0-indexed) for pagination. Returns at most 10 logs per tag per page.
+   * @param upToBlockNumber - If set, only return logs from blocks up to and including this block number.
    */
-  getPrivateLogsByTags(tags: SiloedTag[], page?: number): Promise<TxScopedL2Log[][]> {
+  getPrivateLogsByTags(tags: SiloedTag[], page?: number, upToBlockNumber?: BlockNumber): Promise<TxScopedL2Log[][]> {
     try {
-      return this.#logStore.getPrivateLogsByTags(tags, page);
+      return this.#logStore.getPrivateLogsByTags(tags, page, upToBlockNumber);
     } catch (err) {
       return Promise.reject(err);
     }
@@ -485,14 +486,16 @@ export class KVArchiverDataStore implements ContractDataSource {
    * @param contractAddress - The contract address to search logs for.
    * @param tags - The tags to search for.
    * @param page - The page number (0-indexed) for pagination. Returns at most 10 logs per tag per page.
+   * @param upToBlockNumber - If set, only return logs from blocks up to and including this block number.
    */
   getPublicLogsByTagsFromContract(
     contractAddress: AztecAddress,
     tags: Tag[],
     page?: number,
+    upToBlockNumber?: BlockNumber,
   ): Promise<TxScopedL2Log[][]> {
     try {
-      return this.#logStore.getPublicLogsByTagsFromContract(contractAddress, tags, page);
+      return this.#logStore.getPublicLogsByTagsFromContract(contractAddress, tags, page, upToBlockNumber);
     } catch (err) {
       return Promise.reject(err);
     }