refactor(pxe): migrate PrivateEventStore to BaseStagingStore

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

@@ -36,6 +36,8 @@ describe('validateAndStoreEvents', () => {
   beforeEach(async () => {
     const store = await openTmpStore('test');
     privateEventStore = new PrivateEventStore(store);
+    // Leave a change set open for the tests to operate under: every store operation requires one.
+    privateEventStore.beginChangeSet('test');
 
     contractAddress = await AztecAddress.random();
     recipient = await AztecAddress.random();
@@ -91,6 +93,7 @@ describe('validateAndStoreEvents', () => {
     await eventService.validateAndStoreEvents([request], recipient, map);
 
     await privateEventStore.commitChangeSet('test');
+    privateEventStore.beginChangeSet('test');
   }
 
   it('should throw when tx does not exist or has no effects', async () => {
@@ -113,12 +116,7 @@ describe('validateAndStoreEvents', () => {
 
     await runStoreEvent({ eventContent: otherContent, eventCommitment: otherCommitment });
 
-    const result = await privateEventStore.getPrivateEvents(eventSelector, {
-      contractAddress,
-      fromBlock: blockNumber,
-      toBlock: blockNumber + 1,
-      scopes: [recipient],
-    });
+    const result = await readEvents();
 
     expect(result.length).toEqual(0);
     expect(logger.warn).toHaveBeenCalledWith(expect.stringMatching(/commitment is not present in its tx/));
@@ -128,12 +126,7 @@ describe('validateAndStoreEvents', () => {
     // Commitment is legitimately present in the tx, but the provided content does not hash to it.
     await runStoreEvent({ eventContent: [Fr.random(), Fr.random()] });
 
-    const result = await privateEventStore.getPrivateEvents(eventSelector, {
-      contractAddress,
-      fromBlock: blockNumber,
-      toBlock: blockNumber + 1,
-      scopes: [recipient],
-    });
+    const result = await readEvents();
 
     expect(result.length).toEqual(0);
     expect(logger.warn).toHaveBeenCalledWith(expect.stringMatching(/content does not hash to the provided commitment/));
@@ -143,12 +136,7 @@ describe('validateAndStoreEvents', () => {
     await runStoreEvent();
 
     // I should be able to retrieve the private event I just saved using getPrivateEvents
-    const result = await privateEventStore.getPrivateEvents(eventSelector, {
-      contractAddress,
-      fromBlock: blockNumber,
-      toBlock: blockNumber + 1,
-      scopes: [recipient],
-    });
+    const result = await readEvents();
 
     expect(result.length).toEqual(1);
     expect(result[0].packedEvent).toEqual(eventContent);
@@ -157,4 +145,13 @@ describe('validateAndStoreEvents', () => {
   function defaultValidationTxDataMap() {
     return new Map([[txEffect.txHash.toString(), validationTxData]]);
   }
+
+  /** Reads the fixture's events through the change set the tests operate under. */
+  function readEvents() {
+    return privateEventStore.getPrivateEvents(
+      eventSelector,
+      { contractAddress, fromBlock: blockNumber, toBlock: blockNumber + 1, scopes: [recipient] },
+      'test',
+    );
+  }
 });

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

@@ -23,6 +23,7 @@ describe('runOperation', () => {
     discarded = [];
     const recordingStore: StagedStore = {
       storeName: 'recording_store',
+      beginChangeSet: () => {},
       commitChangeSet: id => {
         committed.push(id);
         return Promise.resolve();
@@ -141,6 +142,7 @@ describe('runOperation', () => {
       stagedStores: [
         {
           storeName: 'undiscardable_store',
+          beginChangeSet: () => {},
           commitChangeSet: () => Promise.resolve(),
           discardChangeSet: () => {
             throw new Error('cannot discard');

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

@@ -29,6 +29,7 @@ describe('OperationQueue', () => {
     discarded = [];
     const recordingStore: StagedStore = {
       storeName: 'recording_store',
+      beginChangeSet: () => {},
       commitChangeSet: id => {
         committed.push(id);
         return Promise.resolve();

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

@@ -425,6 +425,8 @@ describe('PXE', () => {
       scope = await AztecAddress.random();
 
       privateEventStore = new PrivateEventStore(kvStore);
+      // Leave a change set open for the tests to operate under: every store operation requires one.
+      privateEventStore.beginChangeSet('test');
     });
 
     let eventCounter = 0;

yarn-project/pxe/src/pxe.ts

@@ -1366,15 +1366,8 @@ export class PXE {
    *    Defaults to the latest known block to PXE + 1.
    * @returns - The packed events with block and tx metadata.
    */
-  public async getPrivateEvents(
-    eventSelector: EventSelector,
-    filter: PrivateEventFilter,
-  ): Promise<PackedPrivateEvent[]> {
-    let anchorBlockNumber: BlockNumber;
-
-    await this.operationQueue.runSynced(async ({ changeSetId, anchorBlockHeader }) => {
-      anchorBlockNumber = anchorBlockHeader.getBlockNumber();
-
+  public getPrivateEvents(eventSelector: EventSelector, filter: PrivateEventFilter): Promise<PackedPrivateEvent[]> {
+    return this.operationQueue.runSynced(async ({ changeSetId, anchorBlockHeader }) => {
       const contractFunctionSimulator = this.#getSimulatorForTx();
 
       await this.contractSyncService.ensureContractSynced({
@@ -1394,16 +1387,15 @@ export class PXE {
         scopes: filter.scopes,
         triggeredBy: undefined,
       });
-    });
 
-    // anchorBlockNumber is set during the operation and fixed to whatever it is after a block sync
-    const sanitizedFilter = new PrivateEventFilterValidator(anchorBlockNumber!).validate(filter);
+      const sanitizedFilter = new PrivateEventFilterValidator(anchorBlockHeader.getBlockNumber()).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);
+      return this.privateEventStore.getPrivateEvents(eventSelector, sanitizedFilter, changeSetId);
+    });
   }
 
   /**

yarn-project/pxe/src/storage/backwards_compatibility_tests/schema_tests.ts

@@ -436,6 +436,7 @@ export const SCHEMA_TESTS: readonly SchemaTest[] = [
       const privateEventStore = new PrivateEventStore(kvStore);
 
       const changeSetId = 'fixture-change-set';
+      privateEventStore.beginChangeSet(changeSetId);
 
       // Two (contract, selector) pairs and two block numbers so each multimap exhibits both a multi-value row
       // (contractA/selectorA → {e1, e2} and blockN1 → {e1, e2}) and a contrasting single-value row.