refactor: more robust tagging index sync as recipient

yarn-project/pxe/src/storage/tagging_data_provider/sender_tagging_data_provider.test.ts

@@ -3,7 +3,7 @@ import { openTmpStore } from '@aztec/kv-store/lmdb-v2';
 import { DirectionalAppTaggingSecret, type PreTag } from '@aztec/stdlib/logs';
 import { TxHash } from '@aztec/stdlib/tx';
 
-import { WINDOW_LEN } from '../../tagging/sync/sync_sender_tagging_indexes.js';
+import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../../tagging/sync/sync_sender_tagging_indexes.js';
 import { SenderTaggingDataProvider } from './sender_tagging_data_provider.js';
 
 describe('SenderTaggingDataProvider', () => {
@@ -147,7 +147,7 @@ describe('SenderTaggingDataProvider', () => {
         const txHash1 = TxHash.random();
         const txHash2 = TxHash.random();
         const finalizedIndex = 10;
-        const indexBeyondWindow = finalizedIndex + WINDOW_LEN + 1;
+        const indexBeyondWindow = finalizedIndex + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN + 1;
 
         // First store and finalize an index
         await taggingDataProvider.storePendingIndexes([{ secret: secret1, index: finalizedIndex }], txHash1);
@@ -165,7 +165,7 @@ describe('SenderTaggingDataProvider', () => {
         const txHash1 = TxHash.random();
         const txHash2 = TxHash.random();
         const finalizedIndex = 10;
-        const indexAtBoundary = finalizedIndex + WINDOW_LEN;
+        const indexAtBoundary = finalizedIndex + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN;
 
         // First store and finalize an index
         await taggingDataProvider.storePendingIndexes([{ secret: secret1, index: finalizedIndex }], txHash1);

yarn-project/pxe/src/storage/tagging_data_provider/sender_tagging_data_provider.ts

@@ -3,7 +3,7 @@ import type { AztecAsyncKVStore, AztecAsyncMap } from '@aztec/kv-store';
 import type { DirectionalAppTaggingSecret, PreTag } from '@aztec/stdlib/logs';
 import { TxHash } from '@aztec/stdlib/tx';
 
-import { WINDOW_LEN as SENDER_TAGGING_INDEXES_SYNC_WINDOW_LEN } from '../../tagging/sync/sync_sender_tagging_indexes.js';
+import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../../tagging/sync/sync_sender_tagging_indexes.js';
 
 /**
  * Data provider of tagging data used when syncing the sender tagging indexes. The recipient counterpart of this class
@@ -68,10 +68,10 @@ export class SenderTaggingDataProvider {
       // First we check that for any secret the highest used index in tx is not further than window length from
       // the highest finalized index.
       const finalizedIndex = (await this.getLastFinalizedIndex(secret)) ?? 0;
-      if (index > finalizedIndex + SENDER_TAGGING_INDEXES_SYNC_WINDOW_LEN) {
+      if (index > finalizedIndex + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN) {
         throw new Error(
           `Highest used index ${index} is further than window length from the highest finalized index ${finalizedIndex}.
-          Tagging window length ${SENDER_TAGGING_INDEXES_SYNC_WINDOW_LEN} is configured too low. Contact the Aztec team
+          Tagging window length ${UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN} is configured too low. Contact the Aztec team
           to increase it!`,
         );
       }

yarn-project/pxe/src/tagging/recipient_sync/load_private_logs_for_sender_recipient_pair.test.ts

@@ -0,0 +1,242 @@
+import { MAX_INCLUDE_BY_TIMESTAMP_DURATION } from '@aztec/constants';
+import { BlockNumber } from '@aztec/foundation/branded-types';
+import { Fr } from '@aztec/foundation/curves/bn254';
+import { openTmpStore } from '@aztec/kv-store/lmdb-v2';
+import { AztecAddress } from '@aztec/stdlib/aztec-address';
+import { L2BlockHash } from '@aztec/stdlib/block';
+import type { AztecNode } from '@aztec/stdlib/interfaces/server';
+import { DirectionalAppTaggingSecret, PrivateLog, TxScopedL2Log } from '@aztec/stdlib/logs';
+import { makeBlockHeader } from '@aztec/stdlib/testing';
+import { TxHash } from '@aztec/stdlib/tx';
+
+import { type MockProxy, mock } from 'jest-mock-extended';
+
+import { SiloedTag } from '../siloed_tag.js';
+import { UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN } from '../sync/sync_sender_tagging_indexes.js';
+import { Tag } from '../tag.js';
+import { loadPrivateLogsForSenderRecipientPair } from './load_private_logs_for_sender_recipient_pair.js';
+import { NewRecipientTaggingDataProvider } from './new_recipient_tagging_data_provider.js';
+
+// In this test suite we don't care about the anchor block behavior as that is sufficiently tested by
+// the loadLogsForRange test suite, so we use a high block number to ensure it occurs after all logs.
+const NON_INTERFERING_ANCHOR_BLOCK_NUMBER = BlockNumber(100);
+
+describe('loadPrivateLogsForSenderRecipientPair', () => {
+  let secret: DirectionalAppTaggingSecret;
+  let app: AztecAddress;
+
+  let aztecNode: MockProxy<AztecNode>;
+  let taggingDataProvider: NewRecipientTaggingDataProvider;
+
+  const currentTimestamp = BigInt(Math.floor(Date.now() / 1000));
+
+  async function computeSiloedTagForIndex(index: number) {
+    const tag = await Tag.compute({ secret, index });
+    return SiloedTag.compute(tag, app);
+  }
+
+  function makeLog(blockHash: Fr, blockNumber: number, tag: Fr) {
+    return new TxScopedL2Log(
+      TxHash.random(),
+      0,
+      0,
+      BlockNumber(blockNumber),
+      L2BlockHash.fromField(blockHash),
+      PrivateLog.random(tag),
+    );
+  }
+
+  beforeAll(async () => {
+    secret = DirectionalAppTaggingSecret.fromString(Fr.random().toString());
+    app = await AztecAddress.random();
+    aztecNode = mock<AztecNode>();
+  });
+
+  beforeEach(async () => {
+    aztecNode.getLogsByTags.mockReset();
+    aztecNode.getBlockHeaderByHash.mockReset();
+    aztecNode.getL2Tips.mockReset();
+    aztecNode.getBlockHeader.mockReset();
+    taggingDataProvider = new NewRecipientTaggingDataProvider(await openTmpStore('test'));
+  });
+
+  it('returns empty array when no logs found', async () => {
+    aztecNode.getL2Tips.mockResolvedValue({
+      finalized: { number: BlockNumber(10) },
+    } as any);
+
+    aztecNode.getBlockHeader.mockResolvedValue(makeBlockHeader(0, { timestamp: currentTimestamp }));
+
+    // no logs found for any tag
+    aztecNode.getLogsByTags.mockImplementation((tags: Fr[]) => {
+      return Promise.resolve(tags.map((_tag: Fr) => []));
+    });
+
+    const logs = await loadPrivateLogsForSenderRecipientPair(
+      secret,
+      app,
+      aztecNode,
+      taggingDataProvider,
+      NON_INTERFERING_ANCHOR_BLOCK_NUMBER,
+    );
+
+    expect(logs).toHaveLength(0);
+    expect(await taggingDataProvider.getHighestAgedIndex(secret)).toBeUndefined();
+    expect(await taggingDataProvider.getHighestFinalizedIndex(secret)).toBeUndefined();
+  });
+
+  it('loads log and updates highest finalized index but not highest aged index', async () => {
+    const finalizedBlockNumber = 10;
+
+    const logBlockTimestamp = currentTimestamp - 5000n; // not aged
+    const logIndex = 5;
+    const logTag = await computeSiloedTagForIndex(logIndex);
+    const logBlockHeader = makeBlockHeader(0, { timestamp: logBlockTimestamp });
+
+    aztecNode.getL2Tips.mockResolvedValue({
+      finalized: { number: BlockNumber(finalizedBlockNumber) },
+    } as any);
+
+    aztecNode.getBlockHeader.mockResolvedValue(makeBlockHeader(0, { timestamp: currentTimestamp }));
+
+    // The log is finalized
+    aztecNode.getLogsByTags.mockImplementation((tags: Fr[]) => {
+      return Promise.all(
+        tags.map(async (t: Fr) =>
+          t.equals(logTag.value) ? [makeLog(await logBlockHeader.hash(), finalizedBlockNumber, logTag.value)] : [],
+        ),
+      );
+    });
+
+    aztecNode.getBlockHeaderByHash.mockImplementation(async (hash: Fr) => {
+      if (hash.equals(await logBlockHeader.hash())) {
+        return logBlockHeader;
+      }
+      return undefined;
+    });
+
+    const logs = await loadPrivateLogsForSenderRecipientPair(
+      secret,
+      app,
+      aztecNode,
+      taggingDataProvider,
+      NON_INTERFERING_ANCHOR_BLOCK_NUMBER,
+    );
+
+    expect(logs).toHaveLength(1);
+    expect(await taggingDataProvider.getHighestFinalizedIndex(secret)).toBe(logIndex);
+    expect(await taggingDataProvider.getHighestAgedIndex(secret)).toBeUndefined();
+  });
+
+  it('loads log and updates both highest aged and highest finalized indexes', async () => {
+    const finalizedBlockNumber = 10;
+
+    const logBlockTimestamp = currentTimestamp - BigInt(MAX_INCLUDE_BY_TIMESTAMP_DURATION) - 1000n; // aged
+    const logIndex = 7;
+    const logTag = await computeSiloedTagForIndex(logIndex);
+    const logBlockHeader = makeBlockHeader(0, { timestamp: logBlockTimestamp });
+
+    aztecNode.getL2Tips.mockResolvedValue({
+      finalized: { number: BlockNumber(finalizedBlockNumber) },
+    } as any);
+
+    aztecNode.getBlockHeader.mockResolvedValue(makeBlockHeader(0, { timestamp: currentTimestamp }));
+
+    // The log is finalized
+    aztecNode.getLogsByTags.mockImplementation((tags: Fr[]) => {
+      return Promise.all(
+        tags.map(async (t: Fr) =>
+          t.equals(logTag.value) ? [makeLog(await logBlockHeader.hash(), finalizedBlockNumber, logTag.value)] : [],
+        ),
+      );
+    });
+
+    aztecNode.getBlockHeaderByHash.mockImplementation(async (hash: Fr) => {
+      if (hash.equals(await logBlockHeader.hash())) {
+        return logBlockHeader;
+      }
+      return undefined;
+    });
+
+    const logs = await loadPrivateLogsForSenderRecipientPair(
+      secret,
+      app,
+      aztecNode,
+      taggingDataProvider,
+      NON_INTERFERING_ANCHOR_BLOCK_NUMBER,
+    );
+
+    expect(logs).toHaveLength(1);
+    expect(await taggingDataProvider.getHighestAgedIndex(secret)).toBe(logIndex);
+    expect(await taggingDataProvider.getHighestFinalizedIndex(secret)).toBe(logIndex);
+  });
+
+  it('logs at boundaries are properly loaded, window and highest indexes advance as expected', async () => {
+    const finalizedBlockNumber = 10;
+
+    const log1BlockTimestamp = currentTimestamp - BigInt(MAX_INCLUDE_BY_TIMESTAMP_DURATION) - 1000n; // Aged
+    const log2BlockTimestamp = currentTimestamp - 5000n; // Not aged
+    const highestAgedIndex = 3;
+    const highestFinalizedIndex = 5;
+    const log1Index = highestAgedIndex + 1; // At the beginning of the range
+    const log2Index = highestFinalizedIndex + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN; // At the window boundary
+    const log1Tag = await computeSiloedTagForIndex(log1Index);
+    const log2Tag = await computeSiloedTagForIndex(log2Index);
+    const log1BlockHeader = makeBlockHeader(0, { timestamp: log1BlockTimestamp });
+    const log2BlockHeader = makeBlockHeader(1, { timestamp: log2BlockTimestamp });
+
+    // Set existing highest aged index and highest finalized index
+    await taggingDataProvider.updateHighestAgedIndex(secret, highestAgedIndex);
+    await taggingDataProvider.updateHighestFinalizedIndex(secret, highestFinalizedIndex);
+
+    aztecNode.getL2Tips.mockResolvedValue({
+      finalized: { number: BlockNumber(finalizedBlockNumber) },
+    } as any);
+
+    aztecNode.getBlockHeader.mockResolvedValue(makeBlockHeader(0, { timestamp: currentTimestamp }));
+
+    // We record the number of queried tags to be able to verify that the window was moved forward correctly.
+    let numQueriedTags = 0;
+
+    aztecNode.getLogsByTags.mockImplementation((tags: Fr[]) => {
+      numQueriedTags += tags.length;
+      return Promise.all(
+        tags.map(async (t: Fr) => {
+          if (t.equals(log1Tag.value)) {
+            return [makeLog(await log1BlockHeader.hash(), finalizedBlockNumber, log1Tag.value)];
+          } else if (t.equals(log2Tag.value)) {
+            return [makeLog(await log2BlockHeader.hash(), finalizedBlockNumber, log2Tag.value)];
+          }
+          return [];
+        }),
+      );
+    });
+
+    aztecNode.getBlockHeaderByHash.mockImplementation(async (hash: Fr) => {
+      if (hash.equals(await log1BlockHeader.hash())) {
+        return log1BlockHeader;
+      } else if (hash.equals(await log2BlockHeader.hash())) {
+        return log2BlockHeader;
+      }
+      return undefined;
+    });
+
+    const logs = await loadPrivateLogsForSenderRecipientPair(
+      secret,
+      app,
+      aztecNode,
+      taggingDataProvider,
+      NON_INTERFERING_ANCHOR_BLOCK_NUMBER,
+    );
+
+    // Verify that both logs at the boundaries of the range were found and processed
+    expect(logs).toHaveLength(2);
+    expect(await taggingDataProvider.getHighestFinalizedIndex(secret)).toBe(log2Index);
+    expect(await taggingDataProvider.getHighestAgedIndex(secret)).toBe(log1Index);
+
+    // Verify that the window was moved forward correctly
+    // Total range queried: from (highestAgedIndex + 1) to (log2Index + WINDOW_LEN + 1) exclusive
+    const expectedNumQueriedTags = log2Index + UNFINALIZED_TAGGING_INDEXES_WINDOW_LEN - highestAgedIndex;
+    expect(numQueriedTags).toBe(expectedNumQueriedTags);
+  });
+});