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)
This commit is contained in:
shinchan-zhai
2026-03-23 11:10:19 +08:00
parent 7ccac89e2b
commit b9c9f05603
5 changed files with 79 additions and 22 deletions

View File

@@ -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"] }

View File

@@ -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
}
}

View File

@@ -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{}

View File

@@ -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