feat: Broadcast Off to FDv2 status listeners when the data system stops

internal/broadcasters.go

@@ -51,11 +51,18 @@ func (p *channelPair[V]) send(value V) {
 	}
 }
 
-// close stops any send in progress and closes sendCh. Call it once per subscriber.
-func (p *channelPair[V]) close() {
+// close stops any send in progress and closes sendCh. If final is not nil, close first sends *final to
+// sendCh if it has room. Call it once per subscriber.
+func (p *channelPair[V]) close(final *V) {
 	close(p.done)
 	p.lock.Lock()
 	defer p.lock.Unlock()
+	if final != nil {
+		select {
+		case p.sendCh <- *final:
+		default:
+		}
+	}
 	close(p.sendCh)
 }
 
@@ -92,7 +99,7 @@ func (b *Broadcaster[V]) RemoveListener(ch <-chan V) {
 	})
 	b.lock.Unlock()
 	for _, pair := range removed {
-		pair.close()
+		pair.close(nil)
 	}
 }
 
@@ -115,11 +122,21 @@ func (b *Broadcaster[V]) Broadcast(value V) {
 
 // Close closes all current subscriber channels.
 func (b *Broadcaster[V]) Close() {
+	b.closeAll(nil)
+}
+
+// CloseWith sends value to each current subscriber that has room for it, then closes all current subscriber
+// channels. It does not wait for a subscriber to read.
+func (b *Broadcaster[V]) CloseWith(value V) {
+	b.closeAll(&value)
+}
+
+func (b *Broadcaster[V]) closeAll(final *V) {
 	b.lock.Lock()
 	subscribers := b.subscribers
 	b.subscribers = nil
 	b.lock.Unlock()
 	for _, pair := range subscribers {
-		pair.close()
+		pair.close(final)
 	}
 }

internal/broadcasters_test.go

@@ -174,6 +174,43 @@ func testBroadcasterGenerically[V any](t *testing.T, broadcasterFactory func() *
 		})
 	}
 
+	t.Run("close with value delivers it to a listener with room", func(t *testing.T) {
+		withBroadcaster(t, func(b *Broadcaster[V]) {
+			ch := b.AddListener()
+
+			// Close with a final value.
+			value := valueFactory()
+			b.CloseWith(value)
+
+			// The listener should receive the value, and then the channel should close.
+			assert.Equal(t, value, th.RequireValue(t, ch, timeout))
+			th.AssertChannelClosed(t, ch, timeout)
+		})
+	})
+
+	t.Run("close with value does not wait for a full listener", func(t *testing.T) {
+		withBroadcaster(t, func(b *Broadcaster[V]) {
+			ch := b.AddListener()
+			for i := 0; i < subscriberChannelBufferLength; i++ {
+				b.Broadcast(valueFactory())
+			}
+
+			// Close with a final value without reading from the listener.
+			closeDone := make(chan struct{})
+			go func() {
+				b.CloseWith(valueFactory())
+				close(closeDone)
+			}()
+
+			// CloseWith should return, and the listener should get only the values that fit.
+			th.AssertChannelClosed(t, closeDone, timeout)
+			for i := 0; i < subscriberChannelBufferLength; i++ {
+				th.RequireValue(t, ch, timeout)
+			}
+			th.AssertChannelClosed(t, ch, timeout)
+		})
+	})
+
 	t.Run("hasListeners", func(t *testing.T) {
 		withBroadcaster(t, func(b *Broadcaster[V]) {
 			assert.False(t, b.HasListeners())

internal/datasystem/fdv2_datasystem.go

@@ -27,8 +27,9 @@ type broadcasters struct {
 	changeSetBroadcaster *internal.Broadcaster[subsystems.ChangeSet]
 }
 
-func (b *broadcasters) Close() {
-	b.dataSourceStatus.Close()
+// Close sends status to each data source status listener that has room for it, then closes every broadcaster.
+func (b *broadcasters) Close(status interfaces.DataSourceStatus) {
+	b.dataSourceStatus.CloseWith(status)
 	b.dataStoreStatus.Close()
 	b.flagChangeEvent.Close()
 	b.changeSetBroadcaster.Close()
@@ -682,10 +683,9 @@ func (f *FDv2) Stop() error {
 		f.cancel()
 	}
 	f.wg.Wait()
-	// Do not broadcast Off. Status listeners learn of shutdown when their channels close.
-	_, _ = f.maybeUpdateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{})
+	status, _ := f.maybeUpdateStatus(interfaces.DataSourceStateOff, interfaces.DataSourceErrorInfo{})
 	_ = f.store.Close()
-	f.broadcasters.Close()
+	f.broadcasters.Close(status)
 
 	return nil
 }

internal/datasystem/fdv2_datasystem_test.go

@@ -7,6 +7,7 @@ import (
 
 	th "github.com/launchdarkly/go-test-helpers/v3"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/launchdarkly/go-server-sdk/v7/interfaces"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
@@ -69,6 +70,34 @@ func TestFDv2WaitForEndsWhenStopped(t *testing.T) {
 	assert.Less(t, time.Since(timeStart), 500*time.Millisecond)
 }
 
+func TestFDv2StatusListenerReceivesOffWhenStopped(t *testing.T) {
+	system := makeFallbackTestFDv2(t)
+	provider := &dataStatusProvider{system: system}
+	listener := provider.AddStatusListener()
+
+	// Stop the system.
+	require.NoError(t, system.Stop())
+
+	// The listener should receive Off, and then its channel should close.
+	assert.Equal(t, interfaces.DataSourceStateOff, th.RequireValue(t, listener, time.Second).State)
+	th.AssertChannelClosed(t, listener, time.Second)
+}
+
+func TestFDv2WaitForOffReturnsTrueWhenStopped(t *testing.T) {
+	system := makeFallbackTestFDv2(t)
+	provider := &dataStatusProvider{system: system}
+	go func() {
+		<-time.After(100 * time.Millisecond)
+		_ = system.Stop()
+	}()
+
+	// Wait for Off while the system stops.
+	success := provider.WaitFor(interfaces.DataSourceStateOff, time.Second)
+
+	// WaitFor should see Off.
+	assert.True(t, success)
+}
+
 func TestFDv2GetStatusReturnsWhileStatusListenerIsFull(t *testing.T) {
 	system := makeFallbackTestFDv2(t)
 	t.Cleanup(func() { _ = system.Stop() })