Syncing blocks (#36)

package.json

@@ -6,13 +6,15 @@
     "lint": "eslint \"**/*.ts*\""
   },
   "dependencies": {
-    "@buf/penumbra-zone_penumbra.bufbuild_es": "1.3.1-20230911215829-072df077bbc1.1",
-    "@buf/penumbra-zone_penumbra.connectrpc_es": "0.13.2-20230911215829-072df077bbc1.1",
+    "@buf/penumbra-zone_penumbra.bufbuild_es": "1.3.1-20230915230831-726bde1a9447.1",
+    "@buf/penumbra-zone_penumbra.connectrpc_es": "0.13.2-20230915230831-726bde1a9447.1",
     "@connectrpc/connect": "^0.13.2",
     "@connectrpc/connect-web": "^0.13.2",
+    "buffer": "^6.0.3",
     "eslint": "^8.49.0",
     "eslint-config-custom": "workspace:*",
     "penumbra-types": "workspace:*",
+    "penumbra-wasm-ts": "workspace:*",
     "tsconfig": "workspace:*",
     "typescript": "^5.2.2"
   }

src/block-processor.ts

@@ -1,6 +1,12 @@
 import { ObliviousQuerier } from './oblivious';
 import { SpecificQuerier } from './specific';
-import { IndexedDbInterface, SpendableNoteRecord, ViewServerInterface } from 'penumbra-types';
+import {
+  base64ToUint8Array,
+  IndexedDbInterface,
+  NewNoteRecord,
+  ViewServerInterface,
+} from 'penumbra-types';
+import { CompactBlock } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/chain/v1alpha1/chain_pb';
 
 interface QueryClientProps {
   oblQuerier: ObliviousQuerier;
@@ -23,23 +29,91 @@ export class BlockProcessor {
   }
 
   async syncBlocks() {
-    const lastBlock = await this.indexedDb.getLastBlockSynced();
-    const startHeight = lastBlock ? lastBlock + 1n : 0n;
-
-    for await (const res of this.oblQuerier.compactBlockRange(startHeight, true)) {
-      const scanResult = await this.viewServer.scanBlock(res.compactBlock!);
-      await this.handleNewNotes(scanResult.new_notes);
-      await this.indexedDb.saveLastBlockSynced(res.compactBlock!.height);
+    try {
+      await this.syncAndStore();
+    } catch (e) {
+      await this.viewServer.resetTreeToStored();
+      console.error(e);
+      throw e;
     }
   }
 
-  async handleNewNotes(notes: SpendableNoteRecord[]) {
+  async storeNewNotes(notes: NewNoteRecord[]) {
     for (const n of notes) {
       await this.indexedDb.saveSpendableNote(n);
-      const metadata = await this.specQuerier.denomMetadata(n.note.value.assetId);
-      if (metadata) {
-        await this.indexedDb.saveAssetsMetadata(metadata);
+
+      // We need to query separately to convert assetId's into readable denom strings. Persisting those to storage.
+      const assetId = base64ToUint8Array(n.note.value.assetId.inner);
+      const storedDenomData = await this.indexedDb.getAssetsMetadata(assetId);
+      if (!storedDenomData) {
+        const metadata = await this.specQuerier.denomMetadata(n.note.value.assetId.inner);
+        if (metadata) {
+          await this.indexedDb.saveAssetsMetadata(metadata);
+        } // TODO: add base64_to_bech32 to store
+      }
+    }
+  }
+
+  // TODO: Next PR
+  // Each nullifier has a corresponding note stored. This marks them as spent at a specific block height.
+  // async markNotesSpent(nullifiers: Nullifier[], blockHeight: bigint) {
+  //   for (const nullifier of nullifiers) {
+  //     const stringId = uint8ArrayToBase64(nullifier.inner);
+  //     const matchingNote = await this.indexedDb.getNoteByNullifier(stringId);
+  //     if (!matchingNote)
+  //       throw new Error(`No corresponding note for nullifier: ${nullifier.inner.toString()}`);
+  //
+  //     matchingNote.heightSpent = blockHeight;
+  //     await this.indexedDb.saveSpendableNote(matchingNote);
+  //   }
+  // }
+
+  async saveSyncProgress(height: bigint) {
+    const updates = await this.viewServer.updatesSinceCheckpoint();
+    await this.indexedDb.updateStateCommitmentTree(updates, height);
+  }
+
+  // TODO: Put behind debugger flag
+  // private async assertRootValid(blockHeight: bigint): Promise<void> {
+  //   const sourceOfTruth = await this.specQuerier.keyValue(`sct/anchor/${blockHeight}`);
+  //   const inMemoryRoot = this.viewServer.getNctRoot();
+  //
+  //   if (decodeNctRoot(sourceOfTruth) !== inMemoryRoot) {
+  //     throw new Error(
+  //       `Block height: ${blockHeight}. Wasm root does not match remote source of truth. Programmer error.`,
+  //     );
+  //   }
+  // }
+
+  private async syncAndStore() {
+    const lastBlockSynced = await this.indexedDb.getLastBlockSynced();
+    const startHeight = lastBlockSynced ? lastBlockSynced + 1n : 0n;
+    const { lastBlockHeight } = await this.oblQuerier.info();
+
+    // Continuously runs as new blocks are committed
+    for await (const res of this.oblQuerier.compactBlockRange(startHeight, true)) {
+      if (!res.compactBlock) throw new Error('No compant block in response');
+
+      // Scanning has a side effect of updating viewServer's internal tree.
+      const scanResult = this.viewServer.scanBlock(res.compactBlock);
+
+      // TODO: We should not store new blocks as we find them, but only when sync progress is saved: https://github.com/penumbra-zone/web/issues/34
+      //       However, the current wasm crate discards the new notes on every block scan.
+      await this.storeNewNotes(scanResult.new_notes);
+      // await this.markNotesSpent(res.compactBlock.nullifiers, res.compactBlock.height);
+
+      // await this.assertRootValid(res.compactBlock.height); // TODO: Put behind debug flag
+
+      if (shouldStoreProgress(res.compactBlock, lastBlockHeight)) {
+        await this.saveSyncProgress(res.compactBlock.height);
       }
     }
   }
 }
+
+// Writing to disc is expensive, so storing progress occurs:
+// - if syncing is up-to-date, on every block
+// - if not, every 1000th block
+const shouldStoreProgress = (block: CompactBlock, upToDateBlock: bigint): boolean => {
+  return block.height >= upToDateBlock || block.height % 1000n === 0n;
+};

src/oblivious.ts

@@ -4,6 +4,7 @@ import { createGrpcWebTransport } from '@connectrpc/connect-web';
 import {
   ChainParametersRequest,
   CompactBlockRangeRequest,
+  InfoRequest,
 } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/client/v1alpha1/client_pb';
 
 interface ObliviousQuerierProps {
@@ -26,11 +27,16 @@ export class ObliviousQuerier {
   }
 
   async chainParameters() {
-    const req = new ChainParametersRequest({});
+    const req = new ChainParametersRequest();
     const res = await this.client.chainParameters(req);
     return res.chainParameters!;
   }
 
+  async info() {
+    const req = new InfoRequest();
+    return this.client.info(req);
+  }
+
   private createClient(grpcEndpoint: string): PromiseClient<typeof ObliviousQueryService> {
     const transport = createGrpcWebTransport({
       baseUrl: grpcEndpoint,

src/specific.ts

@@ -1,8 +1,12 @@
 import { SpecificQueryService } from '@buf/penumbra-zone_penumbra.connectrpc_es/penumbra/client/v1alpha1/client_connect';
 import { createPromiseClient, PromiseClient } from '@connectrpc/connect';
 import { createGrpcWebTransport } from '@connectrpc/connect-web';
-import { DenomMetadataByIdRequest } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/client/v1alpha1/client_pb';
-import { AssetId } from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/core/crypto/v1alpha1/crypto_pb';
+import {
+  DenomMetadataByIdRequest,
+  KeyValueRequest,
+} from '@buf/penumbra-zone_penumbra.bufbuild_es/penumbra/client/v1alpha1/client_pb';
+import { base64ToUint8Array } from 'penumbra-types/src/utils';
+import { Base64Str } from 'penumbra-types';
 
 interface SpecificQuerierProps {
   grpcEndpoint: string;
@@ -15,14 +19,20 @@ export class SpecificQuerier {
     this.client = this.createClient(grpcEndpoint);
   }
 
-  async denomMetadata(assetId: AssetId) {
+  async denomMetadata(assetId: Base64Str) {
     const request = new DenomMetadataByIdRequest({
-      assetId: { inner: assetId.inner },
+      assetId: { inner: base64ToUint8Array(assetId) },
     });
     const res = await this.client.denomMetadataById(request);
     return res.denomMetadata;
   }
 
+  async keyValue(key: string): Promise<Uint8Array> {
+    const keyValueRequest = new KeyValueRequest({ key });
+    const keyValue = await this.client.keyValue(keyValueRequest);
+    return keyValue.value!.value;
+  }
+
   private createClient(grpcEndpoint: string): PromiseClient<typeof SpecificQueryService> {
     const transport = createGrpcWebTransport({
       baseUrl: grpcEndpoint,