From e7a9e980c885448f99953a15ef29bbc67e2e0c51 Mon Sep 17 00:00:00 2001
From: Bee Klimt <bklimt@launchdarkly.com>
Date: Fri, 30 Oct 2026 14:35:32 -0700
Subject: [PATCH 1/6] fix: Let RemoveListener and Close unblock a broadcast to
 a full listener

---
 internal/broadcasters.go      | 39 ++++++++++++++++++---
 internal/broadcasters_test.go | 64 +++++++++++++++++++++++++++++++++++
 2 files changed, 98 insertions(+), 5 deletions(-)

diff --git a/internal/broadcasters.go b/internal/broadcasters.go
index 2af31c52..d7af4d71 100644
--- a/internal/broadcasters.go
+++ b/internal/broadcasters.go
@@ -17,16 +17,28 @@ import (
 const subscriberChannelBufferLength = 10
 
 // Broadcaster is our generalized implementation of broadcasters.
+//
+// Lock order: acquire releaseLock before lock.
 type Broadcaster[V any] struct {
+	// Reading subscribers requires lock (read or write) or releaseLock.
+	// Modifying subscribers requires both releaseLock and lock (write).
 	subscribers []channelPair[V]
-	lock        sync.RWMutex
+
+	// Sending to a subscriber requires lock (read).
+	lock sync.RWMutex
+
+	// Closing a subscriber's done channel requires releaseLock.
+	// Do not hold releaseLock while sending to a subscriber.
+	releaseLock sync.Mutex
 }
 
-// We need to keep track of both the channel we use for sending (stored as a reflect.Value, because Value
-// has methods for sending and closing), and also the channel for receiving.
+// We need to keep track of both the channel we use for sending and the channel for receiving.
 type channelPair[V any] struct {
 	sendCh    chan<- V
 	receiveCh <-chan V
+
+	// A send to sendCh must stop when done is closed.
+	done chan struct{}
 }
 
 // NewBroadcaster creates a Broadcaster that operates on the specified value type.
@@ -38,7 +50,9 @@ func NewBroadcaster[V any]() *Broadcaster[V] {
 func (b *Broadcaster[V]) AddListener() <-chan V {
 	ch := make(chan V, subscriberChannelBufferLength)
 	var receiveCh <-chan V = ch
-	chPair := channelPair[V]{sendCh: ch, receiveCh: receiveCh}
+	chPair := channelPair[V]{sendCh: ch, receiveCh: receiveCh, done: make(chan struct{})}
+	b.releaseLock.Lock()
+	defer b.releaseLock.Unlock()
 	b.lock.Lock()
 	defer b.lock.Unlock()
 	b.subscribers = append(b.subscribers, chPair)
@@ -48,6 +62,13 @@ func (b *Broadcaster[V]) AddListener() <-chan V {
 // RemoveListener removes a subscriber. The parameter is the same channel that was returned by
 // AddListener.
 func (b *Broadcaster[V]) RemoveListener(ch <-chan V) {
+	b.releaseLock.Lock()
+	defer b.releaseLock.Unlock()
+	for _, pair := range b.subscribers {
+		if pair.receiveCh == ch {
+			close(pair.done)
+		}
+	}
 	b.lock.Lock()
 	defer b.lock.Unlock()
 	b.subscribers = slices.DeleteFunc(b.subscribers, func(pair channelPair[V]) bool {
@@ -74,12 +95,20 @@ func (b *Broadcaster[V]) Broadcast(value V) {
 	b.lock.RLock()
 	defer b.lock.RUnlock()
 	for _, ch := range b.subscribers {
-		ch.sendCh <- value
+		select {
+		case ch.sendCh <- value:
+		case <-ch.done:
+		}
 	}
 }
 
 // Close closes all current subscriber channels.
 func (b *Broadcaster[V]) Close() {
+	b.releaseLock.Lock()
+	defer b.releaseLock.Unlock()
+	for _, s := range b.subscribers {
+		close(s.done)
+	}
 	b.lock.Lock()
 	defer b.lock.Unlock()
 	for _, s := range b.subscribers {
diff --git a/internal/broadcasters_test.go b/internal/broadcasters_test.go
index 9e7c7962..d50e3e4b 100644
--- a/internal/broadcasters_test.go
+++ b/internal/broadcasters_test.go
@@ -48,6 +48,28 @@ func testBroadcasterGenerically[V any](t *testing.T, broadcasterFactory func() *
 		})
 	})
 
+	t.Run("broadcast to full listener delivers every value", func(t *testing.T) {
+		withBroadcaster(t, func(b *Broadcaster[V]) {
+			ch := b.AddListener()
+			values := make([]V, subscriberChannelBufferLength+1)
+			for i := range values {
+				values[i] = valueFactory()
+			}
+
+			// Broadcast one more value than the channel can hold.
+			go func() {
+				for _, v := range values {
+					b.Broadcast(v)
+				}
+			}()
+
+			// The listener should receive every value in order.
+			for _, v := range values {
+				assert.Equal(t, v, th.RequireValue(t, ch, timeout))
+			}
+		})
+	})
+
 	t.Run("unregister subscriber", func(t *testing.T) {
 		withBroadcaster(t, func(b *Broadcaster[V]) {
 			ch1 := b.AddListener()
@@ -63,6 +85,48 @@ func testBroadcasterGenerically[V any](t *testing.T, broadcasterFactory func() *
 		})
 	})
 
+	unblockTests := []struct {
+		name    string
+		release func(b *Broadcaster[V], ch <-chan V)
+	}{
+		{"remove listener unblocks broadcast to full listener",
+			func(b *Broadcaster[V], ch <-chan V) { b.RemoveListener(ch) }},
+		{"close unblocks broadcast to full listener",
+			func(b *Broadcaster[V], _ <-chan V) { b.Close() }},
+	}
+	for _, tt := range unblockTests {
+		t.Run(tt.name, func(t *testing.T) {
+			withBroadcaster(t, func(b *Broadcaster[V]) {
+				ch := b.AddListener()
+				for i := 0; i < subscriberChannelBufferLength; i++ {
+					b.Broadcast(valueFactory())
+				}
+
+				broadcastDone := make(chan struct{})
+				go func() {
+					b.Broadcast(valueFactory())
+					close(broadcastDone)
+				}()
+				// Give Broadcast time to block on the full channel.
+				time.Sleep(10 * time.Millisecond)
+
+				// Release the listener without reading from it.
+				releaseDone := make(chan struct{})
+				go func() {
+					tt.release(b, ch)
+					close(releaseDone)
+				}()
+
+				// Both calls should return.
+				ok := th.AssertChannelClosed(t, broadcastDone, timeout) &&
+					th.AssertChannelClosed(t, releaseDone, timeout)
+				if !ok {
+					<-ch
+				}
+			})
+		})
+	}
+
 	t.Run("hasListeners", func(t *testing.T) {
 		withBroadcaster(t, func(b *Broadcaster[V]) {
 			assert.False(t, b.HasListeners())

From 622a5a6bc3bc00b5d8bd0b8c592037fab118e1de Mon Sep 17 00:00:00 2001
From: Bee Klimt <bklimt@launchdarkly.com>
Date: Fri, 30 Oct 2026 14:55:13 -0700
Subject: [PATCH 2/6] fix: Broadcast FDv2 data source status outside the status
 lock

---
 internal/datasystem/fdv2_datasystem.go      | 12 ++++++++---
 internal/datasystem/fdv2_datasystem_test.go | 24 +++++++++++++++++++++
 2 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go
index c8eda5ba..a0512fd8 100644
--- a/internal/datasystem/fdv2_datasystem.go
+++ b/internal/datasystem/fdv2_datasystem.go
@@ -718,6 +718,14 @@ func (f *FDv2) Offline() bool {
 
 //nolint:revive // DataSourceStatusReporter method.
 func (f *FDv2) UpdateStatus(state interfaces.DataSourceState, err interfaces.DataSourceErrorInfo) {
+	if status, changed := f.maybeUpdateStatus(state, err); changed {
+		f.broadcasters.dataSourceStatus.Broadcast(status)
+	}
+}
+
+func (f *FDv2) maybeUpdateStatus(
+	state interfaces.DataSourceState, err interfaces.DataSourceErrorInfo,
+) (interfaces.DataSourceStatus, bool) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
@@ -733,9 +741,7 @@ func (f *FDv2) UpdateStatus(state interfaces.DataSourceState, err interfaces.Dat
 		changed = true
 	}
 
-	if changed {
-		f.broadcasters.dataSourceStatus.Broadcast(f.status)
-	}
+	return f.status, changed
 }
 
 // restartStatusClock sets the time the current status has been held since to now.
diff --git a/internal/datasystem/fdv2_datasystem_test.go b/internal/datasystem/fdv2_datasystem_test.go
index 59a17c34..00bcd375 100644
--- a/internal/datasystem/fdv2_datasystem_test.go
+++ b/internal/datasystem/fdv2_datasystem_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	th "github.com/launchdarkly/go-test-helpers/v3"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/launchdarkly/go-server-sdk/v7/interfaces"
@@ -51,6 +52,29 @@ func TestFDv2WaitForReturnsAfterTimeoutWhileStatusUpdatesAreQueued(t *testing.T)
 	flooders.Wait()
 }
 
+func TestFDv2GetStatusReturnsWhileStatusListenerIsFull(t *testing.T) {
+	system := makeFallbackTestFDv2(t)
+	t.Cleanup(func() { _ = system.Stop() })
+	provider := &dataStatusProvider{system: system}
+	listener := provider.AddStatusListener()
+	t.Cleanup(func() { provider.RemoveStatusListener(listener) })
+
+	// Fill the listener's 10-value buffer, then block one more status update on it.
+	for i := 0; i < 10; i++ {
+		system.UpdateStatus(interfaces.DataSourceStateInterrupted, interfaces.DataSourceErrorInfo{StatusCode: 500 + i})
+	}
+	go system.UpdateStatus(interfaces.DataSourceStateInterrupted, interfaces.DataSourceErrorInfo{StatusCode: 510})
+	time.Sleep(10 * time.Millisecond)
+
+	// GetStatus should not wait for the listener.
+	returned := make(chan struct{})
+	go func() {
+		provider.GetStatus()
+		close(returned)
+	}()
+	th.AssertChannelClosed(t, returned, time.Second)
+}
+
 func TestFDv2SynchronizerSignalsReadinessOnlyForAFullPayload(t *testing.T) {
 	primary := newScriptedSynchronizer("primary")
 	system := makeSDKKeyTestFDv2(t, primary)

From a1849cd40978e0d855e34081b70d9bfc09b7dd2f Mon Sep 17 00:00:00 2001
From: Bee Klimt <bklimt@launchdarkly.com>
Date: Fri, 30 Oct 2026 14:59:59 -0700
Subject: [PATCH 3/6] fix: Do not broadcast Off when the FDv2 data system stops

---
 internal/datasystem/fdv2_datasystem.go      |  3 ++-
 internal/datasystem/fdv2_datasystem_test.go | 24 +++++++++++++++++++++
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go
index a0512fd8..8aff5e1a 100644
--- a/internal/datasystem/fdv2_datasystem.go
+++ b/internal/datasystem/fdv2_datasystem.go
@@ -645,7 +645,8 @@ func (f *FDv2) Stop() error {
 		f.cancel()
 	}
 	f.wg.Wait()
-	f.UpdateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{})
+	// Do not broadcast Off. Status listeners learn of shutdown when their channels close.
+	_, _ = f.maybeUpdateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{})
 	_ = f.store.Close()
 	f.broadcasters.Close()
 
diff --git a/internal/datasystem/fdv2_datasystem_test.go b/internal/datasystem/fdv2_datasystem_test.go
index 00bcd375..678cbf31 100644
--- a/internal/datasystem/fdv2_datasystem_test.go
+++ b/internal/datasystem/fdv2_datasystem_test.go
@@ -75,6 +75,30 @@ func TestFDv2GetStatusReturnsWhileStatusListenerIsFull(t *testing.T) {
 	th.AssertChannelClosed(t, returned, time.Second)
 }
 
+func TestFDv2StopReturnsWhileStatusListenerIsFull(t *testing.T) {
+	system := makeFallbackTestFDv2(t)
+	provider := &dataStatusProvider{system: system}
+	listener := provider.AddStatusListener()
+	t.Cleanup(func() { provider.RemoveStatusListener(listener) })
+
+	// Fill the listener's 10-value buffer.
+	for i := 0; i < 10; i++ {
+		system.UpdateStatus(interfaces.DataSourceStateInterrupted, interfaces.DataSourceErrorInfo{StatusCode: 500 + i})
+	}
+
+	// Stop without reading from the listener.
+	stopped := make(chan struct{})
+	go func() {
+		_ = system.Stop()
+		close(stopped)
+	}()
+
+	// Stop should return, and the status should be Off.
+	if th.AssertChannelClosed(t, stopped, time.Second) {
+		assert.Equal(t, interfaces.DataSourceStateOff, provider.GetStatus().State)
+	}
+}
+
 func TestFDv2SynchronizerSignalsReadinessOnlyForAFullPayload(t *testing.T) {
 	primary := newScriptedSynchronizer("primary")
 	system := makeSDKKeyTestFDv2(t, primary)

From 964c5a16790c5a69fc3dd7984ab621bee8b93b5b Mon Sep 17 00:00:00 2001
From: Bee Klimt <bklimt@launchdarkly.com>
Date: Fri, 30 Oct 2026 15:04:20 -0700
Subject: [PATCH 4/6] refactor: Remove drain workarounds that RemoveListener no
 longer needs

---
 internal/datasource/data_source_update_sink_impl.go | 10 +---------
 internal/datasystem/fdv2_datasystem.go              | 10 +---------
 2 files changed, 2 insertions(+), 18 deletions(-)

diff --git a/internal/datasource/data_source_update_sink_impl.go b/internal/datasource/data_source_update_sink_impl.go
index 66f01051..8fdc9c1f 100644
--- a/internal/datasource/data_source_update_sink_impl.go
+++ b/internal/datasource/data_source_update_sink_impl.go
@@ -216,15 +216,7 @@ func (d *DataSourceUpdateSinkImpl) waitFor(desiredState intf.DataSourceState, ti
 	}
 
 	statusCh := d.dataSourceStatusBroadcaster.AddListener()
-	defer func() {
-		// A broadcast that waits on this full channel blocks RemoveListener. Drain the
-		// channel until RemoveListener closes it, so that the removal always completes.
-		go func() {
-			for range statusCh {
-			}
-		}()
-		d.dataSourceStatusBroadcaster.RemoveListener(statusCh)
-	}()
+	defer d.dataSourceStatusBroadcaster.RemoveListener(statusCh)
 	d.lock.Unlock()
 
 	var deadline <-chan time.Time
diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go
index 8aff5e1a..33379ced 100644
--- a/internal/datasystem/fdv2_datasystem.go
+++ b/internal/datasystem/fdv2_datasystem.go
@@ -776,15 +776,7 @@ func (d *dataStatusProvider) RemoveStatusListener(listener <-chan interfaces.Dat
 
 func (d *dataStatusProvider) WaitFor(desiredState interfaces.DataSourceState, timeout time.Duration) bool {
 	ch := d.AddStatusListener()
-	defer func() {
-		// A broadcast that waits on this full channel blocks RemoveStatusListener. Drain the
-		// channel until the removal closes it, so that the removal always completes.
-		go func() {
-			for range ch {
-			}
-		}()
-		d.RemoveStatusListener(ch)
-	}()
+	defer d.RemoveStatusListener(ch)
 
 	switch d.system.getStatus().State {
 	case desiredState:

From cb21f509e82a4437a05ec5774bf9ce6e04b42d6b Mon Sep 17 00:00:00 2001
From: Bee Klimt <bklimt@launchdarkly.com>
Date: Fri, 30 Oct 2026 16:05:39 -0700
Subject: [PATCH 5/6] fix: Return from WaitFor when the status listener channel
 closes

---
 .../data_source_status_provider_impl_test.go    | 13 +++++++++++++
 .../datasource/data_source_update_sink_impl.go  |  5 ++++-
 internal/datasystem/fdv2_datasystem.go          |  5 ++++-
 internal/datasystem/fdv2_datasystem_test.go     | 17 +++++++++++++++++
 4 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/internal/datasource/data_source_status_provider_impl_test.go b/internal/datasource/data_source_status_provider_impl_test.go
index 14a3d25b..c7356872 100644
--- a/internal/datasource/data_source_status_provider_impl_test.go
+++ b/internal/datasource/data_source_status_provider_impl_test.go
@@ -168,5 +168,18 @@ func TestDataSourceStatusProviderImpl(t *testing.T) {
 				assert.True(t, time.Now().Sub(timeStart) < 500*time.Millisecond)
 			})
 		})
+
+		t.Run("ends if status broadcaster is closed", func(t *testing.T) {
+			dataSourceStatusProviderImplTest(func(p dataSourceStatusProviderImplTestParams) {
+				go func() {
+					<-time.After(100 * time.Millisecond)
+					p.dataSourceUpdates.dataSourceStatusBroadcaster.Close()
+				}()
+				timeStart := time.Now()
+				success := p.dataSourceStatusProvider.WaitFor(interfaces.DataSourceStateValid, 500*time.Millisecond)
+				assert.False(t, success)
+				assert.True(t, time.Now().Sub(timeStart) < 500*time.Millisecond)
+			})
+		})
 	})
 }
diff --git a/internal/datasource/data_source_update_sink_impl.go b/internal/datasource/data_source_update_sink_impl.go
index 8fdc9c1f..bc3b8509 100644
--- a/internal/datasource/data_source_update_sink_impl.go
+++ b/internal/datasource/data_source_update_sink_impl.go
@@ -226,7 +226,10 @@ func (d *DataSourceUpdateSinkImpl) waitFor(desiredState intf.DataSourceState, ti
 
 	for {
 		select {
-		case newStatus := <-statusCh:
+		case newStatus, ok := <-statusCh:
+			if !ok {
+				return false
+			}
 			if newStatus.State == desiredState {
 				return true
 			}
diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go
index 33379ced..d8a438bd 100644
--- a/internal/datasystem/fdv2_datasystem.go
+++ b/internal/datasystem/fdv2_datasystem.go
@@ -789,7 +789,10 @@ func (d *dataStatusProvider) WaitFor(desiredState interfaces.DataSourceState, ti
 
 	for {
 		select {
-		case status := <-ch:
+		case status, ok := <-ch:
+			if !ok {
+				return false
+			}
 			if status.State == desiredState {
 				return true
 			}
diff --git a/internal/datasystem/fdv2_datasystem_test.go b/internal/datasystem/fdv2_datasystem_test.go
index 678cbf31..631d1244 100644
--- a/internal/datasystem/fdv2_datasystem_test.go
+++ b/internal/datasystem/fdv2_datasystem_test.go
@@ -52,6 +52,23 @@ func TestFDv2WaitForReturnsAfterTimeoutWhileStatusUpdatesAreQueued(t *testing.T)
 	flooders.Wait()
 }
 
+func TestFDv2WaitForEndsWhenStopped(t *testing.T) {
+	system := makeFallbackTestFDv2(t)
+	provider := &dataStatusProvider{system: system}
+	go func() {
+		<-time.After(100 * time.Millisecond)
+		_ = system.Stop()
+	}()
+
+	// Wait for a state that never arrives.
+	timeStart := time.Now()
+	success := provider.WaitFor(interfaces.DataSourceStateValid, 500*time.Millisecond)
+
+	// WaitFor should return false when the system stops, before the timeout.
+	assert.False(t, success)
+	assert.Less(t, time.Since(timeStart), 500*time.Millisecond)
+}
+
 func TestFDv2GetStatusReturnsWhileStatusListenerIsFull(t *testing.T) {
 	system := makeFallbackTestFDv2(t)
 	t.Cleanup(func() { _ = system.Stop() })

From c1e6e39b46a814cfca276f3bc0d7fe4080394dfd Mon Sep 17 00:00:00 2001
From: Bee Klimt <bklimt@launchdarkly.com>
Date: Mon, 2 Nov 2026 13:57:14 -0800
Subject: [PATCH 6/6] fix: Hold no broadcaster lock while sending to a listener

---
 internal/broadcasters.go      | 89 +++++++++++++++++++----------------
 internal/broadcasters_test.go | 47 ++++++++++++++++++
 2 files changed, 95 insertions(+), 41 deletions(-)

diff --git a/internal/broadcasters.go b/internal/broadcasters.go
index d7af4d71..cde0056b 100644
--- a/internal/broadcasters.go
+++ b/internal/broadcasters.go
@@ -17,19 +17,10 @@ import (
 const subscriberChannelBufferLength = 10
 
 // Broadcaster is our generalized implementation of broadcasters.
-//
-// Lock order: acquire releaseLock before lock.
 type Broadcaster[V any] struct {
-	// Reading subscribers requires lock (read or write) or releaseLock.
-	// Modifying subscribers requires both releaseLock and lock (write).
-	subscribers []channelPair[V]
-
-	// Sending to a subscriber requires lock (read).
-	lock sync.RWMutex
-
-	// Closing a subscriber's done channel requires releaseLock.
-	// Do not hold releaseLock while sending to a subscriber.
-	releaseLock sync.Mutex
+	// Guarded by lock. Do not hold lock while sending to a subscriber.
+	subscribers []*channelPair[V]
+	lock        sync.RWMutex
 }
 
 // We need to keep track of both the channel we use for sending and the channel for receiving.
@@ -37,8 +28,35 @@ type channelPair[V any] struct {
 	sendCh    chan<- V
 	receiveCh <-chan V
 
-	// A send to sendCh must stop when done is closed.
+	// Closed when the subscriber is removed. A send to sendCh must stop when done is closed.
 	done chan struct{}
+
+	// Sending to sendCh requires lock (read). Closing sendCh requires lock (write), after done is closed.
+	lock sync.RWMutex
+}
+
+// send sends value to sendCh, unless the subscriber is removed first.
+func (p *channelPair[V]) send(value V) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	// Check that it's not done first, so we don't send on a closed channel.
+	select {
+	case <-p.done:
+		return
+	default:
+	}
+	select {
+	case p.sendCh <- value:
+	case <-p.done:
+	}
+}
+
+// close stops any send in progress and closes sendCh. Call it once per subscriber.
+func (p *channelPair[V]) close() {
+	close(p.done)
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	close(p.sendCh)
 }
 
 // NewBroadcaster creates a Broadcaster that operates on the specified value type.
@@ -50,9 +68,7 @@ func NewBroadcaster[V any]() *Broadcaster[V] {
 func (b *Broadcaster[V]) AddListener() <-chan V {
 	ch := make(chan V, subscriberChannelBufferLength)
 	var receiveCh <-chan V = ch
-	chPair := channelPair[V]{sendCh: ch, receiveCh: receiveCh, done: make(chan struct{})}
-	b.releaseLock.Lock()
-	defer b.releaseLock.Unlock()
+	chPair := &channelPair[V]{sendCh: ch, receiveCh: receiveCh, done: make(chan struct{})}
 	b.lock.Lock()
 	defer b.lock.Unlock()
 	b.subscribers = append(b.subscribers, chPair)
@@ -62,25 +78,22 @@ func (b *Broadcaster[V]) AddListener() <-chan V {
 // RemoveListener removes a subscriber. The parameter is the same channel that was returned by
 // AddListener.
 func (b *Broadcaster[V]) RemoveListener(ch <-chan V) {
-	b.releaseLock.Lock()
-	defer b.releaseLock.Unlock()
-	for _, pair := range b.subscribers {
-		if pair.receiveCh == ch {
-			close(pair.done)
-		}
-	}
+	var removed []*channelPair[V]
 	b.lock.Lock()
-	defer b.lock.Unlock()
-	b.subscribers = slices.DeleteFunc(b.subscribers, func(pair channelPair[V]) bool {
+	b.subscribers = slices.DeleteFunc(b.subscribers, func(pair *channelPair[V]) bool {
 		// The following equality test is the reason why we have to store both the sendCh (chan X) and
 		// the receiveCh (<-chan X) for each subscriber; "s.sendCh == ch" would not be true because
 		// they're of two different types.
 		if pair.receiveCh == ch {
-			close(pair.sendCh)
+			removed = append(removed, pair)
 			return true
 		}
 		return false
 	})
+	b.lock.Unlock()
+	for _, pair := range removed {
+		pair.close()
+	}
 }
 
 // HasListeners returns true if there are any current subscribers.
@@ -93,26 +106,20 @@ func (b *Broadcaster[V]) HasListeners() bool {
 // Broadcast broadcasts a value to all current subscribers.
 func (b *Broadcaster[V]) Broadcast(value V) {
 	b.lock.RLock()
-	defer b.lock.RUnlock()
-	for _, ch := range b.subscribers {
-		select {
-		case ch.sendCh <- value:
-		case <-ch.done:
-		}
+	subscribers := slices.Clone(b.subscribers)
+	b.lock.RUnlock()
+	for _, pair := range subscribers {
+		pair.send(value)
 	}
 }
 
 // Close closes all current subscriber channels.
 func (b *Broadcaster[V]) Close() {
-	b.releaseLock.Lock()
-	defer b.releaseLock.Unlock()
-	for _, s := range b.subscribers {
-		close(s.done)
-	}
 	b.lock.Lock()
-	defer b.lock.Unlock()
-	for _, s := range b.subscribers {
-		close(s.sendCh)
-	}
+	subscribers := b.subscribers
 	b.subscribers = nil
+	b.lock.Unlock()
+	for _, pair := range subscribers {
+		pair.close()
+	}
 }
diff --git a/internal/broadcasters_test.go b/internal/broadcasters_test.go
index d50e3e4b..bd541d5c 100644
--- a/internal/broadcasters_test.go
+++ b/internal/broadcasters_test.go
@@ -127,6 +127,53 @@ func testBroadcasterGenerically[V any](t *testing.T, broadcasterFactory func() *
 		})
 	}
 
+	concurrentCallTests := []struct {
+		name string
+		call func(b *Broadcaster[V], other <-chan V)
+	}{
+		{"close unblocks broadcast to full listener while a listener is added",
+			func(b *Broadcaster[V], _ <-chan V) { b.AddListener() }},
+		{"close unblocks broadcast to full listener while another listener is removed",
+			func(b *Broadcaster[V], other <-chan V) { b.RemoveListener(other) }},
+	}
+	for _, tt := range concurrentCallTests {
+		t.Run(tt.name, func(t *testing.T) {
+			withBroadcaster(t, func(b *Broadcaster[V]) {
+				other := b.AddListener()
+				ch := b.AddListener()
+				for i := 0; i < subscriberChannelBufferLength; i++ {
+					b.Broadcast(valueFactory())
+				}
+
+				go b.Broadcast(valueFactory())
+				// Give Broadcast time to block on a full channel.
+				time.Sleep(10 * time.Millisecond)
+				go tt.call(b, other)
+				// Give the call time to start.
+				time.Sleep(10 * time.Millisecond)
+
+				// Close without reading from either listener.
+				closeDone := make(chan struct{})
+				go func() {
+					b.Close()
+					close(closeDone)
+				}()
+
+				// Close should return.
+				if !th.AssertChannelClosed(t, closeDone, timeout) {
+					go func() {
+						for range ch {
+						}
+					}()
+					go func() {
+						for range other {
+						}
+					}()
+				}
+			})
+		})
+	}
+
 	t.Run("hasListeners", func(t *testing.T) {
 		withBroadcaster(t, func(b *Broadcaster[V]) {
 			assert.False(t, b.HasListeners())