Improved perfs

internal/broker/broker.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"log/slog"
 	"sync"
+	"time"
 
 	"github.com/uopi/uopi/internal/datasource"
 )
@@ -41,18 +42,38 @@ type Broker struct {
 	sources map[string]datasource.DataSource
 	subs    map[SignalRef]*signalSub
 
-	log *slog.Logger
+	maxInterval time.Duration // 0 = unlimited
+	log         *slog.Logger
+}
+
+// Option is a functional option for Broker configuration.
+type Option func(*Broker)
+
+// WithMaxUpdateRate limits fan-out to at most hz updates per second per signal.
+// When upstream delivers faster, intermediate values are coalesced: the most
+// recent value in each interval is forwarded once the interval elapses.
+// A value ≤ 0 disables rate limiting (the default).
+func WithMaxUpdateRate(hz float64) Option {
+	return func(b *Broker) {
+		if hz > 0 {
+			b.maxInterval = time.Duration(float64(time.Second) / hz)
+		}
+	}
 }
 
 // New creates a Broker whose upstream subscriptions are bound to ctx.
 // Cancel ctx (or the parent context passed to main) to shut everything down.
-func New(ctx context.Context, log *slog.Logger) *Broker {
-	return &Broker{
+func New(ctx context.Context, log *slog.Logger, opts ...Option) *Broker {
+	b := &Broker{
 		ctx:     ctx,
 		sources: make(map[string]datasource.DataSource),
 		subs:    make(map[SignalRef]*signalSub),
 		log:     log,
 	}
+	for _, opt := range opts {
+		opt(b)
+	}
+	return b
 }
 
 // Register adds a DataSource to the broker. Must be called before Subscribe.
@@ -169,7 +190,36 @@ func (b *Broker) unsubscribe(ref SignalRef, ch chan<- Update) {
 
 // fanOut reads values from rawCh and dispatches them to all registered clients.
 // It exits when sub.done is closed or rawCh is closed.
+//
+// When b.maxInterval > 0, updates that arrive faster than the interval are
+// coalesced: only the most recent value in each interval window is forwarded,
+// delivered at the end of the interval by a flush ticker. This ensures the
+// latest value is always eventually seen even when the source fires rapidly.
 func (b *Broker) fanOut(ref SignalRef, sub *signalSub, rawCh <-chan datasource.Value) {
+	maxInterval := b.maxInterval
+
+	var lastSent time.Time
+	var pending *Update
+
+	// Only allocate a ticker when rate-limiting is configured.
+	var flushC <-chan time.Time
+	if maxInterval > 0 {
+		t := time.NewTicker(maxInterval)
+		defer t.Stop()
+		flushC = t.C
+	}
+
+	dispatch := func(u Update) {
+		sub.mu.RLock()
+		for ch := range sub.clients {
+			select {
+			case ch <- u:
+			default: // slow consumer: drop rather than block
+			}
+		}
+		sub.mu.RUnlock()
+	}
+
 	for {
 		select {
 		case v, ok := <-rawCh:
@@ -177,15 +227,23 @@ func (b *Broker) fanOut(ref SignalRef, sub *signalSub, rawCh <-chan datasource.V
 				return
 			}
 			update := Update{Ref: ref, Value: v}
-			sub.mu.RLock()
-			for ch := range sub.clients {
-				select {
-				case ch <- update:
-				default:
-					// slow consumer: drop rather than block
-				}
+			// Rate-limit: if an interval is configured and the last delivery
+			// was recent, hold this update as pending (coalesce).
+			if maxInterval > 0 && !lastSent.IsZero() && time.Since(lastSent) < maxInterval {
+				pending = &update
+				continue
+			}
+			lastSent = time.Now()
+			pending = nil
+			dispatch(update)
+
+		case <-flushC:
+			// Deliver the most recent coalesced value, if any.
+			if pending != nil {
+				lastSent = time.Now()
+				dispatch(*pending)
+				pending = nil
 			}
-			sub.mu.RUnlock()
 
 		case <-sub.done:
 			return

internal/broker/broker_stress_test.go

@@ -0,0 +1,269 @@
+package broker_test
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/uopi/uopi/internal/broker"
+	"github.com/uopi/uopi/internal/datasource/stub"
+)
+
+// newBrokerN builds a broker backed by a stub with n dynamically generated signals.
+func newBrokerN(tb testing.TB, n int) (*broker.Broker, context.CancelFunc) {
+	tb.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	log := slog.Default()
+	b := broker.New(ctx, log)
+	ds := stub.NewN(n)
+	if err := ds.Connect(ctx); err != nil {
+		cancel()
+		tb.Fatal(err)
+	}
+	b.Register(ds)
+	return b, cancel
+}
+
+// TestStress_ManySignalsManyClients subscribes 20 clients to 500 signals each,
+// runs for 2 seconds, then verifies delivery counts and zero subscription leaks.
+func TestStress_ManySignalsManyClients(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping stress test in short mode")
+	}
+
+	const (
+		nSignals = 500
+		nClients = 20
+		duration = 2 * time.Second
+	)
+
+	goroutinesBefore := runtime.NumGoroutine()
+
+	brk, cancel := newBrokerN(t, nSignals)
+	defer cancel()
+
+	// Each client has its own channel and subscribes to all nSignals signals.
+	type clientState struct {
+		ch    chan broker.Update
+		unsub []func()
+	}
+	clients := make([]clientState, nClients)
+
+	for i := range nClients {
+		ch := make(chan broker.Update, 2048)
+		unsubs := make([]func(), nSignals)
+		for j := range nSignals {
+			ref := broker.SignalRef{DS: "stub", Name: fmt.Sprintf("pv_%d", j)}
+			unsub, err := brk.Subscribe(ref, ch)
+			if err != nil {
+				t.Fatalf("client %d subscribe pv_%d: %v", i, j, err)
+			}
+			unsubs[j] = unsub
+		}
+		clients[i] = clientState{ch: ch, unsub: unsubs}
+	}
+
+	if n := brk.ActiveSubscriptions(); n != nSignals {
+		t.Errorf("expected %d active upstream subscriptions, got %d", nSignals, n)
+	}
+
+	// Drain updates concurrently for the test duration.
+	var totalReceived atomic.Int64
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+
+	for i := range nClients {
+		wg.Add(1)
+		ch := clients[i].ch
+		go func() {
+			defer wg.Done()
+			var n int64
+			for {
+				select {
+				case <-ch:
+					n++
+				case <-stop:
+					totalReceived.Add(n)
+					return
+				}
+			}
+		}()
+	}
+
+	time.Sleep(duration)
+	close(stop)
+	wg.Wait()
+
+	total := totalReceived.Load()
+	// At 10 Hz for duration seconds: nSignals * nClients * duration.Seconds() * 10
+	minExpected := int64(nSignals) * int64(nClients) * int64(duration.Seconds()) * 5 // conservative: 50%
+	t.Logf("received %d updates (%d clients × %d signals × %.0fs @ 10Hz, min=%d)",
+		total, nClients, nSignals, duration.Seconds(), minExpected)
+	if total < minExpected {
+		t.Errorf("too few updates: got %d, want >= %d", total, minExpected)
+	}
+
+	// Unsubscribe all clients.
+	for _, c := range clients {
+		for _, unsub := range c.unsub {
+			unsub()
+		}
+	}
+
+	// Give the broker time to tear down upstream subscriptions.
+	time.Sleep(200 * time.Millisecond)
+
+	if n := brk.ActiveSubscriptions(); n != 0 {
+		t.Errorf("subscription leak: %d active subscriptions remain after all clients unsubscribed", n)
+	}
+
+	cancel()
+	time.Sleep(300 * time.Millisecond)
+
+	goroutinesAfter := runtime.NumGoroutine()
+	leaked := goroutinesAfter - goroutinesBefore
+	t.Logf("goroutines: before=%d after=%d delta=%d", goroutinesBefore, goroutinesAfter, leaked)
+	if leaked > 20 {
+		t.Errorf("goroutine leak: started with %d, ended with %d (%d extra)", goroutinesBefore, goroutinesAfter, leaked)
+	}
+}
+
+// TestStress_RapidSubscribeUnsubscribe hammers subscribe/unsubscribe on a small
+// signal set from many goroutines concurrently to expose races and deadlocks.
+func TestStress_RapidSubscribeUnsubscribe(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping stress test in short mode")
+	}
+
+	const (
+		nSignals   = 20
+		nGoroutines = 50
+		duration   = 2 * time.Second
+	)
+
+	brk, cancel := newBrokerN(t, nSignals)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+	var ops atomic.Int64
+
+	for range nGoroutines {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ch := make(chan broker.Update, 64)
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				// Pick a signal and rapidly subscribe/unsubscribe.
+				idx := int(ops.Load()) % nSignals
+				ref := broker.SignalRef{DS: "stub", Name: fmt.Sprintf("pv_%d", idx)}
+				unsub, err := brk.Subscribe(ref, ch)
+				if err != nil {
+					t.Errorf("subscribe pv_%d: %v", idx, err)
+					return
+				}
+				ops.Add(1)
+				// Drain a bit before unsubscribing.
+				select {
+				case <-ch:
+				case <-time.After(50 * time.Millisecond):
+				}
+				unsub()
+			}
+		}()
+	}
+
+	time.Sleep(duration)
+	close(stop)
+	wg.Wait()
+
+	t.Logf("completed %d subscribe/unsubscribe cycles in %s", ops.Load(), duration)
+
+	time.Sleep(200 * time.Millisecond)
+	if n := brk.ActiveSubscriptions(); n != 0 {
+		t.Errorf("subscription leak after rapid churn: %d remain", n)
+	}
+}
+
+// TestStress_SharedSignalManyClients verifies that a single high-rate signal
+// fans out correctly to many concurrent clients with no drops for fast consumers.
+func TestStress_SharedSignalManyClients(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping stress test in short mode")
+	}
+
+	const (
+		nClients = 50
+		duration = 2 * time.Second
+	)
+
+	brk, cancel := newBroker(t) // uses stub.New() which has counter_fast at 1 ms
+	defer cancel()
+
+	ref := broker.SignalRef{DS: "stub", Name: "counter_fast"}
+	channels := make([]chan broker.Update, nClients)
+	unsubs := make([]func(), nClients)
+
+	for i := range nClients {
+		ch := make(chan broker.Update, 4096)
+		channels[i] = ch
+		unsub, err := brk.Subscribe(ref, ch)
+		if err != nil {
+			t.Fatalf("subscribe client %d: %v", i, err)
+		}
+		unsubs[i] = unsub
+	}
+	defer func() {
+		for _, u := range unsubs {
+			u()
+		}
+	}()
+
+	if n := brk.ActiveSubscriptions(); n != 1 {
+		t.Errorf("expected exactly 1 upstream subscription for shared signal, got %d", n)
+	}
+
+	var totalReceived atomic.Int64
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+
+	for i := range nClients {
+		wg.Add(1)
+		ch := channels[i]
+		go func() {
+			defer wg.Done()
+			var n int64
+			for {
+				select {
+				case <-ch:
+					n++
+				case <-stop:
+					totalReceived.Add(n)
+					return
+				}
+			}
+		}()
+	}
+
+	time.Sleep(duration)
+	close(stop)
+	wg.Wait()
+
+	total := totalReceived.Load()
+	// counter_fast ticks at 1ms → ~1000 Hz → 2000 updates × 50 clients = 100000 expected minimum
+	minExpected := int64(nClients) * int64(duration.Seconds()) * 500 // 50% delivery ok due to buffering
+	t.Logf("shared signal: received %d updates across %d clients (min=%d)", total, nClients, minExpected)
+	if total < minExpected {
+		t.Errorf("too few updates: got %d, want >= %d", total, minExpected)
+	}
+}

internal/broker/broker_test.go

@@ -122,6 +122,68 @@ func TestUnknownSignal(t *testing.T) {
 	}
 }
 
