wip: meta-corruption in attachable-subscription.test.ts

Author: mabels

core/base/crdt-clock.ts

@@ -64,7 +64,8 @@ export class CRDTClockImpl {
       prevHead,
       updates,
     })) {
-      return this.processUpdates(updatesAcc, all, prevHead);
+      await this.processUpdates(updatesAcc, all, prevHead);
+      return
     }
   }
 
@@ -216,7 +217,7 @@ async function validateBlocks(logger: Logger, newHead: ClockHead, blockstore?: B
   newHead.map(async (cid) => {
     const got = await blockstore.get(cid);
     if (!got) {
-      throw logger.Error().Str("cid", cid.toString()).Msg("int_applyHead missing block").AsError();
+      throw logger.Error().Str("cid", cid.toString()).Msg("validateBlocks missing block").AsError();
     }
   });
 }

core/base/crdt-helpers.ts

@@ -40,7 +40,7 @@ import {
   PARAM,
   NotFoundError,
 } from "@fireproof/core-types-base";
-import { Logger } from "@adviser/cement";
+import { exception2Result, Logger } from "@adviser/cement";
 import { Link, Version } from "multiformats";
 
 function toString<K extends IndexKeyType>(key: K, logger: Logger): string {
@@ -58,7 +58,11 @@ export function toPailFetcher(tblocks: BlockFetcher): PailBlockFetcher {
     get: async <T = unknown, C extends number = number, A extends number = number, V extends Version = 1>(
       link: Link<T, C, A, V>,
     ) => {
-      const block = await tblocks.get(link);
+      const rBlock = await exception2Result(() => tblocks.get(link));
+      if (rBlock.isErr()) {
+        return undefined
+      }
+      const block = rBlock.Ok()
       return block
         ? ({
             cid: block.cid,

core/base/database.ts

@@ -25,6 +25,7 @@ import {
   Attached,
   NotFoundError,
   QueryResult,
+  isNotFoundError,
 } from "@fireproof/core-types-base";
 import { ensureLogger, makeName } from "@fireproof/core-runtime";
 
@@ -78,7 +79,10 @@ export class DatabaseImpl implements Database {
       const { doc } = got;
       return { ...(doc as unknown as DocWithId<T>), _id: id };
     } catch (e) {
-      throw new NotFoundError(`Not found: ${id} - ${e instanceof Error ? e.message : String(e)}`);
+      if (isNotFoundError(e)) {
+        throw e
+      }
+      throw this.logger.Error().Err(e).Msg("unexpect error").AsError()
     }
   }
 
@@ -157,7 +161,6 @@ export class DatabaseImpl implements Database {
     if (typeof opts.limit === "number" && opts.limit >= 0) {
       rows = rows.slice(0, opts.limit);
     }
-
     return { rows, clock: head, name: this.name };
   }
 

core/blockstore/loader.ts

@@ -320,9 +320,7 @@ export class Loader implements Loadable {
         this.blockstoreParent?.crdtParent?.ledgerParent?.name,
       );
       const local = this.attachedStores.local();
-      // console.log("ready", this.id);
       this.metaStreamReader = local.active.meta.stream().getReader();
-      // console.log("attach-local", local.active.car.url().pathname);
       await this.waitFirstMeta(this.metaStreamReader, local, { meta: this.ebOpts.meta, origin: local.active.car.url() });
     });
   }
@@ -765,7 +763,6 @@ export class Loader implements Loadable {
   }
 
   async getBlock(cid: AnyLink): Promise<FPBlock | Falsy> {
-    await this.ready();
     const got = this.cidCache.get(cid.toString());
     return got.value;
   }

core/blockstore/register-store-protocol.ts

@@ -1,6 +1,6 @@
 import { BuildURI, ResolveOnce, runtimeFn, URI } from "@adviser/cement";
 import { SuperThis, PARAM } from "@fireproof/core-types-base";
-import { SerdeGateway, Gateway } from "@fireproof/core-types-blockstore";
+import { SerdeGateway, Gateway, FPEnvelope } from "@fireproof/core-types-blockstore";
 import { MemoryGateway } from "@fireproof/core-gateways-memory";
 import { FileGateway, FILESTORE_VERSION, sysFileSystemFactory } from "@fireproof/core-gateways-file";
 import { DefSerdeGateway, INDEXEDDB_VERSION } from "@fireproof/core-gateways-base";
@@ -165,14 +165,14 @@ if (runtimeFn().isBrowser) {
   });
 }
 
-const memory = new Map<string, Uint8Array>();
+const memory = new Map<string, FPEnvelope<unknown>>();
 registerStoreProtocol({
   protocol: "memory:",
   isDefault: false,
   defaultURI: () => {
     return BuildURI.from("memory://").pathname("ram").URI();
   },
-  gateway: async (sthis) => {
+  serdegateway: async (sthis) => {
     return new MemoryGateway(sthis, memory);
   },
 });

core/blockstore/store.ts

@@ -219,7 +219,6 @@ export class MetaStoreImpl extends BaseStoreImpl implements MetaStore {
           // console.log(
           //   "subscribe",
           //   this.myId,
-          //   id.str,
           //   this.url().pathname,
           //   dbMetas.map((i) => i.cars.map((i) => i.toString())).flat(2),
           // );

core/gateways/base/fp-envelope-serialize.ts

@@ -50,7 +50,7 @@ export async function dbMetaEvent2Serialized(
   );
 }
 
-function WALState2Serialized(sthis: SuperThis, wal: WALState): SerializedWAL {
+export function WALState2Serialized(sthis: SuperThis, wal: WALState): SerializedWAL {
   const serializedWAL: SerializedWAL = {
     fileOperations: wal.fileOperations.map((fop) => ({
       cid: fop.cid.toString(),

core/gateways/base/meta-key-hack.ts

@@ -218,10 +218,10 @@ export function addKeyToDbMetaEncoder(ctx: SerdeGatewayCtx, version: "v1" | "v2"
 }
 
 export class AddKeyToDbMetaGateway implements SerdeGateway {
-  private readonly sdGw: DefSerdeGateway;
+  private readonly sdGw: SerdeGateway;
   readonly version: "v1" | "v2";
-  constructor(gw: Gateway, version: "v1" | "v2") {
-    this.sdGw = new DefSerdeGateway(gw);
+  constructor(gw: SerdeGateway, version: "v1" | "v2") {
+    this.sdGw = gw;
     this.version = version;
   }