Add entry-cleaning up behavior when stream got dropped

[solus161]

Description

What's new:

• Add a guard for SubEntry for RAII behavior. Also update stream in ClientManager.subscribe() to take in the guard and the Receiver;
• Add test for the entry-cleaned-up behavior when stream is dropped;

Related Issue

• Closes [BUG] Dropped WS subscriptions remain registered until a matching broadcast #206

Checklist

• I have read the CONTRIBUTING.md document.
• My code follows the project's coding standards.
• I have added tests to cover my changes.
• All new and existing tests passed (make check).
• I have updated the documentation accordingly.

[lxsaah]

Thanks @solus161, I verified the bug on main (38dc500) and confirmed your PR fixes it.

The drop guard is the right approach. The only real caller throws away the returned id (dispatch.rs:158), so an explicit unsubscribe(id) would have nowhere to be called from. And it does fire where it matters, pump_subscription drops the stream on Unsubscribe, cancel and connection teardown.

Keeping the old cleanup in broadcast is also correct. The stream drops rx before the guard runs, so a broadcast can still see Closed while the entry is there. Removing twice is harmless and ids are never reused.

Two things before merge:

1. The doc comment on subscribe is now wrong. It still says the next matching broadcast prunes the entry.

2. dropped_stream_is_pruned (line 237) no longer tests anything. It broadcasts after dropping, but the guard already removed the entry, so the Closed branch in broadcast is now untested.

Also worth a code comment: Drop takes a DashMap lock now, so dropping a stream while holding a reference into subs would deadlock. Can't happen today, but it's easy to miss when editing broadcast later.

[solus161]

Thanks @solus161, I verified the bug on main (38dc500) and confirmed your PR fixes it.

The drop guard is the right approach. The only real caller throws away the returned id (dispatch.rs:158), so an explicit unsubscribe(id) would have nowhere to be called from. And it does fire where it matters, pump_subscription drops the stream on Unsubscribe, cancel and connection teardown.

Keeping the old cleanup in broadcast is also correct. The stream drops rx before the guard runs, so a broadcast can still see Closed while the entry is there. Removing twice is harmless and ids are never reused.

Two things before merge:

1. **The doc comment on `subscribe` is now wrong**. It still says the next matching broadcast prunes the entry.

2. **`dropped_stream_is_pruned` (line 237) no longer tests anything.** It broadcasts after dropping, but the guard already removed the entry, so the `Closed` branch in `broadcast` is now untested.

Also worth a code comment: Drop takes a DashMap lock now, so dropping a stream while holding a reference into subs would deadlock. Can't happen today, but it's easy to miss when editing broadcast later.

Really appreciate your comment. I just push a fix, docs and test for broadcast prunning behavior. Plz check.

[lxsaah]

@solus161 All three points addressed and I verified the new state locally: 7/7 tests pass, clippy/fmt/doc clean, CI green.

Only wording left, no code issues:

1. The new first line of the subscribe doc says it returns "a pair of stream, topic-tagged record-value updates, and its guard object". It returns (u64, BoxStream) — the guard isn't returned, it's captured inside the stream's state, which is the whole point. A reader would expect a 3-tuple. Maybe: "returns its id and the stream of topic-tagged updates; the stream owns a guard that removes the entry when dropped."

2. The Drop warning says broadcast "defers the Drop till the iter finished". broadcast never drops a guard — it defers its own subs.remove() calls. Same hazard, different mechanism. Suggest "defers its removals until after the iteration".

3. Typos: "This could server as a safety net to make sure nothing leak" → "serve" / "nothing leaks"; "keeps trying to lazily prunes" → "prune"; and in broadcast, "why trying to acquire write-lock" → "while".

LGTM once those are tidied.