feat!: Expose clientID on ReadTransaction (#629)

Exposes clientID on ReadTransaction (and subclasses, e.g. WriteTransaction, IndexTransaction, etc).

This change was previously reverted due to it causing perf test regressions (976bf98, 4da3f52).  The
underlying cause of those regressions was addressed in commit cc2bbeb.

BREAKING CHANGE: Due to the new property on ReadTransaction the Replicache class no longer 
duck-types as a ReadTransaction (prior to 52bab5d Replicache explicitly implemented ReadTransaction).
Any place where Replicache is assigned or passed as a ReadTransaction will need to be updated, most 
likely to use Replicache.query.

Closes #541

Author: grgbkr

src/replicache.test.ts

@@ -151,7 +151,7 @@ testWithBothStores('get, has, scan on empty db', async () => {
     expect(scanItems).to.have.length(0);
   }
 
-  await t(rep);
+  await rep.query(t);
 });
 
 testWithBothStores('put, get, has, del inside tx', async () => {
@@ -2986,3 +2986,24 @@ testWithBothStores('subscribe perf test regression', async () => {
     expect(data[i]).to.equal(i < mut ? i ** 2 + rand : i);
   }
 });
+
+testWithBothStores('client ID is set correctly on transactions', async () => {
+  const rep = await replicacheForTesting(
+    'client-id-is-set-correctly-on-transactions',
+    {
+      mutators: {
+        async expectClientID(tx, expectedClientID: string) {
+          expect(tx.clientID).to.equal(expectedClientID);
+        },
+      },
+    },
+  );
+
+  const repClientID = await rep.clientID;
+
+  await rep.query(tx => {
+    expect(tx.clientID).to.equal(repClientID);
+  });
+
+  await rep.mutate.expectClientID(repClientID);
+});

src/replicache.ts

@@ -148,7 +148,6 @@ export class Replicache<MD extends MutatorDefs = {}> {
   private _online = true;
   private readonly _logger: Logger;
   private readonly _ready: Promise<void>;
-  private readonly _resolveReady: () => void;
   private readonly _clientIDPromise: Promise<string>;
   private _root: Promise<string | undefined> = Promise.resolve(undefined);
   private readonly _mutatorRegistry = new Map<
@@ -295,9 +294,8 @@ export class Replicache<MD extends MutatorDefs = {}> {
 
     // Use a promise-resolve pair so that we have a promise to use even before
     // we call the Open RPC.
-    const {promise, resolve} = resolver<void>();
-    this._ready = promise;
-    this._resolveReady = resolve;
+    const readyResolver = resolver<void>();
+    this._ready = readyResolver.promise;
 
     const {minDelayMs = MIN_DELAY_MS, maxDelayMs = MAX_DELAY_MS} =
       requestOptions;
@@ -331,23 +329,31 @@ export class Replicache<MD extends MutatorDefs = {}> {
 
     this._lc = new LogContext(logLevel).addContext('db', name);
 
-    this._clientIDPromise = this._open();
+    const clientIDResolver = resolver<string>();
+    this._clientIDPromise = clientIDResolver.promise;
+
+    void this._open(clientIDResolver.resolve, readyResolver.resolve);
   }
 
-  private async _open(): Promise<string> {
+  private async _open(
+    resolveClientID: (clientID: string) => void,
+    resolveReady: () => void,
+  ): Promise<void> {
     // If we are currently closing a Replicache instance with the same name,
     // wait for it to finish closing.
     await closingInstances.get(this.name);
 
     await Promise.all([initHasher(), migrate(this._kvStore, this._lc)]);
 
-    const [clientID] = await Promise.all([
-      sync.initClientID(this._kvStore),
+    await Promise.all([
+      sync.initClientID(this._kvStore).then(clientID => {
+        resolveClientID(clientID);
+      }),
       db.maybeInitDefaultDB(this._dagStore),
     ]);
 
     // Now we have both a clientID and DB!
-    this._resolveReady();
+    resolveReady();
 
     if (hasBroadcastChannel) {
       this._broadcastChannel = new BroadcastChannel(storageKeyName(this.name));
@@ -361,7 +367,6 @@ export class Replicache<MD extends MutatorDefs = {}> {
 
     this.pull();
     this._push();
-    return clientID;
   }
 
   /**
@@ -585,13 +590,15 @@ export class Replicache<MD extends MutatorDefs = {}> {
     f: (tx: IndexTransactionImpl) => Promise<void>,
   ): Promise<void> {
     await this._ready;
+    // clientID must be awaited ouside dag transaction to avoid a premature
+    // auto-commit of the idb transaction.
+    const clientID = await this._clientIDPromise;
     await this._dagStore.withWrite(async dagWrite => {
       const dbWrite = await db.Write.newIndexChange(
         db.whenceHead(db.DEFAULT_HEAD_NAME),
         dagWrite,
       );
-
-      const tx = new IndexTransactionImpl(dbWrite, this._lc);
+      const tx = new IndexTransactionImpl(clientID, dbWrite, this._lc);
       await f(tx);
       await tx.commit();
     });
@@ -994,9 +1001,12 @@ export class Replicache<MD extends MutatorDefs = {}> {
    */
   async query<R>(body: (tx: ReadTransaction) => Promise<R> | R): Promise<R> {
     await this._ready;
+    // clientID must be awaited ouside dag transaction to avoid a premature
+    // auto-commit of the idb transaction.
+    const clientID = await this._clientIDPromise;
     return await this._dagStore.withRead(async dagRead => {
       const dbRead = await db.readFromDefaultHead(dagRead);
-      const tx = new ReadTransactionImpl(dbRead, this._lc);
+      const tx = new ReadTransactionImpl(clientID, dbRead, this._lc);
       return await body(tx);
     });
   }
@@ -1079,6 +1089,9 @@ export class Replicache<MD extends MutatorDefs = {}> {
     }
 
     await this._ready;
+    // clientID must be awaited ouside dag transaction to avoid a premature
+    // auto-commit of the idb transaction.
+    const clientID = await this._clientIDPromise;
     return await this._dagStore.withWrite(async dagWrite => {
       let whence: db.Whence | undefined;
       let originalHash: string | null = null;
@@ -1098,7 +1111,7 @@ export class Replicache<MD extends MutatorDefs = {}> {
         dagWrite,
       );
 
-      const tx = new WriteTransactionImpl(dbWrite, this._lc);
+      const tx = new WriteTransactionImpl(clientID, dbWrite, this._lc);
       const result: R = await mutatorImpl(tx, args);
 
       const [ref, changedKeys] = await tx.commit(!isReplay);

src/transactions.ts

@@ -16,6 +16,8 @@ import type {LogContext} from './logger';
  * database.
  */
 export interface ReadTransaction {
+  readonly clientID: string;
+
   /**
    * Get a single value from the database. If the `key` is not present this
    * returns `undefined`.
@@ -65,14 +67,17 @@ export class ReadTransactionImpl<
   Value extends ReadonlyJSONValue = ReadonlyJSONValue,
 > implements ReadTransaction
 {
+  readonly clientID: string;
   protected readonly _dbtx: db.Read;
   protected readonly _lc: LogContext;
 
   constructor(
+    clientID: string,
     dbRead: db.Read,
     lc: LogContext,
     rpcName = 'openReadTransaction',
   ) {
+    this.clientID = clientID;
     this._dbtx = dbRead;
     this._lc = lc
       .addContext('rpc', rpcName)
@@ -122,6 +127,10 @@ export class SubscriptionTransactionWrapper implements ReadTransaction {
     this._tx = tx;
   }
 
+  get clientID(): string {
+    return this._tx.clientID;
+  }
+
   isEmpty(): Promise<boolean> {
     // Any change to the subscription requires rerunning it.
     this._scans.push({});
@@ -194,11 +203,12 @@ export class WriteTransactionImpl
   protected declare readonly _dbtx: db.Write;
 
   constructor(
+    clientID: string,
     dbWrite: db.Write,
     lc: LogContext,
     rpcName = 'openWriteTransaction',
   ) {
-    super(dbWrite, lc, rpcName);
+    super(clientID, dbWrite, lc, rpcName);
   }
 
   async put(key: string, value: JSONValue): Promise<void> {
@@ -276,8 +286,8 @@ export class IndexTransactionImpl
   extends WriteTransactionImpl
   implements IndexTransaction
 {
-  constructor(dbWrite: db.Write, lc: LogContext) {
-    super(dbWrite, lc, 'openIndexTransaction');
+  constructor(clientID: string, dbWrite: db.Write, lc: LogContext) {
+    super(clientID, dbWrite, lc, 'openIndexTransaction');
   }
 
   async createIndex(options: CreateIndexDefinition): Promise<void> {