fix: eliminate duplicate HTTP listener on gateway port

The shared HTTP server in ChannelManager and the HealthServer both
tried to bind to the same gateway port, causing "address already in use".

Consolidate to a single listener: register channel webhooks and health
checkers on the HealthServer's mux via SetupHTTPServer, then let
HealthServer be the sole HTTP listener. Remove the separate httpServer
from ChannelManager entirely.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
dj-oyu 2026-03-19 12:09:49 +09:00
parent ea77996ddf
commit c2e14db165
2 changed files with 16 additions and 46 deletions

View file

@ -11,7 +11,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"math" "math"
"net/http"
"sync" "sync"
"time" "time"
@ -84,8 +83,6 @@ type Manager struct {
config *config.Config config *config.Config
mediaStore media.MediaStore mediaStore media.MediaStore
dispatchTask *asyncTask dispatchTask *asyncTask
mux *http.ServeMux
httpServer *http.Server
mu sync.RWMutex mu sync.RWMutex
placeholders sync.Map // "channel:chatID" → placeholderID (string) placeholders sync.Map // "channel:chatID" → placeholderID (string)
typingStops sync.Map // "channel:chatID" → func() typingStops sync.Map // "channel:chatID" → func()
@ -342,41 +339,31 @@ func (m *Manager) initChannels() error {
return nil return nil
} }
// SetupHTTPServer creates a shared HTTP server with the given listen address. // SetupHTTPServer registers channel webhook handlers and health checkers onto
// It registers health endpoints from the health server and discovers channels // the health server's mux so everything is served by a single HTTP listener.
// that implement WebhookHandler and/or HealthChecker to register their handlers.
func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server) { func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server) {
m.mux = http.NewServeMux() if healthServer == nil {
return
// Register health endpoints
if healthServer != nil {
healthServer.RegisterOnMux(m.mux)
} }
mux := healthServer.Mux()
// Discover and register webhook handlers and health checkers // Discover and register webhook handlers and health checkers
for name, ch := range m.channels { for name, ch := range m.channels {
if wh, ok := ch.(WebhookHandler); ok { if wh, ok := ch.(WebhookHandler); ok {
m.mux.Handle(wh.WebhookPath(), wh) mux.Handle(wh.WebhookPath(), wh)
logger.InfoCF("channels", "Webhook handler registered", map[string]any{ logger.InfoCF("channels", "Webhook handler registered", map[string]any{
"channel": name, "channel": name,
"path": wh.WebhookPath(), "path": wh.WebhookPath(),
}) })
} }
if hc, ok := ch.(HealthChecker); ok { if hc, ok := ch.(HealthChecker); ok {
m.mux.HandleFunc(hc.HealthPath(), hc.HealthHandler) mux.HandleFunc(hc.HealthPath(), hc.HealthHandler)
logger.InfoCF("channels", "Health endpoint registered", map[string]any{ logger.InfoCF("channels", "Health endpoint registered", map[string]any{
"channel": name, "channel": name,
"path": hc.HealthPath(), "path": hc.HealthPath(),
}) })
} }
} }
m.httpServer = &http.Server{
Addr: addr,
Handler: m.mux,
ReadTimeout: 30 * time.Second,
WriteTimeout: 30 * time.Second,
}
} }
func (m *Manager) StartAll(ctx context.Context) error { func (m *Manager) StartAll(ctx context.Context) error {
@ -417,20 +404,6 @@ func (m *Manager) StartAll(ctx context.Context) error {
// Start the TTL janitor that cleans up stale typing/placeholder entries // Start the TTL janitor that cleans up stale typing/placeholder entries
go m.runTTLJanitor(dispatchCtx) go m.runTTLJanitor(dispatchCtx)
// Start shared HTTP server if configured
if m.httpServer != nil {
go func() {
logger.InfoCF("channels", "Shared HTTP server listening", map[string]any{
"addr": m.httpServer.Addr,
})
if err := m.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logger.FatalCF("channels", "Shared HTTP server error", map[string]any{
"error": err.Error(),
})
}
}()
}
logger.InfoC("channels", "All channels started") logger.InfoC("channels", "All channels started")
return nil return nil
} }
@ -441,18 +414,6 @@ func (m *Manager) StopAll(ctx context.Context) error {
logger.InfoC("channels", "Stopping all channels") logger.InfoC("channels", "Stopping all channels")
// Shutdown shared HTTP server first
if m.httpServer != nil {
shutdownCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
if err := m.httpServer.Shutdown(shutdownCtx); err != nil {
logger.ErrorCF("channels", "Shared HTTP server shutdown error", map[string]any{
"error": err.Error(),
})
}
m.httpServer = nil
}
// Cancel dispatcher // Cancel dispatcher
if m.dispatchTask != nil { if m.dispatchTask != nil {
m.dispatchTask.cancel() m.dispatchTask.cancel()

View file

@ -356,6 +356,8 @@ func setupAndStartServices(
} }
} }
// HealthServer is the single HTTP listener. Channel webhooks and health
// checkers were registered on its mux via SetupHTTPServer.
go func() { go func() {
var serverErr error var serverErr error
if useTLS { if useTLS {
@ -571,6 +573,13 @@ func restartServices(
if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil { if err = runningServices.ChannelManager.StartAll(context.Background()); err != nil {
return fmt.Errorf("error restarting channels: %w", err) return fmt.Errorf("error restarting channels: %w", err)
} }
// Start the HealthServer listener (single HTTP server for all routes)
go func() {
if sErr := runningServices.HealthServer.Start(); sErr != nil && sErr != http.ErrServerClosed {
logger.ErrorCF("health", "Health server error", map[string]any{"error": sErr.Error()})
}
}()
fmt.Printf( fmt.Printf(
" ✓ Channels restarted, health endpoints at http://%s:%d/health and ready\n", " ✓ Channels restarted, health endpoints at http://%s:%d/health and ready\n",
cfg.Gateway.Host, cfg.Gateway.Host,