delay errors in observe callbacks to throw after cleanup is done

src/types/AbstractType.js

@@ -30,7 +30,7 @@ import * as encoding from 'lib0/encoding.js' // eslint-disable-line
  * @param {EventType} event
  */
 export const callTypeObservers = (type, transaction, event) => {
-  callEventHandlerListeners(type._eH, event, transaction)
+  const changedType = type
   const changedParentTypes = transaction.changedParentTypes
   while (true) {
     // @ts-ignore
@@ -40,6 +40,7 @@ export const callTypeObservers = (type, transaction, event) => {
     }
     type = type._item.parent
   }
+  callEventHandlerListeners(changedType._eH, event, transaction)
 }
 
 /**

src/utils/Transaction.js

@@ -17,6 +17,7 @@ import * as encoding from 'lib0/encoding.js'
 import * as map from 'lib0/map.js'
 import * as math from 'lib0/math.js'
 import * as set from 'lib0/set.js'
+import { callAll } from 'lib0/function.js'
 
 /**
  * A transaction is created for every change on the Yjs model. It is possible
@@ -145,46 +146,40 @@ export const addChangedTypeToTransaction = (transaction, type, parentSub) => {
 }
 
 /**
- * Implements the functionality of `y.transact(()=>{..})`
+ * @param {Array<Transaction>} transactionCleanups
- *
+ * @param {number} i
- * @param {Doc} doc
- * @param {function(Transaction):void} f
- * @param {any} [origin=true]
- *
- * @private
- * @function
  */
