refactor: more robust tagging index sync as recipient

Fixes #17775

In this PR I implement a new log sync algorithm we've come up with in Buenos Aires that should be fully robust against any kind of private log losses. The algorithm is explained in the body of `loadPrivateLogsForSenderRecipientPair` function. That function should be the entrypoint when reviewing this PR (if checking the utility functions first you would not have enough context).

Unfortunately this PR introduces a performance regression that is tracked by [this issue](https://linear.app/aztec-labs/issue/F-229/improve-log-sync-performance). I am fairly certain that the regression is caused by an unreasonable number of requests performed by the `loadLogsForRange` function - that will be tackled in a followup PR.

In this PR the algorithm is not yet integrated into the system. That is done in a [PR up the stack](#19125).

The directory structure is not yet final - I wanted to keep this PR contained in one place to not have conflicts with Martin's PR. So please ignore that for now (will move stuff around in a final pass).

My plan is as follows:
1. Merge this PR,
2. fix the regression by modifying the `getLogsByTag` API such that it gives me all the info I need (now it doesn't give me block timestamp),
3. once that PR is merged most likely wait for Martin's refactor to be merged and then rebase and polish my integrating new log sync PR,
4. move some files around to make it all cuter.

## Update on regression of time in e2e tests
Unfortunately realized that this has caused a regression and log sync is now significantly slower.

Did 2 runs of the `e2e_l1_with_wall_time` test and on `next` the results are 337 and 334 seconds but with the new code it's 661 and 658 seconds.

Will work on dropping that down before merging that "integrating new log sync" PR.

Author: benesjan

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);
+  });
+});