relay: give subscribe-initiated forwarders the LF publisher chain + tl slot

In LF mode a subscribe-triggered upstream subscription created its publisher
forwarder on relayExec_ with a MoqxRelay-direct callback and never placed it in
tlForwarders_. That forwarder lives on the upstream session's iothread, so its
onEmpty/forwardChanged/newGroupRequested ran there and touched registry_ off
relayExec_, and same-thread subscribers missed the zero-hop fast path.

Factor the publisher-forwarder callback chain (Weak -> CrossExec -> Local) and
tlForwarders_ slot claim out of createPublisherForwarder into a shared
installPublisherForwarderCallbackChain helper. The first subscriber now installs
that same Weak->CrossExec(relayExec_)->Local chain and claims the tl slot on the
publisher's exec (removeOnEmpty=true, since subscribe-initiated tracks drop on
empty). getOrCreateFromSubscribe accepts a null callback so the LF caller
installs the real chain on the forwarder's exec; non-LF still passes the relay
directly.

Document the resulting symmetry in the callback-chain overview and dev doc:
subscribe-initiated now uses the same chain as publish-initiated, with the
removeOnEmpty contract differing (true vs false).

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

Author: afrind

docs/dev/local-forwarder-flow.md

@@ -135,6 +135,8 @@ upstream work, and nests a single sortie to `[Pub]` to wire the channel sub.
 [Relay]                 ├─▶ joinOrPrepareUpstreamSubscription()  # registry: first vs subsequent
 [Relay]                 ├─▶ buildLocalToPublisherCallbacks()
 [Relay]                 └─▶ ⇢⇢▶ [Pub]  single sortie:
+[Pub]                        ├─ if FIRST subscriber:
+[Pub]                        │    └─▶ installPublisherForwarderCallbackChain()  # chain + tlForwarders_ slot
 [Pub]                        ├─▶ installChannelSubscriber(localFwd ↔ publisherFwd)
 [Pub]                        └─ if FIRST subscriber:
 [Pub]                             ├─▶ addChannelSubscriber(relayChain, passive)
@@ -151,9 +153,20 @@ upstream work, and nests a single sortie to `[Pub]` to wire the channel sub.
 ```
 
 The first subscriber does the heavy lifting (upstream subscribe + installing the relay chain,
-the same passive cache/Top-N chain described in [Data flow](#data-flow)). Subsequent
-subscribers on the same thread hit the `attachSubscriber` fast path; on other threads they take
-only the `installChannelSubscriber` half of the sortie.
+the same passive cache/Top-N chain described in [Data flow](#data-flow)). It also installs the
+publisher-forwarder control chain and claims the `tlForwarders_` slot via
+`installPublisherForwarderCallbackChain` — the **same** wiring the publish path uses, so a
+subscribe-initiated publisher forwarder is symmetric with a publish-initiated one. Subsequent
+subscribers on the same thread hit the `attachSubscriber` fast path (now also for
+subscribe-initiated tracks); on other threads they take only the `installChannelSubscriber` half
+of the sortie.
+
+The forwarder is created on `relayExec_` (in `joinOrPrepareUpstreamSubscription`) with a **null**
+callback, then gets its real callback + tl slot installed on `[Pub]` in the first-subscriber
+sortie — `tlForwarders_.get()` must run on the forwarder's own exec. It uses `removeOnEmpty=true`
+(subscribe-initiated tracks drop on empty, unlike publish): `LocalForwarderCallback` vacates the
+tl slot on `[Pub]` when the last subscriber leaves, while `onEmptyImpl` unsubscribes upstream and
+removes the registry entry on `[Relay]`.
 
 ### Why each guard exists
 
@@ -250,8 +263,10 @@ The reusable adapter layers:
 
 ### Publisher forwarder → relay state
 
-Built by `createPublisherForwarder`. Lifecycle events on the publisher's own forwarder must
-reach relay-global state on `relayExec_`:
+Built by `installPublisherForwarderCallbackChain` for both publish-initiated (via
+`createPublisherForwarder`) and subscribe-initiated (via `attachNewLocalForwarderOnRelayExec`'s
+first-subscriber sortie) tracks. Lifecycle events on the publisher's own forwarder must reach
+relay-global state on `relayExec_`:
 
 ```
 [Pub]  publisherFwd fires onEmpty / forwardChanged / newGroupRequested

