Fix: Client task heartbeat changes - #392
Conversation
cancel task context if heartbeat stops add tests
j-nowakowski
changed the title
| tw.On("Execute", mock.Anything, t1).Return(nil, context.Canceled).Once().Run(func(args mock.Arguments) { | ||
| ctx := args.Get(0).(context.Context) | ||
| select { | ||
| case <-time.After(2 * heartbeatInterval): |
There was a problem hiding this comment.
thought(non-blocking): this kind of test is a great candidate for synctest if the maintainer is willing to use an experimental flag
|
Thanks for the report and great writeup! I'm wondering if the separation between the polling goroutines and the dispatcher still make sense 🤔 If there is no available slot to work on a task, why even continue polling and claiming more work that can't be worked on. Combining the two and only fetching new work when there is capacity to actually work on and complete those tasks should help with tasks sitting in the queue for a long time and then being already expired or expiring before the first heartbeat. |
I was thinking about this too. The reason I actually like the poller & dispatcher model is that it allows for a lower latency pipeline of event processing, especially in the case of low number of slots. If we have to poll the task once a slot opens up, we are waiting on that response before working. If we have eagerly buffered a task, we get to work right away. We can always regain throughput by increasing concurrency, but we would have no way to regain latency if we can't buffer. In fact, that makes me think about the task buffer size. Right now, we have an effective buffer hardcoded to size 1 due to the task channel being unbuffered. Might workers benefit from having a configurable buffer size? |
That is true, but I'm not sure how much we'd actually lose. If there is a task available, polling and fetching a new task should be a relatively fast operation and we'd avoid the complexity of having to deal with expired tasks that might've sat in the local queue for a while. We should still abort execution if heartbeats are enabled and fail, agree with that change.
The buffer is unlimited right? Workers will pull work as quickly as they can and add it to the worker's |
The taskQueue channel is unbuffered, so the write to that channel that the poll performs blocks until the dispatcher reads it. In effect, it is a local buffer of 1 task. Edit: and the dispatcher only reads it once a slot is available |
It depends. If polling and working both take about the same amount of time, then making those operations sequential rather than parallel would double the latency. |
|
This could be configurable behavior... e.g., set the buffer to 0 means we only poll as a slot becomes available. This could be useful for flows where task work time is unpredictable. A positive value gives such a buffer which is more useful for quick, consistent tasks. |
4 participants
Problem
While using go-workflows in our platform, we observed a cascading failure scenario due to how the clients poll, heartbeat, and work tasks when the client task concurrency is limited.
How the client worker currently works...
The server also does not tell the client worker how long it has ownership of a task. Exposing this would mean a breaking change in the Backend API, so this proposal will work with this limitation.
The problem is easiest to see if the number of slots is 1. Suppose a task lock is 60s, we heartbeat every 30s, and a task takes 45s to work.
"Soon to expire task" scenario
The worker polls a first task and starts working it immediately. It then buffers a second task. By the time the second task begins work, we are now 45s through its 60s lock, and won't be heartbeating for another 30s, so it will expire before we extend it. A different worker might claim this task, so we will waste a lot of time trying to work it only for the completion event to be rejected due to lack of task ownership. Task after task continues to fail in this way. If we knew to heartbeat more proactively on that second task, we could have finished the task without issue.
"Already expired task" scenario
Similar to last scenario but suppose the task takes 70s to complete. We complete the first task without issue. We start the second task, but this one is doomed to fail no matter what because it is expired. Nonetheless, we spend time trying to work it only for the completion event to be rejected due to lack of task ownership. Task after task continues to fail in this way. If we had given up on that task sooner, we would not have wasted that time.
Solutions we considered
Our initial thought was to simply move the heartbeat earlier in the process so that tasks get heartbeated as soon as the task gets polled. However this can lead to a task being "hoarded" by a worker who is working a long activity (successfully or erroneously). Because it never releases the task, an open worker would never get the opportunity to work it. For that reason, we declined this solution.
Proposal
Considerations
The change with the most impact is bullet point 2 in the proposal, "Cancel task when heartbeat stops", as it is more pessimistic than the existing behavior. We could consider a more middle-ground approach where we only cancel the task if the first heartbeat fails.