From b76fd546fe5a5ae227dfbca8fb7ba566d065e980 Mon Sep 17 00:00:00 2001 From: Valentin Popov Date: Wed, 8 Jul 2026 00:59:07 +0000 Subject: [PATCH] fix(telegram): add request timeout for notifications --- .env.example | 2 + README.md | 2 + README.ru.md | 2 + internal/app/app.go | 16 ++-- internal/config/config.go | 20 +++-- internal/config/config_test.go | 37 +++++++++ internal/notify/notify.go | 63 +++++++++------ internal/notify/telegram_test.go | 129 +++++++++++++++++++++++++++++++ 8 files changed, 236 insertions(+), 35 deletions(-) diff --git a/.env.example b/.env.example index 8ddfe3a..8964f67 100644 --- a/.env.example +++ b/.env.example @@ -26,6 +26,8 @@ TELEGRAM_NOTIFY_INFO=true TELEGRAM_NOTIFY_WARN=true TELEGRAM_NOTIFY_ALERT=true TELEGRAM_NOTIFY_REPORT=true +TELEGRAM_REQUEST_TIMEOUT_SEC=10 +TELEGRAM_QUEUE_SIZE=256 STRATEGY_ROLLING_SHORT=60 STRATEGY_ROLLING_LONG=252 diff --git a/README.md b/README.md index f1adb12..81bc08a 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,8 @@ Common formats: | `TELEGRAM_NOTIFY_WARN` | `true` | Send warnings. | | `TELEGRAM_NOTIFY_ALERT` | `true` | Send critical alerts. | | `TELEGRAM_NOTIFY_REPORT` | `true` | Send daily reports. | +| `TELEGRAM_REQUEST_TIMEOUT_SEC` | `10` | HTTP timeout for Telegram API calls, including the startup `getMe` check and message sends. | +| `TELEGRAM_QUEUE_SIZE` | `256` | In-memory Telegram notification queue size. The queue smooths transient Telegram slowness but is not persisted across process restarts. | ### STRATEGY diff --git a/README.ru.md b/README.ru.md index c8a5647..49cd413 100644 --- a/README.ru.md +++ b/README.ru.md @@ -77,6 +77,8 @@ APP_MODE=backtest go run ./cmd/bot | `TELEGRAM_NOTIFY_WARN` | `true` или `false` | `true` | boolean | Включает предупреждения. | | `TELEGRAM_NOTIFY_ALERT` | `true` или `false` | `true` | boolean | Включает alert-сообщения. Они считаются критичными для доставки и ждут место в очереди. | | `TELEGRAM_NOTIFY_REPORT` | `true` или `false` | `true` | boolean | Включает дневные отчёты. | +| `TELEGRAM_REQUEST_TIMEOUT_SEC` | целое число секунд | `10` | должно быть `> 0` | Таймаут HTTP-запросов к Telegram API, включая стартовую проверку `getMe` и отправку сообщений. | +| `TELEGRAM_QUEUE_SIZE` | целое число | `256` | должно быть `> 0` | Размер in-memory очереди Telegram-уведомлений. Очередь сглаживает временную медленную доставку, но не сохраняется между рестартами процесса. | ### STRATEGY diff --git a/internal/app/app.go b/internal/app/app.go index 3b10143..8d66c9e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -172,13 +172,15 @@ func Run(ctx context.Context, opts Options) error { return err } notifier, err := notify.NewTelegram(notify.TelegramConfig{ - BotToken: cfg.Telegram.BotToken, - ChatID: cfg.Telegram.ChatID, - NotifyInfo: cfg.Telegram.NotifyInfo, - NotifyWarn: cfg.Telegram.NotifyWarn, - NotifyAlert: cfg.Telegram.NotifyAlert, - NotifyReport: cfg.Telegram.NotifyReport, - AuditSink: repo, + BotToken: cfg.Telegram.BotToken, + ChatID: cfg.Telegram.ChatID, + NotifyInfo: cfg.Telegram.NotifyInfo, + NotifyWarn: cfg.Telegram.NotifyWarn, + NotifyAlert: cfg.Telegram.NotifyAlert, + NotifyReport: cfg.Telegram.NotifyReport, + RequestTimeout: time.Duration(cfg.Telegram.RequestTimeoutSec) * time.Second, + QueueSize: cfg.Telegram.QueueSize, + AuditSink: repo, }, log) if err != nil { return fmt.Errorf("create notifier: %w", err) diff --git a/internal/config/config.go b/internal/config/config.go index d7f4a7f..fec2a91 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -66,12 +66,14 @@ type DBConfig struct { } type TelegramConfig struct { - BotToken string `env:"BOT_TOKEN"` - ChatID int64 `env:"CHAT_ID"` - NotifyInfo bool `env:"NOTIFY_INFO" envDefault:"true"` - NotifyWarn bool `env:"NOTIFY_WARN" envDefault:"true"` - NotifyAlert bool `env:"NOTIFY_ALERT" envDefault:"true"` - NotifyReport bool `env:"NOTIFY_REPORT" envDefault:"true"` + BotToken string `env:"BOT_TOKEN"` + ChatID int64 `env:"CHAT_ID"` + NotifyInfo bool `env:"NOTIFY_INFO" envDefault:"true"` + NotifyWarn bool `env:"NOTIFY_WARN" envDefault:"true"` + NotifyAlert bool `env:"NOTIFY_ALERT" envDefault:"true"` + NotifyReport bool `env:"NOTIFY_REPORT" envDefault:"true"` + RequestTimeoutSec int `env:"REQUEST_TIMEOUT_SEC" envDefault:"10"` + QueueSize int `env:"QUEUE_SIZE" envDefault:"256"` } type StrategyConfig struct { @@ -201,6 +203,12 @@ func (c *Config) Validate() error { if c.TInvest.RequestTimeoutSec <= 0 { return errors.New("TINVEST_REQUEST_TIMEOUT_SEC must be positive") } + if c.Telegram.RequestTimeoutSec <= 0 { + return errors.New("TELEGRAM_REQUEST_TIMEOUT_SEC must be positive") + } + if c.Telegram.QueueSize <= 0 { + return errors.New("TELEGRAM_QUEUE_SIZE must be positive") + } if c.TInvest.TradingCalendarExchange == "" { c.TInvest.TradingCalendarExchange = "MOEX" } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 45ccb37..4216cae 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -86,6 +86,39 @@ func TestLoadSchedulerKnobsFromEnv(t *testing.T) { } } +func TestLoadTelegramQueueDefaults(t *testing.T) { + t.Setenv("APP_MODE", "backtest") + + cfg, err := Load() + if err != nil { + t.Fatal(err) + } + if cfg.Telegram.RequestTimeoutSec != 10 { + t.Fatalf("telegram request timeout=%d, want 10", cfg.Telegram.RequestTimeoutSec) + } + if cfg.Telegram.QueueSize != 256 { + t.Fatalf("telegram queue size=%d, want 256", cfg.Telegram.QueueSize) + } +} + +func TestValidateRejectsInvalidTelegramRequestTimeout(t *testing.T) { + cfg := minimalBrokerConfig(domain.ModeSandbox) + cfg.Telegram.RequestTimeoutSec = 0 + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "TELEGRAM_REQUEST_TIMEOUT_SEC") { + t.Fatalf("Validate err=%v, want TELEGRAM_REQUEST_TIMEOUT_SEC requirement", err) + } +} + +func TestValidateRejectsInvalidTelegramQueueSize(t *testing.T) { + cfg := minimalBrokerConfig(domain.ModeSandbox) + cfg.Telegram.QueueSize = 0 + err := cfg.Validate() + if err == nil || !strings.Contains(err.Error(), "TELEGRAM_QUEUE_SIZE") { + t.Fatalf("Validate err=%v, want TELEGRAM_QUEUE_SIZE requirement", err) + } +} + func minimalBrokerConfig(mode domain.Mode) Config { return Config{ App: AppConfig{ @@ -99,6 +132,10 @@ func minimalBrokerConfig(mode domain.Mode) Config { RequestTimeoutSec: 10, }, DB: DBConfig{DSN: "user:pass@tcp(localhost:3306)/bot"}, + Telegram: TelegramConfig{ + RequestTimeoutSec: 10, + QueueSize: 256, + }, Execution: ExecutionConfig{ EntrySignalTime: mustTOD("18:10:00"), EntryWindowStart: mustTOD("18:20:00"), diff --git a/internal/notify/notify.go b/internal/notify/notify.go index 53a27ba..ea979e3 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "net/http" "strings" "time" @@ -13,7 +14,11 @@ import ( "overnight-trading-bot/internal/domain" ) -const mustDeliverEnqueueTimeout = 2 * time.Second +const ( + defaultQueueSize = 256 + defaultRequestTimeout = 10 * time.Second + mustDeliverEnqueueTimeout = 2 * time.Second +) type Notifier interface { Info(ctx context.Context, msg string) error @@ -32,13 +37,15 @@ func (Noop) Report(context.Context, string) error { return nil } func (Noop) Close() error { return nil } type TelegramConfig struct { - BotToken string - ChatID int64 - NotifyInfo bool - NotifyWarn bool - NotifyAlert bool - NotifyReport bool - AuditSink AuditSink + BotToken string + ChatID int64 + NotifyInfo bool + NotifyWarn bool + NotifyAlert bool + NotifyReport bool + RequestTimeout time.Duration + QueueSize int + AuditSink AuditSink } type AuditSink interface { @@ -46,12 +53,13 @@ type AuditSink interface { } type Telegram struct { - cfg TelegramConfig - bot *tgbotapi.BotAPI - log *slog.Logger - queue chan outbound - done chan struct{} - closed chan struct{} + cfg TelegramConfig + bot *tgbotapi.BotAPI + log *slog.Logger + queue chan outbound + done chan struct{} + closed chan struct{} + retryDelay func(error, int) time.Duration } type outbound struct { @@ -61,23 +69,34 @@ type outbound struct { } func NewTelegram(cfg TelegramConfig, log *slog.Logger) (Notifier, error) { + return newTelegramWithEndpoint(cfg, log, tgbotapi.APIEndpoint) +} + +func newTelegramWithEndpoint(cfg TelegramConfig, log *slog.Logger, apiEndpoint string) (Notifier, error) { if cfg.BotToken == "" || cfg.ChatID == 0 { if log != nil { log.Warn("telegram notifier disabled; TELEGRAM_BOT_TOKEN or TELEGRAM_CHAT_ID is empty") } return Noop{}, nil } - bot, err := tgbotapi.NewBotAPI(cfg.BotToken) + if cfg.RequestTimeout <= 0 { + cfg.RequestTimeout = defaultRequestTimeout + } + if cfg.QueueSize <= 0 { + cfg.QueueSize = defaultQueueSize + } + bot, err := tgbotapi.NewBotAPIWithClient(cfg.BotToken, apiEndpoint, &http.Client{Timeout: cfg.RequestTimeout}) if err != nil { return nil, err } t := &Telegram{ - cfg: cfg, - bot: bot, - log: log, - queue: make(chan outbound, 256), - done: make(chan struct{}), - closed: make(chan struct{}), + cfg: cfg, + bot: bot, + log: log, + queue: make(chan outbound, cfg.QueueSize), + done: make(chan struct{}), + closed: make(chan struct{}), + retryDelay: telegramRetryDelay, } go t.dispatch() return t, nil @@ -181,7 +200,7 @@ func (t *Telegram) send(item outbound) { for attempt := 0; attempt < 3; attempt++ { if _, err := t.bot.Send(msg); err != nil { lastErr = err - delay := telegramRetryDelay(err, attempt) + delay := t.retryDelay(err, attempt) if t.log != nil { t.log.Warn("telegram send failed", "attempt", attempt+1, "err", err, "retry_in", delay) } diff --git a/internal/notify/telegram_test.go b/internal/notify/telegram_test.go index 36c1be6..d7ec2fd 100644 --- a/internal/notify/telegram_test.go +++ b/internal/notify/telegram_test.go @@ -1,6 +1,10 @@ package notify import ( + "context" + "net/http" + "net/http/httptest" + "strings" "testing" "time" @@ -44,6 +48,131 @@ func TestTelegramRetryDelayUsesRetryAfter(t *testing.T) { } } +func TestNewTelegramUsesRequestTimeout(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(200 * time.Millisecond) + })) + defer server.Close() + + start := time.Now() + _, err := newTelegramWithEndpoint(TelegramConfig{ + BotToken: "token", + ChatID: 123, + RequestTimeout: 20 * time.Millisecond, + QueueSize: 1, + }, nil, telegramTestEndpoint(server)) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("NewTelegram err=nil, want timeout") + } + if elapsed > 150*time.Millisecond { + t.Fatalf("NewTelegram took %s, want request timeout to fire quickly", elapsed) + } +} + +func TestNewTelegramUsesConfiguredQueueSize(t *testing.T) { + server := telegramTestServer(t, func(w http.ResponseWriter, r *http.Request) { + writeTelegramOK(w, r) + }) + defer server.Close() + + notifier, err := newTelegramWithEndpoint(TelegramConfig{ + BotToken: "token", + ChatID: 123, + RequestTimeout: time.Second, + QueueSize: 3, + }, nil, telegramTestEndpoint(server)) + if err != nil { + t.Fatal(err) + } + defer notifier.Close() + + telegram, ok := notifier.(*Telegram) + if !ok { + t.Fatalf("notifier type=%T, want *Telegram", notifier) + } + if got := cap(telegram.queue); got != 3 { + t.Fatalf("queue cap=%d, want 3", got) + } +} + +func TestTelegramSendTimeoutAuditsMustDeliverFailure(t *testing.T) { + audit := &recordingAuditSink{events: make(chan domain.RiskEvent, 1)} + server := telegramTestServer(t, func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/sendMessage") { + time.Sleep(50 * time.Millisecond) + writeTelegramOK(w, r) + return + } + writeTelegramOK(w, r) + }) + defer server.Close() + + notifier, err := newTelegramWithEndpoint(TelegramConfig{ + BotToken: "token", + ChatID: 123, + NotifyAlert: true, + RequestTimeout: 10 * time.Millisecond, + QueueSize: 1, + AuditSink: audit, + }, nil, telegramTestEndpoint(server)) + if err != nil { + t.Fatal(err) + } + telegram := notifier.(*Telegram) + telegram.retryDelay = func(error, int) time.Duration { return time.Millisecond } + defer notifier.Close() + + if err := notifier.Alert(context.Background(), "critical"); err != nil { + t.Fatal(err) + } + + select { + case event := <-audit.events: + if event.EventType != "notification_undeliverable" { + t.Fatalf("event type=%s, want notification_undeliverable", event.EventType) + } + if event.Severity != domain.SeverityCritical { + t.Fatalf("event severity=%s, want critical", event.Severity) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("timed out waiting for must-deliver audit event") + } +} + type assertErr struct{} func (assertErr) Error() string { return "boom" } + +type recordingAuditSink struct { + events chan domain.RiskEvent +} + +func (s *recordingAuditSink) InsertRiskEvent(ctx context.Context, event domain.RiskEvent) error { + select { + case s.events <- event: + case <-ctx.Done(): + return ctx.Err() + } + return nil +} + +func telegramTestEndpoint(server *httptest.Server) string { + return server.URL + "/bot%s/%s" +} + +func telegramTestServer(t *testing.T, handler http.HandlerFunc) *httptest.Server { + t.Helper() + return httptest.NewServer(handler) +} + +func writeTelegramOK(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasSuffix(r.URL.Path, "/getMe"): + _, _ = w.Write([]byte(`{"ok":true,"result":{"id":1,"is_bot":true,"first_name":"test","username":"test_bot"}}`)) + default: + _, _ = w.Write([]byte(`{"ok":true,"result":{"message_id":1,"date":0,"chat":{"id":123,"type":"private"},"text":"ok"}}`)) + } +}