Concurrency Control in Hatchet Tasks

Hatchet provides powerful concurrency control features to help you manage the execution of your tasks. This is particularly useful when you have tasks that may be triggered frequently or have long-running steps, and you want to limit the number of concurrent executions to prevent overloading your system, ensure fairness, or avoid race conditions. Concurrency strategies can be added to both tasks and workflows.

Why use concurrency control?

You should primarily use concurrency control when you need to ensure fair access to resources across your application’s users, projects, or organizations. By limiting the number of in-flight tasks for a particular user, you can prevent that user from monopolizing the system.

Concurrency control also lets you limit the number of runs for a workflow globally, if you use a static CEL expression, such as 'global'. This and rate limiting are the recommended mechanisms for setting per-workflow throughput limits.

Available Strategies:

• Group Round Robin queues incoming task and workflow runs and only dispatches them to workers and triggers them once an available slot is open.
• Cancel In Progress cancels in-progress instances of the task or workflow with matching concurrency keys in order to free up slots for the newly-triggered task or workflow run.
• Cancel Newest cancels any incoming task or workflow runs for a key once the number of runs in a running state for that key has reached a provided limit.

Group Round Robin

When a new task instance is triggered, the Group Round Robin strategy will:

1. Determine the key that the run belongs to based on the CEL expression defined in the task or workflow’s concurrency configuration.
2. Check if there are any available slots for the computed concurrency key based on the maximum number of concurrent runs allowed by the concurrency configuration.
3. If a slot is available, the new task or workflow starts executing immediately.
4. If no slots are available, the new task or workflow is added to a queue for its key.
5. When a running task instance completes and a slot becomes available for a group, the next queued instance for that group (in round-robin order) is dequeued and starts executing.

Group round robin ensures that task instances are processed fairly across different groups, preventing any one group from monopolizing the available resources. It also helps to reduce latency for instances within each group, as they are processed in a round-robin fashion rather than strictly in the order they were triggered. Group round robin is also useful as a global concurrency control for the maximum number of runs of a single task or workflow that you want executing at any given time, regardless of any grouping. You can set a constant concurrency key as the expression, such as '*', to enable this global concurrency control behavior.

To use this strategy, set the GROUP_ROUND_ROBIN limit strategy along with a max_runs limit and a key expression:

class WorkflowInput(BaseModel):
    group: str


concurrency_limit_rr_workflow = hatchet.workflow(
    name="ConcurrencyDemoWorkflowRR",
    concurrency=ConcurrencyExpression(
        expression="input.group",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
    ),
    input_validator=WorkflowInput,
)

export const simpleConcurrency = hatchet.workflow<SimpleInput, SimpleOutput>({
  name: 'simple-concurrency',
  concurrency: {
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
    expression: 'input.GroupKey',
  },
});

var maxRuns int32 = 1
strategy := types.GroupRoundRobin

return client.NewStandaloneTask("simple-concurrency",
	func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet.WithWorkflowConcurrency(types.Concurrency{
		Expression:    "input.GroupKey",
		MaxRuns:       &maxRuns,
		LimitStrategy: &strategy,
	}),
)

CONCURRENCY_LIMIT_RR_WORKFLOW = HATCHET.workflow(
  name: "ConcurrencyDemoWorkflowRR",
  concurrency: Hatchet::ConcurrencyExpression.new(
    expression: "input.group",
    max_runs: 1,
    limit_strategy: :group_round_robin
  )
)

CONCURRENCY_LIMIT_RR_WORKFLOW.task(:step1) do |input, ctx|
  puts "starting step1"
  sleep 2
  puts "finished step1"
end

Cancel In Progress

When a new task instance is triggered, the Cancel In Progress strategy will:

1. Determine the key that the run belongs to based on the CEL expression defined in the task or workflow’s concurrency configuration.
2. Check if there are any available slots for the computed concurrency key based on the maximum number of concurrent runs allowed by the concurrency configuration.
3. If a slot is available, the new task or workflow starts executing immediately.
4. If no slots are available, a running instance of the task or workflow with the same concurrency key will be cancelled to free up a slot, and the new instance will start executing immediately.

Cancel In Progress ensures that the most recently triggered runs always take priority over older ones, which is useful when newer inputs supersede older ones and in-progress work becomes stale or irrelevant as soon as a newer run arrives. It’s particularly well suited for user-facing interactions where only the latest input matters (such as chat messages, form submissions, or search-as-you-type), for resource-intensive tasks where it’s more efficient to abandon an old run than wait for it to complete, and for any scenario where you want to prioritize processing the most recent data or events over older ones.

To use this strategy, set the CANCEL_IN_PROGRESS limit strategy along with a max_runs limit and a key expression:

class WorkflowInput(BaseModel):
    group: str


concurrency_cancel_in_progress_workflow = hatchet.workflow(
    name="ConcurrencyCancelInProgress",
    concurrency=ConcurrencyExpression(
        expression="input.group",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
    ),
    input_validator=WorkflowInput,
)

export const concurrencyCancelInProgressWorkflow = hatchet.workflow<WorkflowInput, WorkflowOutput>({
  name: 'concurrencycancelinprogress',
  concurrency: {
    expression: 'input.group',
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.CANCEL_IN_PROGRESS,
  },
});

var maxRuns int32 = 1
strategy := types.CancelInProgress

return client.NewStandaloneTask("cancel-in-progress",
	func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet.WithWorkflowConcurrency(types.Concurrency{
		Expression:    "input.GroupKey",
		MaxRuns:       &maxRuns,
		LimitStrategy: &strategy,
	}),
)

