Batch task queue user data persistence updates #7039
Conversation
common/persistence/sql/task.go
Outdated
| if err != nil { | ||
| return err | ||
| if m.Db.IsDupEntryError(err) { | ||
| return &persistence.ConditionFailedError{Msg: err.Error()} |
There was a problem hiding this comment.
with this type of error handling you will end up with "partially updated" state, but you don't really know which one are passed.
(unless I miss something).
Is there a reason to stop on error? or may it make sense to move forward, and return an array of failed updates?
There was a problem hiding this comment.
it's a transaction so you can't end up with things partially updated, right?
There was a problem hiding this comment.
it was not clear that all of this happening inside transaction. Probably worth a comment
| // try to add more items. stop after a gap of MaxGap, total time of MaxTotalWait, or | ||
| // MaxItems items. | ||
| maxWaitC, maxWaitT := s.clock.NewTimer(s.opts.MaxDelay) | ||
| loop: |
There was a problem hiding this comment.
I'm not sure I understand all the possible implications, so I guess I will trust this work and covered by tests.
| // No error, but not applied. That means we had a conflict. | ||
| // Iterate through results to identify first conflicting row. | ||
| for { | ||
| name, nameErr := getTypedFieldFromRow[string]("task_queue_name", previous) | ||
| previousVersion, verErr := getTypedFieldFromRow[int64]("version", previous) | ||
| update, hasUpdate := request.Updates[name] | ||
| if nameErr == nil && verErr == nil && hasUpdate && update.Version != previousVersion { | ||
| if update.Conflicting != nil { | ||
| *update.Conflicting = true | ||
| } | ||
| return &p.ConditionFailedError{ | ||
| Msg: fmt.Sprintf("Failed to update task queues: task queue %q version %d != %d", | ||
| name, update.Version, previousVersion), | ||
| } | ||
| } | ||
| clear(previous) | ||
| if !iter.MapScan(previous) { | ||
| break | ||
| } | ||
| } | ||
| return &p.ConditionFailedError{Msg: "Failed to update task queues: unknown conflict"} |
There was a problem hiding this comment.
It's possible that I missed it (it's a big PR), but this is untested, isn't it? If so, that makes me a little nervous.
| } | ||
| if len(request.BuildIdsAdded) > 0 { | ||
| err = tx.AddToBuildIdToTaskQueueMapping(ctx, sqlplugin.AddToBuildIdToTaskQueueMapping{ | ||
| for taskQueue, update := range request.Updates { |
There was a problem hiding this comment.
This could be up to 100 separate Update writes, right? I assume self-hosted users won't usually see a lot of parallel updates?
What changed?
Multiple user data updates coming in for task queues in the same namespace within a short period of time get batched into a smaller number of persistence operations (transactions).
Since multiple updates are batched into a transaction, conflicts in one can cause unrelated ones to fail. This is detected and a non-retryable error is returned for the conflicting one, while a retryable error is returned for the other ones.
Why?
With deployments, we sometimes have to update user data on multiple task queues at once (all in the same namespace), and on cassandra, these updates go through an LWT. This could cause a backup since the throughput of LWTs is fairly low.
This change allows batching of multiple updates in one persistence operation (LWT on cassandra or transaction on sql). The batching is transparent: updates that come in within a short period of time automatically get batched (in matching engine).
How did you test it?
Potential risks