Handle task queue dispatch rate limiting in WCI autoscaling - #64
Handle task queue dispatch rate limiting in WCI autoscaling#64smuneebahmad wants to merge 4 commits into
Conversation
| TaskQueueName: th.taskQueueName, | ||
| TaskQueueType: th.taskQueueType, | ||
| IsSyncMatch: event.IsSyncMatch, | ||
| IsSyncMatch: noSyncMatchBatchCount == 0, // backward compatibility: true when no genuine not-matched events so old WCI workers don't scale up |
There was a problem hiding this comment.
is this really the same? I am a bit confused why the change is needed as part of the PR
| // configNoSyncRateLimitedSuppressQuietMsKey: in ProcessMetricsPoll, suppress scale-up for this many ms | ||
| // after the last observed rate-limited signal. 0 = disabled. | ||
| // Default is 2× the poll interval. | ||
| configNoSyncRateLimitedSuppressQuietMsKey = "rate_limited_suppress_quiet_ms" | ||
| configNoSyncRateLimitedSuppressQuietMsDefault = 2 * configNoSyncMetricsPollIntervalMsDefault | ||
|
|
||
| stateLastRateLimitedTimestampKey = "last_rate_limited_time_ms" |
There was a problem hiding this comment.
what is this needed for?
There was a problem hiding this comment.
Added more explanation around how these values are used to adjust ProcessMetricsPoll to handle TQ dispatch rate limiting.
There was a problem hiding this comment.
The explanation helps, but now I am unclear whether you are leveraging an implementation detail here. Did you confirm that these rate limit events will be created reliably?
67a83bb to
231c21d
Compare
231c21d to
4859762
Compare
58f5b1e to
27e03e5
Compare
| // Rate-limited events (worker is available but there's no task handoff due to rate-limiting) are a | ||
| // distinct outcome tracked separately in RateLimitedSignalsSinceLast. |
There was a problem hiding this comment.
invert the sentence structure and move it down to be the comment above RateLimitedSignalsSinceLast?
| updatedState[stateRateLimitedCountSinceLastPollKey] = prevCount + int64(event.RateLimitedSignalsSinceLast) | ||
| } | ||
|
|
||
| return &TaskAddResponse{Actions: actions, Status: updatedState, ThrottledCount: throttledCount, RateLimitedCount: rateLimitedCount}, nil |
There was a problem hiding this comment.
why do we return the input data RateLimitedCount here? do you expect the algo to change the number?
| // configNoSyncRateLimitedSuppressQuietMsKey: in ProcessMetricsPoll, suppress scale-up for this many ms | ||
| // after the last observed rate-limited signal. 0 = disabled. | ||
| // Default is 2× the poll interval. | ||
| configNoSyncRateLimitedSuppressQuietMsKey = "rate_limited_suppress_quiet_ms" | ||
| configNoSyncRateLimitedSuppressQuietMsDefault = 2 * configNoSyncMetricsPollIntervalMsDefault | ||
|
|
||
| stateLastRateLimitedTimestampKey = "last_rate_limited_time_ms" |
There was a problem hiding this comment.
The explanation helps, but now I am unclear whether you are leveraging an implementation detail here. Did you confirm that these rate limit events will be created reliably?
| // In older or existing workflows, handle scenario when SyncMatchOutcome | ||
| // is not available in the hooks API |
There was a problem hiding this comment.
Note that "In older or existing workflows" immediately is out-of-date because there is no indication of what "currently" is. Better to be specific here. Something like "SyncMatchOutcome was added in server release X. Here, we handle situations for workflows that were started before that server release."
| } | ||
|
|
||
| func (th *taskHookImpl) batchMatchSignals(_ context.Context, workflowID string, isSyncMatch bool) (int, int, bool) { | ||
| func (th *taskHookImpl) batchMatchSignals( |
There was a problem hiding this comment.
This function could really use a docstring, especially with the many unrelated input and output parameters.
Especially curious is the "skip" parameter. I honestly have no idea what that is doing, even after reading through the Claude-generated unit tests.
And, come to think of it maybe rename the function to storeTaskAddSignalResults or something like that? "batch" is an implementation detail and "match" doesn't have anything to do with the the signal (no sync match is just one of multiple signals, no?)
| configNoSyncMetricsPollIntervalMsKey = "metrics_poll_interval_ms" | ||
| configNoSyncMetricsPollIntervalMsDefault = int64(60_000) // 60s | ||
|
|
||
| // configNoSyncRateLimitedSuppressQuietMsKey is the duration to block backlog-driven scale-up |
There was a problem hiding this comment.
"suppress quiet"? I don't know what Claude/Cursor is attempting to convey with that. Sounds like gibberish to me.
Questions:
- Is this something that really needs to be configurable?
- "backlog-driven scale-up after rate limiting is no longer actively signalled" sounds similarly obtuse to me. Would it be clearer to say "This is the duration that WCI will wait after encountering rate-limiting attempting to invoke a serverless worker"?
3 participants
What was changed
Extend WCI to correctly handle task queue dispatch rate limiting for Lambda compute providers. Rate-limited events based on task queue dispatch rate, were previously incorrectly treated as "no worker available" and triggered worker invocations.
This change classifies rate-limited events separately using the new
SyncMatchOutcomesignal from PR #10045, preventing spurious Lambda invocations when rate limiting is the bottleneck.The rate_based algorithm (GCP Cloud Run) will be tracked as a follow-up change.
Note: The current changes validate rate limiting during task submission (sync match path), and do not currently handle rate limiting during backlog drain (async dispatch path)
Why?
Adding more workers when the rate limiting is the bottleneck does nothing. Before this change, WCI couldn't distinguish "no worker available" from "worker available but rate-limited", so it scaled up in both cases unnecessarily.
Checklist
Closes: https://temporalio.atlassian.net/browse/COM-125
How was this tested: