fix: pxe synch

yarn-project/aztec.js/src/wallet/base_wallet.ts

@@ -42,7 +42,7 @@ export abstract class BaseWallet implements Wallet {
   registerAccount(privKey: GrumpkinPrivateKey, partialAddress: PartialAddress): Promise<CompleteAddress> {
     return this.pxe.registerAccount(privKey, partialAddress);
   }
-  registerRecipient(account: CompleteAddress): Promise<void> {
+  registerRecipient(account: CompleteAddress): Promise<boolean> {
     return this.pxe.registerRecipient(account);
   }
   getRegisteredAccounts(): Promise<CompleteAddress[]> {

yarn-project/pxe/src/database/memory_db.ts

@@ -138,6 +138,7 @@ export class MemoryDB extends MemoryContractDatabase implements PxeDatabase {
 
   public setSynchronizedBlock(blockNumber: number, blockHeader: BlockHeader): Promise<void> {
     this.globalVariablesHash = blockHeader.globalVariablesHash;
+    this.blockNumber = blockNumber;
     this.setTreeRoots({
       [MerkleTreeId.NOTE_HASH_TREE]: blockHeader.noteHashTreeRoot,
       [MerkleTreeId.NULLIFIER_TREE]: blockHeader.nullifierTreeRoot,

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

@@ -167,13 +167,15 @@ export class PXEService implements PXE {
     return Promise.resolve(account);
   }
 
-  public async registerRecipient(recipient: CompleteAddress): Promise<void> {
+  public async registerRecipient(recipient: CompleteAddress): Promise<boolean> {
     const wasAdded = await this.db.addCompleteAddress(recipient);
     if (wasAdded) {
       this.log.info(`Added recipient:\n ${recipient.toReadableString()}`);
     } else {
       this.log.info(`Recipient:\n "${recipient.toReadableString()}"\n already registered.`);
     }
+
+    return wasAdded;
   }
 
   public async getRecipients(): Promise<CompleteAddress[]> {

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

@@ -52,25 +52,27 @@ export const pxeTestSuite = (testName: string, pxeSetup: () => Promise<PXE>) =>
       const keyPair = ConstantKeyPair.random(new Grumpkin());
       const completeAddress = CompleteAddress.fromPrivateKeyAndPartialAddress(keyPair.getPrivateKey(), Fr.random());
 
-      await pxe.registerAccount(keyPair.getPrivateKey(), completeAddress.partialAddress);
-      await pxe.registerAccount(keyPair.getPrivateKey(), completeAddress.partialAddress);
+      await expect(pxe.registerAccount(keyPair.getPrivateKey(), completeAddress.partialAddress)).resolves.toEqual(
+        completeAddress,
+      );
+      await expect(pxe.registerAccount(keyPair.getPrivateKey(), completeAddress.partialAddress)).resolves.toEqual(
+        completeAddress,
+      );
     });
 
     it('cannot register a recipient with the same aztec address but different pub key or partial address', async () => {
       const recipient1 = CompleteAddress.random();
       const recipient2 = new CompleteAddress(recipient1.address, Point.random(), Fr.random());
 
       await pxe.registerRecipient(recipient1);
-      await expect(() => pxe.registerRecipient(recipient2)).rejects.toThrow(
-        `Complete address with aztec address ${recipient1.address}`,
-      );
+      await expect(pxe.registerRecipient(recipient2)).resolves.toEqual(false);
     });
 
     it('does not throw when registering the same recipient twice (just ignores the second attempt)', async () => {
       const completeAddress = CompleteAddress.random();
 
-      await pxe.registerRecipient(completeAddress);
-      await pxe.registerRecipient(completeAddress);
+      await expect(pxe.registerRecipient(completeAddress)).resolves.toEqual(true);
+      await expect(pxe.registerRecipient(completeAddress)).resolves.toEqual(false);
     });
 
     it('successfully adds a contract', async () => {

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

@@ -2,7 +2,7 @@ import { BlockHeader, CompleteAddress, EthAddress, Fr, GrumpkinScalar } from '@a
 import { Grumpkin } from '@aztec/circuits.js/barretenberg';
 import { TestKeyStore } from '@aztec/key-store';
 import { AztecLmdbStore } from '@aztec/kv-store';
-import { AztecNode, INITIAL_L2_BLOCK_NUM, L2Block, MerkleTreeId } from '@aztec/types';
+import { AztecNode, L2Block, MerkleTreeId } from '@aztec/types';
 
 import { MockProxy, mock } from 'jest-mock-extended';
 import omit from 'lodash.omit';
@@ -47,7 +47,7 @@ describe('Synchronizer', () => {
     aztecNode.getBlocks.mockResolvedValue([L2Block.fromFields(omit(block, 'newEncryptedLogs', 'newUnencryptedLogs'))]);
     aztecNode.getLogs.mockResolvedValueOnce([block.newEncryptedLogs!]).mockResolvedValue([block.newUnencryptedLogs!]);
 
-    await synchronizer.work(INITIAL_L2_BLOCK_NUM - 1);
+    await synchronizer.work();
 
     const roots = database.getTreeRoots();
     expect(roots[MerkleTreeId.CONTRACT_TREE]).toEqual(block.endContractTreeSnapshot.root);
@@ -69,7 +69,7 @@ describe('Synchronizer', () => {
     ]);
     aztecNode.getLogs.mockResolvedValue([block1.newEncryptedLogs!]).mockResolvedValue([block1.newUnencryptedLogs!]);
 
-    await synchronizer.work(INITIAL_L2_BLOCK_NUM - 1);
+    await synchronizer.work();
     const roots1 = database.getTreeRoots();
     expect(roots1[MerkleTreeId.CONTRACT_TREE]).toEqual(roots[MerkleTreeId.CONTRACT_TREE]);
     expect(roots1[MerkleTreeId.CONTRACT_TREE]).not.toEqual(block1.endContractTreeSnapshot.root);
@@ -80,7 +80,7 @@ describe('Synchronizer', () => {
       L2Block.fromFields(omit(block5, 'newEncryptedLogs', 'newUnencryptedLogs')),
     ]);
 
-    await synchronizer.work(INITIAL_L2_BLOCK_NUM - 1);
+    await synchronizer.work();
     const roots5 = database.getTreeRoots();
     expect(roots5[MerkleTreeId.CONTRACT_TREE]).not.toEqual(roots[MerkleTreeId.CONTRACT_TREE]);
     expect(roots5[MerkleTreeId.CONTRACT_TREE]).toEqual(block5.endContractTreeSnapshot.root);
@@ -97,7 +97,7 @@ describe('Synchronizer', () => {
       .mockResolvedValueOnce([block.newEncryptedLogs!]); // called by synchronizer.workNoteProcessorCatchUp
 
     // Sync the synchronizer so that note processor has something to catch up to
-    await synchronizer.work(INITIAL_L2_BLOCK_NUM - 1);
+    await synchronizer.work();
 
     // Used in synchronizer.isAccountStateSynchronized
     aztecNode.getBlockNumber.mockResolvedValueOnce(1);
@@ -113,22 +113,22 @@ describe('Synchronizer', () => {
     // Add the account which will add the note processor to the synchronizer
     synchronizer.addAccount(completeAddress.publicKey, keyStore);
 
-    await synchronizer.workNoteProcessorCatchUp(synchronizer.getSyncStatus().blocks);
+    await synchronizer.workNoteProcessorCatchUp();
 
     expect(await synchronizer.isAccountStateSynchronized(completeAddress.address)).toBe(true);
   });
 });
 
 class TestSynchronizer extends Synchronizer {
-  public work(from: number) {
-    return super.work(from);
+  public work() {
+    return super.work();
   }
 
   public initialSync(): Promise<void> {
     return super.initialSync();
   }
 
-  public workNoteProcessorCatchUp(to: number): Promise<void> {
-    return super.workNoteProcessorCatchUp(to);
+  public workNoteProcessorCatchUp(): Promise<void> {
+    return super.workNoteProcessorCatchUp();
   }
 }

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

@@ -45,14 +45,12 @@ export class Synchronizer {
 
     const run = async () => {
       while (this.running) {
-        const synchedToBlock = this.db.getBlockNumber();
-
         if (this.noteProcessorsToCatchUp.length > 0) {
           // There is a note processor that needs to catch up. We hijack the main loop to catch up the note processor.
-          await this.workNoteProcessorCatchUp(synchedToBlock, limit, retryInterval);
+          await this.workNoteProcessorCatchUp(limit, retryInterval);
         } else {
           // No note processor needs to catch up. We continue with the normal flow.
-          await this.work(synchedToBlock + 1, limit, retryInterval);
+          await this.work(limit, retryInterval);
         }
       }
     };
@@ -66,7 +64,8 @@ export class Synchronizer {
     await this.db.setSynchronizedBlock(blockNumber, blockHeader);
   }
 
-  protected async work(from: number, limit = 1, retryInterval = 1000): Promise<void> {
+  protected async work(limit = 1, retryInterval = 1000): Promise<void> {
+    const from = this.db.getBlockNumber() + 1;
     try {
       let encryptedLogs = await this.node.getLogs(from, limit, LogType.ENCRYPTED);
       if (!encryptedLogs.length) {
@@ -103,8 +102,8 @@ export class Synchronizer {
         block.attachLogs(unencryptedLogs[i], LogType.UNENCRYPTED);
       });
 
-      // Wrap blocks in block contexts.
-      const blockContexts = blocks.map(block => new L2BlockContext(block));
+      // Wrap blocks in block contexts & only keep those that match our query
+      const blockContexts = blocks.filter(block => block.number >= from).map(block => new L2BlockContext(block));
 
       // Update latest tree roots from the most recent block
       const latestBlock = blockContexts[blockContexts.length - 1];
@@ -121,8 +120,9 @@ export class Synchronizer {
     }
   }
 
-  protected async workNoteProcessorCatchUp(toBlockNumber: number, limit = 1, retryInterval = 1000): Promise<void> {
+  protected async workNoteProcessorCatchUp(limit = 1, retryInterval = 1000): Promise<void> {
     const noteProcessor = this.noteProcessorsToCatchUp[0];
+    const toBlockNumber = this.db.getBlockNumber();
     if (noteProcessor.status.syncedToBlock >= toBlockNumber) {
       // Note processor already synched, nothing to do
       this.noteProcessorsToCatchUp.shift();

yarn-project/types/src/interfaces/pxe.ts

@@ -72,7 +72,7 @@ export interface PXE {
    * the recipient's notes. We can send notes to this account because we can encrypt them with the recipient's
    * public key.
    */
-  registerRecipient(recipient: CompleteAddress): Promise<void>;
+  registerRecipient(recipient: CompleteAddress): Promise<boolean>;
 
   /**
    * Retrieves the user accounts registered on this PXE Service.