From 5e06037fa2fdeee246e4cd7eebc9774907ecf088 Mon Sep 17 00:00:00 2001 From: shinchan-zhai Date: Mon, 23 Mar 2026 10:30:13 +0800 Subject: [PATCH] reliability: wrap all 27 bare goroutines with safe.Go/GoNamed panic recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Applied safe.GoNamed to: - 9 exchange order_sync goroutines (OKX, Hyperliquid, Aster, Bybit, KuCoin, Gate, Bitget, Lighter, BinanceΓ—2) - Drawdown monitor (auto_trader_risk.go) - Brain news scanner + market briefs (agent/brain.go) - Sentinel scanner (agent/sentinel.go) - Agent scheduler (agent/scheduler.go) - x402 idle watchdog (mcp/payment/x402.go) - MCP stream idle watchdog (mcp/client.go) - Rate limiter cleanup (api/rate_limiter.go) - 3 telemetry fire-and-forget sends (telemetry/experience.go) - CoinAnk WS handler (provider/coinank/coinank_api/kline_ws.go) - API server goroutine (main.go) Added manual defer/recover with error reporting to: - Telegram AI agent handler (sends error msg to user on panic) - Trader data fetch (returns error result on panic to prevent deadlock) Before: a panic in ANY of these 27 goroutines would crash the entire trading process with zero diagnostics. Now all panics are caught, logged with stack traces, and the process continues running. --- agent/brain.go | 9 +++++---- agent/scheduler.go | 5 +++-- agent/sentinel.go | 5 +++-- api/rate_limiter.go | 5 +++-- main.go | 5 +++-- manager/trader_manager.go | 14 ++++++++++++-- mcp/client.go | 5 +++-- mcp/payment/x402.go | 5 +++-- provider/coinank/coinank_api/kline_ws.go | 5 +++-- telegram/bot.go | 7 +++++++ telemetry/experience.go | 13 +++++++------ trader/aster/trader_sync.go | 5 +++-- trader/auto_trader_risk.go | 5 +++-- trader/binance/order_sync.go | 9 +++++---- trader/bitget/order_sync.go | 5 +++-- trader/bybit/order_sync.go | 5 +++-- trader/gate/order_sync.go | 5 +++-- trader/hyperliquid/order_sync.go | 5 +++-- trader/kucoin/order_sync.go | 5 +++-- trader/lighter/order_sync.go | 5 +++-- trader/okx/order_sync.go | 5 +++-- 21 files changed, 84 insertions(+), 48 deletions(-) diff --git a/agent/brain.go b/agent/brain.go index fd5fd70c..3c9acd27 100644 --- a/agent/brain.go +++ b/agent/brain.go @@ -6,6 +6,7 @@ import ( "io" "log/slog" "net/http" + "nofx/safe" "strings" "sync" "time" @@ -49,7 +50,7 @@ func (b *Brain) HandleSignal(sig Signal) { func (b *Brain) StartNewsScan(interval time.Duration) { seen := make(map[string]bool) - go func() { + safe.GoNamed("brain-news-scan", func() { ticker := time.NewTicker(interval) defer ticker.Stop() for { @@ -59,7 +60,7 @@ func (b *Brain) StartNewsScan(interval time.Duration) { b.scanNews(seen) } } - }() + }) } func (b *Brain) scanNews(seen map[string]bool) { @@ -117,7 +118,7 @@ func (b *Brain) scanNews(seen map[string]bool) { } func (b *Brain) StartMarketBriefs(hours []int) { - go func() { + safe.GoNamed("brain-market-briefs", func() { ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() sent := make(map[string]bool) @@ -134,7 +135,7 @@ func (b *Brain) StartMarketBriefs(hours []int) { } } } - }() + }) } func (b *Brain) sendBrief(hour int) { diff --git a/agent/scheduler.go b/agent/scheduler.go index ab31e004..dc251b98 100644 --- a/agent/scheduler.go +++ b/agent/scheduler.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "nofx/safe" "strings" "time" ) @@ -19,7 +20,7 @@ func NewScheduler(a *Agent, l *slog.Logger) *Scheduler { } func (s *Scheduler) Start(ctx context.Context) { - go func() { + safe.GoNamed("agent-scheduler", func() { ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() lastReport := time.Time{} @@ -50,7 +51,7 @@ func (s *Scheduler) Start(ctx context.Context) { } } } - }() + }) } func (s *Scheduler) Stop() { close(s.stopCh) } diff --git a/agent/sentinel.go b/agent/sentinel.go index f8de41ae..eb7e4f3f 100644 --- a/agent/sentinel.go +++ b/agent/sentinel.go @@ -7,6 +7,7 @@ import ( "log/slog" "math" "net/http" + "nofx/safe" "strconv" "strings" "sync" @@ -61,7 +62,7 @@ func NewSentinel(symbols []string, cb SignalCallback, logger *slog.Logger) *Sent } func (s *Sentinel) Start() { - go func() { + safe.GoNamed("sentinel", func() { ticker := time.NewTicker(60 * time.Second) defer ticker.Stop() s.scan() @@ -73,7 +74,7 @@ func (s *Sentinel) Start() { s.scan() } } - }() + }) } func (s *Sentinel) Stop() { close(s.stopCh) } diff --git a/api/rate_limiter.go b/api/rate_limiter.go index 5039b96f..7affd8b2 100644 --- a/api/rate_limiter.go +++ b/api/rate_limiter.go @@ -2,6 +2,7 @@ package api import ( "net/http" + "nofx/safe" "sync" "time" @@ -35,13 +36,13 @@ func NewLoginRateLimiter() *LoginRateLimiter { } // Clean up stale entries every 10 minutes - go func() { + safe.GoNamed("rate-limiter-cleanup", func() { ticker := time.NewTicker(10 * time.Minute) defer ticker.Stop() for range ticker.C { rl.cleanup() } - }() + }) return rl } diff --git a/main.go b/main.go index bca465e7..03297c98 100644 --- a/main.go +++ b/main.go @@ -10,6 +10,7 @@ import ( "nofx/telemetry" "nofx/logger" "nofx/manager" + "nofx/safe" _ "nofx/mcp/payment" _ "nofx/mcp/provider" "nofx/store" @@ -136,11 +137,11 @@ func main() { telegramReloadCh := make(chan struct{}, 1) server.SetTelegramReloadCh(telegramReloadCh) - go func() { + safe.GoNamed("api-server", func() { if err := server.Start(); err != nil { logger.Fatalf("❌ Failed to start API server: %v", err) } - }() + }) // Start NOFXi Agent (proactive intelligence layer) β€” must be created BEFORE Telegram bot nofxiAgent := nofxiagent.New(traderManager, st, nil, slog.Default()) diff --git a/manager/trader_manager.go b/manager/trader_manager.go index fa151b6e..bed8f0a6 100644 --- a/manager/trader_manager.go +++ b/manager/trader_manager.go @@ -273,6 +273,16 @@ func (tm *TraderManager) getConcurrentTraderData(traders []*trader.AutoTrader) [ // Concurrently fetch data for each trader for i, t := range traders { go func(index int, trader *trader.AutoTrader) { + defer func() { + if r := recover(); r != nil { + logger.Errorf("πŸ”₯ [trader-data-fetch] panic for trader %s: %v", trader.GetName(), r) + resultChan <- traderResult{index: index, data: map[string]interface{}{ + "trader_id": trader.GetID(), + "trader_name": trader.GetName(), + "error": "internal panic", + }} + } + }() // Set timeout to 10 seconds for single trader (increased from 3s for DEX reliability) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -281,14 +291,14 @@ func (tm *TraderManager) getConcurrentTraderData(traders []*trader.AutoTrader) [ accountChan := make(chan map[string]interface{}, 1) errorChan := make(chan error, 1) - go func() { + safe.GoNamed("trader-account-info", func() { account, err := trader.GetAccountInfo() if err != nil { errorChan <- err } else { accountChan <- account } - }() + }) status := trader.GetStatus() var traderData map[string]interface{} diff --git a/mcp/client.go b/mcp/client.go index 2d5c864c..4b96c773 100644 --- a/mcp/client.go +++ b/mcp/client.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net/http" + "nofx/safe" "strings" "time" ) @@ -689,7 +690,7 @@ func (client *Client) CallWithRequestStream(req *Request, onChunk func(string)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() resetCh := make(chan struct{}, 1) - go func() { + safe.GoNamed("mcp-stream-idle-watchdog", func() { t := time.NewTimer(idleTimeout) defer t.Stop() for { @@ -710,7 +711,7 @@ func (client *Client) CallWithRequestStream(req *Request, onChunk func(string)) t.Reset(idleTimeout) } } - }() + }) httpReq = httpReq.WithContext(ctx) resp, err := client.HTTPClient.Do(httpReq) diff --git a/mcp/payment/x402.go b/mcp/payment/x402.go index 7649de11..207fffb3 100644 --- a/mcp/payment/x402.go +++ b/mcp/payment/x402.go @@ -14,6 +14,7 @@ import ( "github.com/ethereum/go-ethereum/crypto" "nofx/mcp" + "nofx/safe" ) const ( @@ -405,7 +406,7 @@ func X402CallStream(c *mcp.Client, signFn X402SignFunc, tag string, systemPrompt // Start idle-timeout watchdog AFTER the 402 dance is done. resetCh := make(chan struct{}, 1) - go func() { + safe.GoNamed("x402-idle-watchdog", func() { t := time.NewTimer(x402StreamIdleTimeout) defer t.Stop() for { @@ -426,7 +427,7 @@ func X402CallStream(c *mcp.Client, signFn X402SignFunc, tag string, systemPrompt t.Reset(x402StreamIdleTimeout) } } - }() + }) onLine := func() { select { diff --git a/provider/coinank/coinank_api/kline_ws.go b/provider/coinank/coinank_api/kline_ws.go index 7702cd10..d0a0db39 100644 --- a/provider/coinank/coinank_api/kline_ws.go +++ b/provider/coinank/coinank_api/kline_ws.go @@ -5,6 +5,7 @@ import ( "encoding/json" "nofx/provider/coinank" "nofx/provider/coinank/coinank_enum" + "nofx/safe" "strconv" "strings" @@ -105,7 +106,7 @@ func read(conn *websocket.Conn, ch chan string) { func handleResponse(ch <-chan string, needKline bool, needTicker bool) (<-chan *WsResult[coinank.KlineResult], <-chan *WsResult[KlineTickers]) { klineCh := make(chan *WsResult[coinank.KlineResult], 1024) tickersCh := make(chan *WsResult[KlineTickers], 1024) - go func() { + safe.GoNamed("coinank-ws-handler", func() { if needKline { defer close(klineCh) } else { @@ -147,7 +148,7 @@ func handleResponse(ch <-chan string, needKline bool, needTicker bool) (<-chan * } } } - }() + }) return klineCh, tickersCh } diff --git a/telegram/bot.go b/telegram/bot.go index 3b63bdbe..ae30831e 100644 --- a/telegram/bot.go +++ b/telegram/bot.go @@ -227,6 +227,13 @@ func runBot(token string, cfg *config.Config, st *store.Store) bool { // ── AI agent ───────────────────────────────────────────────────────── go func(chatID int64, text string) { + defer func() { + if r := recover(); r != nil { + logger.Errorf("πŸ”₯ [telegram-ai-agent] panic: %v", r) + msg := tgbotapi.NewMessage(chatID, "⚠️ Internal error. Please try again.") + bot.Send(msg) //nolint:errcheck + } + }() sent, err := bot.Send(tgbotapi.NewMessage(chatID, "⏳")) placeholderID := 0 if err == nil { diff --git a/telemetry/experience.go b/telemetry/experience.go index 8fae9d79..8b7254d4 100644 --- a/telemetry/experience.go +++ b/telemetry/experience.go @@ -5,6 +5,7 @@ import ( "bytes" "encoding/json" "net/http" + "nofx/safe" "sync" "time" ) @@ -108,9 +109,9 @@ func TrackTrade(event TradeEvent) { } // Send asynchronously to not block trading - go func() { + safe.Go(func() { _ = sendTradeEvent(event) - }() + }) } // sendTradeEvent sends the trade event to GA4 @@ -165,7 +166,7 @@ func TrackStartup(version string) { return } - go func() { + safe.Go(func() { client.mu.RLock() installationID := client.installationID client.mu.RUnlock() @@ -194,7 +195,7 @@ func TrackStartup(version string) { resp.Body.Close() } } - }() + }) } func TrackAIUsage(event AIUsageEvent) { @@ -202,7 +203,7 @@ func TrackAIUsage(event AIUsageEvent) { return } - go func() { + safe.Go(func() { client.mu.RLock() installationID := client.installationID client.mu.RUnlock() @@ -238,5 +239,5 @@ func TrackAIUsage(event AIUsageEvent) { resp.Body.Close() } } - }() + }) } diff --git a/trader/aster/trader_sync.go b/trader/aster/trader_sync.go index 2a1c0d51..205a58e2 100644 --- a/trader/aster/trader_sync.go +++ b/trader/aster/trader_sync.go @@ -3,6 +3,7 @@ package aster import ( "fmt" "nofx/logger" + "nofx/safe" "nofx/market" "nofx/store" "sort" @@ -182,12 +183,12 @@ func deriveAsterOrderAction(side, positionSide string, realizedPnL float64) stri // StartOrderSync starts background order sync task for Aster func (t *AsterTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("aster-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromAster(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Aster order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ Aster order sync started (interval: %v)", interval) } diff --git a/trader/auto_trader_risk.go b/trader/auto_trader_risk.go index b8c94fb3..f7166070 100644 --- a/trader/auto_trader_risk.go +++ b/trader/auto_trader_risk.go @@ -3,6 +3,7 @@ package trader import ( "fmt" "nofx/logger" + "nofx/safe" "strings" "time" ) @@ -10,7 +11,7 @@ import ( // startDrawdownMonitor starts drawdown monitoring func (at *AutoTrader) startDrawdownMonitor() { at.monitorWg.Add(1) - go func() { + safe.GoNamed("drawdown-monitor", func() { defer at.monitorWg.Done() ticker := time.NewTicker(1 * time.Minute) // Check every minute @@ -27,7 +28,7 @@ func (at *AutoTrader) startDrawdownMonitor() { return } } - }() + }) } // checkPositionDrawdown checks position drawdown situation diff --git a/trader/binance/order_sync.go b/trader/binance/order_sync.go index 76705395..c762d068 100644 --- a/trader/binance/order_sync.go +++ b/trader/binance/order_sync.go @@ -4,6 +4,7 @@ import ( "fmt" "nofx/logger" "nofx/market" + "nofx/safe" "nofx/store" "nofx/trader/types" "sort" @@ -351,21 +352,21 @@ func (t *FuturesTrader) determineOrderAction(side, positionSide string, realized // StartOrderSync starts background order sync task for Binance func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { // Run first sync immediately - go func() { + safe.GoNamed("binance-order-sync-initial", func() { logger.Infof("πŸ”„ Running initial Binance order sync...") if err := t.SyncOrdersFromBinance(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Initial Binance order sync failed: %v", err) } - }() + }) // Then run periodically ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("binance-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromBinance(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Binance order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ Binance order sync started (interval: %v)", interval) } diff --git a/trader/bitget/order_sync.go b/trader/bitget/order_sync.go index e8f10789..65d9f0d8 100644 --- a/trader/bitget/order_sync.go +++ b/trader/bitget/order_sync.go @@ -5,6 +5,7 @@ import ( "fmt" "nofx/logger" "nofx/market" + "nofx/safe" "nofx/store" "sort" "strconv" @@ -281,12 +282,12 @@ func (t *BitgetTrader) SyncOrdersFromBitget(traderID string, exchangeID string, // StartOrderSync starts background order sync task for Bitget func (t *BitgetTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("bitget-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromBitget(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Bitget order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ Bitget order sync started (interval: %v)", interval) } diff --git a/trader/bybit/order_sync.go b/trader/bybit/order_sync.go index cf54afac..147585b0 100644 --- a/trader/bybit/order_sync.go +++ b/trader/bybit/order_sync.go @@ -10,6 +10,7 @@ import ( "net/http" "nofx/logger" "nofx/market" + "nofx/safe" "nofx/store" "sort" "strconv" @@ -300,12 +301,12 @@ func (t *BybitTrader) SyncOrdersFromBybit(traderID string, exchangeID string, ex // StartOrderSync starts background order sync task for Bybit func (t *BybitTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("bybit-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromBybit(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Bybit order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ Bybit order sync started (interval: %v)", interval) } diff --git a/trader/gate/order_sync.go b/trader/gate/order_sync.go index 85d49ebf..59a9bae9 100644 --- a/trader/gate/order_sync.go +++ b/trader/gate/order_sync.go @@ -4,6 +4,7 @@ import ( "fmt" "nofx/logger" "nofx/market" + "nofx/safe" "nofx/store" "sort" "strconv" @@ -293,12 +294,12 @@ func (t *GateTrader) SyncOrdersFromGate(traderID string, exchangeID string, exch // StartOrderSync starts background order sync task for Gate func (t *GateTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("gate-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromGate(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Gate order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ Gate order sync started (interval: %v)", interval) } diff --git a/trader/hyperliquid/order_sync.go b/trader/hyperliquid/order_sync.go index 752e5c3d..65b3225e 100644 --- a/trader/hyperliquid/order_sync.go +++ b/trader/hyperliquid/order_sync.go @@ -3,6 +3,7 @@ package hyperliquid import ( "fmt" "nofx/logger" + "nofx/safe" "nofx/market" "nofx/store" "sort" @@ -138,12 +139,12 @@ func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeI // StartOrderSync starts background order sync task func (t *HyperliquidTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("hyperliquid-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromHyperliquid(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ Hyperliquid order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ Hyperliquid order sync started (interval: %v)", interval) } diff --git a/trader/kucoin/order_sync.go b/trader/kucoin/order_sync.go index a817075e..1038a97b 100644 --- a/trader/kucoin/order_sync.go +++ b/trader/kucoin/order_sync.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "nofx/logger" + "nofx/safe" "nofx/store" "nofx/trader/types" "sort" @@ -401,12 +402,12 @@ func (t *KuCoinTrader) SyncOrdersFromKuCoin(traderID string, exchangeID string, // StartOrderSync starts background order sync task for KuCoin func (t *KuCoinTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("kucoin-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromKuCoin(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ KuCoin order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ KuCoin order sync started (interval: %v)", interval) } diff --git a/trader/lighter/order_sync.go b/trader/lighter/order_sync.go index 24fab52d..f0bccd84 100644 --- a/trader/lighter/order_sync.go +++ b/trader/lighter/order_sync.go @@ -4,6 +4,7 @@ import ( "fmt" "nofx/logger" "nofx/market" + "nofx/safe" "nofx/store" "sort" "strings" @@ -147,7 +148,7 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, exchangeID stri // StartOrderSync starts background order sync task func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("lighter-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromLighter(traderID, exchangeID, exchangeType, st); err != nil { // Only log non-404 errors to reduce log spam @@ -156,6 +157,6 @@ func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exc } } } - }() + }) logger.Infof("πŸ”„ Lighter order+position sync started (interval: %v)", interval) } diff --git a/trader/okx/order_sync.go b/trader/okx/order_sync.go index a8246ecb..4028e637 100644 --- a/trader/okx/order_sync.go +++ b/trader/okx/order_sync.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "nofx/logger" + "nofx/safe" "nofx/market" "nofx/store" "sort" @@ -274,12 +275,12 @@ func (t *OKXTrader) SyncOrdersFromOKX(traderID string, exchangeID string, exchan // StartOrderSync starts background order sync task for OKX func (t *OKXTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { ticker := time.NewTicker(interval) - go func() { + safe.GoNamed("okx-order-sync", func() { for range ticker.C { if err := t.SyncOrdersFromOKX(traderID, exchangeID, exchangeType, st); err != nil { logger.Infof("⚠️ OKX order sync failed: %v", err) } } - }() + }) logger.Infof("πŸ”„ OKX order sync started (interval: %v)", interval) }