fix: on-demand lease manager for cold start concurrency queue#4281
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Benchmark resultsCompared against |
There was a problem hiding this comment.
Pull request overview
This PR reduces cold-start latency for concurrency-keyed scheduling by acquiring concurrency-strategy leases on-demand when a task arrives for a strategy that currently has no in-memory ConcurrencyManager, avoiding the periodic (5s) lease-poll delay.
Changes:
- Add an on-demand, blocking concurrency-strategy lease acquisition path and immediately notify the newly created
ConcurrencyManagerto schedule waiting tasks promptly. - Add an integration regression test that reproduces the “cold strategy waits for lease poll” behavior and asserts the task is scheduled within a short deadline.
- Remove the now-redundant RMQ message type/handler for “new concurrency strategy”, and re-enable the stale concurrency deactivation operation.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/scheduling/v1/tenant_manager.go | Creates/awakens concurrency managers on-demand for cold strategies and triggers immediate scheduling. |
| pkg/scheduling/v1/lease_manager.go | Adds a blocking on-demand lease acquisition method that returns the strategy directly (bypassing the lossy channel). |
| pkg/scheduling/v1/concurrency_integration_test.go | Adds regression coverage for cold-start scheduling latency on concurrency strategies. |
| internal/services/shared/tasktypes/v1/scheduler.go | Removes the now-unused “new concurrency strategy” message payload/builder. |
| internal/services/scheduler/v1/scheduler.go | Removes the handler for the deleted “new concurrency strategy” message. |
| internal/services/controllers/task/controller.go | Re-enables the periodic operation that deactivates stale step concurrency strategies. |
| internal/msgqueue/message_ids.go | Removes the message ID constant for the deleted message type. |
| return nil | ||
| } | ||
|
|
||
| func (s *Scheduler) handleNewConcurrencyStrategy(ctx context.Context, msg *msgqueue.Message) error { |
There was a problem hiding this comment.
Why are we removing this? It feels like this is a different case that we should also handle: a completely new concurrency strategy being registered from PutWorkflow. We likely are not propagating this message yet, but we probably should.
There was a problem hiding this comment.
Restored the handler. In terms of the PutWorkflow change - I think my changes already address this when the very first task arrives with concurrency? So IMO the publishing of this RMQ message from PutWorkflow is only proactive optimization?
| // hot path in their own goroutine, so blocking is fine -- and it guarantees the acquisition is not | ||
| // silently dropped under contention (the previous TryLock + non-blocking-send approach would no-op | ||
| // when several cold strategies raced each other or the periodic poll, falling back to the 5s poll). | ||
| func (l *LeaseManager) acquireConcurrencyStrategyOnDemand(ctx context.Context, strategyId int64) (*sqlcv1.V1StepConcurrency, error) { |
There was a problem hiding this comment.
I'm not sure why this needs to use a different path than the notifyNewQueue mechanism. I think we can rethink the non-blocking write to the concurrencyLeasesCh to potentially make it blocking, if that's what we're worried about.
From what I can tell there are two important things here:
- That we call
notifyafter we instantiate the concurrency strategy (which we should probably do anyway) - That we call
notifyNewConcurrencyStrategywhen a message arrives with a strategy ID which isn't currently in our leases.
|
|
|
|
| // notify the new manager so it schedules its waiting task now instead of waiting for its first tick. | ||
| c.notify(ctx) |
There was a problem hiding this comment.
nit: we should consider doing this on concurrency manager instantiation instead? I think the manager has a notifier channel as well, on instantiation we can just put a message into that channel so that it runs immediately after startup? having to remember to notify each time we call newConcurrencyManager is poor design I think
| // mark this strategy as already managed so we don't try to acquire a lease for it below | ||
| delete(strategyIdsMap, c.strategy.ID) |
There was a problem hiding this comment.
are we sure about this? I see immediately above this we're doing:
if _, ok := strategyIdsMap[c.strategy.ID]; !ok {
continue
}So it seems like we might be doing more processing as a result of deleting this entry. Perhaps we need two copies of the map?
|
|
||
| // notify the new manager so it picks up already-queued work now instead of waiting for its first | ||
| // tick (matters on startup/rebalance). notify() is a non-blocking send, so it's fine under the lock. | ||
| c.notify(ctx) |
There was a problem hiding this comment.
same comment as above/below
| replayEnabled bool | ||
| analyzeCronInterval time.Duration | ||
| signaler *signal.OLAPSignaler | ||
| tw *trigger.TriggerWriter | ||
| promGate *prometheus.Gate |
There was a problem hiding this comment.
nit: is this linted properly?
Description
This PR aims to fix the cold start latency concurrency workflows can face which increases the latency between QUEUED -> ASSIGNED_TO_WORKER status progression.
Type of change
What's Changed
Checklist
Changes have been:
🤖 AI Disclosure