feat(bigtable): add SessionThrottler / AdaptiveSessionThrottler for OpenSession pacing - #20184
Conversation
…penSession pacing
…er.go to match primary Go type
There was a problem hiding this comment.
Code Review
This pull request introduces AdaptiveSessionThrottler to manage pacing and rate-limiting for session creation. The reviewer identified a critical deadlock bug where blocked waiters in Acquire can hang indefinitely when a failure penalty expires because there is no mechanism to wake them up. To resolve this, the reviewer suggested using time.AfterFunc to trigger a broadcast when the penalty expires, and provided a corresponding test case to prevent regressions.
| func (b *AdaptiveSessionThrottler) Release(success bool) { | ||
| b.mu.Lock() | ||
| defer b.mu.Unlock() | ||
| if b.inUse > 0 { | ||
| b.inUse-- | ||
| } | ||
| if !success && b.penaltyDuration > 0 { | ||
| b.penalties = append(b.penalties, b.nowFn().Add(b.penaltyDuration)) | ||
| } | ||
| // A slot may have just opened up (success) or the ceiling may have | ||
| // grown since the waiter parked (UpdateConfig). Wake everyone; the | ||
| // loop re-checks under the lock. | ||
| b.cond.Broadcast() | ||
| } |
There was a problem hiding this comment.
There is a critical bug where blocked waiters in Acquire can deadlock indefinitely when a failure penalty expires.
The Issue
When Release(false) is called with a non-zero penaltyDuration, the slot is held as a penalty. The sum b.inUse + len(b.penalties) remains at maxConcurrent, meaning no slots are immediately freed. Therefore, any blocked waiters will remain asleep on b.cond.Wait().
However, when the penalty duration eventually expires, there is no background timer or mechanism to wake up the blocked waiters. If no other external events (like a successful Release or UpdateConfig) occur, the blocked waiters will hang indefinitely (or until their context times out), even though slots have technically become available.
Additionally, calling b.cond.Broadcast() immediately on a failed release when penaltyDuration > 0 is redundant and causes unnecessary CPU overhead (thundering herd) because no slots are actually freed at that moment.
The Solution
Use time.AfterFunc to schedule a b.cond.Broadcast() exactly when the penalty expires. This ensures that waiters are woken up precisely when a slot becomes available, and avoids redundant broadcasts on failed releases.
func (b *AdaptiveSessionThrottler) Release(success bool) {
b.mu.Lock()
defer b.mu.Unlock()
if b.inUse > 0 {
b.inUse--
}
if !success && b.penaltyDuration > 0 {
b.penalties = append(b.penalties, b.nowFn().Add(b.penaltyDuration))
time.AfterFunc(b.penaltyDuration, func() {
b.mu.Lock()
b.cond.Broadcast()
b.mu.Unlock()
})
} else {
// A slot opened up immediately. Wake up waiters.
b.cond.Broadcast()
}
}There was a problem hiding this comment.
Confirmed the deadlock — traced it: Release(false) leaves inUse+len(penalties) at max, so parked waiters keep failing tryAcquireLocked and only wake if an unrelated Release/UpdateConfig event fires the broadcast. Fixed in 7c59cde via time.AfterFunc(penaltyDuration, cond.Broadcast), and dropped the unconditional broadcast on the failure path since no slot has freed. Broadcast dispatched without holding b.mu (sync.Cond permits it) so the callback doesn't fight the Acquire loop for the lock.
| if got := tr.Snapshot(); got.InUse != 1 { | ||
| t.Fatalf("InUse after penalty drained + reacquire = %d, want 1", got.InUse) | ||
| } | ||
| } |
There was a problem hiding this comment.
To prevent regressions and verify that blocked waiters are correctly woken up when a penalty expires (without requiring any external triggers), we should add a dedicated test case.
Here is a suggested test implementation:
func TestThrottler_BlockedWaiterWokenByPenaltyExpiry(t *testing.T) {
tr := NewAdaptiveSessionThrottler(1, 50*time.Millisecond)
if err := tr.Acquire(context.Background()); err != nil {
t.Fatalf("Acquire: %v", err)
}
tr.Release(false) // failure -> slot held for 50ms
acquired := make(chan error, 1)
go func() {
acquired <- tr.Acquire(context.Background())
}()
select {
case err := <-acquired:
t.Fatalf("Acquire succeeded too early: %v", err)
case <-time.After(20 * time.Millisecond):
}
select {
case err := <-acquired:
if err != nil {
t.Fatalf("Acquire failed: %v", err)
}
case <-time.After(200 * time.Millisecond):
t.Fatal("Acquire blocked forever; penalty expiry did not wake it up")
}
}There was a problem hiding this comment.
Added TestThrottler_BlockedWaiterWokenByPenaltyExpiry in 7c59cde — real-time, asserts the parked waiter stays parked past 20ms and wakes before the 500ms fail-deadline. Passes under -race.
…pires Release(false) reserves the slot for penaltyDuration but did not schedule a wake, so a waiter parked on cond.Wait would hang until either an unrelated Release/UpdateConfig event fired the broadcast or the caller's ctx timed out — even though the slot was technically free once the penalty timer elapsed. Fix: schedule a time.AfterFunc(penaltyDuration, cond.Broadcast) on failed Release, and skip the immediate broadcast on that path since no slot has freed yet. Test asserts a parked waiter wakes at penalty expiry without any external event.
| } | ||
|
|
||
| // AdaptiveSessionThrottler is a concurrency governor with adaptive | ||
| // failure penalties. Semantics match Java's SessionCreationBudget |
There was a problem hiding this comment.
nit, we can remove the comment about java
…penSession pacing (googleapis#20184) ## Summary Adds `SessionThrottler` interface and `AdaptiveSessionThrottler` implementation under `bigtable/internal/transport/`. This is a concurrency governor with adaptive failure penalties for pacing `OpenSession` calls, matching Java's `SessionCreationBudget` (google-cloud-java: `SessionCreationBudget.java`). - A failed `OpenSession` keeps its slot reserved for `penaltyDuration` before returning it to the pool, so repeated failures throttle further attempts. - Counter+slice representation (rather than a chan-based semaphore) lets `UpdateConfig` raise or lower the ceiling at runtime without leaking in-flight callers. - `Snapshot()` exposes `InUse` / `Capacity` / `PenaltyDuration` for the debug UI. This is a self-contained, unwired primitive (stdlib-only: `context`, `sync`, `time`). A follow-up will wire it into `SessionPoolImpl.UpdateConfig` and the session-creation path. ## Test plan - [x] `go build ./internal/transport/...` - [x] `go vet ./internal/transport/...` - [x] `go test ./internal/transport/... -run 'Throttler' -count=1` (7 tests: acquire/release success, block-at-cap, ctx respect, failure-penalty slot hold, UpdateConfig grow unblocks, UpdateConfig shrink honored, concurrent smoke) - [x] `goimports -d` — no diffs - [x] `golint ./internal/transport/session_creation_budget{,_test}.go` — clean
🤖 I have created a release *beep* *boop* --- ## [1.51.0](bigtable/v1.50.0...bigtable/v1.51.0) (2026-07-23) ### Features * **bigtable:** Add ChainInterceptors and RetryingVRpc for vRPC pipeline ([#20185](#20185)) ([c7a832a](c7a832a)) * **bigtable:** Add ClientConfigurationManager ([#19986](#19986)) ([3a8f927](3a8f927)) * **bigtable:** Add debug tag counter (recordDebugTag / assertDebugTag) ([#20114](#20114)) ([3c97590](3c97590)) * **bigtable:** Add lazyPool helper for on-demand session pool opening ([#20182](#20182)) ([f6ae3fb](f6ae3fb)) * **bigtable:** Add PeakEwma continuous time-decay latency tracker ([#20187](#20187)) ([9d124ef](9d124ef)) * **bigtable:** Add PoolSizer for server-driven session pool capacity ([#20189](#20189)) ([57ebbeb](57ebbeb)) * **bigtable:** Add session package with SessionClient + SessionTableAPI interfaces ([#20180](#20180)) ([4b82fd2](4b82fd2)) * **bigtable:** Add Session primitives (AttemptOutcome, vRPC ctx, msgtype) ([#20116](#20116)) ([e1011e2](e1011e2)) * **bigtable:** Add Session state enum ([#19981](#19981)) ([0748972](0748972)) * **bigtable:** Add SessionThrottler / AdaptiveSessionThrottler for OpenSession pacing ([#20184](#20184)) ([02e3c6d](02e3c6d)) * **bigtable:** Add SessionThrottler / AdaptiveSessionThrottler for OpenSession pacing ([#20184](#20184)) ([29be83e](29be83e)) * **bigtable:** Add sessionTracer for per-Session lifecycle + vRPC metrics ([#20190](#20190)) ([a466345](a466345)) * **bigtable:** Enable new auth library and JWT for instance admin client ([#20013](#20013)) ([21c4a44](21c4a44)) * **bigtable:** Modularize channel priming behind a ChannelPrimer interface ([#20027](#20027)) ([5214ab7](5214ab7)) * **bigtable:** Modularize Direct Access compatibility check ([#19987](#19987)) ([a25e93d](a25e93d)) * **o11y:** Regenerate clients for LRO tracing ([#20107](#20107)) ([779074e](779074e)) ### Bug Fixes * **bigtable:** Default cluster/zone in toOtelMetricAttrs to avoid Monitoring reject ([#20178](#20178)) ([14493f4](14493f4)) * **bigtable:** Eliminate stats-handler MD race in internal/metrics tracer ([#20158](#20158)) ([c387066](c387066)) --- This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please). Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>
Summary
Adds
SessionThrottlerinterface andAdaptiveSessionThrottlerimplementation under
bigtable/internal/transport/. This is aconcurrency governor with adaptive failure penalties for pacing
OpenSessioncalls, matching Java'sSessionCreationBudget(google-cloud-java:
SessionCreationBudget.java).OpenSessionkeeps its slot reserved forpenaltyDurationbefore returning it to the pool, so repeated failures throttle
further attempts.
lets
UpdateConfigraise or lower the ceiling at runtime withoutleaking in-flight callers.
Snapshot()exposesInUse/Capacity/PenaltyDurationforthe debug UI.
This is a self-contained, unwired primitive (stdlib-only:
context,sync,time). A follow-up will wire it intoSessionPoolImpl.UpdateConfigand the session-creation path.Test plan
go build ./internal/transport/...go vet ./internal/transport/...go test ./internal/transport/... -run 'Throttler' -count=1(7 tests: acquire/release success, block-at-cap, ctx respect,
failure-penalty slot hold, UpdateConfig grow unblocks,
UpdateConfig shrink honored, concurrent smoke)
goimports -d— no diffsgolint ./internal/transport/session_creation_budget{,_test}.go— clean