Concurrency key doesn't handle Hash: TypeError (can't cast Hash)

[Earlopain]

As the title says. The job arguments are serialized by ActiveJob to something the database understands but the same is not true for the concurrency key.

class ExampleJob < ApplicationJob
  include GoodJob::ActiveJobExtensions::Concurrency
  good_job_control_concurrency_with(total_limit: 1, key: -> { arguments.first })

  def perform(arg)
  end
end

Job.new.perform({a: "test"})
# => All good
Job.perform_later({a: "test"})
# => Failed enqueuing Job to GoodJob(queue): TypeError (can't cast Hash)

Not sure if what I was trying to do should even be supported, but the error message I get as a result of doing this is not very clear and lead me to believe that the issues was with the parameters I was passing to the job.

Edit: Updated the issue description after figuring out what was actually going on.

[bensheldon]

@Earlopain interesting! I'll have to look into this. GoodJob doesn't directly serialize attributes to Postgres. ActiveJob serializes all of the job data to a hash, that GoodJob then inserts into a Postgres jsonb column.

Could you possibly be setting a different Job value (queue name, priority, delay) to an unexpected value?

These are the attributes that GoodJob tries to insert into the database:

good_job/app/models/good_job/execution.rb

Lines 253 to 260 in f683f37

	execution_args = {
	active_job_id: active_job.job_id,
	queue_name: active_job.queue_name.presence || DEFAULT_QUEUE_NAME,
	priority: active_job.priority || DEFAULT_PRIORITY,
	serialized_params: active_job.serialize,
	scheduled_at: scheduled_at,
	create_with_advisory_lock: create_with_advisory_lock,
	}

[bensheldon]

btw, I am successfully able to pass hash arguments to a job:

[8] pry(main)> ExampleJob.perform_later({a: "test"})
  TRANSACTION (0.9ms)  BEGIN
  GoodJob::Execution Create (1.8ms)  INSERT INTO "good_jobs" ("queue_name", "priority", "serialized_params", "created_at", "updated_at", "active_job_id") VALUES ($1, $2, $3, $4, $5, $6) RETURNING "id"  [["queue_name", "default"], ["priority", 0], ["serialized_params", "{\"job_class\":\"ExampleJob\",\"job_id\":\"1937675a-81a1-439b-8cb9-b3beb90b12cd\",\"provider_job_id\":null,\"queue_name\":\"default\",\"priority\":null,\"arguments\":[{\"a\":\"test\",\"_aj_symbol_keys\":[\"a\"]}],\"executions\":0,\"exception_executions\":{},\"locale\":\"en\",\"timezone\":\"UTC\",\"enqueued_at\":\"2023-01-04T20:27:23Z\"}"], ["created_at", "2023-01-04 20:27:23.598502"], ["updated_at", "2023-01-04 20:27:23.598502"], ["active_job_id", "1937675a-81a1-439b-8cb9-b3beb90b12cd"]]
  TRANSACTION (1.3ms)  COMMIT
  SQL (0.3ms)  NOTIFY good_job, '{"queue_name":"default"}'
Enqueued ExampleJob (Job ID: 1937675a-81a1-439b-8cb9-b3beb90b12cd) to GoodJob(default) with arguments: {:a=>"test"}
=> #<ExampleJob:0x000000011b6ffc40
 @arguments=[{:a=>"test"}],
 @exception_executions={},
 @executions=0,
 @job_id="1937675a-81a1-439b-8cb9-b3beb90b12cd",
 @priority=nil,
 @provider_job_id="c307e271-8b0b-4596-859f-4bc0a143eef2",
 @queue_name="default",
 @timezone="UTC">
[9] pry(main)> ExampleJob.perform_later({a: "test"}.with_indifferent_access)
  TRANSACTION (0.3ms)  BEGIN
  GoodJob::Execution Create (13.3ms)  INSERT INTO "good_jobs" ("queue_name", "priority", "serialized_params", "created_at", "updated_at", "active_job_id") VALUES ($1, $2, $3, $4, $5, $6) RETURNING "id"  [["queue_name", "default"], ["priority", 0], ["serialized_params", "{\"job_class\":\"ExampleJob\",\"job_id\":\"bf2f5df7-6c82-4ba7-a368-a710981fa8f7\",\"provider_job_id\":null,\"queue_name\":\"default\",\"priority\":null,\"arguments\":[{\"a\":\"test\",\"_aj_hash_with_indifferent_access\":true}],\"executions\":0,\"exception_executions\":{},\"locale\":\"en\",\"timezone\":\"UTC\",\"enqueued_at\":\"2023-01-04T20:27:35Z\"}"], ["created_at", "2023-01-04 20:27:35.788395"], ["updated_at", "2023-01-04 20:27:35.788395"], ["active_job_id", "bf2f5df7-6c82-4ba7-a368-a710981fa8f7"]]
  TRANSACTION (0.8ms)  COMMIT
  SQL (0.2ms)  NOTIFY good_job, '{"queue_name":"default"}'
Enqueued ExampleJob (Job ID: bf2f5df7-6c82-4ba7-a368-a710981fa8f7) to GoodJob(default) with arguments: {"a"=>"test"}
=> #<ExampleJob:0x000000011b6b91c8
 @arguments=[{"a"=>"test"}],
 @exception_executions={},
 @executions=0,
 @job_id="bf2f5df7-6c82-4ba7-a368-a710981fa8f7",
 @priority=nil,
 @provider_job_id="a0ce0d52-83e6-4e8c-837d-d819b9e69fc6",
 @queue_name="default",
 @timezone="UTC">

Can you share how you're defining your Job and ApplicationJob classes?

[Earlopain]

I will try to get it down to a small reproducable sample tomorrow. If you're able to do this no problem then I'm probably doing something silly myself.

[Earlopain]

Here's a sample job which contains the problem:

class ExampleJob < ApplicationJob
  include GoodJob::ActiveJobExtensions::Concurrency
  good_job_control_concurrency_with(total_limit: 1, key: -> { arguments.first })

  def perform(arg)
  end
end

The issue is with the concurrency control I wrote, remove that and it works as expected. Not sure if what I was trying to do should even be supported, but the error message I get as a result of doing this is not very clear and lead me to believe that the issues was with the parameters I was passing the to job.