Auto closing broadcast

[shsms]

Pull request overview

This PR introduces a new auto-closing broadcast channel API (BroadcastChannel) and deprecates the legacy Broadcast/Anycast channel classes, aiming to make broadcast channels close automatically when one side (senders or receivers) is fully gone/closed.

Changes:

• Add BroadcastChannel (tuple-style constructor returning an initial BroadcastSender/BroadcastReceiver) and rename broadcast endpoint classes to BroadcastSender/BroadcastReceiver.
• Add new sender capability interfaces (SubscribableSender, ClonableSender, ClonableSubscribableSender) and export them publicly.
• Deprecate Broadcast and Anycast, update release notes, and add tests covering broadcast auto-close behavior.

[shsms]

Based on the one shot PR, which also needs more docs and examples.

cc: @simonvoelcker

[Copilot]

Pull request overview

This PR introduces a new auto-closing broadcast channel API (BroadcastChannel) and deprecates the legacy Broadcast/Anycast channel classes, aiming to make broadcast channels close automatically when one side (senders or receivers) is fully gone/closed.

Changes:

• Add BroadcastChannel (tuple-style constructor returning an initial BroadcastSender/BroadcastReceiver) and rename broadcast endpoint classes to BroadcastSender/BroadcastReceiver.
• Add new sender capability interfaces (SubscribableSender, ClonableSender, ClonableSubscribableSender) and export them publicly.
• Deprecate Broadcast and Anycast, update release notes, and add tests covering broadcast auto-close behavior.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 11 comments.

Show a summary per file

File	Description
tests/test_broadcast.py	Updates overflow test type assertions and adds auto-close tests for BroadcastChannel.
src/frequenz/channels/_sender.py	Adds new sender capability ABCs (subscribe/clone) for use by broadcast senders.
src/frequenz/channels/_broadcast.py	Implements auto-close mechanics, introduces BroadcastChannel, and deprecates Broadcast.
src/frequenz/channels/_anycast.py	Deprecates Anycast.
src/frequenz/channels/init.py	Exports new broadcast types and sender capability interfaces.
RELEASE_NOTES.md	Documents the new channel APIs and deprecations.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[llucax]

I think there might be a few bugs left. If those are real bugs, it would be good to add test cases for them too.

[llucax]

This can probably fail because when this instance is deleted by the GC because there is no guarantee in order, self._channel might be already destroyed when this runs. I think we should probably leave the __del__ out and required users to properly close receivers explicitly/using them as context managers. Also not calling the channel's aclose() could still leave things in a bad state.

[shsms]

I see, that's why copilot was suggesting getattr instead I guess. It also recommended wrapping the whole thing under try:except Exception:pass, which I did. So it might work or might not, but won't make things worse, right?

aclose() on channel wakes up all receivers, so they can stop waiting. If all senders are gone, we call channel.aclose():

frequenz-channels-python/src/frequenz/channels/_broadcast.py

Lines 415 to 419 in ec05bfb

	if (
	self._channel._sender_count == 0 # pylint: disable=protected-access
	and self._channel._auto_close_enabled # pylint: disable=protected-access
	):
	await self._channel.aclose()

There's no other place where we need to call channel.aclose(). Not when all the receivers have gone.

[llucax]

This seems to be a real problem. The __del__() can still end up in blocked receivers:

This times out after a second:

import gc

async def test_broadcast_receive_unblocks_when_last_sender_is_garbage_collected() -> None:
    """A blocked receiver should stop promptly when the last sender disappears."""
    sender, receiver = BroadcastChannel[int](name="auto-close-test")
    sender_2 = sender.clone()

    await sender.aclose()

    receive_task = asyncio.create_task(receiver.receive())
    await asyncio.sleep(0)

    # Adding await sender_2.aclose() here fixes it, of course
    del sender_2
    gc.collect()

    with pytest.raises(ReceiverStoppedError):
        async with asyncio.timeout(1):
            await receive_task

I would remove the __del__, it doesn't really help much and it is wrong. You should never access other mutable objects in the __del__ anyway, it is intended only to free resources like from external C libraries or stuff like that, they are not proper destructors like in other languages).

