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

[afrind]

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).

---

This change is 

[afrind]

@afrind made 6 comments.
Reviewable status: 0 of 8 files reviewed, 5 unresolved discussions.

---

src/MoqxRelay.cpp line 479 at r1 (raw file):

  // 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.

Seeded is used without definition here, so it's a bit confusing. I'm loath to make this comment longer, but as is it's not readable.

---

src/SubscriptionRegistry.cpp line 51 at r1 (raw file):

    auto forwarder = std::make_shared<moxygen::MoQForwarder>(ftn, largest);
    // Null callback: caller installs the real chain later on the forwarder's exec (LF path).
    if (callback) {

Is setting a null callback on forwarder harmful? It might be better to set it unilaterally and make no changes to SubscriptionRegistry.

---

src/MoqxRelay.cpp line 1983 at r1 (raw file):

  }

  // Fulfill on every exit (incl. error/cancel below) so isNew=false waiters never hang.

Do waiters need to differentiate between a successful exit and an unsuccessful one? We could used readiness->setException?

---

src/relay/LocalForwarderCallback.h line 25 at r1 (raw file):

//
// removeOnEmpty distinguishes the two roles:
//   - subscribe-path local & subscribe-initiated publisher forwarders (removeOnEmpty=true): when

Is this change adding value?

---

src/relay/LocalForwarderRegistry.h line 33 at r1 (raw file):

  struct Entry {
    std::shared_ptr<moxygen::MoQForwarder> forwarder;
    std::shared_ptr<folly::SharedPromise<folly::Unit>> readiness;

Would this read better as "ready" rather than "readiness"?

---

src/relay/LocalForwarderRegistry.h line 56 at r1 (raw file):

  // 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

maybe rename seeded to largestKnown or something similar?

[michalhosna]

I (and claude) see races.

See 43425c1

@michalhosna made 1 comment.
Reviewable status: 0 of 8 files reviewed, 5 unresolved discussions.

[afrind]

@afrind made 4 comments and resolved 4 discussions.
Reviewable status: 0 of 8 files reviewed, 1 unresolved discussion.

---

src/MoqxRelay.cpp line 1983 at r1 (raw file):

Previously, afrind wrote…

Do waiters need to differentiate between a successful exit and an unsuccessful one? We could used readiness->setException?

Fixed

---

src/relay/LocalForwarderCallback.h line 25 at r1 (raw file):

Previously, afrind wrote…

Is this change adding value?

Fixed

---

src/relay/LocalForwarderRegistry.h line 56 at r1 (raw file):

Previously, afrind wrote…

maybe rename seeded to largestKnown or something similar?

Decided on ready (since largest is unknown when no content published yet)

---

src/SubscriptionRegistry.cpp line 51 at r1 (raw file):

Previously, afrind wrote…

Is setting a null callback on forwarder harmful? It might be better to set it unilaterally and make no changes to SubscriptionRegistry.

Fixed

[afrind]

replaced this with the stack at #541