Module: ActiveJob::Temporal::Payload

Extended by:
Payload
Included in:
Payload
Defined in:
lib/activejob/temporal/payload.rb

Overview

Note:

Payload Size Limit Temporal enforces a maximum payload size (configurable via max_payload_size_kb, default 250 KB). Large payloads will raise a SerializationError with a human-readable message indicating the actual size and the limit. Consider passing database IDs or S3 keys instead of large objects.

Note:

GlobalID Serialization ActiveRecord models are automatically serialized using GlobalID. This requires the model to exist in the database at enqueue time and still exist at execution time. If the record is deleted, deserialization will fail.

rubocop:disable Metrics/ModuleLength Payload serialization and deserialization for ActiveJob.

This module converts ActiveJob instances into JSON-serializable hash payloads for transmission to Temporal workflows and activities. It also handles argument deserialization back into Ruby objects.

Examples:

Payload structure

{
  job_class: "MyJob",
  job_id: "abc-123",
  queue_name: "default",
  active_job: {
    "arguments" => [{"_aj_serialized"=>"ActiveJob::Serializers::ObjectSerializer", ...}],
    ...
  },
  executions: 0,
  exception_executions: {},
  scheduled_at: "2025-10-29T12:00:00Z" # optional
}

See Also:

Constant Summary collapse

PAYLOAD_WARNING_THRESHOLD = 
0.8
PAYLOAD_NEAR_LIMIT_THRESHOLD = 
0.9
WORKFLOW_CONTROL_FIELDS = 
%i[
  scheduled_at
  default_activity_options
  retry_policy
  temporal_options
  workflow_identity
  dead_letter
  rate_limits
  workflow_interactions
  child_workflows
  chain
  dependencies
  dependency_failure_policy
  activity_task_queue
  continue_as_new
  workflow_state
  local_activity_helpers
  observability
  schedule_id
  schedule_workflow_id_prefix
  schedule_execution_job_id
  payload_encryption_context
].freeze

Instance Method Summary collapse

Instance Method Details

#delete_external_payload(payload, config: ActiveJob::Temporal.config) ⇒ Object 



238
239
240
 
# File 'lib/activejob/temporal/payload.rb', line 238

def delete_external_payload(payload, config: ActiveJob::Temporal.config)
  PayloadStorage.delete(payload, config: config)
end

#deserialize_args(payload, config: ActiveJob::Temporal.config, encryption_context: nil) ⇒ Array

Deserializes job arguments from a payload hash.

Extracts the serialized arguments array from ActiveJob's canonical serialized job data, falling back to legacy top-level arguments, then uses ActiveJob's built-in deserialization to reconstruct Ruby objects (including GlobalID references to ActiveRecord models).

Examples:

Deserialize arguments

payload = { active_job: { "arguments" => [{"_aj_serialized"=>"..."}] } }
args = Payload.deserialize_args(payload)
# => [actual_ruby_object]

GlobalID deserialization with deleted record

begin
  payload = { active_job: { "arguments" => [{"_aj_globalid"=>"gid://app/User/999"}] } }
  args = Payload.deserialize_args(payload)
rescue ActiveRecord::RecordNotFound => e
  # Record was deleted between enqueue and execution
  Rails.logger.warn("Job argument no longer exists: #{e.message}")
end

Parameters:

  • payload (Hash) —

    Payload hash containing serialized ActiveJob data

Options Hash (payload):

  • :active_job (Hash) —

    Full ActiveJob serialized payload

  • :arguments (Array) —

    Legacy serialized arguments

Returns:

  • (Array) —

    Deserialized arguments array ready for job.perform(*args)

Raises:

  • (ActiveJob::SerializationError) —

    if deserialization fails

  • (GlobalID::RecordNotFound) —

    if a GlobalID reference points to a deleted record



199
200
201
 
# File 'lib/activejob/temporal/payload.rb', line 199

def deserialize_args(payload, config: ActiveJob::Temporal.config, encryption_context: nil)
  deserialize_payload_args(deserialize_payload(payload, config: config, encryption_context: encryption_context))
end

#deserialize_payload(payload, config: ActiveJob::Temporal.config, encryption_context: nil) ⇒ Object 



212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/activejob/temporal/payload.rb', line 212

def deserialize_payload(payload, config: ActiveJob::Temporal.config, encryption_context: nil)
  stored_payload = PayloadStorage.load(
    payload,
    config: config,
    workflow_control_fields: WORKFLOW_CONTROL_FIELDS
  )
  transport_payload = decrypt_transport_payload(stored_payload, config, encryption_context)

  execution_payload = serializer_for_transport_payload(transport_payload, config).load(transport_payload)
  preserve_workflow_control_fields(transport_payload, execution_payload)
end

#deserialize_payload_args(payload) ⇒ Object 



203
204
205
206
207
208
209
210
 
# File 'lib/activejob/temporal/payload.rb', line 203

def deserialize_payload_args(payload)
  serialized_args = active_job_arguments(payload) || payload[:arguments] || payload["arguments"]
  ActiveJob::Arguments.deserialize(serialized_args)
rescue ActiveJob::SerializationError, ActiveJob::Temporal::ConfigurationError
  raise
rescue StandardError => e
  raise ActiveJob::SerializationError, e.message
end

#encrypt_payload(payload, config: ActiveJob::Temporal.config, encryption_context: nil) ⇒ Object 



224
225
226
 
# File 'lib/activejob/temporal/payload.rb', line 224

def encrypt_payload(payload, config: ActiveJob::Temporal.config, encryption_context: nil)
  encrypt_payload_if_configured(payload, config, encryption_context: encryption_context)
