Class: Hatchet::ConcurrencyExpression

Inherits:
Object
  • Object
show all
Defined in:
lib/hatchet/concurrency.rb,
sig/hatchet/concurrency.rbs

Overview

Defines a concurrency expression for workflow or task-level concurrency control

Examples:

Workflow-level concurrency

Hatchet::ConcurrencyExpression.new(
  expression: "input.group_key",
  max_runs: 5,
  limit_strategy: :cancel_in_progress
)

Dynamic per-group limits via a CEL expression

Hatchet::ConcurrencyExpression.new(
  expression: "input.tier",
  max_runs: "input.tier == 'premium' ? 10 : 1",
  limit_strategy: :group_round_robin
)

Tenant-scoped, shared across workflows

Hatchet::ConcurrencyExpression.new(
  expression: "input.group",
  max_runs: 1,
  limit_strategy: :group_round_robin,
  name: "tenant-wide-limit",
  is_tenant_scoped: true
)

Task-level concurrency with multiple keys

[
  Hatchet::ConcurrencyExpression.new(expression: "input.digit", max_runs: 8, limit_strategy: :group_round_robin),
  Hatchet::ConcurrencyExpression.new(expression: "input.name", max_runs: 3, limit_strategy: :group_round_robin)
]

Constant Summary collapse

LIMIT_STRATEGY_MAP =
Deprecated.

kept for backwards compatibility; use Hatchet::ConcurrencyProto::LIMIT_STRATEGY_MAP

Returns:

  • (Hash[Symbol, Symbol]) 
ConcurrencyProto::LIMIT_STRATEGY_MAP

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(expression:, max_runs: 1, limit_strategy: :cancel_in_progress, name: nil, is_tenant_scoped: false) ⇒ ConcurrencyExpression

Returns a new instance of ConcurrencyExpression.

Parameters:

  • expression (String) —

    CEL expression evaluated against input

  • max_runs (Integer, String) (defaults to: 1) —

    Maximum concurrent runs, or a CEL expression computing them per group

  • limit_strategy (Symbol) (defaults to: :cancel_in_progress) —

    Strategy when limit is reached

  • name (String, nil) (defaults to: nil) —

    Unique (per tenant) strategy name; required when tenant-scoped

  • is_tenant_scoped (Boolean) (defaults to: false) —

    Share this strategy across workflows, keyed by name

  • expression: (String)
  • max_runs: (Integer, String) (defaults to: 1)
  • limit_strategy: (Symbol) (defaults to: :cancel_in_progress)
  • name: (String, nil) (defaults to: nil)
  • is_tenant_scoped: (Boolean) (defaults to: false)

Raises:

  • (ArgumentError) 


99
100
101
102
103
104
105
106
107
 
# File 'lib/hatchet/concurrency.rb', line 99

def initialize(expression:, max_runs: 1, limit_strategy: :cancel_in_progress, name: nil, is_tenant_scoped: false)
  raise ArgumentError, "a name is required for tenant-scoped concurrency" if is_tenant_scoped && (name.nil? || name.empty?)

  @expression = expression
  @max_runs = max_runs
  @limit_strategy = limit_strategy
  @name = name
  @is_tenant_scoped = is_tenant_scoped
end

Instance Attribute Details

#expression ⇒ String (readonly)

Returns CEL expression evaluated against the workflow input.

Returns:

  • (String) —

    CEL expression evaluated against the workflow input



73
74
75
 
# File 'lib/hatchet/concurrency.rb', line 73

def expression
  @expression
end

#is_tenant_scoped ⇒ Boolean (readonly)

Returns When true, the entry defines (or updates in place) a tenant-scoped strategy shared across workflows, keyed by name: every task declaring the same name consumes the same concurrency limit. The position in the concurrency list is the chain order, and chains sharing tenant-scoped strategies must order them consistently.

Returns:

  • (Boolean) —

    When true, the entry defines (or updates in place) a tenant-scoped strategy shared across workflows, keyed by name: every task declaring the same name consumes the same concurrency limit. The position in the concurrency list is the chain order, and chains sharing tenant-scoped strategies must order them consistently.



92
93
94
 
# File 'lib/hatchet/concurrency.rb', line 92

def is_tenant_scoped
  @is_tenant_scoped
end

#limit_strategy ⇒ Symbol (readonly)

Returns Strategy when limit is exceeded (:cancel_in_progress, :cancel_newest, :group_round_robin, :queue, :cancel_queued_except_newest, :cancel_queued_except_oldest).

Returns:

  • (Symbol) —

    Strategy when limit is exceeded (:cancel_in_progress, :cancel_newest, :group_round_robin, :queue, :cancel_queued_except_newest, :cancel_queued_except_oldest)



82
83
84
 
# File 'lib/hatchet/concurrency.rb', line 82

def limit_strategy
  @limit_strategy
end

#max_runs ⇒ Integer, String (readonly)

Returns Maximum concurrent runs for this key: a fixed number, or a CEL expression over task input computing the max runs for that task's concurrency group. With an expression, a group's effective limit is the value from its most recently created task.

Returns:

  • (Integer, String) —

    Maximum concurrent runs for this key: a fixed number, or a CEL expression over task input computing the max runs for that task's concurrency group. With an expression, a group's effective limit is the value from its most recently created task.



79
80
81
 
# File 'lib/hatchet/concurrency.rb', line 79

def max_runs
  @max_runs
end

#name ⇒ String? (readonly)

Returns Unique (per tenant) strategy name; required when tenant-scoped.

Returns:

  • (String, nil) —

    Unique (per tenant) strategy name; required when tenant-scoped



85
86
87
 
# File 'lib/hatchet/concurrency.rb', line 85

def name
  @name
end

Instance Method Details

#to_h ⇒ Hash

Convert to a hash for API serialization

Returns:

  • (Hash) 


114
115
116
117
118
119
120
121
122
123
 
# File 'lib/hatchet/concurrency.rb', line 114

def to_h
  h = {
    expression: @expression,
    max_runs: @max_runs,
    limit_strategy: @limit_strategy.to_s.upcase,
  }
  h[:name] = @name if @name
  h[:is_tenant_scoped] = true if @is_tenant_scoped
  h
end

#to_proto ⇒ V1::Concurrency

Convert to a V1::Concurrency protobuf message

Returns:



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/hatchet/concurrency.rb', line 127

def to_proto
  proto_strategy = ConcurrencyProto::LIMIT_STRATEGY_MAP[@limit_strategy] || :CANCEL_IN_PROGRESS
  static_max, max_runs_expression = ConcurrencyProto.split_max_runs(@max_runs)

  args = {
    expression: @expression,
    max_runs: static_max,
    limit_strategy: proto_strategy,
  }
  args[:max_runs_expression] = max_runs_expression if max_runs_expression
  args[:name] = @name if @name
  args[:is_tenant_scoped] = true if @is_tenant_scoped

  ::V1::Concurrency.new(**args)
end