From 79ab165ffda8b21b4e94f97523d0ae03501fe3d1 Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Tue, 6 Oct 2026 13:32:56 +0200 Subject: [PATCH] server: fix event dropping - use blocking callback instead of TrySend The pubsub library's SubscribeChan uses non-blocking TrySend which drops events when the channel is full. Switch to Subscribe with a blocking callback to ensure no events are lost. Also increased channel buffers from 64 to 256. --- cmd/olliesrv/internal/session/event.go | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/cmd/olliesrv/internal/session/event.go b/cmd/olliesrv/internal/session/event.go index 4474d01..5f91d26 100644 --- a/cmd/olliesrv/internal/session/event.go +++ b/cmd/olliesrv/internal/session/event.go @@ -31,9 +31,16 @@ func PublishEvent(topic, payload string) { } // SubscribeEvents returns a channel that receives all events matching the topic. +// Events are never dropped — publishing blocks if the channel is full. func SubscribeEvents(ctx context.Context, topic string) <-chan Event { - ch := make(chan Event, 64) - sub := eventBus.SubscribeChan(topic, ch, pubsub.KeepOpen) + // Use a callback instead of SubscribeChan to avoid non-blocking TrySend + ch := make(chan Event, 256) + sub := eventBus.Subscribe(topic, func(ev Event) { + select { + case ch <- ev: + case <-ctx.Done(): + } + }) go func() { <-ctx.Done() eventBus.Unsubscribe(sub) @@ -48,7 +55,7 @@ func SubscribeEvents(ctx context.Context, topic string) <-chan Event { func SubscribeEventsFiltered(ctx context.Context, filter string) <-chan Event { // Subscribe to all events and filter client-side for complex patterns // (pubsub only supports simple topic matching) - ch := make(chan Event, 64) + ch := make(chan Event, 256) allEvents := SubscribeEvents(ctx, "*") go func() { defer close(ch) @@ -56,7 +63,8 @@ func SubscribeEventsFiltered(ctx context.Context, filter string) <-chan Event { if MatchTopic(filter, ev.Topic) { select { case ch <- ev: - default: // drop if full + case <-ctx.Done(): + return } } }