From b9c9f05603ae08b02beed11e6552e36eea3c2ce6 Mon Sep 17 00:00:00 2001 From: shinchan-zhai Date: Mon, 23 Mar 2026 11:10:19 +0800 Subject: [PATCH] robustness: shared HTTP client, HTTP status checks, abort controller, streaming perf - web.go: reuse shared binanceClient with connection pooling instead of per-request client - sentinel.go: check HTTP status code before parsing Binance ticker response - brain.go: check HTTP status codes in news scan and market brief APIs - brain.go: add cleanStaleSignals() to prevent unbounded sync.Map growth - scheduler.go: periodically clean expired pending trades - AgentChatPage.tsx: add AbortController to cancel in-flight requests - AgentChatPage.tsx: check response.ok before parsing JSON - AgentChatPage.tsx: batch word streaming (~40 frames max) to reduce re-renders - AgentChatPage.tsx: handle AbortError gracefully (remove orphan bot message) --- agent/brain.go | 23 +++++++++++++- agent/scheduler.go | 4 +++ agent/sentinel.go | 4 +++ agent/web.go | 14 +++++++-- web/src/pages/AgentChatPage.tsx | 56 ++++++++++++++++++++++----------- 5 files changed, 79 insertions(+), 22 deletions(-) diff --git a/agent/brain.go b/agent/brain.go index 3c9acd27..e3961d49 100644 --- a/agent/brain.go +++ b/agent/brain.go @@ -32,6 +32,17 @@ func NewBrain(agent *Agent, logger *slog.Logger) *Brain { func (b *Brain) Stop() { close(b.stopCh) } +// cleanStaleSignals removes debounce entries older than 30 minutes. +func (b *Brain) cleanStaleSignals() { + cutoff := time.Now().Add(-30 * time.Minute) + b.recentSignals.Range(func(key, value any) bool { + if t, ok := value.(time.Time); ok && t.Before(cutoff) { + b.recentSignals.Delete(key) + } + return true + }) +} + func (b *Brain) HandleSignal(sig Signal) { key := fmt.Sprintf("%s:%s", sig.Type, sig.Symbol) if v, ok := b.recentSignals.Load(key); ok { @@ -53,11 +64,16 @@ func (b *Brain) StartNewsScan(interval time.Duration) { safe.GoNamed("brain-news-scan", func() { ticker := time.NewTicker(interval) defer ticker.Stop() + cleanTick := 0 for { select { case <-b.stopCh: return case <-ticker.C: b.scanNews(seen) + cleanTick++ + if cleanTick%6 == 0 { // every ~30 min + b.cleanStaleSignals() + } } } }) @@ -67,6 +83,10 @@ func (b *Brain) scanNews(seen map[string]bool) { resp, err := b.http.Get("https://min-api.cryptocompare.com/data/v2/news/?lang=EN&sortOrder=latest") if err != nil { return } defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + b.logger.Debug("news API non-200", "status", resp.StatusCode) + return + } body, err := io.ReadAll(io.LimitReader(resp.Body, 1024*1024)) // 1MB limit if err != nil { return } @@ -148,8 +168,9 @@ func (b *Brain) sendBrief(hour int) { resp, err := b.http.Get(fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=%s", sym)) if err != nil { continue } body, readErr := io.ReadAll(io.LimitReader(resp.Body, 64*1024)) // 64KB limit + statusOK := resp.StatusCode == http.StatusOK resp.Body.Close() - if readErr != nil { continue } + if readErr != nil || !statusOK { continue } var t map[string]string if err := json.Unmarshal(body, &t); err != nil { continue } if sym == "BTCUSDT" { btcPrice = t["lastPrice"]; btcChg = t["priceChangePercent"] } diff --git a/agent/scheduler.go b/agent/scheduler.go index dc251b98..6eba00ff 100644 --- a/agent/scheduler.go +++ b/agent/scheduler.go @@ -43,10 +43,14 @@ func (s *Scheduler) Start(ctx context.Context) { lastCheck = now } // Clean stale chat history every hour (sessions idle > 24h) + // and expired pending trades if now.Sub(lastCleanup) > 1*time.Hour { if s.agent.history != nil { s.agent.history.CleanOld(24 * time.Hour) } + if s.agent.pending != nil { + s.agent.pending.CleanExpired() + } lastCleanup = now } } diff --git a/agent/sentinel.go b/agent/sentinel.go index eb7e4f3f..186dbc56 100644 --- a/agent/sentinel.go +++ b/agent/sentinel.go @@ -116,6 +116,10 @@ func (s *Sentinel) check(symbol string) { resp, err := s.http.Get(fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=%s", symbol)) if err != nil { return } defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + s.logger.Debug("sentinel ticker non-200", "symbol", symbol, "status", resp.StatusCode) + return + } body, err := io.ReadAll(io.LimitReader(resp.Body, 256*1024)) // 256KB limit if err != nil { return } var t map[string]interface{} diff --git a/agent/web.go b/agent/web.go index 413dca6f..bf079571 100644 --- a/agent/web.go +++ b/agent/web.go @@ -17,6 +17,17 @@ var validSymbolRe = regexp.MustCompile(`^[A-Za-z0-9\-_]{1,20}$`) // validIntervalRe matches only valid kline intervals (e.g. 1m, 5m, 1h, 4h, 1d, 1w). var validIntervalRe = regexp.MustCompile(`^[0-9]{1,2}[mhHdDwWM]$`) +// binanceClient is a shared HTTP client for proxying Binance API requests. +// Reused across requests to benefit from connection pooling. +var binanceClient = &http.Client{ + Timeout: 10 * time.Second, + Transport: &http.Transport{ + MaxIdleConns: 20, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + }, +} + // WebHandler provides HTTP endpoints for the NOFXi agent. type WebHandler struct { agent *Agent @@ -105,8 +116,7 @@ func (w *WebHandler) HandleTicker(rw http.ResponseWriter, r *http.Request) { } func proxyBinance(rw http.ResponseWriter, url string) { - client := &http.Client{Timeout: 10 * time.Second} - resp, err := client.Get(url) + resp, err := binanceClient.Get(url) if err != nil { writeJSON(rw, 502, map[string]string{"error": "upstream request failed"}) return diff --git a/web/src/pages/AgentChatPage.tsx b/web/src/pages/AgentChatPage.tsx index 015e1933..f7359400 100644 --- a/web/src/pages/AgentChatPage.tsx +++ b/web/src/pages/AgentChatPage.tsx @@ -39,6 +39,7 @@ export function AgentChatPage() { const [loading, setLoading] = useState(false) const messagesEndRef = useRef(null) const chatInputRef = useRef(null) + const abortRef = useRef(null) // Sidebar section collapse state const [sections, setSections] = useState({ @@ -98,6 +99,11 @@ export function AgentChatPage() { setLoading(true) try { + // Abort any in-flight request + abortRef.current?.abort() + const controller = new AbortController() + abortRef.current = controller + const res = await fetch('/api/agent/chat', { method: 'POST', headers: { @@ -105,15 +111,22 @@ export function AgentChatPage() { ...(token ? { Authorization: `Bearer ${token}` } : {}), }, body: JSON.stringify({ message: text, lang: language }), + signal: controller.signal, }) + if (!res.ok) { + const errData = await res.json().catch(() => ({})) + throw new Error(errData.error || `Server error (${res.status})`) + } const data = await res.json() const responseText = data.response || data.error || 'No response' - // Simulate streaming by revealing text progressively + // Simulate streaming by revealing text in chunks (batched to reduce renders) const words = responseText.split(/(\s+)/) + const chunkSize = Math.max(3, Math.ceil(words.length / 40)) // ~40 frames max let displayed = '' - for (let i = 0; i < words.length; i++) { - displayed += words[i] + for (let i = 0; i < words.length; i += chunkSize) { + const chunk = words.slice(i, i + chunkSize).join('') + displayed += chunk const current = displayed setMessages((prev) => prev.map((m) => @@ -129,8 +142,8 @@ export function AgentChatPage() { : m ) ) - if (i < words.length - 1) { - await new Promise((r) => setTimeout(r, 10)) + if (i + chunkSize < words.length) { + await new Promise((r) => setTimeout(r, 20)) } } // Mark streaming done @@ -138,21 +151,26 @@ export function AgentChatPage() { prev.map((m) => (m.id === botId ? { ...m, streaming: false } : m)) ) } catch (e: any) { - setMessages((prev) => - prev.map((m) => - m.id === botId - ? { - ...m, - text: '⚠️ Error: ' + e.message, - time: new Date().toLocaleTimeString([], { - hour: '2-digit', - minute: '2-digit', - }), - streaming: false, - } - : m + if (e.name === 'AbortError') { + // Request was cancelled (e.g. user sent a new message), clean up silently + setMessages((prev) => prev.filter((m) => m.id !== botId)) + } else { + setMessages((prev) => + prev.map((m) => + m.id === botId + ? { + ...m, + text: '⚠️ Error: ' + e.message, + time: new Date().toLocaleTimeString([], { + hour: '2-digit', + minute: '2-digit', + }), + streaming: false, + } + : m + ) ) - ) + } } setLoading(false) chatInputRef.current?.focus()