From d7084052b778734807415d8771835da0de4ec925 Mon Sep 17 00:00:00 2001 From: Yasuhiro Matsumoto Date: Mon, 16 Feb 2026 00:59:09 +0900 Subject: [PATCH] Eliminate busy-loop in dispatchOutbound when context is cancelled --- pkg/channels/manager.go | 47 ++++++++++++++++++----------------------- 1 file changed, 21 insertions(+), 26 deletions(-) diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 15f8c6037..b5af573ab 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -246,38 +246,33 @@ func (m *Manager) dispatchOutbound(ctx context.Context) { logger.InfoC("channels", "Outbound dispatcher started") for { - select { - case <-ctx.Done(): + msg, ok := m.bus.SubscribeOutbound(ctx) + if !ok { logger.InfoC("channels", "Outbound dispatcher stopped") return - default: - msg, ok := m.bus.SubscribeOutbound(ctx) - if !ok { - continue - } + } - // Silently skip internal channels - if constants.IsInternalChannel(msg.Channel) { - continue - } + // Silently skip internal channels + if constants.IsInternalChannel(msg.Channel) { + continue + } - m.mu.RLock() - channel, exists := m.channels[msg.Channel] - m.mu.RUnlock() + m.mu.RLock() + channel, exists := m.channels[msg.Channel] + m.mu.RUnlock() - if !exists { - logger.WarnCF("channels", "Unknown channel for outbound message", map[string]interface{}{ - "channel": msg.Channel, - }) - continue - } + if !exists { + logger.WarnCF("channels", "Unknown channel for outbound message", map[string]interface{}{ + "channel": msg.Channel, + }) + continue + } - if err := channel.Send(ctx, msg); err != nil { - logger.ErrorCF("channels", "Error sending message to channel", map[string]interface{}{ - "channel": msg.Channel, - "error": err.Error(), - }) - } + if err := channel.Send(ctx, msg); err != nil { + logger.ErrorCF("channels", "Error sending message to channel", map[string]interface{}{ + "channel": msg.Channel, + "error": err.Error(), + }) } } }