relay: hop cross-exec handle teardown to the owning exec

The reciprocal peer SubscribeNamespaceHandle was acquired via a direct
session->subscribeNamespace() and stored raw, so peerSubNsHandles_.erase()
destroyed it on relayExec_ — running its teardown (isClosed() read + control
RST) on the wrong thread while the peer session closed on its own io exec
(TSAN-confirmed data race on MoQSession).

Acquire it through maybeWrapPublisher so the handle is a
CrossExecSubscribeNamespaceHandle bound to the session's exec, and give every
cross-exec wrapper in PublisherCrossExecFilter a destructor that hops the inner
handle's destruction to that exec — so destructor-driven teardown is
exec-correct, not just the explicit unsubscribe/cancel calls.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: afrind

src/MoqxRelay.cpp

@@ -1364,11 +1364,10 @@ folly::coro::Task<Publisher::SubscribeNamespaceResult> MoqxRelay::subscribeNames
     // Tag with the peer's relay ID so we suppress echoing these namespaces
     // back to that peer on reconnect.
     auto handle = makeNamespaceBridgeHandle(weak_from_this(), session, incomingPeerID, relayExec_);
-    // subscribeNamespace must run on the peer session's executor.
-    auto recipResult = co_await folly::coro::co_withExecutor(
-        folly::getKeepAliveToken(session->getExecutor()),
-        session->subscribeNamespace(makePeerSubNs(), handle)
-    ); // no token: reciprocal, prevents loop
+    // maybeWrapPublisher runs the call on the peer session's executor and wraps
+    // the returned handle so its teardown hops there too (no token: reciprocal).
+    auto recipResult =
+        co_await maybeWrapPublisher(relayExec_, session)->subscribeNamespace(makePeerSubNs(), handle);
     if (recipResult.hasError()) {
       XLOG(ERR) << "Reciprocal peer subNs failed: " << recipResult.error().reasonPhrase;
     } else {

src/relay/PublisherCrossExecFilter.cpp

@@ -22,6 +22,13 @@ class CrossExecSubscriptionHandle : public moxygen::SubscriptionHandle {
   )
       : inner_(std::move(inner)), exec_(exec) {}
 
+  ~CrossExecSubscriptionHandle() override {
+    // Inner dtor may touch session state; destroy it on exec_, not the dropping thread.
+    if (inner_) {
+      exec_->add([inner = std::move(inner_)]() mutable {});
+    }
+  }
+
   const moxygen::SubscribeOk& subscribeOk() const override { return inner_->subscribeOk(); }
 
   void unsubscribe() override {
@@ -48,6 +55,13 @@ class CrossExecFetchHandle : public moxygen::Publisher::FetchHandle {
   )
       : inner_(std::move(inner)), exec_(exec) {}
 
+  ~CrossExecFetchHandle() override {
+    // Inner dtor may touch session state; destroy it on exec_, not the dropping thread.
+    if (inner_) {
+      exec_->add([inner = std::move(inner_)]() mutable {});
+    }
+  }
+
   const moxygen::FetchOk& fetchOk() const override { return inner_->fetchOk(); }
 
   void fetchCancel() override {
@@ -77,6 +91,13 @@ class CrossExecNamespacePublishHandle : public moxygen::Publisher::NamespacePubl
   )
       : inner_(std::move(inner)), exec_(std::move(exec)) {}
 
+  ~CrossExecNamespacePublishHandle() override {
+    // Inner dtor may touch session state; destroy it on exec_, not the dropping thread.
+    if (inner_) {
+      exec_->add([inner = std::move(inner_)]() mutable {});
+    }
+  }
+
   void namespaceMsg(const moxygen::TrackNamespace& ns) override {
     exec_->add([inner = inner_, ns]() mutable { inner->namespaceMsg(ns); });
   }
@@ -98,6 +119,13 @@ class CrossExecSubscribeNamespaceHandle : public moxygen::Publisher::SubscribeNa
   )
       : inner_(std::move(inner)), exec_(exec) {}
 
+  ~CrossExecSubscribeNamespaceHandle() override {
+    // Inner dtor may touch session state; destroy it on exec_, not the dropping thread.
+    if (inner_) {
+      exec_->add([inner = std::move(inner_)]() mutable {});
+    }
+  }
+
   const moxygen::SubscribeNamespaceOk& subscribeNamespaceOk() const override {
     return inner_->subscribeNamespaceOk();
   }
@@ -128,6 +156,13 @@ class CrossExecPublishBlockedHandle : public moxygen::Publisher::PublishBlockedH
   )
       : inner_(std::move(inner)), exec_(std::move(exec)) {}
 
+  ~CrossExecPublishBlockedHandle() override {
+    // Inner dtor may touch session state; destroy it on exec_, not the dropping thread.
+    if (inner_) {
+      exec_->add([inner = std::move(inner_)]() mutable {});
+    }
+  }
+
   void publishBlocked(
       const moxygen::TrackNamespace& trackNamespaceSuffix,
       const std::string& trackName
@@ -150,6 +185,13 @@ class CrossExecSubscribeTracksHandle : public moxygen::Publisher::SubscribeTrack
   )
       : inner_(std::move(inner)), exec_(exec) {}
 
+  ~CrossExecSubscribeTracksHandle() override {
+    // Inner dtor may touch session state; destroy it on exec_, not the dropping thread.
+    if (inner_) {
+      exec_->add([inner = std::move(inner_)]() mutable {});
+    }
+  }
+
   const moxygen::SubscribeTracksOk& subscribeTracksOk() const override {
     return inner_->subscribeTracksOk();
   }

test/MoqxRelaySubNsTests.cpp

@@ -167,6 +167,7 @@ TEST_P(MoQRelayTest, ExactNamespaceSubscriberReceivesPublishNamespace) {
 
   removeSession(publisher);
   removeSession(subscriber);
+  driveIfMultiThread(); // flush exec-hopped handle destruction before mocks are destroyed
 }
 
 // Bug: when a subscriber with forward=true joins a namespace whose track