Merge pull request #544 from streamich/hamt-crdt

HAMT CRDT implementation

package.json

@@ -83,7 +83,7 @@
     "arg": "^5.0.2",
     "hyperdyperid": "^1.2.0",
     "multibase": "^4.0.6",
-    "thingies": "^1.17.0"
+    "thingies": "^1.18.0"
   },
   "devDependencies": {
     "@automerge/automerge": "2.1.7",

src/json-pack/cbor/CborDecoderBase.ts

@@ -13,15 +13,14 @@ export class CborDecoderBase<R extends IReader & IReaderResettable = IReader & I
 {
   public constructor(
     public reader: R = new Reader() as any,
-    protected readonly keyDecoder: CachedUtf8Decoder = sharedCachedUtf8Decoder,
+    public readonly keyDecoder: CachedUtf8Decoder = sharedCachedUtf8Decoder,
   ) {}
 
   public read(uint8: Uint8Array): PackValue {
     this.reader.reset(uint8);
     return this.val() as PackValue;
   }
 
-  /** @deprecated */
   public decode(uint8: Uint8Array): unknown {
     this.reader.reset(uint8);
     return this.val();

src/json-pack/cbor/__tests__/CborEncoderDag.spec.ts

@@ -2,6 +2,7 @@ import {Writer} from '../../../util/buffers/Writer';
 import {CborEncoderDag} from '../CborEncoderDag';
 import {CborDecoder} from '../CborDecoder';
 import {JsonPackExtension} from '../../JsonPackExtension';
+import {CborDecoderDag} from '../CborDecoderDag';
 
 const writer = new Writer(1);
 const encoder = new CborEncoderDag(writer);
@@ -50,37 +51,50 @@ describe('only extension = 42 is permitted', () => {
     expect(val).toStrictEqual({a: 'a', b: 'b'});
   });
 
-  test('can encode CID using inlined custom class', () => {
-    class CID {
-      constructor(public readonly value: string) {}
+  class CID {
+    constructor(public readonly value: string) {}
+  }
+  class NotCID {
+    constructor(public readonly value: string) {}
+  }
+
+  class IpfsCborEncoder extends CborEncoderDag {
+    public writeUnknown(val: unknown): void {
+      if (val instanceof CID) this.writeTag(42, val.value);
+      else throw new Error('Unknown value type');
     }
-    const encoder = new (class extends CborEncoderDag {
-      public writeUnknown(val: unknown): void {
-        if (val instanceof CID) encoder.writeTag(42, val.value);
-        else throw new Error('Unknown value type');
-      }
-    })();
+  }
+
+  class IpfsCborDecoder extends CborDecoderDag {
+    public readTagRaw(tag: number): CID | unknown {
+      if (tag === 42) return new CID(this.val() as any);
+      throw new Error('UNKNOWN_TAG');
+    }
+  }
+
+  test('can encode CID using inlined custom class', () => {
+    const encoder = new IpfsCborEncoder();
     const encoded = encoder.encode({a: 'a', b: new JsonPackExtension(42, 'b')});
     const val = decoder.read(encoded);
     expect(val).toStrictEqual({a: 'a', b: new JsonPackExtension(42, 'b')});
     const encoded2 = encoder.encode({a: 'a', b: new CID('b')});
-    const val2 = decoder.read(encoded2);
+    const val2 = decoder.decode(encoded2);
     expect(val).toStrictEqual({a: 'a', b: new JsonPackExtension(42, 'b')});
+    expect(val2).toStrictEqual({a: 'a', b: new JsonPackExtension(42, 'b')});
+  });
+
+  test('can encode CID inside a nested array', () => {
+    const encoder = new IpfsCborEncoder();
+    const decoder = new IpfsCborDecoder();
+    const cid = new CID('my-cid');
+    const data = [1, [2, [3, cid, 4], 5], 6];
+    const encoded = encoder.encode(data);
+    const decoded = decoder.decode(encoded);
+    expect(decoded).toStrictEqual(data);
   });
 
   test('can throw on unknown custom class', () => {
-    class CID {
-      constructor(public readonly value: string) {}
-    }
-    class NotCID {
-      constructor(public readonly value: string) {}
-    }
-    const encoder = new (class extends CborEncoderDag {
-      public writeUnknown(val: unknown): void {
-        if (val instanceof CID) encoder.writeTag(42, val.value);
-        else throw new Error('Unknown value type');
-      }
-    })();
+    const encoder = new IpfsCborEncoder();
     const encoded1 = encoder.encode({a: 'a', b: new CID('b')});
     expect(() => encoder.encode({a: 'a', b: new NotCID('b')})).toThrowError(new Error('Unknown value type'));
   });

src/json-pack/json/JsonDecoder.ts

@@ -203,6 +203,11 @@ export class JsonDecoder implements BinaryJsonDecoder {
     return this.readAny();
   }
 
+  public decode(uint8: Uint8Array): unknown {
+    this.reader.reset(uint8);
+    return this.readAny();
+  }
+
   public readAny(): PackValue {
     this.skipWhitespace();
     const reader = this.reader;

src/json-pack/types.ts

@@ -15,6 +15,7 @@ type PackArray = PackValue[] | readonly PackValue[];
 type PackObject = {[key: string]: PackValue} | Readonly<{[key: string]: PackValue}>;
 
 export interface BinaryJsonEncoder {
+  encode(value: unknown): Uint8Array;
   writer: IWriter & IWriterGrowable;
   writeAny(value: unknown): void;
   writeNull(): void;
@@ -52,6 +53,7 @@ export interface TlvBinaryJsonEncoder {
 }
 
 export interface BinaryJsonDecoder {
+  decode(uint8: Uint8Array): unknown;
   reader: IReader & IReaderResettable;
   read(uint8: Uint8Array): PackValue;
 }

src/json-pack/ubjson/UbjsonDecoder.ts

@@ -11,6 +11,11 @@ export class UbjsonDecoder implements BinaryJsonDecoder {
     return this.readAny();
   }
 
+  public decode(uint8: Uint8Array): unknown {
+    this.reader.reset(uint8);
+    return this.readAny();
+  }
+
   public readAny(): PackValue {
     const reader = this.reader;
     const octet = reader.u8();

src/util/buffers/b.ts

@@ -0,0 +1 @@
+export const b = (...args: number[]) => new Uint8Array(args);

src/util/buffers/cmpUint8Array2.ts

@@ -0,0 +1,10 @@
+export const cmpUint8Array2 = (a: Uint8Array, b: Uint8Array): number => {
+  const len1 = a.length;
+  const len2 = b.length;
+  const len = Math.min(len1, len2);
+  for (let i = 0; i < len; i++) {
+    const diffChar = a[i] - b[i];
+    if (diffChar !== 0) return diffChar;
+  }
+  return len1 - len2;
+};

src/util/buffers/toBuf.ts

@@ -0,0 +1,8 @@
+import {encode} from './utf8/encode';
+
+export const toBuf = (str: string): Uint8Array => {
+  const maxLength = str.length * 4;
+  const arr = new Uint8Array(maxLength);
+  const strBufferLength = encode(arr, str, 0, maxLength);
+  return arr.slice(0, strBufferLength);
+};

src/web3/adl/hamt-crdt/Hamt.ts

@@ -0,0 +1,106 @@
+import {Defer} from 'thingies/es2020/Defer';
+import {concurrency} from 'thingies/es2020/concurrencyDecorator';
+import {HamtFrame} from './HamtFrame';
+import * as hlc from '../../hlc';
+import {Cid} from '../../multiformats';
+import {sha256} from '../../crypto';
+import {toBuf} from '../../../util/buffers/toBuf';
+import type {CidCasStruct} from '../../store/cas/CidCasStruct';
+import type * as types from './types';
+
+export interface HamtDependencies {
+  cas: CidCasStruct;
+  hlcs: hlc.HlcFactory;
+}
+
+export class Hamt implements types.HamtApi {
+  protected _root: HamtFrame;
+  protected _dirty: boolean = false;
+  protected _loading: Promise<void> | null = null;
+
+  public cid: Cid | null = null;
+  public prev: Cid | null = null;
+  public seq: number = 0;
+  public ops: types.HamtOp[] = [];
+
+  constructor(protected readonly deps: HamtDependencies) {
+    this._root = new HamtFrame(deps.cas, null);
+  }
+
+  public hasChanges(): boolean {
+    return this._dirty;
+  }
+
+  // ------------------------------------------------------------------ HamtApi
+
+  public async load(cid: Cid): Promise<void> {
+    this.cid = cid;
+    const future = new Defer<void>();
+    this._loading = future.promise;
+    try {
+      const [prev, seq, ops, data] = (await this.deps.cas.get(cid)) as types.HamtRootFrameDto;
+      this.prev = prev;
+      this.seq = seq;
+      this._root.loadDto(data, null);
+      future.resolve();
+    } catch (err) {
+      future.reject(err);
+    } finally {
+      this._loading = null;
+    }
+    return future.promise;
+  }
+
+  @concurrency(1)
+  public async put(key: Uint8Array | string, val: unknown): Promise<boolean> {
+    if (this._loading) await this._loading;
+    const hashedKey = await this._key(key);
+    const id = this.deps.hlcs.inc();
+    const idDto = hlc.toDto(id);
+    const op: types.HamtOp = [hashedKey, val, idDto];
+    const success = await this._root.put(op);
+    if (success) this.ops.push(op);
+    return success;
+  }
+
+  public async get(key: Uint8Array | string): Promise<unknown | undefined> {
+    if (this._loading) await this._loading;
+    const hashedKey = await this._key(key);
+    return await this._root.get(hashedKey);
+  }
+
+  /** Convert any key to buffer and prefix with 4-byte hash. */
+  protected async _key(key: Uint8Array | string): Promise<Uint8Array> {
+    const keyBuf = typeof key === 'string' ? toBuf(key) : key;
+    const hash = await sha256(keyBuf);
+    const buf = new Uint8Array(4 + keyBuf.length);
+    buf.set(hash.subarray(0, 4), 0);
+    buf.set(keyBuf, 4);
+    return buf;
+  }
+
+  public async has(key: Uint8Array | string): Promise<boolean> {
+    if (this._loading) await this._loading;
+    return (await this.get(key)) !== undefined;
+  }
+
+  public async del(key: Uint8Array | string): Promise<boolean> {
+    if (this._loading) await this._loading;
+    return await this.put(key, undefined);
+  }
+
+  public async save(): Promise<[head: Cid, affected: Cid[]]> {
+    const [, affected] = await this._root.saveChildren();
+    const prev = this.cid;
+    const seq = this.seq + 1;
+    const frameDto = this._root.toDto();
+    const dto: types.HamtRootFrameDto = [prev, seq, this.ops, frameDto];
+    const cid = await this.deps.cas.put(dto);
+    this.cid = cid;
+    this.prev = prev;
+    this.seq = seq;
+    this.ops = [];
+    affected.push(cid);
+    return [cid, affected];
+  }
+}

src/web3/adl/hamt-crdt/HamtFactory.ts

@@ -0,0 +1,12 @@
+import {Hamt, type HamtDependencies} from './Hamt';
+
+export interface HamtFactoryDependencies extends HamtDependencies {}
+
+export class HamtFactory {
+  constructor(protected readonly deps: HamtFactoryDependencies) {}
+
+  public make(): Hamt {
+    const hamt = new Hamt(this.deps);
+    return hamt;
+  }
+}