refactor: staged writes in private events

Fourth part of the series started with #19445.

This makes the PrivateEventStore work based on staged writes. With this, private events aren't written to persistent storage until PXE decides to commit the job.

Author: mverzilli

yarn-project/pxe/src/contract_function_simulator/oracle/utility_execution_oracle.ts

@@ -409,7 +409,7 @@ export class UtilityExecutionOracle implements IMiscOracle, IUtilityExecutionOra
       ),
     );
 
-    const eventService = new EventService(this.anchorBlockStore, this.aztecNode, this.privateEventStore);
+    const eventService = new EventService(this.anchorBlockStore, this.aztecNode, this.privateEventStore, this.jobId);
     const eventStorePromises = eventValidationRequests.map(request =>
       eventService.storeEvent(
         request.contractAddress,

yarn-project/pxe/src/events/event_service.test.ts

@@ -80,15 +80,15 @@ describe('storeEvent', () => {
 
     aztecNode.getTxEffect.mockImplementation(() => Promise.resolve(indexedTxEffect));
 
-    eventService = new EventService(anchorBlockStore, aztecNode, privateEventStore);
+    eventService = new EventService(anchorBlockStore, aztecNode, privateEventStore, 'test');
   });
 
-  function runStoreEvent(
+  async function runStoreEvent(
     overrides: {
       eventCommitment?: Fr;
     } = {},
   ) {
-    return eventService.storeEvent(
+    await eventService.storeEvent(
       contractAddress,
       eventSelector,
       randomness,
@@ -97,6 +97,8 @@ describe('storeEvent', () => {
       txEffect.txHash,
       recipient,
     );
+
+    await privateEventStore.commit('test');
   }
 
   it('should throw when tx does not exist or has no effects', async () => {

yarn-project/pxe/src/events/event_service.ts

@@ -13,6 +13,7 @@ export class EventService {
     private readonly anchorBlockStore: AnchorBlockStore,
     private readonly aztecNode: AztecNode,
     private readonly privateEventStore: PrivateEventStore,
+    private readonly jobId: string,
   ) {}
 
   public async storeEvent(
@@ -52,14 +53,21 @@ export class EventService {
       );
     }
 
-    return this.privateEventStore.storePrivateEventLog(selector, randomness, content, siloedEventCommitment, {
-      contractAddress,
-      scope,
-      txHash,
-      l2BlockNumber: txEffect.l2BlockNumber,
-      l2BlockHash: txEffect.l2BlockHash,
-      txIndexInBlock: txEffect.txIndexInBlock,
-      eventIndexInTx,
-    });
+    return this.privateEventStore.storePrivateEventLog(
+      selector,
+      randomness,
+      content,
+      siloedEventCommitment,
+      {
+        contractAddress,
+        scope,
+        txHash,
+        l2BlockNumber: txEffect.l2BlockNumber,
+        l2BlockHash: txEffect.l2BlockHash,
+        txIndexInBlock: txEffect.txIndexInBlock,
+        eventIndexInTx,
+      },
+      this.jobId,
+    );
   }
 }

yarn-project/pxe/src/events/private_event_filter_validator.test.ts

@@ -25,17 +25,17 @@ describe('PrivateEventFilterValidator', () => {
     anchorBlockStore = mock<AnchorBlockStore>();
 
     anchorBlockStore.getBlockHeader.mockResolvedValue(lastKnownBlock);
-    validator = new PrivateEventFilterValidator(anchorBlockStore);
+    validator = new PrivateEventFilterValidator(lastKnownBlockNumber);
   });
 
-  it('rejects empty scope', async () => {
-    await expect(validator.validate({ contractAddress, fromBlock: lastKnownBlockNumber, scopes: [] })).rejects.toThrow(
+  it('rejects empty scope', () => {
+    expect(() => validator.validate({ contractAddress, fromBlock: lastKnownBlockNumber, scopes: [] })).toThrow(
       /At least one scope is required to get private events/,
     );
   });
 
-  it('defaults to whole range', async () => {
-    const dataProviderFilter = await validator.validate({ contractAddress, scopes: [scope] });
+  it('defaults to whole range', () => {
+    const dataProviderFilter = validator.validate({ contractAddress, scopes: [scope] });
     expect(dataProviderFilter).toEqual({
       contractAddress,
       scopes: [scope],
@@ -45,8 +45,8 @@ describe('PrivateEventFilterValidator', () => {
     });
   });
 
-  it('toBlock defaults to lastKnownBlock + 1', async () => {
-    const dataProviderFilter = await validator.validate({
+  it('toBlock defaults to lastKnownBlock + 1', () => {
+    const dataProviderFilter = validator.validate({
       contractAddress,
       scopes: [scope],
       fromBlock: INITIAL_L2_BLOCK_NUM,
@@ -59,8 +59,8 @@ describe('PrivateEventFilterValidator', () => {
     });
   });
 
-  it('toBlock without fromBlock defaults to [INITIAL_L2_BLOCK_NUM, toBlock)', async () => {
-    const dataProviderFilter = await validator.validate({
+  it('toBlock without fromBlock defaults to [INITIAL_L2_BLOCK_NUM, toBlock)', () => {
+    const dataProviderFilter = validator.validate({
       contractAddress,
       scopes: [scope],
       toBlock: BlockNumber(lastKnownBlockNumber + 1),
@@ -73,28 +73,28 @@ describe('PrivateEventFilterValidator', () => {
     });
   });
 
-  it('rejects fromBlock >= toBlock', async () => {
-    await expect(
+  it('rejects fromBlock >= toBlock', () => {
+    expect(() =>
       validator.validate({
         contractAddress,
         scopes: [scope],
         fromBlock: lastKnownBlockNumber,
         toBlock: lastKnownBlockNumber,
       }),
-    ).rejects.toThrow(/toBlock must be strictly greater than fromBlock/);
+    ).toThrow(/toBlock must be strictly greater than fromBlock/);
 
-    await expect(
+    expect(() =>
       validator.validate({
         contractAddress,
         scopes: [scope],
         fromBlock: lastKnownBlockNumber,
         toBlock: BlockNumber(lastKnownBlockNumber - 1),
       }),
-    ).rejects.toThrow(/toBlock must be strictly greater than fromBlock/);
+    ).toThrow(/toBlock must be strictly greater than fromBlock/);
   });
 
-  it('preserves txHash', async () => {
-    let dataProviderFilter = await validator.validate({
+  it('preserves txHash', () => {
+    let dataProviderFilter = validator.validate({
       contractAddress,
       scopes: [scope],
       fromBlock: INITIAL_L2_BLOCK_NUM,
@@ -108,7 +108,7 @@ describe('PrivateEventFilterValidator', () => {
     });
 
     const txHash = TxHash.random();
-    dataProviderFilter = await validator.validate({
+    dataProviderFilter = validator.validate({
       contractAddress,
       scopes: [scope],
       fromBlock: INITIAL_L2_BLOCK_NUM,

yarn-project/pxe/src/events/private_event_filter_validator.ts

@@ -2,23 +2,21 @@ import type { PrivateEventFilter } from '@aztec/aztec.js/wallet';
 import { INITIAL_L2_BLOCK_NUM } from '@aztec/constants';
 import { BlockNumber } from '@aztec/foundation/branded-types';
 
-import { AnchorBlockStore } from '../storage/anchor_block_store/anchor_block_store.js';
 import type { PrivateEventStoreFilter } from '../storage/private_event_store/private_event_store.js';
 
 export class PrivateEventFilterValidator {
-  constructor(private anchorBlockStore: AnchorBlockStore) {}
+  constructor(private lastBlock: BlockNumber) {}
 
-  async validate(filter: PrivateEventFilter): Promise<PrivateEventStoreFilter> {
+  validate(filter: PrivateEventFilter): PrivateEventStoreFilter {
     let { fromBlock, toBlock } = filter;
 
     // Block range filters in Aztec Node are defined as closed-open intervals [fromBlock, toBlock), so
     // we respect that convention here for consistency.
     // We then default to [INITIAL_L2_BLOCK_NUM, latestKnownBlock + 1), ie: by default return events from
     // the first block to the latest known block.
     if (!fromBlock || !toBlock) {
-      const lastKnownBlock = (await this.anchorBlockStore.getBlockHeader()).getBlockNumber();
       fromBlock = fromBlock ?? BlockNumber(INITIAL_L2_BLOCK_NUM);
-      toBlock = toBlock ?? BlockNumber(lastKnownBlock + 1);
+      toBlock = toBlock ?? BlockNumber(this.lastBlock + 1);
     }
 
     if (filter.scopes.length === 0) {

yarn-project/pxe/src/pxe.test.ts

@@ -238,6 +238,7 @@ describe('PXE', () => {
           txIndexInBlock: 0,
           eventIndexInTx: eventCounter++,
         },
+        'test',
       );
 
       return event;
@@ -247,6 +248,7 @@ describe('PXE', () => {
       // Store a couple of events to exercise `getPrivateEvents`
       const event1 = await storeEvent();
       const event2 = await storeEvent();
+      await privateEventStore.commit('test');
 
       const events = await pxe.getPrivateEvents(eventSelector, {
         contractAddress,
@@ -287,6 +289,8 @@ describe('PXE', () => {
           storeEvent(lastKnownBlockNumber + 1),
           storeEvent(lastKnownBlockNumber + 1),
         ]);
+
+        await privateEventStore.commit('test');
       });
 
       it('filters by txHash', async () => {

yarn-project/pxe/src/pxe.ts

@@ -1,4 +1,5 @@
 import type { PrivateEventFilter } from '@aztec/aztec.js/wallet';
+import { BlockNumber } from '@aztec/foundation/branded-types';
 import { Fr } from '@aztec/foundation/curves/bn254';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import { SerialQueue } from '@aztec/foundation/queue';
@@ -157,7 +158,7 @@ export class PXE {
     );
 
     const jobCoordinator = new JobCoordinator(store);
-    jobCoordinator.registerStores([capsuleStore, senderTaggingStore, recipientTaggingStore]);
+    jobCoordinator.registerStores([capsuleStore, senderTaggingStore, recipientTaggingStore, privateEventStore]);
 
     const debugUtils = new PXEDebugUtils(contractStore, noteStore);
 
@@ -1009,9 +1010,17 @@ export class PXE {
    *    Defaults to the latest known block to PXE + 1.
    * @returns - The packed events with block and tx metadata.
    */
-  public getPrivateEvents(eventSelector: EventSelector, filter: PrivateEventFilter): Promise<PackedPrivateEvent[]> {
-    return this.#putInJobQueue(async jobId => {
+  public async getPrivateEvents(
+    eventSelector: EventSelector,
+    filter: PrivateEventFilter,
+  ): Promise<PackedPrivateEvent[]> {
+    let anchorBlockNumber: BlockNumber;
+
+    await this.#putInJobQueue(async jobId => {
       await this.blockStateSynchronizer.sync();
+
+      anchorBlockNumber = (await this.anchorBlockStore.getBlockHeader()).getBlockNumber();
+
       const contractFunctionSimulator = this.#getSimulatorForTx();
 
       await this.contractStore.syncPrivateState(
@@ -1020,15 +1029,16 @@ export class PXE {
         async privateSyncCall =>
           await this.#simulateUtility(contractFunctionSimulator, privateSyncCall, [], undefined, jobId),
       );
+    });
 
-      const sanitizedFilter = await new PrivateEventFilterValidator(this.anchorBlockStore).validate(filter);
+    const sanitizedFilter = new PrivateEventFilterValidator(anchorBlockNumber!).validate(filter);
 
-      this.log.debug(
-        `Getting private events for ${sanitizedFilter.contractAddress.toString()} from ${sanitizedFilter.fromBlock} to ${sanitizedFilter.toBlock}`,
-      );
+    this.log.debug(
+      `Getting private events for ${sanitizedFilter.contractAddress.toString()} from ${sanitizedFilter.fromBlock} to ${sanitizedFilter.toBlock}`,
+    );
 
-      return this.privateEventStore.getPrivateEvents(eventSelector, sanitizedFilter);
-    });
+    // sanitizedFilter is assigned during the synchro
+    return this.privateEventStore.getPrivateEvents(eventSelector, sanitizedFilter);
   }
 
   /**