+func TestMaxUpdateRate(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	const maxHz = 5.0 // 5 Hz → 200 ms interval
+
+	ds := stub.New() // counter_fast fires at 1 kHz
+	if err := ds.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+	b := broker.New(ctx, slog.Default(), broker.WithMaxUpdateRate(maxHz))
+	b.Register(ds)
+
+	ref := broker.SignalRef{DS: "stub", Name: "counter_fast"}
+	ch := make(chan broker.Update, 512)
+	unsub, err := b.Subscribe(ref, ch)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer unsub()
+
+	// Collect updates for 2 seconds.
+	var count int
+	deadline := time.After(2 * time.Second)
+loop:
+	for {
+		select {
+		case <-ch:
+			count++
+		case <-deadline:
+			break loop
+		}
+	}
+
+	// At 5 Hz for 2 s we expect ~10 updates (±3 for timer jitter + coalescing).
+	t.Logf("MaxUpdateRate: received %d updates at %.0f Hz limit over 2s (expect ~10)", count, maxHz)
+	if count < 5 || count > 20 {
+		t.Errorf("expected ~10 updates at 5 Hz, got %d", count)
+	}
+}
+
+func TestMaxUpdateRate_UnlimitedByDefault(t *testing.T) {
+	b, cancel := newBroker(t) // no WithMaxUpdateRate → unlimited
+	defer cancel()
+
+	ref := broker.SignalRef{DS: "stub", Name: "counter_fast"}
+	ch := make(chan broker.Update, 4096)
+	unsub, err := b.Subscribe(ref, ch)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer unsub()
+
+	// counter_fast fires at 1 kHz; over 500 ms we should see ≥ 400 updates.
+	time.Sleep(500 * time.Millisecond)
+	got := len(ch)
+	t.Logf("Unlimited: received %d updates in 500ms from 1kHz signal", got)
+	if got < 400 {
+		t.Errorf("expected >= 400 updates with no rate limit, got %d", got)
+	}
+}
+
 func TestMultipleSignals(t *testing.T) {
 	b, cancel := newBroker(t)
 	defer cancel()