Module: Shoryuken::ActiveJob::CurrentAttributes::Loading

Defined in:
lib/shoryuken/active_job/current_attributes.rb

Overview

Module prepended to JobWrapper to restore CurrentAttributes on execute.

Instance Method Summary collapse

Instance Method Details

#perform(sqs_msg, hash) ⇒ void

This method returns an undefined value.

Performs the job after restoring CurrentAttributes

Parameters:

  • sqs_msg (Shoryuken::Message) —

    the SQS message

  • hash (Hash) —

    the deserialized job data



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/shoryuken/active_job/current_attributes.rb', line 120

def perform(sqs_msg, hash)
  CurrentAttributes.cattrs&.each do |key, klass_name|
    next unless hash.key?(key)

    klass = klass_name.constantize

    begin
      attrs = Serializer.deserialize(hash[key])
      attrs.each do |attr_name, value|
        klass.public_send(:"#{attr_name}=", value) if klass.respond_to?(:"#{attr_name}=")
      end
    rescue => e
      # Log but don't fail if attributes can't be restored
      # (e.g., attribute removed between enqueue and execute)
      Shoryuken.logger.warn("Failed to restore CurrentAttributes #{klass_name}: #{e.message}")
    end
  end

  super
ensure
  # Reset every registered CurrentAttributes class after the job - not
  # only the ones whose key was in this message.
  #
  # Why unconditional (and why this differs from Sidekiq): Sidekiq's
  # loader only touches classes present in the job and relies on the
  # Rails executor - which it runs every job inside - to reset all
  # CurrentAttributes between units of work. Shoryuken has no such safety
  # net: it wraps a job in the reloader/executor only when
  # `enable_reloading` is set, which is off by default, so nothing else
  # clears CurrentAttributes between jobs.
  #
  # A blanket reset here is therefore the only thing guaranteeing a clean
  # thread. Resetting only the present keys leaks whenever a value ends up
  # set during a job whose message carried no cattr key - e.g. the worker
  # (or code it calls) writes to Current, on a keyless message (empty
  # context at enqueue, a different producer, or persist configured after
  # the message was queued). CurrentAttributes are thread-local and the
  # pool reuses threads, so that value would surface in the next job.
  CurrentAttributes.cattrs&.each_value do |klass_name|
    klass_name.constantize.reset
  rescue => e
    Shoryuken.logger.warn("Failed to reset CurrentAttributes #{klass_name}: #{e.message}")
  end
end