From 958da76d33dbb8c968c2fd7a9ee81bf4b87876c1 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Thu, 25 Jun 2026 14:07:59 +0700 Subject: [PATCH] fix(modules): reserve reply budget and cap price-fetch timeouts in stats handlers Stats handlers reused the single bounded update context for both upstream price fetches and the final Telegram reply, while per-upstream HTTP timeouts equalled the whole-handler budget. One slow upstream drained the deadline and the reply failed with "context deadline exceeded" (observed on /stock_stats). - add chathelper.FetchContext: fetches run under a child context that reserves a tail of the deadline for the reply, which is sent on the original context - cap kbs/coin/gold/vnappmob HTTP timeouts at 3s so one hung upstream cannot consume the reply budget - fetch held stock/coin prices concurrently (bounded) so total latency tracks the slowest single fetch instead of their sum; per-symbol failures degrade to a "no price" line and the summary still sends --- go.mod | 2 +- internal/modules/coin/prices.go | 7 +- internal/modules/coin/views.go | 49 ++++++++++--- .../modules/coin/views_reply_budget_test.go | 62 ++++++++++++++++ internal/modules/gold/handlers.go | 12 +++- internal/modules/gold/prices.go | 2 +- internal/modules/gold/vnappmob_client.go | 2 +- internal/modules/stock/handlers.go | 50 +++++++++---- internal/modules/stock/prices.go | 7 +- .../modules/util/chathelper/chathelper.go | 25 +++++++ .../util/chathelper/chathelper_test.go | 41 +++++++++++ .../plan.md | 72 +++++++++++++++++++ 12 files changed, 297 insertions(+), 34 deletions(-) create mode 100644 internal/modules/coin/views_reply_budget_test.go create mode 100644 plans/260625-1337-stats-reply-budget-timeout/plan.md diff --git a/go.mod b/go.mod index a694a14..ba9eafe 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ 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 @@ -52,7 +53,6 @@ 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/prices.go b/internal/modules/coin/prices.go index 38fc43e..5fcfe21 100644 --- a/internal/modules/coin/prices.go +++ b/internal/modules/coin/prices.go @@ -14,8 +14,11 @@ const ( binanceDefaultURL = "https://data-api.binance.vision/api/v3/ticker/price" coinbaseDefaultURL = "https://api.coinbase.com/v2/exchange-rates" coinGeckoDefaultURL = "https://api.coingecko.com/api/v3/simple/price" - coinHTTPTimeout = 10 * time.Second - coinPriceCacheTTL = 30 * time.Second + // coinHTTPTimeout caps a single provider call, kept under the handler + // deadline so one slow provider cannot starve the Telegram reply budget + // (see chathelper.FetchContext). + coinHTTPTimeout = 3 * time.Second + coinPriceCacheTTL = 30 * time.Second ) var ErrNoCoinPrice = errors.New("coin: no price available") diff --git a/internal/modules/coin/views.go b/internal/modules/coin/views.go index 99367e8..7111a27 100644 --- a/internal/modules/coin/views.go +++ b/internal/modules/coin/views.go @@ -7,6 +7,7 @@ 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" @@ -24,19 +25,45 @@ 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 - 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(ctx, coin); err == nil { - value := held * price.USD - totalValue += value - line += " = " + FormatUSD(value) + " @ " + FormatUSD(price.USD) + " (" + price.Source + ")" - } else { + + // 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) + 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 + } line += " (price unavailable)" } - } - lines = append(lines, line) + 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, "Total value: "+FormatUSD(totalValue)) lines = append(lines, "Invested: "+FormatUSD(p.Meta.Invested)) diff --git a/internal/modules/coin/views_reply_budget_test.go b/internal/modules/coin/views_reply_budget_test.go new file mode 100644 index 0000000..04366b8 --- /dev/null +++ b/internal/modules/coin/views_reply_budget_test.go @@ -0,0 +1,62 @@ +package coin + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/tiennm99/miti99bot/internal/testutil" +) + +// blockingPriceFetcher simulates an upstream that never responds: it blocks +// until the fetch context is cancelled, then returns its error. This is the +// exact failure that made /coin_stats (and /stock_stats) time out — the fetch +// must not be allowed to consume the budget the reply needs. +type blockingPriceFetcher struct{} + +func (blockingPriceFetcher) FetchUSD(ctx context.Context, _ CoinSymbol) (CoinPrice, error) { + <-ctx.Done() + return CoinPrice{}, ctx.Err() +} + +// TestHandleStatsDeliversReplyWhenUpstreamHangs proves the reply-reserve fix: +// even when the price upstream hangs for the entire fetch budget, handleStats +// still delivers a summary (with a "price unavailable" line) on the original +// context instead of failing the whole reply with "context deadline exceeded". +func TestHandleStatsDeliversReplyWhenUpstreamHangs(t *testing.T) { + // Seed a holding using a fast fetcher, then swap in the hanging upstream. + s := newTestState(map[string]CoinPrice{"BTC": {USD: 100}}, nil) + rb := testutil.NewRecordingBot(t) + setupCtx := context.Background() + if err := s.handleTopup(setupCtx, rb.Bot, testutil.NewPrivateMessage(7, "/coin_topup 1000")); err != nil { + t.Fatalf("topup: %v", err) + } + if err := s.handleBuy(setupCtx, rb.Bot, testutil.NewPrivateMessage(7, "/coin_buy 500 BTC")); err != nil { + t.Fatalf("buy: %v", err) + } + + s.prices = blockingPriceFetcher{} + rb.Reset() + + // 4s parent deadline → ~1s fetch budget (replyReserve is 3s), leaving the + // reply ample headroom. + statsCtx, cancel := context.WithTimeout(context.Background(), 4*time.Second) + defer cancel() + start := time.Now() + if err := s.handleStats(statsCtx, rb.Bot, testutil.NewPrivateMessage(7, "/coin_stats")); err != nil { + t.Fatalf("handleStats returned error (reply not delivered): %v", err) + } + elapsed := time.Since(start) + + if statsCtx.Err() != nil { + t.Fatalf("parent context expired before reply (no budget reserved): %v", statsCtx.Err()) + } + if elapsed >= 3*time.Second { + t.Fatalf("handleStats took %v — fetch was not bounded below the reply reserve", elapsed) + } + sent := rb.LastSent().Text() + if !strings.Contains(sent, "Coin Account Summary") || !strings.Contains(sent, "price unavailable") { + t.Fatalf("reply missing summary / degraded line; got:\n%s", sent) + } +} diff --git a/internal/modules/gold/handlers.go b/internal/modules/gold/handlers.go index 320c530..2a7e55e 100644 --- a/internal/modules/gold/handlers.go +++ b/internal/modules/gold/handlers.go @@ -23,7 +23,12 @@ func (s *state) handlePrice(ctx context.Context, b *bot.Bot, update *models.Upda if len(args) != 0 { return chathelper.Reply(ctx, b, update.Message, "Usage: /gold_price") } - p, err := s.prices.FetchPrice(ctx) + // Fetch under a reply-reserved sub-context (the composite fetcher may try + // providers sequentially); reply on the original ctx so delivery keeps its + // budget headroom. + fetchCtx, cancel := chathelper.FetchContext(ctx) + defer cancel() + p, err := s.prices.FetchPrice(fetchCtx) if err != nil { return s.replyPriceError(ctx, b, update, err) } @@ -176,7 +181,10 @@ func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Upda lines := []string{"Gold Account Summary\n", "VND: " + FormatVND(p.VND), "Gold: " + FormatLuong(p.Luong) + " luong"} totalValue := p.VND - if buyPrice, _, err := s.prices.FetchLuongPrices(ctx); err == nil { + // Fetch under a reply-reserved sub-context; reply on the original ctx. + fetchCtx, cancel := chathelper.FetchContext(ctx) + defer cancel() + if buyPrice, _, err := s.prices.FetchLuongPrices(fetchCtx); err == nil { goldValue := p.Luong * buyPrice totalValue += goldValue lines = append(lines, "Price: "+FormatVND(buyPrice)+"/luong") diff --git a/internal/modules/gold/prices.go b/internal/modules/gold/prices.go index 934c35a..8525862 100644 --- a/internal/modules/gold/prices.go +++ b/internal/modules/gold/prices.go @@ -15,7 +15,7 @@ import ( const ( fxDefaultURL = "https://open.er-api.com/v6/latest/USD" - goldHTTPTimeout = 10 * time.Second + goldHTTPTimeout = 3 * time.Second // kept under the handler deadline; see chathelper.FetchContext fxFallbackCacheTTL = time.Hour gramsPerLuong = 37.5 gramsPerTroyOunce = 31.1034768 diff --git a/internal/modules/gold/vnappmob_client.go b/internal/modules/gold/vnappmob_client.go index 51bb66b..7b6b2cd 100644 --- a/internal/modules/gold/vnappmob_client.go +++ b/internal/modules/gold/vnappmob_client.go @@ -22,7 +22,7 @@ const ( vnappmobDefaultURL = "https://api.vnappmob.com" vnappmobKeyCacheKey = "vnappmob:api_key" vnappmobRefreshBuffer = 24 * time.Hour - vnappmobHTTPTimeout = 10 * time.Second + vnappmobHTTPTimeout = 3 * time.Second // kept under the handler deadline; see chathelper.FetchContext ) // VNAppMobClient fetches Vietnam SJC gold prices from api.vnappmob.com. diff --git a/internal/modules/stock/handlers.go b/internal/modules/stock/handlers.go index 0029efc..59d83e9 100644 --- a/internal/modules/stock/handlers.go +++ b/internal/modules/stock/handlers.go @@ -9,6 +9,7 @@ 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" @@ -365,19 +366,42 @@ func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Upda if len(heldList) > 0 { lines = append(lines, "\nStocks:") - // Sequential price fetch (Lambda has no concurrency benefit at small N - // and goroutines complicate test seams). For typical <10 holdings, - // total latency is bounded by sum-of-fetches; KBS responds in <500ms. - for _, h := range heldList { - price, err := s.prices.FetchPrice(ctx, h.symbol) - if err != nil { - 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)) + // 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. + 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 } } diff --git a/internal/modules/stock/prices.go b/internal/modules/stock/prices.go index cc19a76..5a9a92c 100644 --- a/internal/modules/stock/prices.go +++ b/internal/modules/stock/prices.go @@ -18,9 +18,10 @@ const kbsDefaultURL = "https://kbbuddywts.kbsec.com.vn/iis-server/investment/sto // market holidays — KBS returns the latest bar within the window in [0]. const kbsLookbackDays = 14 -// kbsHTTPTimeout caps the price fetch. KBS is generally fast; 10s leaves -// headroom for TLS + DNS on a Lambda cold start. -const kbsHTTPTimeout = 10 * time.Second +// kbsHTTPTimeout caps a single ticker's price fetch. Kept well under the +// handler's overall deadline so one slow/hung ticker cannot drain the budget +// the handler needs to deliver its Telegram reply (see chathelper.FetchContext). +const kbsHTTPTimeout = 3 * time.Second // PriceClient is the KBS price fetcher. Zero value uses the default URL + // `&{Timeout: kbsHTTPTimeout}` HTTP client; tests inject HTTP + URL. diff --git a/internal/modules/util/chathelper/chathelper.go b/internal/modules/util/chathelper/chathelper.go index f9fc4bf..5428886 100644 --- a/internal/modules/util/chathelper/chathelper.go +++ b/internal/modules/util/chathelper/chathelper.go @@ -51,6 +51,31 @@ func ArgAfterCommand(text string) string { // NowMillis returns current UTC ms-since-epoch. func NowMillis() int64 { return time.Now().UTC().UnixMilli() } +// replyReserve is the slice of the handler's deadline kept aside for delivering +// the Telegram reply. The whole update handler runs under one bounded context +// (see internal/telegram/webhook.go); if upstream price fetches consume all of +// it, the final SendMessage fails with "context deadline exceeded" and the user +// sees no response. Reserving a fixed tail guarantees delivery headroom. +const replyReserve = 3 * time.Second + +// FetchContext derives a child of ctx for upstream data fetches, leaving +// replyReserve of the parent's deadline for the subsequent Reply (which must be +// called with the original ctx, not this child). If ctx has no deadline, or +// less than replyReserve remains, the child gets a small positive floor so a +// fetch still attempts rather than failing instantly. Callers must call the +// returned cancel. +func FetchContext(ctx context.Context) (context.Context, context.CancelFunc) { + dl, ok := ctx.Deadline() + if !ok { + return context.WithCancel(ctx) + } + budget := time.Until(dl) - replyReserve + if budget < time.Second { + budget = time.Second + } + return context.WithTimeout(ctx, budget) +} + // Reply sends a plain-text response to the chat the inbound message came from. // // Forwards MessageThreadID so replies in a forum-supergroup topic stay in the diff --git a/internal/modules/util/chathelper/chathelper_test.go b/internal/modules/util/chathelper/chathelper_test.go index 9079065..b4be0de 100644 --- a/internal/modules/util/chathelper/chathelper_test.go +++ b/internal/modules/util/chathelper/chathelper_test.go @@ -3,12 +3,53 @@ package chathelper import ( "context" "testing" + "time" "github.com/go-telegram/bot/models" "github.com/tiennm99/miti99bot/internal/testutil" ) +func TestFetchContext(t *testing.T) { + t.Run("reserves reply budget from parent deadline", func(t *testing.T) { + parent, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + child, childCancel := FetchContext(parent) + defer childCancel() + dl, ok := child.Deadline() + if !ok { + t.Fatal("child has no deadline") + } + // budget = parent remaining (~10s) - replyReserve (3s) ≈ 7s. + if d := time.Until(dl); d > 8*time.Second || d < 6*time.Second { + t.Fatalf("fetch budget = %v, want ≈7s (10s parent - 3s reserve)", d) + } + }) + + t.Run("floors to 1s when parent deadline is within the reserve", func(t *testing.T) { + parent, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + child, childCancel := FetchContext(parent) + defer childCancel() + dl, _ := child.Deadline() + if d := time.Until(dl); d < 900*time.Millisecond || d > 1100*time.Millisecond { + t.Fatalf("floored budget = %v, want ≈1s", d) + } + }) + + t.Run("no parent deadline yields a cancelable child", func(t *testing.T) { + child, childCancel := FetchContext(context.Background()) + defer childCancel() + if _, ok := child.Deadline(); ok { + t.Fatal("child unexpectedly has a deadline") + } + childCancel() + if child.Err() == nil { + t.Fatal("cancel did not propagate to child") + } + }) +} + func TestSubjectFor(t *testing.T) { tests := []struct { name string diff --git a/plans/260625-1337-stats-reply-budget-timeout/plan.md b/plans/260625-1337-stats-reply-budget-timeout/plan.md new file mode 100644 index 0000000..6c10806 --- /dev/null +++ b/plans/260625-1337-stats-reply-budget-timeout/plan.md @@ -0,0 +1,72 @@ +# Fix: stats handlers starve Telegram reply of context budget + +Status: DONE (2026-06-25) — implemented, `make vet` + `make test -race` green, code-review DONE (no critical/high). + +## Problem + +Update handler runs under one 10s ctx (`telegram/webhook.go:81`). Stats handlers reuse +that same ctx for both upstream price fetches and the final `sendMessage`. Per-upstream +HTTP timeout is also 10s (== handler budget), so one slow upstream drains the deadline and +the reply fails: `context deadline exceeded`. Confirmed in CloudWatch for `/stock_stats` +(dispatch→error exactly 10.0s; coin/gold succeeded same window — KBS was slow). + +## Remedies (all approved) + +1. **Reserve reply budget** — fetches run under a derived sub-ctx that leaves a reserve for + the reply; the reply itself uses the original handler ctx. +2. **Lower per-fetch HTTP timeout** 10s → 3s, so one hung upstream can't eat the budget. +3. **Parallelize** the per-symbol fetch loops (stock, coin) so total ≈ slowest single fetch. + +## Acceptance criteria + +- `/stock_stats`, `/coin_stats` with N holdings: total fetch time bounded by slowest single + fetch (~3s), not N×. Reply always delivered if any upstream responds within budget. +- A single unresponsive upstream → that line shows "(no price)" / "(price unavailable)"; + the summary still sends. No more whole-reply `context deadline exceeded`. +- `make vet` + `make test` green. No public-contract changes. Output text/order unchanged. + +## Scope + +IN: stock/coin/gold stats handlers; gold price handler; the 4 HTTP-timeout consts. +OUT: buy/sell single-fetch handlers (the 3s timeout alone leaves ~7s reply headroom); +`incomeEventsHTTPTimeout` (separate command, not a stats loop); webhook 10s budget itself. + +## Changes + +### New shared helper — `internal/modules/util/chathelper/chathelper.go` +- `const replyReserve = 3 * time.Second` +- `func FetchContext(ctx) (context.Context, context.CancelFunc)` — derive child ctx leaving + `replyReserve` before the parent deadline (floor 1s); pass-through if parent has no deadline. + +### `internal/modules/stock/prices.go` +- `kbsHTTPTimeout` 10s → 3s. + +### `internal/modules/stock/handlers.go` (`handleStats`) +- Build `heldList`, fetch prices concurrently via `errgroup` (SetLimit 8) into an indexed + results slice (preserves order), using `chathelper.FetchContext(ctx)`. Reply on original ctx. +- Fix the stale "in parallel" comment (now actually parallel). + +### `internal/modules/coin/prices.go` + `internal/modules/coin/views.go` (`handleStats`) +- `coinHTTPTimeout` 10s → 3s. +- Parallelize the `sortedAssetSymbols` loop the same way; reply on original ctx. + +### `internal/modules/gold/prices.go`, `vnappmob_client.go`, `handlers.go` +- `goldHTTPTimeout` + `vnappmobHTTPTimeout` 10s → 3s. +- `handleStats` + `handlePrice`: fetch under `chathelper.FetchContext(ctx)`, reply on original ctx. + (Single fetch but composite tries providers sequentially → reserve guarantees reply headroom.) + +### `go.mod` +- `errgroup` (golang.org/x/sync) moves indirect → direct via `go mod tidy`. + +## Risks / rollback + +- Concurrency: `PriceClient`/coin client HTTP clients are safe for concurrent reuse (shared + pool via sync.Once). Result slice written by index → no shared-write race. +- 3s may be tight on Lambda cold-start TLS to a slow upstream; mitigated by reserve+parallel + and graceful "(no price)" fallback. Revert = restore constants to 10s. +- Tests inject HTTP client + URL; lowering real-client timeout doesn't affect injected ones. + +## Validation + +`make vet`, `make test`. Add focused test: stats handler with a hanging upstream returns a +reply (not a deadline error) within budget.