src/MoqxRelay.cpp

@@ -14,6 +14,7 @@
 #include "relay/PublisherCrossExecFilter.h"
 #include "relay/SubscriberCrossExecFilter.h"
 #include "relay/WeakRelayForwarderCallback.h"
+#include <folly/ScopeGuard.h>
 #include <folly/container/F14Set.h>
 #include <moxygen/MoQFilters.h>
 #include <moxygen/MoQTrackProperties.h>
@@ -451,28 +452,44 @@ MoqxRelay::validatePublishNamespace(const FullTrackName& ftn, RequestID requestI
   return std::nullopt;
 }
 
-// Constructs the publisher's local forwarder and installs its callback chain (Weak ->
-// CrossExec -> LocalForwarder) on publisherExec, before the reply hops to relayExec_.
-// tlForwarders_ must already be initialized.
-std::shared_ptr<MoQForwarder> MoqxRelay::createPublisherForwarder(const PublishRequest& pub) {
-  const auto& ftn = pub.fullTrackName;
-  auto localPubFwd = std::make_shared<MoQForwarder>(ftn, pub.largest);
-  localPubFwd->setExtensions(pub.extensions);
-
-  // removeOnEmpty=false: the publisher's forwarder must survive subscriber churn, so
-  // LocalForwarderCallback removes it from tlForwarders_ only when the source ends.
+// Installs the publisher-forwarder callback chain (Weak -> CrossExec(relayExec_) ->
+// LocalForwarder) and claims the tlForwarders_ slot. Must run on the forwarder's exec.
+void MoqxRelay::installPublisherForwarderCallbackChain(
+    const FullTrackName& ftn,
+    const std::shared_ptr<MoQForwarder>& publisherFwd,
+    bool removeOnEmpty
+) {
+  if (!tlForwarders_.get()) {
+    tlForwarders_.reset(new LocalForwarderRegistry());
+  }
   auto relayAdapter = std::make_shared<WeakRelayForwarderCallback>(weak_from_this());
   auto crossExec = std::make_shared<CrossExecForwarderCallback>(
       relayExec_,
-      localPubFwd,
+      publisherFwd,
       std::move(relayAdapter)
   );
-  localPubFwd->setCallback(std::make_shared<LocalForwarderCallback>(
+  publisherFwd->setCallback(std::make_shared<LocalForwarderCallback>(
       tlForwarders_.get(),
       ftn,
       std::move(crossExec),
-      /*removeOnEmpty=*/false
+      removeOnEmpty
   ));
+  // Authoritative slot claim; displaces any stale subscribe-path local forwarder so
+  // same-thread subscribers reuse THIS forwarder via the fast path. Publish forwarders
+  // (removeOnEmpty=false) are seeded at creation; subscribe claims await upstream OK.
+  tlForwarders_->set(ftn, publisherFwd, /*seeded=*/!removeOnEmpty);
+}
+
+// Constructs the publisher's local forwarder and installs its callback chain (Weak ->
+// CrossExec -> LocalForwarder) on publisherExec, before the reply hops to relayExec_.
+std::shared_ptr<MoQForwarder> MoqxRelay::createPublisherForwarder(const PublishRequest& pub) {
+  const auto& ftn = pub.fullTrackName;
+  auto localPubFwd = std::make_shared<MoQForwarder>(ftn, pub.largest);
+  localPubFwd->setExtensions(pub.extensions);
+
+  // removeOnEmpty=false: the publisher's forwarder must survive subscriber churn, so
+  // LocalForwarderCallback removes it from tlForwarders_ only when the source ends.
+  installPublisherForwarderCallbackChain(ftn, localPubFwd, /*removeOnEmpty=*/false);
 
   return localPubFwd;
 }
@@ -488,17 +505,9 @@ Subscriber::PublishResult MoqxRelay::publishFromPublisherExec(
     return folly::makeUnexpected(std::move(*err));
   }
 
-  if (!tlForwarders_.get()) {
-    tlForwarders_.reset(new LocalForwarderRegistry());
-  }
-
+  // createPublisherForwarder claims the tlForwarders_ slot on this exec.
   auto localPubFwd = createPublisherForwarder(pub);
 
-  // The publisher's forwarder is authoritative — claim the slot, displacing any
-  // stale subscribe-path local forwarder so same-thread subscribers reuse THIS
-  // forwarder via the fast path.
-  tlForwarders_->set(pub.fullTrackName, localPubFwd);
-
   // crossExecFilter is a channel subscriber for the relay exec
   // regulsterPublishOnRelay exec completes wiring the chain (topNFilter → terminationFilter →
   // cache).
@@ -886,14 +895,17 @@ void teardownLocalForwarderOnFailure(
 // Single-threaded mode:
 //   forwarder.callback = MoqxRelay (direct, no hop)
 //
-// Multi-threaded — publisher forwarder (lives on publisherExec):
+// Multi-threaded — publisher forwarder (lives on publisherExec). Built by
+// installPublisherForwarderCallbackChain for BOTH publish-initiated (removeOnEmpty=false,
+// survives churn) and subscribe-initiated (removeOnEmpty=true, drops on empty) tracks:
 //   publisherFwd.callback =
-//     CrossExecForwarderCallback(relayExec_, publisherFwd,
-//       WeakRelayForwarderCallback(relay))
+//     LocalForwarderCallback(tlForwarders_, ftn,
+//       CrossExecForwarderCallback(relayExec_, publisherFwd,
+//         WeakRelayForwarderCallback(relay)))
 //
-//   [publisherExec] CrossExecForwarderCallback: captures ftn by value,
-//                 dispatches to relayExec_ fire-and-forget
-//       ↓
+//   [publisherExec] LocalForwarderCallback: removes from tlForwarders_ (onPublishDone
+//                 always; onEmpty if removeOnEmpty), passes the rest through
+//       ↓ (CrossExecForwarderCallback dispatches to relayExec_ fire-and-forget)
 //   [relayExec_]  WeakRelayForwarderCallback: recovers relay via weak_ptr,
 //                 calls onEmptyImpl / forwardChangedImpl / newGroupRequestedImpl
 //
@@ -1762,6 +1774,7 @@ folly::coro::Task<MoqxRelay::PublisherAttachment> MoqxRelay::attachNewLocalForwa
     bool forward
 ) {
   // Runs on relayExec_.
+  XCHECK(mode() == Mode::LocalForwarder) << "subscribe-init chain install is LF-only";
   const auto& ftn = subReq.fullTrackName;
   PublisherAttachment attach;
 
@@ -1796,6 +1809,10 @@ folly::coro::Task<MoqxRelay::PublisherAttachment> MoqxRelay::attachNewLocalForwa
   co_await folly::coro::co_withExecutor(
       folly::getKeepAliveToken(attach.publisherExec),
       [&]() -> folly::coro::Task<void> {
+        // First subscriber installs the publisher chain + tl slot before any sub is added.
+        if (sr.firstSetup) {
+          installPublisherForwarderCallbackChain(ftn, attach.publisherFwd, /*removeOnEmpty=*/true);
+        }
         installChannelSubscriber(
             *cbs.channelCb,
             *attach.publisherFwd,
@@ -1888,7 +1905,7 @@ MoqxRelay::joinOrPrepareUpstreamSubscription(SubscribeRequest subReq) {
 
   auto firstOrSubsequent = registry_.getOrCreateFromSubscribe(
       ftn,
-      shared_from_this(),
+      /*callback=*/nullptr,
       [this, &ftn](std::shared_ptr<MoQForwarder> f) { return buildFilterChain(ftn, std::move(f)); }
   );
 
@@ -1952,14 +1969,27 @@ folly::coro::Task<Publisher::SubscribeResult> MoqxRelay::subscribeFromSubscriber
       acquireLocalForwarder(ftn, [&] { return std::make_shared<MoQForwarder>(ftn); });
 
   if (!isNew) {
+    // Wait for an in-flight isNew=true setup to seed largest, else SUBSCRIBE_OK carries
+    // the pre-seeding value a client reads as a track restart.
+    if (auto ready = localReg->readiness(ftn); ready && !ready->isFulfilled()) {
+      co_await ready->getSemiFuture();
+    }
     if (auto err = checkRangeNotInPast(*localFwd, subReq)) {
       co_return folly::makeUnexpected(std::move(*err));
     }
     co_return attachSubscriber(*localFwd, std::move(session), subReq, std::move(consumer));
   }
 
-  // isNew=true: this thread owns setup. Install PendingForwarderCallback first so
-  // forwardChanged/newGroupRequested/onEmpty events during setup are captured for replay.
+  // Fulfill on every exit (incl. error/cancel below) so isNew=false waiters never hang.
+  auto readiness = localReg->beginReadiness(ftn);
+  auto fulfillReadiness = folly::makeGuard([&readiness]() noexcept {
+    if (!readiness->isFulfilled()) {
+      readiness->setValue();
+    }
+  });
+
+  // Install PendingForwarderCallback first so forwardChanged/newGroupRequested/onEmpty
+  // events during setup are captured for replay.
   auto pendingCb = std::make_shared<PendingForwarderCallback>(localReg, ftn);
   localFwd->setCallback(pendingCb);
 

src/MoqxRelay.h

@@ -349,10 +349,17 @@ class MoqxRelay : public moxygen::Publisher,
   );
 
   // Constructs the publisher's local forwarder and installs its callback chain on
-  // publisherExec. tlForwarders_ must already be initialized.
+  // publisherExec.
   std::shared_ptr<moxygen::MoQForwarder> createPublisherForwarder(const moxygen::PublishRequest& pub
   );
 
+  // Must run on the forwarder's exec; removeOnEmpty=false survives churn, true drops on empty.
+  void installPublisherForwarderCallbackChain(
+      const moxygen::FullTrackName& ftn,
+      const std::shared_ptr<moxygen::MoQForwarder>& publisherFwd,
+      bool removeOnEmpty
+  );
+
   std::optional<moxygen::PublishError>
   validatePublishNamespace(const moxygen::FullTrackName& ftn, moxygen::RequestID requestID) const;
 

src/SubscriptionRegistry.cpp

@@ -47,7 +47,10 @@ SubscriptionRegistry::getOrCreateFromSubscribe(
   auto it = subscriptions_.find(ftn);
   if (it == subscriptions_.end()) {
     auto forwarder = std::make_shared<moxygen::MoQForwarder>(ftn, largest);
-    forwarder->setCallback(std::move(callback));
+    // Null callback: caller installs the real chain later on the forwarder's exec (LF path).
+    if (callback) {
+      forwarder->setCallback(std::move(callback));
+    }
     auto [consumer, topNFilter] = chainBuilder(forwarder);
     auto [emplaceIt, inserted] = subscriptions_.emplace(
         std::piecewise_construct,

src/relay/LocalForwarderCallback.h

@@ -22,10 +22,11 @@ namespace openmoq::moqx {
 // never clobber a newer one that has claimed the same track name.
 //
 // removeOnEmpty distinguishes the two roles:
-//   - subscribe-path local forwarder (removeOnEmpty=true): when its last
+//   - subscribe-path local & subscribe-initiated publisher forwarders (removeOnEmpty=true): when
+//   last
 //     subscriber leaves, its channel sub is pulled from the publisher and it is
 //     dead — remove on onEmpty as well as onPublishDone.
-//   - publisher's publisher forwarder (removeOnEmpty=false): it must survive
+//   - publish-initiated publisher forwarder (removeOnEmpty=false): it must survive
 //     subscriber churn (new subscribers may arrive while the publisher is
 //     live), so it is removed ONLY when the source terminates (onPublishDone).
 class LocalForwarderCallback : public moxygen::MoQForwarder::Callback {

src/relay/LocalForwarderRegistry.h

@@ -10,6 +10,7 @@
 #include <moxygen/relay/MoQForwarder.h>
 
 #include <folly/container/F14Map.h>
+#include <folly/futures/SharedPromise.h>
 
 namespace openmoq::moqx {
 
@@ -25,6 +26,13 @@ class LocalForwarderRegistry {
     bool isNew;
   };
 
+  // readiness is non-null only while a subscribe-initiated forwarder's largest is
+  // seeded asynchronously; publish-path forwarders seed at creation and leave it null.
+  struct Entry {
+    std::shared_ptr<moxygen::MoQForwarder> forwarder;
+    std::shared_ptr<folly::SharedPromise<folly::Unit>> readiness;
+  };
+
   // Returns the existing local forwarder for ftn, or calls factory() to create
   // one. factory() is called at most once per ftn per thread lifetime.
   GetOrCreateResult getOrCreate(
@@ -33,10 +41,10 @@ class LocalForwarderRegistry {
   ) {
     auto it = forwarders_.find(ftn);
     if (it != forwarders_.end()) {
-      return {it->second, /*isNew=*/false};
+      return {it->second.forwarder, /*isNew=*/false};
     }
     auto forwarder = factory();
-    forwarders_.emplace(ftn, forwarder);
+    forwarders_[ftn].forwarder = forwarder;
     return {std::move(forwarder), /*isNew=*/true};
   }
 
@@ -45,16 +53,24 @@ class LocalForwarderRegistry {
   // a stale subscribe-path local forwarder under the same name is displaced
   // here, and drains itself via the source-termination cascade (its identity-
   // checked removal then no-ops, since this forwarder now owns the slot).
+  // seeded=true releases readiness waiters now (the claiming forwarder's largest is
+  // authoritative); seeded=false leaves them pending until the upstream OK.
   std::shared_ptr<moxygen::MoQForwarder>
-  set(const moxygen::FullTrackName& ftn, std::shared_ptr<moxygen::MoQForwarder> forwarder) {
-    auto prev = std::move(forwarders_[ftn]);
-    forwarders_[ftn] = std::move(forwarder);
+  set(const moxygen::FullTrackName& ftn,
+      std::shared_ptr<moxygen::MoQForwarder> forwarder,
+      bool seeded) {
+    auto& entry = forwarders_[ftn];
+    auto prev = std::move(entry.forwarder);
+    entry.forwarder = std::move(forwarder);
+    if (seeded && entry.readiness && !entry.readiness->isFulfilled()) {
+      entry.readiness->setValue();
+    }
     return prev;
   }
 
   std::shared_ptr<moxygen::MoQForwarder> get(const moxygen::FullTrackName& ftn) const {
     auto it = forwarders_.find(ftn);
-    return it != forwarders_.end() ? it->second : nullptr;
+    return it != forwarders_.end() ? it->second.forwarder : nullptr;
   }
 
   // Identity-checked removal: erase the entry for ftn only if it still points
@@ -64,18 +80,29 @@ class LocalForwarderRegistry {
   // forwarder that has since claimed the same track name.
   void remove(const moxygen::FullTrackName& ftn, const moxygen::MoQForwarder* expected) {
     auto it = forwarders_.find(ftn);
-    if (it == forwarders_.end() || it->second.get() != expected) {
+    if (it == forwarders_.end() || it->second.forwarder.get() != expected) {
       return;
     }
     forwarders_.erase(it);
   }
 
+  // isNew=false attachers await this before reading largest, so they never observe the
+  // pre-seeding value a client reads as a track restart.
+  std::shared_ptr<folly::SharedPromise<folly::Unit>>
+  beginReadiness(const moxygen::FullTrackName& ftn) {
+    auto promise = std::make_shared<folly::SharedPromise<folly::Unit>>();
+    forwarders_[ftn].readiness = promise;
+    return promise;
+  }
+
+  std::shared_ptr<folly::SharedPromise<folly::Unit>> readiness(const moxygen::FullTrackName& ftn
+  ) const {
+    auto it = forwarders_.find(ftn);
+    return it != forwarders_.end() ? it->second.readiness : nullptr;
+  }
+
 private:
-  folly::F14FastMap<
-      moxygen::FullTrackName,
-      std::shared_ptr<moxygen::MoQForwarder>,
-      moxygen::FullTrackName::hash>
-      forwarders_;
+  folly::F14FastMap<moxygen::FullTrackName, Entry, moxygen::FullTrackName::hash> forwarders_;
 };
 
 } // namespace openmoq::moqx

test/MoqxRelayPublishTests.cpp

@@ -451,6 +451,9 @@ TEST_P(MoQRelayTest, PublishReconnectDuringSubscribeScopeGuardCrash) {
   // Relay state mutations must run on the relay executor; doPublishNamespaceDone
   // touches the namespace tree, which publishDone also cleans up via relayEvb_.
   verifyOnRelayExec([&] { relay_->doPublishNamespaceDone(kTestNamespace, publisherSession2); });
+  // Drain the subscriber exec so the detached failure-teardown coro runs and
+  // releases the subscriber session it holds (else: leaked mock at exit).
+  driveIfMultiThread();
 }
 
 // Same reconnect scenario but the upstream subscribe returns OK instead of an