[BUG] Dropped WS subscriptions remain registered until a matching broadcast

[thaodt]

Describe the bug
ClientManager::subscribe inserts each subscription into ClientManager.subs. Dropping the returned stream closes its receiver, but does not remove the corresponding map entry.

Dead subscriptions are detected only inside ClientManager::broadcast, after the broadcast topic matches the subscription pattern and try_send returns Closed. Therefore, a dropped subscription for a quiet topic remains registered indefinitely unless a future broadcast matches that pattern.

This also affects subscriptions dropped during unsubscribe or connection teardown.

To Reproduce
The behavior can be reproduced with a focused unit test in aimdb-websocket-connector/src/server/client_manager.rs:

#[tokio::test]
async fn dropped_quiet_subscription_is_pruned_without_matching_traffic() {
    let mgr = ClientManager::new(256);
    let (_id, stream) = mgr.subscribe("quiet.topic");

    assert_eq!(mgr.subscription_count(), 1);
    drop(stream);

    // A non-matching broadcast does not inspect the closed sender.
    mgr.broadcast("other.topic", b"value").await;

    assert_eq!(mgr.subscription_count(), 0);
}

The final assertion currently fails with subscription_count() == 1. Broadcasting to quiet.topic afterward removes the entry, confirming that cleanup depends on matching traffic.

Expected behavior
Dropping a subscription stream should remove its entry from ClientManager.subs w/o requiring any future broadcast.

Unsubscribe and connection tear-down should therefore release all associated subscription state promptly, including subscriptions for topics that may never receive another update.

Environment

• OS: N/A - source-level and unit-test reproduction
• Rust Version: workspace toolchain
• AimDB Version: feat/retire-aimdb-ws-protocol at 81696c7659f44cbc6c988a112de8eaba76b9a8b7
• Also present on the current main branch, this is not a regression introduced by PR Feat/retire aimdb ws protocol #201

Additional context
Stale entries retain their pattern, sender and counters. Every later broadcast still scans these entries and evaluates topic_matches, so repeated subscribe/disconnect cycles on quiet or unique topics can grow both memory use and fan-out work over the server lifetime. The per-connection subscription limit does not bound this accumulated global state.

A possible fix is to return a stream with an RAII drop guard that removes its subscription ID from the shared map, or provide an explicit removal path used by unsubscribe and session teardown.

Please add a regression proving that dropping a stream removes the entry without requiring matching traffic.

[solus161]

Hi @thaodt Amidamaru, may I try this?

[thaodt]

Hi @thaodt Amidamaru, may I try this?

sure please go ahead