Don't wait for transaction to finish before applying synced transactions (#37)

Author: KyleAMathews

packages/optimistic/src/collection.ts

@@ -7,10 +7,10 @@ import type {
   CollectionConfig,
   InsertConfig,
   OperationConfig,
+  OptimisticChangeMessage,
   PendingMutation,
   StandardSchema,
   Transaction,
-  TransactionState,
 } from "./types"
 
 // Store collections in memory using Tanstack store
@@ -26,7 +26,7 @@ const loadingCollections = new Map<
 
 interface PendingSyncedTransaction<T extends object = Record<string, unknown>> {
   committed: boolean
-  operations: Array<ChangeMessage<T>>
+  operations: Array<OptimisticChangeMessage<T>>
 }
 
 /**
@@ -152,7 +152,7 @@ export class Collection<T extends object = Record<string, unknown>> {
   public transactionManager!: ReturnType<typeof getTransactionManager<T>>
   private transactionStore: TransactionStore
 
-  public optimisticOperations: Derived<Array<ChangeMessage<T>>>
+  public optimisticOperations: Derived<Array<OptimisticChangeMessage<T>>>
   public derivedState: Derived<Map<string, T>>
   public derivedArray: Derived<Array<T>>
 
@@ -200,17 +200,16 @@ export class Collection<T extends object = Record<string, unknown>> {
     this.optimisticOperations = new Derived({
       fn: ({ currDepVals: [transactions] }) => {
         const result = Array.from(transactions.values())
-          .filter(
-            (transaction) =>
-              transaction.state !== `completed` &&
-              transaction.state !== `failed`
-          )
-          .map((transaction) =>
-            transaction.mutations.map((mutation) => {
-              const message: ChangeMessage<T> = {
+          .map((transaction) => {
+            const isActive = ![`completed`, `failed`].includes(
+              transaction.state
+            )
+            return transaction.mutations.map((mutation) => {
+              const message: OptimisticChangeMessage<T> = {
                 type: mutation.type,
                 key: mutation.key,
                 value: mutation.modified as T,
+                isActive,
               }
               if (
                 mutation.metadata !== undefined &&
@@ -220,7 +219,7 @@ export class Collection<T extends object = Record<string, unknown>> {
               }
               return message
             })
-          )
+          })
           .flat()
 
         return result
@@ -233,32 +232,27 @@ export class Collection<T extends object = Record<string, unknown>> {
     this.derivedState = new Derived({
       fn: ({ currDepVals: [syncedData, operations] }) => {
         const combined = new Map<string, T>(syncedData)
+        const optimisticKeys = new Set<string>()
+
         // Apply the optimistic operations on top of the synced state.
         for (const operation of operations) {
-          let existingValue
-          switch (operation.type) {
-            case `insert`:
-              combined.set(operation.key, operation.value)
-              break
-            case `update`:
-              existingValue = syncedData.get(operation.key)
-              combined.set(operation.key, {
-                ...(existingValue || {}),
-                ...operation.value,
-              })
-              break
-            case `delete`:
-              combined.delete(operation.key)
-              break
+          optimisticKeys.add(operation.key)
+          if (operation.isActive) {
+            switch (operation.type) {
+              case `insert`:
+                combined.set(operation.key, operation.value)
+                break
+              case `update`:
+                combined.set(operation.key, operation.value)
+                break
+              case `delete`:
+                combined.delete(operation.key)
+                break
+            }
           }
         }
 
         // Update object => key mappings
-        const optimisticKeys = new Set<string>()
-        for (const operation of operations) {
-          optimisticKeys.add(operation.key)
-        }
-
         optimisticKeys.forEach((key) => {
           if (combined.has(key)) {
             this.objectKeyMap.set(combined.get(key)!, key)
@@ -338,86 +332,64 @@ export class Collection<T extends object = Record<string, unknown>> {
    * This method processes operations from pending transactions and applies them to the synced data
    */
   commitPendingTransactions = () => {
-    // Check if there's any transactions that aren't finished.
-    // If not, proceed.
-    // If so, subscribe to transactions and keep checking if can proceed.
-    //
-    // The plan is to have a finer-grained locking but just blocking applying
-    // synced data until a persisting transaction is finished seems fine.
-    // We also don't yet have support for transactions that don't immediately
-    // persist so right now, blocking sync only delays their application for a
-    // few hundred milliseconds. So not the worse thing in th world.
-    // But something to fix in the future.
-    // Create a Set with only the terminal states
-    const terminalStates = new Set<TransactionState>([`completed`, `failed`])
-
-    // Function to check if a state is NOT a terminal state
-    function isNotTerminalState({ state }: Transaction): boolean {
-      return !terminalStates.has(state)
-    }
-    if (
-      this.transactions.size === 0 ||
-      !Array.from(this.transactions.values()).some(isNotTerminalState)
-    ) {
-      const keys = new Set<string>()
-      batch(() => {
-        for (const transaction of this.pendingSyncedTransactions) {
-          for (const operation of transaction.operations) {
-            keys.add(operation.key)
-            this.syncedMetadata.setState((prevData) => {
-              switch (operation.type) {
-                case `insert`:
-                  prevData.set(operation.key, operation.metadata)
-                  break
-                case `update`:
-                  prevData.set(operation.key, {
-                    ...prevData.get(operation.key)!,
-                    ...operation.metadata,
-                  })
-                  break
-                case `delete`:
-                  prevData.delete(operation.key)
-                  break
-              }
-              return prevData
-            })
-            this.syncedData.setState((prevData) => {
-              switch (operation.type) {
-                case `insert`:
-                  prevData.set(operation.key, operation.value)
-                  break
-                case `update`:
-                  prevData.set(operation.key, {
-                    ...prevData.get(operation.key)!,
-                    ...operation.value,
-                  })
-                  break
-                case `delete`:
-                  prevData.delete(operation.key)
-                  break
-              }
-              return prevData
-            })
-          }
+    const keys = new Set<string>()
+    batch(() => {
+      for (const transaction of this.pendingSyncedTransactions) {
+        for (const operation of transaction.operations) {
+          keys.add(operation.key)
+          this.syncedMetadata.setState((prevData) => {
+            switch (operation.type) {
+              case `insert`:
+                prevData.set(operation.key, operation.metadata)
+                break
+              case `update`:
+                prevData.set(operation.key, {
+                  ...prevData.get(operation.key)!,
+                  ...operation.metadata,
+                })
+                break
+              case `delete`:
+                prevData.delete(operation.key)
+                break
+            }
+            return prevData
+          })
+          this.syncedData.setState((prevData) => {
+            switch (operation.type) {
+              case `insert`:
+                prevData.set(operation.key, operation.value)
+                break
+              case `update`:
+                prevData.set(operation.key, {
+                  ...prevData.get(operation.key)!,
+                  ...operation.value,
+                })
+                break
+              case `delete`:
+                prevData.delete(operation.key)
+                break
+            }
+            return prevData
+          })
         }
-      })
+      }
+    })
 
-      keys.forEach((key) => {
-        const curValue = this.state.get(key)
-        if (curValue) {
-          this.objectKeyMap.set(curValue, key)
-        }
-      })
+    keys.forEach((key) => {
+      const curValue = this.state.get(key)
+      if (curValue) {
+        this.objectKeyMap.set(curValue, key)
+      }
+    })
 
-      this.pendingSyncedTransactions = []
+    this.pendingSyncedTransactions = []
 
-      // Call any registered one-time commit listeners
-      if (!this.hasReceivedFirstCommit) {
-        this.hasReceivedFirstCommit = true
-        const callbacks = [...this.onFirstCommitCallbacks]
-        this.onFirstCommitCallbacks = []
-        callbacks.forEach((callback) => callback())
-      }
+    // Call any registered one-time commit listeners
+    if (!this.hasReceivedFirstCommit) {
+      this.hasReceivedFirstCommit = true
+      const callbacks = [...this.onFirstCommitCallbacks]
+      this.onFirstCommitCallbacks = []
+      callbacks.forEach((callback) => callback())
     }
   }
 
@@ -746,7 +718,9 @@ export class Collection<T extends object = Record<string, unknown>> {
       if (typeof item === `object` && (item as unknown) !== null) {
         const objectKey = this.objectKeyMap.get(item)
         if (objectKey === undefined) {
-          throw new Error(`Object not found in collection`)
+          throw new Error(
+            `Object not found in collection: ${JSON.stringify(item)}`
+          )
         }
         key = objectKey
       } else if (typeof item === `string`) {

packages/optimistic/src/types.ts

@@ -84,6 +84,13 @@ export interface ChangeMessage<T extends object = Record<string, unknown>> {
   metadata?: Record<string, unknown>
 }
 
+export interface OptimisticChangeMessage<
+  T extends object = Record<string, unknown>,
+> extends ChangeMessage<T> {
+  // Is this change message part of an active transaction. Only applies to optimistic changes.
+  isActive?: boolean
+}
+
 export interface MutationFn<T extends object = Record<string, unknown>> {
   persist: (params: {
     transaction: Transaction

packages/optimistic/tests/collection.test.ts

@@ -3,7 +3,11 @@ import "fake-indexeddb/auto"
 import mitt from "mitt"
 import { z } from "zod"
 import { Collection, SchemaValidationError } from "../src/collection"
-import type { ChangeMessage, PendingMutation } from "../src/types"
+import type {
+  ChangeMessage,
+  OptimisticChangeMessage,
+  PendingMutation,
+} from "../src/types"
 
 describe(`Collection`, () => {
   it(`should throw if there's no sync config`, () => {
@@ -185,10 +189,11 @@ describe(`Collection`, () => {
     })
 
     // Check the optimistic operation is there
-    const insertOperation: ChangeMessage = {
+    const insertOperation: OptimisticChangeMessage = {
       key: insertedKey,
       value: { value: `bar` },
       type: `insert`,
+      isActive: true,
     }
     expect(collection.optimisticOperations.state[0]).toEqual(insertOperation)
 
@@ -222,7 +227,9 @@ describe(`Collection`, () => {
       // @ts-expect-error possibly undefined is ok in test
       Array.from(collection.transactions.values())[0].state
     ).toMatchInlineSnapshot(`"completed"`)
-    expect(collection.optimisticOperations.state).toEqual([])
+    expect(
+      collection.optimisticOperations.state.filter((o) => o.isActive)
+    ).toEqual([])
     expect(collection.state).toEqual(new Map([[insertedKey, { value: `bar` }]]))
 
     // Test insert with provided key
@@ -307,7 +314,7 @@ describe(`Collection`, () => {
     expect(collection.state.has(keys[3])).toBe(false)
   })
 
-  it(`synced updates shouldn't be applied while there's an ongoing transaction`, async () => {
+  it(`synced updates should be applied while there's an ongoing transaction`, async () => {
     const emitter = mitt()
 
     // new collection w/ mock sync/mutation
@@ -336,8 +343,15 @@ describe(`Collection`, () => {
           // we're still in the middle of persisting a transaction.
           emitter.emit(`update`, [
             { key: `the-key`, type: `insert`, changes: { bar: `value` } },
+            // This update is ignored because the optimistic update overrides it.
+            { key: `foo`, type: `update`, changes: { bar: `value2` } },
           ])
-          expect(collection.state).toEqual(new Map([[`foo`, { value: `bar` }]]))
+          expect(collection.state).toEqual(
+            new Map([
+              [`foo`, { value: `bar` }],
+              [`the-key`, { bar: `value` }],
+            ])
+          )
           // Remove it so we don't have to assert against it below
           emitter.emit(`update`, [{ key: `the-key`, type: `delete` }])
           return Promise.resolve()
@@ -369,10 +383,11 @@ describe(`Collection`, () => {
     })
 
     // Check the optimistic operation is there
-    const insertOperation: ChangeMessage = {
+    const insertOperation: OptimisticChangeMessage = {
       key: `foo`,
       value: { value: `bar` },
       type: `insert`,
+      isActive: true,
     }
     expect(collection.optimisticOperations.state[0]).toEqual(insertOperation)