CONCURRENCY_CANCEL_IN_PROGRESS_WORKFLOW = HATCHET.workflow(
  name: "ConcurrencyCancelInProgress",
  concurrency: Hatchet::ConcurrencyExpression.new(
    expression: "input.group",
    max_runs: 1,
    limit_strategy: :cancel_in_progress
  )
)

Cancel Newest

When a new task instance is triggered, the Cancel Newest strategy will:

1. Determine the key that the run belongs to based on the CEL expression defined in the task or workflow’s concurrency configuration.
2. Check if there are any available slots for the computed concurrency key based on the maximum number of concurrent runs allowed by the concurrency configuration.
3. If a slot is available, the new task or workflow starts executing immediately.
4. If no slots are available, the newly triggered run is cancelled immediately, allowing the in-progress runs to continue uninterrupted.

Cancel Newest is the inverse of Cancel In Progress: rather than preempting running work in favor of new arrivals, it protects in-progress runs from being disrupted by allowing them to complete before any new work for the same key is started. This is useful when you want to guarantee that long-running task instances finish without interference, when the cost of restarting work outweighs the value of processing newer inputs, and when you want to prevent a single group’s instances from monopolizing the available slots by rejecting excess runs outright instead of queuing them.

To use this strategy, set the CANCEL_NEWEST limit strategy along with a max_runs limit and a key expression:

class WorkflowInput(BaseModel):
    group: str


concurrency_cancel_newest_workflow = hatchet.workflow(
    name="ConcurrencyCancelNewest",
    concurrency=ConcurrencyExpression(
        expression="input.group",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_NEWEST,
    ),
    input_validator=WorkflowInput,
)

export const concurrencyCancelNewestWorkflow = hatchet.workflow<WorkflowInput, WorkflowOutput>({
  name: 'concurrencycancelnewest',
  concurrency: {
    expression: 'input.group',
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.CANCEL_NEWEST,
  },
});

var maxRuns int32 = 1
strategy := types.CancelNewest

return client.NewStandaloneTask("cancel-newest",
	func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet.WithWorkflowConcurrency(types.Concurrency{
		Expression:    "input.GroupKey",
		MaxRuns:       &maxRuns,
		LimitStrategy: &strategy,
	}),
)

CONCURRENCY_CANCEL_NEWEST_WORKFLOW = HATCHET.workflow(
  name: "ConcurrencyCancelNewest",
  concurrency: Hatchet::ConcurrencyExpression.new(
    expression: "input.group",
    max_runs: 1,
    limit_strategy: :cancel_newest
  )
)

Multiple concurrency strategies

You can also combine multiple concurrency strategies to create a more complex concurrency control system. For example, you can use one group key to represent a specific team, and another group to represent a specific resource in that team, giving you more control over the rate at which tasks are executed.

class WorkflowInput(BaseModel):
    name: str
    digit: str


concurrency_workflow_level_workflow = hatchet.workflow(
    name="ConcurrencyWorkflowLevel",
    input_validator=WorkflowInput,
    concurrency=[
        ConcurrencyExpression(
            expression="input.digit",
            max_runs=DIGIT_MAX_RUNS,
            limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
        ),
        ConcurrencyExpression(
            expression="input.name",
            max_runs=NAME_MAX_RUNS,
            limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
        ),
    ],
)

export const multipleConcurrencyKeys = hatchet.workflow<SimpleInput, SimpleOutput>({
  name: 'simple-concurrency',
  concurrency: [
    {
      maxRuns: 1,
      limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
      expression: 'input.Tier',
    },
    {
      maxRuns: 1,
      limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
      expression: 'input.Account',
    },
  ],
});

strategy := types.GroupRoundRobin
var maxRuns int32 = 20

return client.NewStandaloneTask("multi-concurrency",
	func(ctx worker.HatchetContext, input ConcurrencyInput) (*TransformedOutput, error) {
		// Random sleep between 200ms and 1000ms
		time.Sleep(time.Duration(200+rand.Intn(800)) * time.Millisecond)

		return &TransformedOutput{
			TransformedMessage: input.Message,
		}, nil
	},
	hatchet.WithWorkflowConcurrency(
		types.Concurrency{
			Expression:    "input.Tier",
			MaxRuns:       &maxRuns,
			LimitStrategy: &strategy,
		}, types.Concurrency{
			Expression:    "input.Account",
			MaxRuns:       &maxRuns,
			LimitStrategy: &strategy,
		},
	),
)

CONCURRENCY_WORKFLOW_LEVEL_WORKFLOW = HATCHET.workflow(
  name: "ConcurrencyWorkflowLevel",
  concurrency: [
    Hatchet::ConcurrencyExpression.new(
      expression: "input.digit",
      max_runs: DIGIT_MAX_RUNS_WL,
      limit_strategy: :group_round_robin
    ),
    Hatchet::ConcurrencyExpression.new(
      expression: "input.name",
      max_runs: NAME_MAX_RUNS_WL,
      limit_strategy: :group_round_robin
    )
  ]
)

CONCURRENCY_WORKFLOW_LEVEL_WORKFLOW.task(:task_1) do |input, ctx|
  sleep SLEEP_TIME_WL
end

CONCURRENCY_WORKFLOW_LEVEL_WORKFLOW.task(:task_2) do |input, ctx|
  sleep SLEEP_TIME_WL
end