fix(telegram): add request timeout for notifications
Deploy / Test, build and deploy (push) Failing after 6m3s
Deploy / Test, build and deploy (push) Failing after 6m3s
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
+9
-7
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
|
||||
@@ -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"),
|
||||
|
||||
+41
-22
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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"}}`))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user