relay: propagate downstream interest loss with warm-cache linger + upstream UNSUBSCRIBE#180
Conversation
…stream UNSUBSCRIBE
Observed bug
------------
When a relay deduplicates several downstream subscribers onto a single
upstream subscription, it never propagates interest loss back upstream.
After the LAST downstream subscriber for a track unsubscribes or
disconnects, the relay sends NO upstream UNSUBSCRIBE: the upstream
publisher keeps producing and the relay keeps pulling the track until the
whole session dies. Measured live against Cloudflare's deployment: the
upstream subscription stayed active for >80s after the last subscriber
left, with no UNSUBSCRIBE ever emitted.
Root cause
----------
Three things combined:
* No interest tracking. `TracksReader::subscribe` (the dedup point) hands
each downstream subscriber a clone of a cached `TrackReader`, but
nothing counts how many are alive.
* A pinned cache reader. On a cache miss the dedup cache stores a strong
`TrackReader` clone, so the track's reader side never fully drops even
after every real subscriber is gone. There was no TTL/eviction sweep
(`TracksWriter::remove` was dead code).
* `Subscribe::drop` is the ONLY place the transport crate emits an
upstream UNSUBSCRIBE. The upstream-subscribe holders in the relay
(`consumer.rs` locals path and `remote.rs` cross-relay path) hold that
`Subscribe` and await only `subscribe.closed()`, which resolves solely
on upstream events (SUBSCRIBE_ERROR / PUBLISH_DONE / stream error) or
session teardown. Downstream UNSUBSCRIBE / disconnect merely dropped
per-viewer `TrackReader` clones while the cache kept one pinned, so the
`Subscribe` was never dropped and no UNSUBSCRIBE was sent.
Fix design
----------
Add explicit downstream-interest tracking at both dedup points and react to
it at both upstream-subscribe sites:
* Interest refcount. New `serve::TrackInterest` / `TrackInterestGuard`
(moq-transport). Every `TrackReader` handed to a downstream subscriber
carries a guard that increments the per-track count on
creation/clone and decrements on drop. The dedup cache's own pinned
reader clone deliberately carries NO guard, so it is not counted as
interest. The reader flows through `Subscribed::serve`, which holds it
for the whole downstream subscription, so the guard's lifetime exactly
matches real interest.
* Warm-cache linger. When interest reaches zero the upstream-subscribe
task does NOT tear down immediately; it waits `WARM_CACHE_LINGER`
(3s, see moq-relay-ietf/src/lib.rs). If a new subscriber arrives during
the linger it is cancelled and serving continues from the still-warm
upstream subscription, preserving the intentional instant-late-joiner
caching behavior (page reloads, reconnects, channel flips) without a
fresh upstream SUBSCRIBE and its startup latency.
* Eviction + UNSUBSCRIBE. If the linger elapses still idle, the task
evicts the dedup-cache entry and returns, dropping the `Subscribe`
handle so `Subscribe::drop` emits the upstream UNSUBSCRIBE. The
eviction check is serialized with `subscribe()` under the shared cache
lock, so a subscriber arriving at the same instant is never stranded on
a torn-down upstream: it either keeps the entry warm or takes the
cache-miss path and starts a fresh upstream subscription.
* Both paths. The same select-on-interest-idle + linger + eviction logic
is applied to the locals path (`consumer.rs`, via `TracksReader`'s new
`TrackRequest`/`TrackLease`) and the cross-relay path (`remote.rs`,
whose own per-track cache now stores the interest alongside the reader).
Existing deduplication-while-alive behavior is preserved (multiple
concurrent subscribers still share one upstream subscription;
`test_track_deduplication_while_alive` still passes).
Tests
-----
moq-transport gains unit tests for: interest guard increment/decrement and
idle/busy signalling; refcount reaching zero + linger allowing eviction +
re-subscribe re-requesting upstream; re-subscribe during the linger keeping
the warm subscription (no new upstream request); and re-subscribe after
eviction creating a genuinely fresh upstream subscription.
cargo build + cargo test pass for moq-transport and moq-relay-ietf, and
`cargo build --workspace` is clean.
|
Independent confirmation of this exact failure is recorded in #191. Against The same behavior is deterministic against the It would be valuable to add a full relay integration regression alongside the
The current CI failure appears limited to |
Observed bug
When a relay deduplicates several downstream subscribers onto a single upstream subscription, it never propagates interest loss back upstream. After the LAST downstream subscriber for a track unsubscribes or disconnects, the relay sends NO upstream UNSUBSCRIBE: the upstream publisher keeps producing and the relay keeps pulling the track until the whole session dies. Measured live against Cloudflare's deployment: the upstream subscription stayed active for >80s after the last subscriber left, with no UNSUBSCRIBE ever emitted.
Root cause
Three things combined:
TracksReader::subscribe(the dedup point) hands each downstream subscriber a clone of a cachedTrackReader, but nothing counts how many are alive.TrackReaderclone, so the track's reader side never fully drops even after every real subscriber is gone. There was no TTL/eviction sweep (TracksWriter::removewas dead code).Subscribe::dropis the ONLY place the transport crate emits an upstream UNSUBSCRIBE. The upstream-subscribe holders in the relay (consumer.rslocals path andremote.rscross-relay path) hold thatSubscribeand await onlysubscribe.closed(), which resolves solely on upstream events (SUBSCRIBE_ERROR / PUBLISH_DONE / stream error) or session teardown. Downstream UNSUBSCRIBE / disconnect merely dropped per-viewerTrackReaderclones while the cache kept one pinned, so theSubscribewas never dropped and no UNSUBSCRIBE was sent.Fix design
Add explicit downstream-interest tracking at both dedup points and react to it at both upstream-subscribe sites:
Interest refcount. New
serve::TrackInterest/TrackInterestGuard(moq-transport). EveryTrackReaderhanded to a downstream subscriber carries a guard that increments the per-track count on creation/clone and decrements on drop. The dedup cache's own pinned reader clone deliberately carries NO guard, so it is not counted as interest. The reader flows throughSubscribed::serve, which holds it for the whole downstream subscription, so the guard's lifetime exactly matches real interest.Warm-cache linger. When interest reaches zero the upstream-subscribe task does NOT tear down immediately; it waits
WARM_CACHE_LINGER(3s, see moq-relay-ietf/src/lib.rs). If a new subscriber arrives during the linger it is cancelled and serving continues from the still-warm upstream subscription, preserving the intentional instant-late-joiner caching behavior (page reloads, reconnects, channel flips) without a fresh upstream SUBSCRIBE and its startup latency.Eviction + UNSUBSCRIBE. If the linger elapses still idle, the task evicts the dedup-cache entry and returns, dropping the
Subscribehandle soSubscribe::dropemits the upstream UNSUBSCRIBE. The eviction check is serialized withsubscribe()under the shared cache lock, so a subscriber arriving at the same instant is never stranded on a torn-down upstream: it either keeps the entry warm or takes the cache-miss path and starts a fresh upstream subscription.Both paths. The same select-on-interest-idle + linger + eviction logic is applied to the locals path (
consumer.rs, viaTracksReader's newTrackRequest/TrackLease) and the cross-relay path (remote.rs, whose own per-track cache now stores the interest alongside the reader).Existing deduplication-while-alive behavior is preserved (multiple concurrent subscribers still share one upstream subscription;
test_track_deduplication_while_alivestill passes).Tests
moq-transport gains unit tests for: interest guard increment/decrement and idle/busy signalling; refcount reaching zero + linger allowing eviction + re-subscribe re-requesting upstream; re-subscribe during the linger keeping the warm subscription (no new upstream request); and re-subscribe after eviction creating a genuinely fresh upstream subscription.
cargo build + cargo test pass for moq-transport and moq-relay-ietf, and
cargo build --workspaceis clean.