fix(StateManager): race condition between OBSERVE_RESPONSE and OBSERV…

…E_NOTIFICATION

src/common/BaseSharedStateCollection.js

@@ -1,6 +1,8 @@
 /**
  * @private
  */
+let id = 0;
+
 class BaseSharedStateCollection {
   constructor(stateManager, schemaName) {
     this._stateManager = stateManager;
@@ -10,6 +12,8 @@ class BaseSharedStateCollection {
     this._onAttachCallbacks = new Set();
     this._onDetachCallbacks = new Set();
     this._unobserve = null;
+    // testing
+    this._id = id++;
   }
 
   async _init() {
@@ -36,24 +40,32 @@ class BaseSharedStateCollection {
   }
 
   /**
-   * Size of the collection
+   * Size of the collection, alias `size`
    * @type {number}
    */
   get length() {
     return this._states.length;
   }
 
+  /**
+   * Size of the collection, , alias `length`
+   * @type {number}
+   */
+  get size() {
+    return this._states.length;
+  }
+
   /**
    * Detach from the collection, i.e. detach all underlying shared states.
    * @type {number}
    */
   async detach() {
     this._unobserve();
+    this._onUpdateCallbacks.clear();
 
-    const promises = Array.from(this._states).map(state => state.detach());
+    const promises = this._states.map(state => state.detach());
     await Promise.all(promises);
 
-    this._onUpdateCallbacks.clear();
     this._onDetachCallbacks.clear();
   }
 

src/common/BaseStateManager.js

@@ -93,6 +93,10 @@ class BaseStateManager {
       const [callback, filterSchemaName] = this._observeRequestCallbacks.get(reqId);
       this._observeRequestCallbacks.delete(reqId);
 
+      // now that the OBSERVE_REPOSNSE callback is executed, store it in
+      // OBSERVE_NOTIFICATION listeners
+      this._observeListeners.set(callback, filterSchemaName);
+
       const promises = list.map(([schemaName, stateId, nodeId]) => {
         if (filterSchemaName === '*' || filterSchemaName === schemaName) {
           return callback(schemaName, stateId, nodeId);
@@ -231,8 +235,22 @@ class BaseStateManager {
       // store the callback for execution on the response. the returned Promise
       // is fullfiled once callback has been executed with each existing states
       this._observeRequestCallbacks.set(reqId, [callback, filterSchemaName]);
-      // store the callback for execution on subsequent notifications
-      this._observeListeners.set(callback, filterSchemaName);
+
+      // NOTE: do not store in `_observeListeners` yet as it can produce races, e.g.:
+      // cf. test `observe should properly behave in race condition`
+      // ```
+      // await client.stateManager.observe(async (schemaName, stateId, nodeId) => {});
+      // // client now receives OBSERVE_NOTIFICATIONS
+      // await otherClient.stateManager.create('a');
+      // // second observer added in between
+      // client.stateManager.observe(async (schemaName, stateId, nodeId) => {});
+      // ````
+      // OBSERVE_NOTIFICATION is received before the OBSERVE_RESPONSE, then the
+      // second observer is called twice:
+      // - OBSERVE_RESPONSE 1 []
+      // - OBSERVE_NOTIFICATION [ 'a', 1, 0 ]
+      // - OBSERVE_NOTIFICATION [ 'a', 1, 0 ] // this should not be executed
+      // - OBSERVE_RESPONSE 1 [ [ 'a', 1, 0 ] ]
 
       this.client.transport.emit(OBSERVE_REQUEST, reqId);
     });

src/common/SharedStatePrivate.js

@@ -172,6 +172,7 @@ class SharedStatePrivate {
     } else {
       // detach only if not creator
       client.transport.addListener(`${DETACH_REQUEST}-${this.id}-${remoteId}`, (reqId) => {
+        console.log('SharedStatePrivate: receive detach request', reqId);
         this._detachClient(remoteId, client);
         client.transport.emit(`${DETACH_RESPONSE}-${this.id}-${remoteId}`, reqId);
       });

tests/states/SharedState.spec.js

@@ -496,4 +496,22 @@ describe('# ShareState', () => {
       }
     });
   });
+
+  describe(`## Race conditions`, () => {
+    it.skip(`[FIXME #73] should not stuck the program`, async () => {
+      const aCreated = await server.stateManager.create('a');
+      const aAttached = await client.stateManager.attach('a');
+
+      // DELETE_REQUEST is received first on the SharedStatePrivate which deletes all its listeners
+      // The DETACH_REQUEST is sent but have not response, then it never resolves so the program is stuck...
+
+      // Possible fix:
+      // keep track of all requests and reject them if a DELETE_NOTIFICATION or
+      // a DETACH_NOTIFICATION is received
+      // is received
+
+      await aCreated.delete();
+      await aAttached.detach(); // program is stuck here...
+    });
+  });
 });