-export const transact = (doc, f, origin = null, local = true) => {
+const cleanupTransactions = (transactionCleanups, i) => {
-  const transactionCleanups = doc._transactionCleanups
+  if (i < transactionCleanups.length) {
-  let initialCall = false
-  if (doc._transaction === null) {
-    initialCall = true
-    doc._transaction = new Transaction(doc, origin, local)
-    transactionCleanups.push(doc._transaction)
-    doc.emit('beforeTransaction', [doc._transaction, doc])
-  }
-  try {
-    f(doc._transaction)
-  } finally {
-    if (initialCall && transactionCleanups[0] === doc._transaction) {
-      // The first transaction ended, now process observer calls.
-      // Observer call may create new transactions for which we need to call the observers and do cleanup.
-      // We don't want to nest these calls, so we execute these calls one after another
-      for (let i = 0; i < transactionCleanups.length; i++) {
     const transaction = transactionCleanups[i]
-        const store = transaction.doc.store
+    const doc = transaction.doc
+    const store = doc.store
     const ds = transaction.deleteSet
+    try {
       sortAndMergeDeleteSet(ds)
       transaction.afterState = getStateVector(transaction.doc.store)
       doc._transaction = null
       doc.emit('beforeObserverCalls', [transaction, doc])
-        // emit change events on changed types
+      /**
-        transaction.changed.forEach((subs, itemtype) => {
+       * An array of event callbacks.
+       *
+       * Each callback is called even if the other ones throw errors.
+       *
+       * @type {Array<function():void>}
+       */
+      const fs = []
+      // observe events on changed types
+      transaction.changed.forEach((subs, itemtype) =>
+        fs.push(() => {
           if (itemtype._item === null || !itemtype._item.deleted) {
             itemtype._callObserver(transaction, subs)
           }
         })
-        transaction.changedParentTypes.forEach((events, type) => {
+      )
+      fs.push(() => {
+        // deep observe events
+        transaction.changedParentTypes.forEach((events, type) =>
+          fs.push(() => {
             // We need to think about the possibility that the user transforms the
             // Y.Doc in the event.
             if (type._item === null || !type._item.deleted) {
@@ -201,7 +196,11 @@ export const transact = (doc, f, origin = null, local = true) => {
               callEventHandlerListeners(type._dEH, events, transaction)
             }
           })
-        doc.emit('afterTransaction', [transaction, doc])
+        )
+        fs.push(() => doc.emit('afterTransaction', [transaction, doc]))
+      })
+      callAll(fs, [])
+    } finally {
       /**
        * @param {Array<AbstractStruct>} structs
        * @param {number} pos
@@ -295,8 +294,47 @@ export const transact = (doc, f, origin = null, local = true) => {
           doc.emit('update', [encoding.toUint8Array(updateMessage), transaction.origin, doc])
         }
       }
-      }
+      if (transactionCleanups.length <= i + 1) {
         doc._transactionCleanups = []
+      } else {
+        cleanupTransactions(transactionCleanups, i + 1)
+      }
+    }
+  }
+}
+
+/**
+ * Implements the functionality of `y.transact(()=>{..})`
+ *
+ * @param {Doc} doc
+ * @param {function(Transaction):void} f
+ * @param {any} [origin=true]
+ *
+ * @private
+ * @function
+ */
+export const transact = (doc, f, origin = null, local = true) => {
+  const transactionCleanups = doc._transactionCleanups
+  let initialCall = false
+  if (doc._transaction === null) {
+    initialCall = true
+    doc._transaction = new Transaction(doc, origin, local)
+    transactionCleanups.push(doc._transaction)
+    doc.emit('beforeTransaction', [doc._transaction, doc])
+  }
+  try {
+    f(doc._transaction)
+  } finally {
+    if (initialCall && transactionCleanups[0] === doc._transaction) {
+      // The first transaction ended, now process observer calls.
+      // Observer call may create new transactions for which we need to call the observers and do cleanup.
+      // We don't want to nest these calls, so we execute these calls one after
+      // another.
+      // Also we need to ensure that all cleanups are called, even if the
+      // observes throw errors.
+      // This file is full of hacky try {} finally {} blocks to ensure that an
+      // event can throw errors and also that the cleanup is called.
+      cleanupTransactions(transactionCleanups, 0)
     }
   }
 }

tests/y-map.tests.js

@@ -340,6 +340,56 @@ export const testChangeEvent = tc => {
   compare(users)
 }
 
+/**
+ * @param {t.TestCase} tc
+ */
+export const testYmapEventExceptionsShouldCompleteTransaction = tc => {
+  const doc = new Y.Doc()
+  const map = doc.getMap('map')
+
+  let updateCalled = false
+  let throwingObserverCalled = false
+  let throwingDeepObserverCalled = false
+  doc.on('update', () => {
+    updateCalled = true
+  })
+
+  const throwingObserver = () => {
+    throwingObserverCalled = true
+    throw new Error('Failure')
+  }
+
+  const throwingDeepObserver = () => {
+    throwingDeepObserverCalled = true
+    throw new Error('Failure')
+  }
+
+  map.observe(throwingObserver)
+  map.observeDeep(throwingDeepObserver)
+
+  t.fails(() => {
+    map.set('y', '2')
+  })
+
+  t.assert(updateCalled)
+  t.assert(throwingObserverCalled)
+  t.assert(throwingDeepObserverCalled)
+
+  // check if it works again
+  updateCalled = false
+  throwingObserverCalled = false
+  throwingDeepObserverCalled = false
+  t.fails(() => {
+    map.set('z', '3')
+  })
+
+  t.assert(updateCalled)
+  t.assert(throwingObserverCalled)
+  t.assert(throwingDeepObserverCalled)
+
+  t.assert(map.get('z') === '3')
+}
+
 /**
  * @param {t.TestCase} tc
  */

tsconfig.json

@@ -38,7 +38,10 @@
     "moduleResolution": "node",            /* Specify module resolution strategy: 'node' (Node.js) or 'classic' (TypeScript pre-1.6). */
     "baseUrl": "./",                       /* Base directory to resolve non-absolute module names. */
     "paths": {
-      "yjs": ["./src/index.js"]
+      "yjs": ["./src/index.js"],
+      "lib0/*": ["node_modules/lib0/*"],
+      "lib0/set.js": ["node_modules/lib0/set.js"],
+      "lib0/function.js": ["node_modules/lib0/function.js"]
     },                           /* A series of entries which re-map imports to lookup locations relative to the 'baseUrl'. */
     // "rootDirs": [],                        /* List of root folders whose combined content represents the structure of the project at runtime. */
     // "typeRoots": [],                       /* List of folders to include type definitions from. */