EventEmitter: offAndWait()/removeAllListenersAndWait() for the destroy-vs-in-flight-handler race

packages/streamr-dht/modules/connection/simulator/Simulator.cppm

@@ -30,6 +30,7 @@ module;
 #include <condition_variable>
 #include <cstddef>
 #include <cstdint>
+#include <exception>
 #include <functional>
 #include <map>
 #include <memory>
@@ -301,16 +302,30 @@ private:
             this->inFlightOperations++;
             lock.unlock();
             executor->add([this, operation]() {
-                switch (operation.type) {
-                    case OperationType::CONNECT:
-                        this->executeConnectOperation(operation);
-                        break;
-                    case OperationType::CLOSE:
-                        this->executeCloseOperation(operation);
-                        break;
-                    case OperationType::SEND:
-                        this->executeSendOperation(operation);
-                        break;
+                // The decrement below must run even if the call-out
+                // throws: folly's SerialExecutor swallows task exceptions
+                // (SerialExecutor::worker invokeCatchingExns), so a
+                // propagated exception would silently skip the decrement
+                // and park stop()'s drain-wait forever.
+                try {
+                    switch (operation.type) {
+                        case OperationType::CONNECT:
+                            this->executeConnectOperation(operation);
+                            break;
+                        case OperationType::CLOSE:
+                            this->executeCloseOperation(operation);
+                            break;
+                        case OperationType::SEND:
+                            this->executeSendOperation(operation);
+                            break;
+                    }
+                } catch (const std::exception& e) {
+                    SLogger::error(
+                        "Simulator operation threw an exception: " +
+                        std::string(e.what()));
+                } catch (...) {
+                    SLogger::error(
+                        "Simulator operation threw a non-std exception");
                 }
                 // Notify under the lock: once stop()'s drain-wait sees the
                 // count hit zero the Simulator may be destroyed, so this

packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm

@@ -78,13 +78,21 @@ public:
     }
 
     void destroy() {
-        transport.off<transportevents::Message>(this->onMessageHandlerToken);
+        // The waiting variant: the owner destroys this communicator right
+        // after destroy() returns, and a Message/Disconnected handler
+        // mid-invocation on another thread would then dereference freed
+        // memory. offAndWait() returns only once no such invocation is
+        // running (unless destroy() itself is called from inside a
+        // handler invocation, where waiting could deadlock and the
+        // historical non-blocking semantics are kept).
+        transport.offAndWait<transportevents::Message>(
+            this->onMessageHandlerToken);
         // Also detach the Disconnected listener: leaving it registered
         // dangles `this` on the transport after destruction, and a live
         // listener could still add client errors while a retired
         // communicator waits to be destroyed (see
         // RecursiveOperationManager::retiredSessionCommunicators).
-        transport.off<transportevents::Disconnected>(
+        transport.offAndWait<transportevents::Disconnected>(
             this->onDisconnectedHandlerToken);
     }
     using RpcCommunicator::registerRpcMethod;