mirror of
https://github.com/tiennm99/miti99bot.git
synced 2026-09-08 20:20:17 +00:00
fix(modules): fetch stats prices sequentially to reuse pooled connections
Parallelizing the per-symbol fetches opened N simultaneous TLS handshakes into an empty connection pool. On the memory-constrained Lambda (256MB ~0.15 vCPU) those CPU-bound handshakes thrashed and each exceeded the per-fetch timeout, so every ticker rendered "(no price)". Sequential fetches reuse the price client's keep-alive connection (one handshake), which is why the code was sequential by design. - revert stock and coin stats loops to sequential (keep the reply-budget sub-context and 3s per-fetch timeout) - log per-symbol fetch failures instead of silently swallowing them
This commit is contained in:
@@ -9,7 +9,6 @@ require (
|
||||
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.3
|
||||
github.com/aws/aws-sdk-go-v2/service/ssm v1.68.6
|
||||
github.com/go-telegram/bot v1.20.0
|
||||
golang.org/x/sync v0.20.0
|
||||
golang.org/x/time v0.15.0
|
||||
google.golang.org/api v0.274.0
|
||||
google.golang.org/genai v1.56.0
|
||||
@@ -53,6 +52,7 @@ require (
|
||||
golang.org/x/crypto v0.51.0 // indirect
|
||||
golang.org/x/net v0.54.0 // indirect
|
||||
golang.org/x/oauth2 v0.36.0 // indirect
|
||||
golang.org/x/sync v0.20.0 // indirect
|
||||
golang.org/x/sys v0.44.0 // indirect
|
||||
golang.org/x/text v0.37.0 // indirect
|
||||
google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
|
||||
"github.com/go-telegram/bot"
|
||||
"github.com/go-telegram/bot/models"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/tiennm99/miti99bot/internal/log"
|
||||
"github.com/tiennm99/miti99bot/internal/modules/util/chathelper"
|
||||
@@ -26,44 +25,28 @@ func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Upda
|
||||
lines := []string{"Coin Account Summary\n", "USD: " + FormatUSD(p.USD)}
|
||||
totalValue := p.USD
|
||||
|
||||
// Fetch every held coin's price concurrently (bounded), under a
|
||||
// reply-reserved sub-context so a slow provider cannot drain the budget the
|
||||
// final Reply needs. Per-coin errors degrade to "(price unavailable)";
|
||||
// results are written by index (no shared-write race) and rendered in order.
|
||||
symbols := sortedAssetSymbols(p.Assets)
|
||||
// Fetch sequentially (not concurrently) so the price client's keep-alive
|
||||
// connection pool is reused across coins rather than opening N simultaneous
|
||||
// TLS handshakes — the latter thrashes the CPU-constrained Lambda and times
|
||||
// out. The reply-reserved sub-context bounds the whole loop so the final
|
||||
// Reply keeps its budget; a slow/failed provider degrades to "(price
|
||||
// unavailable)" instead of failing the summary.
|
||||
fetchCtx, cancel := chathelper.FetchContext(ctx)
|
||||
defer cancel()
|
||||
type coinResult struct {
|
||||
line string
|
||||
value float64
|
||||
}
|
||||
results := make([]coinResult, len(symbols))
|
||||
var g errgroup.Group
|
||||
g.SetLimit(8)
|
||||
for i, symbol := range symbols {
|
||||
i, symbol := i, symbol
|
||||
g.Go(func() error {
|
||||
held := p.Assets[symbol]
|
||||
line := symbol + ": " + FormatCoinQty(held)
|
||||
if coin, err := ResolveCoinSymbol(symbol); err == nil {
|
||||
if price, err := s.prices.FetchUSD(fetchCtx, coin); err == nil {
|
||||
value := held * price.USD
|
||||
results[i] = coinResult{
|
||||
line: line + " = " + FormatUSD(value) + " @ " + FormatUSD(price.USD) + " (" + price.Source + ")",
|
||||
value: value,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
for _, symbol := range sortedAssetSymbols(p.Assets) {
|
||||
held := p.Assets[symbol]
|
||||
line := symbol + ": " + FormatCoinQty(held)
|
||||
if coin, err := ResolveCoinSymbol(symbol); err == nil {
|
||||
if price, err := s.prices.FetchUSD(fetchCtx, coin); err == nil {
|
||||
value := held * price.USD
|
||||
totalValue += value
|
||||
line += " = " + FormatUSD(value) + " @ " + FormatUSD(price.USD) + " (" + price.Source + ")"
|
||||
} else {
|
||||
log.Error("coin_fetch_price", "symbol", symbol, "err", err)
|
||||
line += " (price unavailable)"
|
||||
}
|
||||
results[i] = coinResult{line: line}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
_ = g.Wait() // closures never return an error; partial results are intended
|
||||
for _, r := range results {
|
||||
lines = append(lines, r.line)
|
||||
totalValue += r.value
|
||||
}
|
||||
lines = append(lines, line)
|
||||
}
|
||||
lines = append(lines, "Total value: "+FormatUSD(totalValue))
|
||||
lines = append(lines, "Invested: "+FormatUSD(p.Meta.Invested))
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
|
||||
"github.com/go-telegram/bot"
|
||||
"github.com/go-telegram/bot/models"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/tiennm99/miti99bot/internal/keylock"
|
||||
"github.com/tiennm99/miti99bot/internal/log"
|
||||
@@ -329,8 +328,9 @@ func (s *state) handleConvert(ctx context.Context, b *bot.Bot, update *models.Up
|
||||
"Currency exchange is not available yet.\n"+s.comingSoonMessage)
|
||||
}
|
||||
|
||||
// handleStats fetches every held ticker's current price (in parallel) and
|
||||
// renders the portfolio. Read-only — no portfolio mutation, so no keylock.
|
||||
// handleStats fetches every held ticker's current price sequentially (reusing
|
||||
// the pooled KBS connection) and renders the portfolio. Read-only — no
|
||||
// portfolio mutation, so no keylock.
|
||||
func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Update) error {
|
||||
userID, ok := senderInfo(update)
|
||||
if !ok {
|
||||
@@ -366,42 +366,28 @@ func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Upda
|
||||
|
||||
if len(heldList) > 0 {
|
||||
lines = append(lines, "\nStocks:")
|
||||
// Fetch every held ticker concurrently so total latency is bounded by
|
||||
// the slowest single fetch, not their sum. Fetches run under a
|
||||
// reply-reserved sub-context (FetchContext) so a slow upstream cannot
|
||||
// drain the budget the final Reply needs; per-fetch errors degrade to
|
||||
// "(no price)" rather than failing the whole summary. Results are
|
||||
// written by index — no shared-write race — and rendered in order.
|
||||
// Fetch sequentially, NOT concurrently. The memoised HTTP client keeps a
|
||||
// keep-alive connection pool across calls (see PriceClient), so serial
|
||||
// fetches to the same KBS host pay one TLS handshake and reuse the
|
||||
// connection. Firing them in parallel instead opens N simultaneous
|
||||
// handshakes into an empty pool; on the memory-constrained Lambda
|
||||
// (256MB ≈ 0.15 vCPU) those CPU-bound handshakes thrash and each blows
|
||||
// past the per-fetch timeout. The reply-reserved sub-context bounds the
|
||||
// whole loop so the final Reply keeps its budget; a failed/slow ticker
|
||||
// degrades to "(no price)" rather than failing the summary.
|
||||
fetchCtx, cancel := chathelper.FetchContext(ctx)
|
||||
defer cancel()
|
||||
type stockResult struct {
|
||||
line string
|
||||
value float64
|
||||
}
|
||||
results := make([]stockResult, len(heldList))
|
||||
var g errgroup.Group
|
||||
g.SetLimit(8)
|
||||
for i, h := range heldList {
|
||||
i, h := i, h
|
||||
g.Go(func() error {
|
||||
price, err := s.prices.FetchPrice(fetchCtx, h.symbol)
|
||||
if err != nil {
|
||||
results[i] = stockResult{line: " " + h.symbol + " x" + FormatStock(float64(h.qty)) + " (no price)"}
|
||||
return nil
|
||||
}
|
||||
val := float64(h.qty) * price
|
||||
results[i] = stockResult{
|
||||
line: " " + h.symbol + " x" + FormatStock(float64(h.qty)) +
|
||||
" @ " + FormatVND(price) + " = " + FormatVND(val),
|
||||
value: val,
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
_ = g.Wait() // closures never return an error; partial results are intended
|
||||
for _, r := range results {
|
||||
lines = append(lines, r.line)
|
||||
totalValue += r.value
|
||||
for _, h := range heldList {
|
||||
price, err := s.prices.FetchPrice(fetchCtx, h.symbol)
|
||||
if err != nil {
|
||||
log.Error("stock_fetch_price", "symbol", h.symbol, "err", err)
|
||||
lines = append(lines, " "+h.symbol+" x"+FormatStock(float64(h.qty))+" (no price)")
|
||||
continue
|
||||
}
|
||||
val := float64(h.qty) * price
|
||||
totalValue += val
|
||||
lines = append(lines, " "+h.symbol+" x"+FormatStock(float64(h.qty))+
|
||||
" @ "+FormatVND(price)+" = "+FormatVND(val))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,18 @@
|
||||
|
||||
Status: DONE (2026-06-25) — implemented, `make vet` + `make test -race` green, code-review DONE (no critical/high).
|
||||
|
||||
## Post-deploy correction (2026-06-25)
|
||||
|
||||
First deploy delivered the reply (original bug fixed) but every ticker showed "(no price)".
|
||||
Cause: parallelizing fetches opened N simultaneous TLS handshakes into an empty connection
|
||||
pool; on the 256MB Lambda (~0.15 vCPU) the CPU-bound handshakes thrashed and each exceeded
|
||||
the 3s timeout. Sequential fetches reuse the pooled KBS/provider connection (one handshake),
|
||||
which is why the original code was sequential by design.
|
||||
|
||||
Fix: reverted stock + coin loops to **sequential** (kept `FetchContext` reply reserve and the
|
||||
3s per-fetch timeout); dropped errgroup (back to indirect); added per-fetch error logging
|
||||
(`stock_fetch_price` / `coin_fetch_price`) to close the swallowed-error diagnostic gap.
|
||||
|
||||
## Problem
|
||||
|
||||
Update handler runs under one 10s ctx (`telegram/webhook.go:81`). Stats handlers reuse
|
||||
|
||||
Reference in New Issue
Block a user