fix(channels): make subscriptions lifecycle-safe

[cofin]

Description

ChannelsPlugin updates subscriber state and backend subscriptions without shared lifecycle synchronization. Concurrent transitions can unsubscribe active subscribers, while failures can leak subscribers and dynamic channel entries.

This change:

• validates and materializes channel inputs before mutation
• serializes subscriber bookkeeping and backend transitions
• registers subscribers before backend subscription and history retrieval
• rolls back state after errors or cancellation
• keeps history retrieval and subscriber shutdown outside the lock
• removes empty dynamic channels while retaining configured channels
• makes repeated unsubscribe operations safe
• subscribes generated WebSockets before accepting the connection

This consolidates the applicable fixes from #4895, #4872, and #4868 while crediting their authors.

Closes #4894
Closes #4871
Closes #4867

The Redis stream cursor behavior noted in #4867 remains unchanged.

---

📚 Documentation preview 📚: https://litestar-org.github.io/litestar-docs-preview/4969

[codecov]

Codecov Report

❌ Patch coverage is 96.15385% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.45%. Comparing base (733d72c) to head (eabbd47).
⚠️ Report is 19 commits behind head on main.

Files with missing lines	Patch %	Lines
litestar/channels/plugin.py	96.15%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4969      +/-   ##
==========================================
+ Coverage   67.32%   67.45%   +0.12%     
==========================================
  Files         293      293              
  Lines       15246    15328      +82     
  Branches     1728     1742      +14     
==========================================
+ Hits        10265    10339      +74     
- Misses       4834     4839       +5     
- Partials      147      150       +3     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Vitaly312]

If the backend is unreachable (e.g. network failure):

1. subscribe for the first subscriber fails, its except block triggers unsubscribe, then _remove_subscriber
2. _remove_subscriber fails, self._channels.setdefault(channel, set()).add(subscriber) in its except block returns the subscriber into the local state
3. So this call of subscribe will end by an exception but a subscriber stays in the local state, and next subscriptions will not trigger backend subscription. So, all subsequent events will not be published since the plugin is unsubscribed

I think it's better to just move backend calls before state mutating instead of compensating state mutation on exceptions, it will simplify code.

[cofin]

@Vitaly312 I agree. I'll make the necessary adjustments.

[Vitaly312]

@cofin Thanks! Actually I opened a PR cofin#8 into this branch with these adjustments (since it's too big to paste as code suggestions). Feel free to merge or adapt it if helpful!

[provinzkraut]

Can we get a more detailed title and description here? It says what was done mechanically, but doesn't really explain the bug that was fixed

[Vitaly312]

History should be fetched after the state mutation. Otherwise, if the event was published after history fetching but before adding a corresponding subscriber in the self._channels (e.g. when backend.subscribe suspends), this event will be lost since it isn't in the history and _sub_worker will ignore it because the subscriber isn't in the self._channels. I think history should be fetched in the end like it was before (missed that in my PR), but it should roll back if failed:

...
if history:
    try:
        await self.put_subscriber_history(subscriber=subscriber, limit=history, channels=channels)
    except:
        await self.unsubscribe(subscriber=subscriber, channels=channels)
        raise
return subscriber

But this approach requires some deduplication. I mean, if some event was published after the plugin puts a subscriber in the state, and before history fetching, this event will be consumed twice. I guess this requires changing abstract ChannelsBackend.stream_events signature to provide some timestamp/ID for each event to distinguish between identical events and duplicates (or just change subscribe signature to not handle history fetching at all/require that consumers must be idempotent).