feat: Batch flush - #3975
Conversation
…nt automatically fixed them
|
|
|
|
|
|
| WHERE tr.tenant_id = br.tenant_id | ||
| AND tr.batch_id = br.batch_id | ||
| ) | ||
| FOR UPDATE |
There was a problem hiding this comment.
always good practice for every FOR UPDATE to have an ORDER BY, or you risk deadlocking
| LIMIT | ||
| COALESCE(sqlc.narg('limit')::integer, 1000); | ||
|
|
||
| -- name: ListDistinctBatchResources :many |
There was a problem hiding this comment.
these leases are going to be acquired for every single step_id, even those which are inactive, correct? Like if a workflow has 100 versions, each with batching configured, we'll acquire 100 leases? this is problematic, we need to limit it to "active" steps somehow
There was a problem hiding this comment.
Nope, it doesn't acquire for steps that are inactive because it joins with v1_batched_queue_item which is just the in-progress batch items. So unless those 100 versions all had items in flight, there wouldn't be 100 leases.
…nd remove task controller timeout logic (replaced by batch scheduler)
| -- +goose Up | ||
| -- +goose StatementBegin | ||
| -- v0 schema alignment | ||
| ALTER TYPE "LeaseKind" ADD VALUE 'BATCH'; |
There was a problem hiding this comment.
I know I'm usually the "don't use if not exists" guy, but we need it here 😅 since there's no way to remove a value from an enum, we have to have it otherwise we can't run this migration twice (same for the others below)
abelanger5
left a comment
abelanger5
left a comment
There was a problem hiding this comment.
Some non-blocking comments, but looks good to me!
| } | ||
| } | ||
|
|
||
| runtimes, err := d.repov1.Tasks().ListTaskRuntimes(ctx, tenantId, bulkDatas) |
There was a problem hiding this comment.
At this point we've added so many things to this method (ListTasks, durable invocations, parent outputs, payloads, and now this) - should we try to write a new repository method so that we can reduce database round-trips?
|
|
||
| // Confirm the worker belongs to the auth-tenant before mutating its labels. | ||
| if _, err := s.repov1.Workers().GetWorkerForEngine(ctx, tenant.ID, workerId); err != nil { | ||
| if _, err := s.repov1.Workers().GetWorkerForEngine(ctx, tenant.ID, workerId); err != nil { //nolint |
There was a problem hiding this comment.
why does lint fail? it's probably just a matter of renaming err or something
There was a problem hiding this comment.
nit: not seeing spans/traces in this file (or really in that many places in this PR) - should definitely do a quick pass and make sure we're instrumenting the highest-risk calls.
|
|
|
|
|
|
|
|
4 participants
Description
#2617
Adds concept of "dynamic batches" that allow the workflow to specify either a max batch size, or a batch timeout time, or both, that will flush the batch to the worker once either of those limits are reached. Batch scheduling is done by moving rows from v1_queue_item to v1_batch_queue_item where they will be picked up by the batch scheduler. The dispatcher will send a "START_BATCH" action to workers, containing all the payloads in the batch. If the payloads in a batch are about to exceed the 4mb gRPC limit, then the batch will automatically be flushed, resulting in lower batch sizes.
Assorted architectural notes:
input.tenant_id.Fixes # (issue)
Type of change
What's Changed
Checklist
Changes have been:
🤖 AI Disclosure