[shsms]

BroadcastChannel (new in 1.12) closed itself on the first send without receivers, and producers had no way to tell whether anyone was listening. This branch keeps the channel open in that case, like tokio's broadcast channel, fixes several gaps in closing, and adds BroadcastSender.has_receivers and BroadcastReceiver.has_senders.

Changes

• Closing a receiver now wakes a task that waits on it. Receivers wait on an Event that is replaced on each wake-up, instead of an asyncio.Condition. Waiters left on a closed event loop no longer make close(), send() or aclose() raise. Please look closely at _wake_receivers() and BroadcastReceiver.ready().
• Closing a receiver drops its buffered messages, so it is never ready, and select() no longer loops on it.
• A send() without receivers raises the new NoReceiversError (a SenderError) and drops the message, but leaves the channel open, so new receivers can still be subscribed. With resend_latest=True, the message is still kept, so a new receiver gets the real latest value. Only closing the senders closes the channel.
• A sender dropped without being closed keeps the channel open. The docs recommend contextlib.aclosing().
• clone() refuses a closed sender or channel. subscribe() uses the limit and warn_on_overflow the channel was created with as its defaults.
• BroadcastChannel no longer emits Broadcast's deprecation warning. The class is now private, and Broadcast is a deprecated alias of it, made with deprecated_aliases() from the new frequenz-core dependency. Its docstring is now short and points to BroadcastChannel.
• The experimental Pipe closes its sender when its receiver stops, and both ends when a send fails. With close_pipe_on_send_error=False, it drops the message after a failed send and keeps sending the messages that follow. stop() still leaves both open.
• Docs: the user guide, README, package overview and all docstring examples use BroadcastChannel. There is a new Oneshot page, the Anycast page is gone, and several false claims about closing are fixed.

Breaking

• An auto-closing channel no longer closes on a send() without receivers. send() raises NoReceiversError instead, and subscribe() keeps working.
• A Pipe now closes its sender when its receiver stops, and both ends when a send fails with a SenderError.
• BroadcastSender.clone() raises SenderClosedError or SenderError on a closed sender or channel.
• BroadcastSender.subscribe() takes its limit and warn_on_overflow defaults from the channel.
• BroadcastReceiver.close() now wakes waiting tasks, so it must be called from the thread running the event loop.
• The Broadcast deprecation warning fires when the name is imported or accessed, not on creation, and type checkers no longer report it as deprecated.
• This package now depends on frequenz-core >= 1.6.0, < 2, and needs typing-extensions >= 4.13.0.

[Marenz]

🤖 LGTM. This fixes real bugs, and the tests cover the hard cases (GC in another thread, closed loops, double detach). The release notes are accurate. ci_checks_max and pytest_min pass locally (152 tests).

Nits:

1. The PR description says "The channel still closes on the next send() without receivers, as in 1.12". The code does the opposite: send() raises NoReceiversError and keeps the channel open (_broadcast.py:499), as commit 8278c97 and the release notes say. Please fix the description and add this to the "Breaking" list (before, that send closed the channel and raised SenderError with a ChannelClosedError cause).
2. BroadcastReceiver.close() wakes receivers directly (_broadcast.py:756), while the sender finalizer uses call_soon_threadsafe(). With asyncio debug off, calling close() from another thread leaves a waiting task stuck without any error; the thread test only covers debug mode. A docstring note is probably enough, or route it through _wake_receivers_soon().
3. _sender_count -= 1 (_broadcast.py:182) runs in the finalizer, which can be any thread, and is not atomic. Worth a comment that only the wake-up is thread-safe.
4. _Broadcast is private, so there is no public type left to check ChannelClosedError.channel against. Maybe say so in the release notes.

[llucax]

My main concern here is I'm not convinced about the messing around with the GC, the rest is mostly minor. I will add another review/comment about the breaking changes and how we should ship this (v1 or v2).

I also did a AI review:

🤖 rAIview (AI review here!)

The lifecycle fixes and expanded documentation address several real problems. I found one remaining garbage-collection regression: the endpoint finalizers retain channels strongly, which can keep otherwise unreachable channels and cached payloads alive indefinitely. Both finalizer callbacks should refer to the channel weakly.

[llucax]

🟡⬇️ Avoid retaining channels strongly in endpoint finalizers

