Synchronize node lifecycle transitions (#1507)

This places protections around node lifecycle transitions with many async steps.  Previously it was possible to
trigger these simultaneously in ways that cause crashes due to interference between them.  They may still be
invoked in parallel but will execute sequentially.  Protected methods are available for composing the operations
when the mutex is already held.

packages/general/src/util/Mutex.ts

@@ -65,6 +65,21 @@ export class Mutex implements PromiseLike<unknown> {
         }
     }
 
+    /**
+     * Enqueue work with an awaitable result.
+     */
+    produce<T>(task: () => PromiseLike<T>, cancel?: () => void): Promise<T> {
+        return new Promise<T>((resolve, reject) => {
+            this.run(async () => {
+                try {
+                    resolve(await task());
+                } catch (e) {
+                    reject(e);
+                }
+            }, cancel);
+        });
+    }
+
     /**
      * Cancel remaining work and perform one last task with the Mutex held.
      */

packages/node/src/node/Node.ts

@@ -82,6 +82,10 @@ export abstract class Node<T extends Node.CommonRootEndpoint = Node.CommonRootEn
      * Bring the node online.
      */
     async start() {
+        await this.lifecycle.mutex.produce(this.startWithMutex.bind(this));
+    }
+
+    protected async startWithMutex() {
         this.env.runtime.add(this);
 
         try {
@@ -123,6 +127,10 @@ export abstract class Node<T extends Node.CommonRootEndpoint = Node.CommonRootEn
      * Once the node is offline you may use {@link start} to bring the node online again.
      */
     async cancel() {
+        await this.lifecycle.mutex.produce(this.cancelWithMutex.bind(this));
+    }
+
+    protected async cancelWithMutex() {
         if (!this.#runtime) {
             return;
         }
@@ -133,17 +141,27 @@ export abstract class Node<T extends Node.CommonRootEndpoint = Node.CommonRootEn
     }
 
     override async close() {
+        await this.lifecycle.mutex.produce(this.closeWithMutex.bind(this));
+    }
+
+    protected async closeWithMutex() {
         // The runtime is not designed to operate with a node that is shutting down so destroy it before performing
         // actual close
-        //
-        // TODO - this should probably block other functions like start()
         if (this.#runtime) {
-            await this.cancel();
+            await this.cancelWithMutex();
         }
 
         await super.close();
     }
 
+    override async reset() {
+        await this.lifecycle.mutex.produce(this.resetWithMutex.bind(this));
+    }
+
+    protected async resetWithMutex() {
+        return super.reset();
+    }
+
     /**
      * Create the network runtime.
      */
@@ -177,7 +195,7 @@ export abstract class Node<T extends Node.CommonRootEndpoint = Node.CommonRootEn
     }
 
     override async [Construction.destruct]() {
-        await this.cancel();
+        await this.cancelWithMutex();
         await super[Construction.destruct]();
         DiagnosticSource.delete(this);
     }

packages/node/src/node/NodeLifecycle.ts

@@ -7,7 +7,7 @@
 import { ActionContext } from "#behavior/context/ActionContext.js";
 import { Endpoint } from "#endpoint/Endpoint.js";
 import { EndpointLifecycle } from "#endpoint/properties/EndpointLifecycle.js";
-import { AsyncObservable, Observable } from "#general";
+import { AsyncObservable, Mutex, Observable } from "#general";
 
 /**
  * Extended lifecycle information that only applies to root endpoints.
@@ -21,10 +21,13 @@ export class NodeLifecycle extends EndpointLifecycle {
     #initialized = Observable<[isCommissioned: boolean]>();
     #isOnline = false;
     #isCommissioned = false;
+    #mutex: Mutex;
 
     constructor(endpoint: Endpoint) {
         super(endpoint);
 
+        this.#mutex = new Mutex(endpoint);
+
         this.#online.on(() => {
             this.#isOnline = true;
         });
@@ -97,4 +100,17 @@ export class NodeLifecycle extends EndpointLifecycle {
     get decommissioned() {
         return this.#decommissioned;
     }
+
+    /**
+     * Mutex for protecting node lifecycle transitions.
+     *
+     * Methods that implement complex async lifecycle transitions use this mutex to ensure conflicting operations cannot
+     * intermingle.
+     *
+     * Generally methods that hold this mutex have a protected "*WithMutex" variant.  This allows for nesting of logic
+     * that requires the mutex without causing deadlock.
+     */
+    get mutex() {
+        return this.#mutex;
+    }
 }

packages/node/src/node/ServerNode.ts

@@ -114,20 +114,24 @@ export class ServerNode<T extends ServerNode.RootEndpoint = ServerNode.RootEndpo
      * Perform a factory reset of the node.
      */
     override async erase() {
+        await this.lifecycle.mutex.produce(this.eraseWithMutex.bind(this));
+    }
+
+    protected async eraseWithMutex() {
         try {
             await this.construction;
 
             // Go offline before performing reset
             const isOnline = this.lifecycle.isOnline;
             if (isOnline) {
-                await this.cancel();
+                await this.cancelWithMutex();
             }
 
             // Inform user
             this.statusUpdate("resetting to factory defaults");
 
             // Reset in-memory state
-            await this.reset();
+            await this.resetWithMutex();
 
             // Reset persistent state
             await this.resetStorage();
@@ -137,7 +141,7 @@ export class ServerNode<T extends ServerNode.RootEndpoint = ServerNode.RootEndpo
 
             // Go back online if we were online at time of reset, otherwise just await reinitialization
             if (isOnline) {
-                await this.start();
+                await this.startWithMutex();
             } else {
                 await this.construction.ready;
             }

packages/node/test/node/ServerNodeTest.ts

@@ -350,7 +350,7 @@ describe("ServerNode", () => {
             await node.cancel();
         }
 
-        await node.erase();
+        await MockTime.resolve(node.erase());
 
         // Confirm previous online state is resumed
         expect(node.lifecycle.isOnline).equals(mode === "online");

packages/protocol/src/mdns/MdnsService.ts

@@ -22,6 +22,7 @@ const logger = Logger.get("MDNS");
 export class MdnsService {
     #broadcaster?: MdnsBroadcaster;
     #scanner?: MdnsScanner;
+    #env: Environment;
     readonly #construction: Construction<MdnsService>;
     readonly #enableIpv4: boolean;
     readonly limitedToNetInterface?: string;
@@ -31,6 +32,7 @@ export class MdnsService {
     }
 
     constructor(environment: Environment, options?: MdnsService.Options) {
+        this.#env = environment;
         environment.set(MdnsService, this);
         environment.runtime.add(this);
 
@@ -78,6 +80,8 @@ export class MdnsService {
     }
 
     async [Symbol.asyncDispose]() {
+        this.#env.delete(MdnsService, this);
+
         await this.#construction.close(async () => {
             const broadcasterDisposal = MaybePromise.then(this.#broadcaster?.close(), undefined, e =>
                 logger.error("Error disposing of MDNS broadcaster", e),