Revert "Revert "*make syncing async""

This reverts commit d37d37048c.

index.js

@@ -50,7 +50,7 @@ class Channel {
     for (const m of messages) this.addMessage(m);
   }
 
-  open(handshake) {
+  async open(handshake) {
     const id =
       this._mux._free.length > 0
         ? this._mux._free.pop()
@@ -81,7 +81,7 @@ class Channel {
     if (this._handshake) this._handshake.encode(state, handshake);
 
     this._mux._write0(state.buffer);
-    this._mux.syncState?.();
+    await this._mux.syncState?.();
   }
 
   _dec() {
@@ -93,10 +93,10 @@ class Channel {
     this._mux._safeDestroy(err);
   }
 
-  _fullyOpenSoon() {
+  async _fullyOpenSoon() {
     this._mux._remote[this._remoteId - 1].session = this;
     queueTick(this._fullyOpen.bind(this));
-    this._mux.syncState?.();
+    await this._mux.syncState?.();
   }
 
   _fullyOpen() {
@@ -133,7 +133,7 @@ class Channel {
     }
   }
 
-  _close(isRemote) {
+  async _close(isRemote) {
     if (this.closed === true) return;
     this.closed = true;
 
@@ -154,7 +154,7 @@ class Channel {
     this._track(this.onclose(isRemote, this));
 
     if (this._active === 0) this._destroy();
-    this._mux.syncState?.();
+    await this._mux.syncState?.();
   }
 
   _destroy() {
@@ -177,7 +177,7 @@ class Channel {
     this._mux.uncork();
   }
 
-  close() {
+  async close() {
     if (this.closed === true) return;
 
     const state = { buffer: null, start: 2, end: 2 };
@@ -190,7 +190,7 @@ class Channel {
     state.buffer[1] = 3;
     c.uint.encode(state, this._localId);
 
-    this._mux._close(false);
+    await this._mux._close(false);
     this._mux._write0(state.buffer);
   }
 
@@ -493,11 +493,11 @@ module.exports = class Protomux {
     this.stream.end();
   }
 
-  _decode(remoteId, state) {
+  async _decode(remoteId, state) {
     const type = c.uint.decode(state);
 
     if (remoteId === 0) {
-      this._oncontrolsession(type, state);
+      await this._oncontrolsession(type, state);
       return;
     }
 
@@ -512,11 +512,11 @@ module.exports = class Protomux {
       return;
     }
 
-    this.syncState?.();
+    await this.syncState?.();
     r.session._recv(type, state);
   }
 
-  _oncontrolsession(type, state) {
+  async _oncontrolsession(type, state) {
     switch (type) {
       case 0:
         this._onbatch(state);
@@ -535,7 +535,7 @@ module.exports = class Protomux {
         break;
     }
 
-    this.syncState?.();
+    await this.syncState?.();
   }
 
   _bufferMessage(r, type, { buffer, start, end }) {