weakref.finalize(self, channel._remove_sender) stores a bound method, so the finalizer registry keeps channel alive. The receiver finalizer does the same with channel._remove_receiver. Although the channel stores its receivers weakly, it stores _latest strongly, even with the default resend_latest=False. A message can therefore create an indirect strong reference back to an endpoint, for example when two channels exchange reply senders, or when a message contains an object that owns an endpoint.

Once every application reference is dropped, the graph is still reachable through finalize registry → bound callback → channel → _latest → endpoint. This is not just a collectable reference cycle: the global registry prevents collection, so the finalizer that would release the graph never runs. Python's weakref.finalize documentation explicitly warns against callback references to the finalized object, including indirect references.

I reproduced this with two BroadcastChannel[object] instances that send each other's senders, drain both receiver buffers, and then drop all four endpoints. After gc.collect(), both senders remain alive on this PR, while all four endpoints are collected on the base branch. The receiver callback independently causes the same problem for a legacy Broadcast whose cached message references its receiver, even after its sender is explicitly closed.

Impact

Applications that repeatedly create and discard channels with endpoint-containing payloads can accumulate channels, finalizers, cached messages, and everything those messages reference until process exit or explicit cleanup. Draining the receiver buffers does not avoid the leak.

Related locations

• Receiver finalizer: this callback also needs to avoid retaining the channel strongly.
• Cached message assignment: every successful send stores the payload in _latest, regardless of resend_latest.

Suggested fix (⬇️ effort)

Use a callback that receives only a weakref.ref(channel), dereferences it once, and calls the removal method only if the channel is still alive. Apply this to both endpoint finalizers, without capturing a bound channel method in the callback or its arguments. Endpoints already keep their channel alive while application references to them exist. A runtime probe using weak channel callbacks collected both reproduced cycles and still woke a live receiver when its last sender was collected.

Suggested tests

Add collection tests for both finalizers, including a cross-channel cycle with drained buffers and a receiver cycle after explicit sender closure.

Suggested new tests

import gc
import weakref

from frequenz.channels import Broadcast, BroadcastChannel


async def test_unreachable_cross_channel_cycle_is_collected() -> None:
    sender1, receiver1 = BroadcastChannel[object](name="requests")
    sender2, receiver2 = BroadcastChannel[object](name="replies")
    refs = [weakref.ref(x) for x in (sender1, receiver1, sender2, receiver2)]

    await sender1.send(sender2)
    await sender2.send(sender1)
    await receiver1.receive()
    await receiver2.receive()
    del sender1, receiver1, sender2, receiver2
    gc.collect()

    assert all(ref() is None for ref in refs)


async def test_unreachable_receiver_cycle_is_collected() -> None:
    channel = Broadcast[object](name="receiver-cycle")
    sender = channel.new_sender()
    receiver = channel.new_receiver()
    ref = weakref.ref(receiver)

    await sender.send(receiver)
    await receiver.receive()
    await sender.aclose()
    del channel, sender, receiver
    gc.collect()

    assert ref() is None

---

🟡 Medium Severity: Probably best to fix before merging; skipping needs a good reason.

⬇️ Low Effort: A small callback change and focused regression tests.

[shsms]

Fixed by removing both finalizers, so nothing global holds the channel any more.

[llucax]

Nitpick: This was sorted alphabetically, I would keep the order. We could also just remove it from here, do don't need to promote deprecated stuff...

[shsms]

Dropped anycast

[llucax]

I'm not sure about this (and the "Auto-close broadcast channels when the last sender is dropped" commit in general). Is this supposed to only fix when someone forgets to call aclose() on the receiver, (a bug basically) or is this handling a valid use case?

Senders that were garbage-collected without aclose() ...

I can't see any cases where a sender could be garbage-collected without aclose() that is not a user forgetting to call aclose() (a bug).

The garbage collector can detach a sender in any thread, or in the
middle of Event.wait().

Why? How can the garbage collector detach anything?

If it is to catch a case where users forgot the aclose() I wouldn't go to the trouble of trying to clean stuff for them. All the weakref.finalize() business is usually very fragile, and I wouldn't be surprised if we introduce more subtle bugs with it, so it is basically extra complexity (storing the loop, calling call_soon_threadsafe(), using weakref.finalize(), having code running inside the GC, etc.) for nothing.

[shsms]

Damn it, it is for handling the user bug of forgetting to close, but it would have been very useful for that, especially with all the agentic coding and it is also get super hard to find out which receiver or sender wasn't closed, and we'll give up too easily instead of trying to debug properly.

But I agree that the implementation is buggy and looks like it can't be perfect, so I've dropped it.

