From d02a110d8e76b882924e2e8eec46c3169835c141 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Thu, 25 Jun 2026 14:28:25 +0700 Subject: [PATCH] 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 --- go.mod | 2 +- internal/modules/coin/views.go | 53 ++++++---------- internal/modules/stock/handlers.go | 60 +++++++------------ .../plan.md | 12 ++++ 4 files changed, 54 insertions(+), 73 deletions(-) diff --git a/go.mod b/go.mod index ba9eafe..a694a14 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/internal/modules/coin/views.go b/internal/modules/coin/views.go index 7111a27..36613e9 100644 --- a/internal/modules/coin/views.go +++ b/internal/modules/coin/views.go @@ -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)) diff --git a/internal/modules/stock/handlers.go b/internal/modules/stock/handlers.go index 59d83e9..1429ea6 100644 --- a/internal/modules/stock/handlers.go +++ b/internal/modules/stock/handlers.go @@ -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)) } } diff --git a/plans/260625-1337-stats-reply-budget-timeout/plan.md b/plans/260625-1337-stats-reply-budget-timeout/plan.md index 6c10806..fe4a51d 100644 --- a/plans/260625-1337-stats-reply-budget-timeout/plan.md +++ b/plans/260625-1337-stats-reply-budget-timeout/plan.md @@ -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