Fix scheduled job handling in adapter #16
Conversation
* They were failing because `timestamp` is a float, not a `ActiveSupport::TimeWithZone` object. * When jobs are created via `set`, they already come in with `scheduled_at` set as a `ActiveSupport::TimeWithZone` object. * However, just in case they don't, and expect #enqueue_at to work when passing a timestamp, we convert it to a `ActiveSupport::TimeWithZone` object and set it on `job.scheduled_at`. 1. `lib/active_job/enqueuing.rb#_raw_enqueue` calls `queue_adapter.enqueue_at self, scheduled_at.to_f` (notice it converts the timestamp to a float) [source](https://github.com/rails/rails/blob/v8.1.0/activejob/lib/active_job/enqueuing.rb#L134) 2. `lib/active_job/queue_adapters/async_job_adapter.rb#enqueue_at` receives `timestamp`, e.g. `1761485982.6905391`. 3. `lib/active_job/queue_adapters/async_job_adapter.rb#enqueue_at` sets it on `job.scheduled_at = timestamp` [source](https://github.com/socketry/async-job-adapter-active_job/blob/v0.18.3/lib/active_job/queue_adapters/async_job_adapter.rb#L34) 4. `lib/active_job/queue_adapters/async_job_adapter.rb#enqueue_at` calls `@dispatcher.call(job)` 5. `lib/async/job/adapter/active_job/dispatcher.rb#call` calls `job.serialize` [source](https://github.com/socketry/async-job-adapter-active_job/blob/v0.18.3/lib/async/job/adapter/active_job/dispatcher.rb#L50) 6. `lib/active_job/core.rb#serialize` calls `"scheduled_at" => scheduled_at ? scheduled_at.utc.iso8601(9) : nil` [source](https://github.com/rails/rails/blob/v8.1.0/activejob/lib/active_job/core.rb#L130) 7. `undefined method 'utc' for an instance of Float` because `scheduled_at` was set to the timestamp.
| before do | ||
| dispatcher.queues["default"] = queue | ||
| job.set(wait:) | ||
| end |
There was a problem hiding this comment.
The other spec tests that calling adapter.enqueue_at(job, timestamp.to_f) without explicitly setting scheduled_at on the job via set will successfully set scheduled_at.
This new spec tests that setting it on the job as you're supposed to (via set) will work.
We might not want to support the former test, but again, leaving it just in case we want to guarantee that.
| # @parameter timestamp [Time] The time at which to enqueue the job. | ||
| def enqueue_at(job, timestamp) | ||
| job.scheduled_at = timestamp | ||
| job.scheduled_at ||= Time.at(timestamp) |
There was a problem hiding this comment.
Again, since this is an Active Job adapter, I'd suggest just removing it since we know jobs will already come with scheduled_at set, but I'm leaving it there just in case we want to guarantee that #enqueue_at sets it on the job when passing a timestamp.
I understand that commit 3b4fd0b - Fix handling of scheduled jobs. added this so there might be a more nuanced reason why we make sure its set here, so that's yet another reason why I'm not blindly removing it.
1 participant
Description
Summary
They were failing because the
timestampargument passed toActiveJob::QueueAdapters::AsyncJobAdapter#enqueue_atis a float, not aActiveSupport::TimeWithZone/Timeobject.The adapter was setting
job.scheduled_at = timestamp, which causedActiveJob::Core#serializeto fail withundefined method 'utc' for an instance of Floatbecause it does:"scheduled_at" => scheduled_at ? scheduled_at.utc.iso8601(9) : nil.Fix
This PR fixes the above by using
||=to set thejob.scheduled_atto aTimeobject viaTime.at.This is not really necessary since when jobs are created via
set, they already come in withscheduled_atset as aActiveSupport::TimeWithZoneobject.Since this is an Active Job adapter, I'd suggest just removing it since we know jobs will already come with
scheduled_atset, but I'm leaving it there just in case we want to guarantee that#enqueue_atsets it on the job when passing a timestamp.Detailed stack trace
lib/active_job/enqueuing.rb#_raw_enqueuecallsqueue_adapter.enqueue_at self, scheduled_at.to_f(notice it converts the timestamp to a float)source
lib/active_job/queue_adapters/async_job_adapter.rb#enqueue_atreceivestimestamp, e.g.1761485982.6905391source.lib/active_job/queue_adapters/async_job_adapter.rb#enqueue_atsets it onjob.scheduled_at = timestampsource
lib/active_job/queue_adapters/async_job_adapter.rb#enqueue_atcalls@dispatcher.call(job)source.lib/async/job/adapter/active_job/dispatcher.rb#callcallsjob.serializesourcelib/active_job/core.rb#serializecalls"scheduled_at" => scheduled_at ? scheduled_at.utc.iso8601(9) : nilsourceundefined method 'utc' for an instance of Floatbecausescheduled_atwas set to the timestamp.Types of Changes
Contribution