mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
fix: avoid zero-rate queue reader throttling (#9619)
## Summary - When `persistenceMaxQPS` is set to `0` (meaning "unlimited"), the queue reader rate limiter receives a zero rate, causing all queue readers to fail with "burst size is smaller than required token count" and never process tasks. - This fixes the rate function to return a high default (100,000) when persistence QPS is unlimited, and adds a missing `return` after the error log to prevent falling through to an invalid lock/read path. ## Bug Details `NewHostRateLimiterRateFn` computes `float64(persistenceMaxRPS()) * persistenceMaxRPSRatio`. When `persistenceMaxRPS()` returns 0 (unlimited), this produces rate=0, which creates a rate limiter with burst=0. The reader then cannot acquire even 1 token, logging the error but continuing into `r.Lock()` without tasks — effectively a tight error loop. ## Testing - Verified with benchmarks that queue readers function correctly with `persistenceMaxQPS: 0` in dynamic config. - Existing unit tests pass.
This commit is contained in:
@@ -226,9 +226,17 @@ func NewHostRateLimiterRateFn(
|
||||
return float64(maxPollHostRps)
|
||||
}
|
||||
|
||||
// ensure queue loading won't consume all persistence tokens
|
||||
// especially upon host restart when we need to perform a load
|
||||
// for all shards
|
||||
return float64(persistenceMaxRPS()) * persistenceMaxRPSRatio
|
||||
if pMaxRPS := persistenceMaxRPS(); pMaxRPS > 0 {
|
||||
// ensure queue loading won't consume all persistence tokens
|
||||
// especially upon host restart when we need to perform a load
|
||||
// for all shards
|
||||
return float64(pMaxRPS) * persistenceMaxRPSRatio
|
||||
}
|
||||
|
||||
// persistenceMaxQPS=0 means "unlimited" — use a high default to avoid
|
||||
// producing a zero rate which would block all queue readers.
|
||||
// NOTE: Do not use math.MaxFloat64 here; int(math.MaxFloat64) overflows
|
||||
// to math.MinInt64, which breaks burst calculation in the rate limiter.
|
||||
return 100_000
|
||||
}
|
||||
}
|
||||
|
||||
@@ -431,6 +431,7 @@ func (r *ReaderImpl) loadAndSubmitTasks() {
|
||||
|
||||
// this should never happen
|
||||
r.logger.Error("Queue reader rate limiter burst size is smaller than required token count")
|
||||
return
|
||||
}
|
||||
|
||||
r.Lock()
|
||||
|
||||
Reference in New Issue
Block a user