end

#enforce_size!(payload, metrics_payload: payload, config: ActiveJob::Temporal.config, byte_size: nil) ⇒ Object

Raises:

  • (ActiveJob::SerializationError) 


242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/activejob/temporal/payload.rb', line 242

def enforce_size!(payload, metrics_payload: payload, config: ActiveJob::Temporal.config, byte_size: nil)
  bytes = byte_size || JSON.generate(payload).bytesize
  max_size_kb = config.max_payload_size_kb || 250
  size_limit_bytes = max_size_kb * 1024
  actual_size_kb = bytes / 1024.0
  usage_ratio = bytes.to_f / size_limit_bytes

  Observability.emit(
    :payload_serialize,
    Observability.attributes_from_payload(metrics_payload, bytes: bytes)
  )
  log_payload_size(metrics_payload, actual_size_kb, max_size_kb, usage_ratio)
  return if bytes <= size_limit_bytes

  message = format(
    "Job payload size (%<actual>.1f KB) exceeds maximum allowed size (%<max>d KB). " \
    "Consider reducing argument size or using references (e.g., database IDs).",
    actual: actual_size_kb,
    max: max_size_kb
  )
  raise ActiveJob::SerializationError, message
end

#from_job(job, scheduled_at: nil, enforce_size: true, config: ActiveJob::Temporal.config, **options) ⇒ Hash

Note:

Record Lifecycle Caveat When using GlobalID serialization for ActiveRecord models, the record MUST exist in the database at both enqueue time AND execution time. If the record is deleted before the job executes, deserialization will fail with ActiveRecord::RecordNotFound.

Note:

Payload Size Optimization To reduce payload size, prefer passing database IDs instead of full ActiveRecord objects. For example, pass user.id instead of user. This is especially important for jobs with large argument lists or complex nested objects.

Converts an ActiveJob instance into a serializable payload hash.

Examples:

Basic job payload

job = MyJob.new
payload = Payload.from_job(job)
# => { job_class: "MyJob", job_id: "...", active_job: { "arguments" => [...] }, ... }

Scheduled job payload

job = MyJob.new
payload = Payload.from_job(job, scheduled_at: 1.hour.from_now)
# => { ..., scheduled_at: "2025-10-29T13:00:00Z" }

Handling payload size errors

begin
  MyJob.perform_later(large_object)
rescue ActiveJob::SerializationError => e
  Rails.logger.error("Payload too large: #{e.message}")
  # Recommendation: Pass ID instead of full object
  MyJob.perform_later(large_object.id)
end

GlobalID serialization (ActiveRecord models)

user = User.find(123)
MyJob.perform_later(user)  # Serializes as GlobalID
# Payload contains: { "_aj_globalid" => "gid://app/User/123" }

Non-serializable object error

begin
  MyJob.perform_later(File.open("/tmp/file.txt"))
rescue ActiveJob::SerializationError => e
  # => "Unsupported argument type: File"
  Rails.logger.error(e.message)
end

Parameters:

  • job (ActiveJob::Base) —

    The job instance to serialize

  • scheduled_at (Time, String, nil) (defaults to: nil) —

    Optional scheduled execution time

Returns:

  • (Hash) —

    Serialized payload with keys:

    • :job_class [String] Fully-qualified job class name
    • :job_id [String] Unique job identifier
    • :queue_name [String] Target queue name
    • :active_job [Hash] ActiveJob serialized payload, including arguments
    • :executions [Integer] Current execution count (default 0)
    • :exception_executions [Hash] Exception execution counts (default {})
    • :scheduled_at [String] ISO8601 timestamp (optional)

Raises:

  • (ActiveJob::SerializationError) —

    if arguments cannot be serialized

  • (ActiveJob::SerializationError) —

    if payload exceeds max_payload_size_kb (includes actual size in message)

  • (ArgumentError) —

    if scheduled_at is not convertible to Time

  • (ArgumentError) —

    if job is nil

  • (NoMethodError) —

    if job does not respond to required attributes

  • (JSON::GeneratorError) —

    if payload cannot be JSON-serialized



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/activejob/temporal/payload.rb', line 143

def from_job(
  job,
  scheduled_at: nil,
  enforce_size: true,
  config: ActiveJob::Temporal.config,
  **options
)
  encrypt = options.fetch(:encrypt, true)
  offload = options.fetch(:offload, enforce_size)
  encryption_context = options[:encryption_context]
  storage_metadata = options[:storage_metadata]
  payload = payload_from_job_attributes(job)
  payload[:scheduled_at] = iso8601_timestamp(scheduled_at) if scheduled_at

  scheduled_timestamp = payload.delete(:scheduled_at)
  final_payload = serializer_for(config).dump(payload)
  final_payload[:scheduled_at] = scheduled_timestamp if scheduled_timestamp
  final_payload = encrypt_payload_for_transport(final_payload, encrypt, config, encryption_context)
  finalize_transport_payload(
    final_payload,
    metrics_payload: payload,
    config: config,
    offload: offload ? storage_metadata || {} : nil,
    enforce_size: enforce_size
  )
end

#offload_payload(payload, metadata:, config: ActiveJob::Temporal.config, byte_size: nil) ⇒ Object 



228
229
230
231
232
233
234
235
236
 
# File 'lib/activejob/temporal/payload.rb', line 228

def offload_payload(payload, metadata:, config: ActiveJob::Temporal.config, byte_size: nil)
  PayloadStorage.offload_if_needed(
    payload,
    config: config,
    metadata: metadata,
    workflow_control_fields: WORKFLOW_CONTROL_FIELDS,
    byte_size: byte_size
  )
end