[llucax]

Same for commit "Detach garbage-collected broadcast receivers at once".

Garbage-collected receivers stayed in the channel until the next send() swept them out. They now detach as soon as they are collected, so the channel doesn't keep dead receivers between sends.

Why do we need this? If an opened receiver is garbage collected, that's a user bug IMHO, they should close them before forgetting the last reference to them.

[shsms]

gone

[llucax]

What about num_senders (and num_receivers) instead? You can still use it as a bool but can also get the amount.

[shsms]

sender_count already exists, IIRC, that was the old way of identifying whether someone should wait on a receiver or not, don't remember.

Any way, num_senders would be weird when there's also sender_count.

Also, sender_count sends a value even if the receiver is closed, which might not be ideal, I guess we should actually deprecate that.

has_receivers etc gives enough info. I think exact counts is too much info, at least for the way we want to be using this field.

[llucax]

Nitpick: We could consider splitting into more files, like _broadcast_sender.py and _broadcast_receiver.py or such.

[shsms]

Fixed by trimming the docstring of the deprecated Broadcast, asking people to use BroadcastChannel instead.

[llucax]

I would remove this, the next repo-config will add the admonition automatically.

• Document deprecations with a Deprecated: admonition frequenz-repo-config-python#641

[shsms]

done

[llucax]

If we remove this file from the index, we should probably remove the file too.

[shsms]

done

[llucax]

If we stop handling garbage collected stuff, we should remove these (and all other gc mentions) too.

[shsms]

done

[llucax]

What? Why not:

Suggested change

	except SenderError as error:
	if isinstance(error, NoReceiversError):
	_logger.warning(
	"Pipe from [%s] to [%s] stopped: the sender has no receivers",
	self._receiver,
	self._sender,
	)
	else:
	_logger.exception(
	"Pipe from [%s] to [%s] stopped: sending failed",
	self._receiver,
	self._sender,
	)
	except NoReceiversError:
	_logger.warning(
	"Pipe from [%s] to [%s] stopped: the sender has no receivers",
	self._receiver,
	self._sender,
	)
	except SenderError:
	_logger.exception(
	"Pipe from [%s] to [%s] stopped: sending failed",
	self._receiver,
	self._sender,
	)

[shsms]

done

[llucax]

I checked several internal consumers to understand the practical compatibility impact. Most still use the legacy, non-auto-closing Broadcast, so these changes appear to affect us less than the breaking-change list initially suggests. Only a couple of the inspected consumers use Pipe, and their normal ownership and teardown patterns don’t look immediately broken by this change. That is reassuring for our current usage, but it doesn’t establish compatibility for other users.

I’d suggest the following release plan:

1. Release the backward-compatible fixes and additive APIs in 1.x, while preserving existing defaults and public contracts.
2. Make the new behavior for sending without receivers opt-in in 1.x. The existing behavior should remain the default; users should explicitly choose to raise NoReceiversError without closing the channel.
3. Optionally, make Pipe’s new endpoint-closing behavior opt-in in 1.x too. Closing caller-supplied endpoints changes ownership expectations, even if the current internal call sites appear compatible. However, given that Pipe is experimental, I’d also be comfortable making an explicit, documented exception and shipping this breaking change in 1.x.

For each compatibility option we introduce, I’d use this transition:

Release	Default and migration
1.x	Preserve the old behavior; allow opting into the new behavior.
2.x	Make the new behavior the default, but retain the compatibility option. Emit a DeprecationWarning when the old behavior is explicitly selected.
3.x	Remove the old behavior and the compatibility option.

Keeping the options in 2.x would let applications adopt the new behavior while still on 1.x and carry that configuration across the major-version upgrade. Once we agree on the approach, we can track their eventual removal in a separate 3.x breaking-changes issue.

If implementing and maintaining this compatibility would be too much work, I’d instead create a v2.x.x branch and target this PR there. We already have some accumulated deprecations, so we could also take the opportunity to remove deprecated APIs and do some cleanup for 2.0.0. That would make it a useful major release rather than introducing compatibility machinery just to keep these changes in 1.x.

Separately, if we already rely on Pipe in real applications, I think it’s probably time to promote it out of experimental. That should include clear documentation of endpoint ownership, normal completion, send failures, and cancellation, plus a migration path for the existing import. It could be a separate focused PR, but it seems worth doing alongside this lifecycle work.

[shsms]

Also added close_pipe_on_send_error: bool = True to pipe.