[shsms]

ok, removed.

[llucax]

Shouldn't this also trigger a closing of the channel if this is the last receiver? I guess to make this happen we need to transition to aclose() for receivers, right?

[shsms]

No, the closing of the channel is really a closing of the receivers. But if none are left, there's nothing for the channel close to do.

We might still consider aclose for receivers, but separate problem I guess.

[llucax]

So, the "issue" here is, if we don't close the channel (don't mark it as closed), a new subscribe() can "resurrect" the channel. I'm not sure if this is intended behavior, it might be, but if it is we should document it more clearly. I'm not sure if it is a good idea though, because if we do this it means once you got SenderClosedError for a sender, a new send() might succeed, which sounds weird to me.

Suggested new test (that doesn't allow for resurrection, could be adapted if we want to allow it)

async def test_broadcast_auto_close_cannot_be_resurrected_by_new_subscription() -> None:
    """Closing the last receiver should keep the channel closed."""
    sender, receiver = BroadcastChannel[int](name="auto-close-test")

    receiver.close()

    with pytest.raises(SenderError) as excinfo:
        await sender.send(1)
    assert isinstance(excinfo.value.__cause__, ChannelClosedError)

    late_receiver = sender.subscribe()  # <--- This should probably raise SenderError/ChannelClosedError

    with pytest.raises(SenderError) as excinfo:
        await sender.send(2)  # <--- This doesn't raise now
    assert isinstance(excinfo.value.__cause__, ChannelClosedError)

[shsms]

Fixed. Also interesting that we allowed new_receiver calls from closed channels. Won't be relevant soon though.

[llucax]

Shouldn't self._channel._latest = message and recv._enqueue(message)be done after all the checks are done (right before we notify all receivers)? Otherwise it might be we report a send failed, but still receivers will receive the message.

[shsms]

If self.channel.receivers is already empty, there won't be a enqueue call. There would just be a sender error.

If self.channels.receivers has only stale receivers, they would get cleaned up first, enqueue will not be called. Then there would be a sender error.

If there are active receivers, enqueue will be called on them. They are not stale, so they will stay. No sender error will be raised.

The sendererror happens below, just to make sure we've dropped stale receivers first.

[llucax]

OK, so it is only about _latest. It seems updating _latest doesn't have a visible effect. This works as expected:

async def test_broadcast_failed_send_does_not_update_latest_message() -> None:
    """A send that fails because the channel is closed should not mutate _latest."""
    sender, receiver = BroadcastChannel[int](
        name="auto-close-test",
        resend_latest=True,
    )
    await sender.send(1)
    await sender.aclose()
    with pytest.raises(SenderError):
        await sender.send(2)
    assert await receiver.receive() == 1

That said, it is weird to me that _latest is updated even if the send failed. Unless there is a very good reason to not move it to after it is really sent, I don't see why we shouldn't do it:

async def test_broadcast_failed_send_does_not_update_latest_message() -> None:
    """A send that fails because the channel is closed should not mutate _latest."""
    sender, receiver = BroadcastChannel[int](
        name="auto-close-test",
        resend_latest=True,
    )
    await sender.send(1)
    assert await receiver.receive() == 1
    receiver.close()

    with pytest.raises(SenderError):
        await sender.send(2)

    assert sender._channel._latest == 1  # <--- fails because it was updated to 2 anyways

[llucax]

Unrelated to this, but I just noticed sometimes we raise SenderClosedError (sender._closed is True) and sometimes SenderError from ChannelClosedError (sender._channel._closed is True). I wonder if we should use SenderClosedError from ChannelClosedError instead, it sounds like if a channel is closed the sender should be closed too. I guess it relates to the question if senders/channels can be resurrected by re-subscribing or not.

That said, I guess this comes from before, and we are doing it like this in other places, so I'm OK to leave the broader question for another time.

[shsms]

I don't see why we shouldn't do it

I wasn't objecting, I just wasn't aware of the issue. I've moved the line updating _latest to after the raise.

[shsms]

Ready again @llucax