feat(broadcast): add a bounded MPMC broadcast channel

CHANGELOG.md

@@ -25,12 +25,14 @@ All notable changes to this project will be documented in this file.
 
 ### New features
 
+* Implement `broadcast::mpmc::bounded`, a lossless bounded broadcast channel that retains at most the requested capacity and makes producers wait for the slowest active receiver.
 * Add opt-in bounded and unbounded `asyncband::mpmc` queues with cloneable producers and competing consumers, delivering each accepted value to exactly one receiver while a receiver remains.
 * Add an opt-in runtime-agnostic `Phaser` with shared observer handles, dynamic RAII participants registered individually or in batches through an owning iterator, `u64` phase numbers, split arrival/wait with cancellation-resilient retries, and a `close` operation that releases unfinished waits with `Closed`.
 * Add bounded MPSC `reserve` and `try_reserve` methods returning a `Permit`, allowing callers to wait for capacity before constructing a message; pending sends and reservations receive capacity in wait-queue order, and unused permits release capacity without claiming message order.
 
 ### Bug fixes
 
+* Complete semaphore permit releases and notify all eligible waiters even if a wake callback panics.
 * Release MPSC receiver wakers when the receiver is dropped, avoiding retained tasks and ownership cycles when a waker holds a sender.
 * Notify all blocked bounded MPSC senders on receiver disconnection even when a buffered message destructor panics.
 * Avoid deadlocks when a bounded MPSC sender's waker clone callback receives from the same channel.

README.md

@@ -102,7 +102,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon
 |                            | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/)                                    | `oneshot`      | Send one value from one sender to one receiver.                                                               |
 |                            | [`mpmc`](https://docs.rs/asyncband/*/asyncband/mpmc/)                                          | `mpmc`         | Distribute each value to exactly one of multiple competing receivers.      |
 |                            | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/)                                          | `mpsc`         | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue.      |
-|                            | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/)                                | `broadcast`    | Deliver every value to receivers active at send time; retain an unbounded backlog until each consumes or drops. |
+|                            | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/)                                | `broadcast`    | Deliver every value to active receivers with bounded backpressure or unbounded retention. |
 |                            | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/)                                        | `watch`        | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. |
 | Object reuse               | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/)                                          | `pool`         | Reuse objects through bounded or unbounded pool variants.                                                     |
 | Sync interop               | [`FutureExt`](https://docs.rs/asyncband/*/asyncband/blocking/trait.FutureExt.html)             | `blocking`     | Drive one runtime-agnostic future from a blocking thread.                                                     |

asyncband/src/broadcast/mpmc/bounded/tests.rs

@@ -0,0 +1,90 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// These run under Miri via `cargo x miri`, so they stay single-threaded and small. Behavior
+// reachable from the public API is covered in `tests-integration/broadcast_mpmc_bounded_test.rs`.
+
+use std::task::Waker;
+
+use super::*;
+
+#[test]
+#[should_panic(expected = "broadcast bounded channel requires capacity > 0")]
+fn bounded_panics_on_zero_capacity() {
+    let _ = bounded::<()>(0);
+}
+
+#[test]
+#[should_panic(expected = "broadcast channel version counter overflowed")]
+fn send_panics_on_version_overflow() {
+    // The receiver is dropped right away: the doctored counter would make its own drop overflow.
+    let (tx, _) = bounded(1);
+    tx.shared.inner.lock().log.set_tail(u64::MAX);
+    let _ = tx.try_send(());
+}
+
+#[test]
+fn buffer_is_preallocated_and_never_shrinks() {
+    let capacity = 128;
+    let (tx, mut rx) = bounded(capacity);
+    let allocated = tx.shared.inner.lock().log.buffer_capacity();
+    assert!(allocated >= capacity);
+
+    // Fill to capacity, drain completely, and repeat with a much smaller cycle. An elastic backlog
+    // would hand the allocation back after the small cycle; a fixed one must not.
+    for i in 0..capacity {
+        tx.try_send(i).unwrap();
+    }
+    for i in 0..capacity {
+        assert_eq!(rx.try_recv(), Ok(i));
+    }
+    tx.try_send(0).unwrap();
+    assert_eq!(rx.try_recv(), Ok(0));
+
+    assert_eq!(tx.retained_message_count(), 0);
+    assert_eq!(tx.shared.inner.lock().log.buffer_capacity(), allocated);
+}
+
+#[test]
+fn capacity_reports_the_requested_value() {
+    let (tx, _rx) = bounded::<i32>(3);
+    assert_eq!(tx.capacity(), 3);
+}
+
+#[test]
+fn a_large_reclaim_leaves_no_permit_slack() {
+    // Dropping a lagging subscription frees the whole backlog in one step, far more slots than the
+    // single parked producer can use. Permits beyond that producer would sit in the semaphore, and
+    // the next send to block would burn each one on a publish attempt that cannot succeed.
+    let capacity = 64;
+    let (tx, mut fast) = bounded(capacity);
+    let lagging = tx.subscribe();
+    for value in 0..capacity {
+        tx.try_send(value).unwrap();
+    }
+    for _ in 0..capacity {
+        fast.try_recv().unwrap();
+    }
+
+    let mut cx = Context::from_waker(Waker::noop());
+    let mut send = Box::pin(tx.send(capacity));
+    assert!(send.as_mut().poll(&mut cx).is_pending());
+
+    drop(lagging);
+    assert!(send.as_mut().poll(&mut cx).is_ready());
+    assert_eq!(tx.shared.tx_permits.available_permits(), 0);
+}