jobtype in polyglot

[cjbottaro]

Hi! I have jobs being enqueued with jobtype "AsyncWork.DocumentJob" and then the Ruby worker client is bombing trying to constantize that.

I thought I could make middleware to handle this, but the middleware gets the instantiated jobtype instead of the "payload".

It would be pretty easy to make a patch that gives the JSON payload to middleware and doesn't instantiate the job class until after middleware is run, but that's a potentially breaking change.

Another solution would be add :jobtype as an option to job classes like

class DocumentJob
  include Faktory::Job
  faktory_options jobtype: "AsyncJob.DocumentJob"
  ...
end

But that's tricky because we'd have to keep track of all job classes and I'm not sure how well that work with Rail's lazy loading of things.

Third option is have a global config for it?

Faktory.configure_worker do |config|
  config.jobtype_map = {
    "AsyncWork.DocumentJob" => "DocumentJob"
  }
end

What do you think? I would be happy to implement any of these or any other idea.

[mperham]

FWR assumes that jobtype maps onto a Ruby Job subclass. I'm not sure we want to break that convention but I might be ok with an adapter that could be plugged into FWR in lieu of the default constantize mechanism. Is there some reason your system can't adapt to the convention?

[cjbottaro]

The Faktory worker package for Elixir is flexible enough to override the jobtype on enqueuing:

# Typical enqueuing
AsyncWork.DocumentJob.perform_async([1, 2]) # jobtype: "AsyncWork.DocumentJob"

# Or use low level function
Faktory.push("Some::Class", [1, 2]) # jobtype: "Some::Class"

But since Faktory is for polyglot environments, I thought any language should be able to enqueue jobs for any other language to consume.

And I guess the convention can be that the enqueuer needs to know the jobtype that the consumer is expecting... but it seems more intuitive to me the other way around.

[mperham]

I think that's the rub: do we add the optional complexity in the Ruby worker package to map from arbitrary string to Ruby class or do we push the issue back to the Elixir client and force it to know the conventions for calling a Ruby job? I don't know the right answer, curious what other people think. If I rename a Ruby class, should the Elixir client code have to change?

You could also have an API like FWR's Faktory::Job#set API which allows overriding the default values:

SomeJob.set(queue: 'foo', jobtype: 'SomeOtherName').perform_async("foo")

[cjbottaro]

curious what other people think.

Same here.

I guess for now, both the Ruby client and the Elixir client (and other clients?) can specify the jobtype when enqueuing, so this isn't a major issue or anything.

You could also have an API like

SomeJob.set(queue: 'foo', jobtype: 'SomeOtherName').perform_async("foo")

I had an API like that, but then removed it in favor of just passing an options hash (err, keyword list), because Elixir doesn't support argument splatting, so it's really easy to distinguish between job args and enqueuing options.

job_args = ["hi", 123]
SomeJob.perform_async(job_args, queue: "not_default")

Just need to expand the options to accept :jobtype. 👍

[mperham]

Sounds like you have a path forward so I'm going to close this.

[fnordfish]

Almost a year late to the party, but ran into the same question. So, I was trying the suggested method and here's the funny thing:

Using set with symbol :jobtype produces the desired outcome:

SomeRubyJob.set(jobtype: 'rs_func', queue: 'rs').perform_async(1, 2, 3)
#=>
{"retry"=>25,
 "queue"=>"rs",
 "jobtype"=>"rs_func",
 "args"=>[1, 2, 3],
 "jid"=>"e15ba3d76445957a27831e24"}

But using set with string jobtype does not:

SomeRubyJob.set('jobtype' => 'rs_func', 'queue' => 'rs').perform_async(1, 2, 3)
#=> 
{"retry"=>25,
 "queue"=>"rs",
 "jobtype"=>PloyglotTest::SomeRubyJob,
 "args"=>[1, 2, 3],
 "jid"=>"7b7253c2f047f170a41f1104"}

That's because Set explicitly overwrites string jobtype with the ruby class name and when the job options are later stringified, the symbol :jobtype happens to win. The whole thing makes me wonder:

Is using Set a stable API for overwriting the jobtype?
Is using Job the right place to schedule a polyglot job, or shouldn't we use Client.push instead?
The only disadvantage in using Client directly is, that Job does all the nifty things like perform_in (calculating the at option), setting default options (not a big deal, imho), handling server pools and running client middlewares.

In the end, it also looks pretty weird to have a FWR job class with no perform method and calling perform_async on it. Using some kind of client class would make things a bit more obvious.

I also understand that the current Client class is just the communication layer, and all that stuff Job does has no real place in it.

Could there be room for an extra class or mixin the Job class and a "MyPolyglotJobClient" class would use (to get at option calculation, server pool handling and client middleware invocation)?

[mperham]

At the end of the day, Faktory::Job is just semantic sugar for building a proper job hash. Build your own helper method to make the hash in a form that is intuitive to you.

cl = Faktory::Client.new
cl.push({"retry"=>25,
 "queue"=>"rs",
 "jobtype"=>"rs_func",
 "args"=>[1, 2, 3],
 "jid"=>"7b7253c2f047f170a41f1104"})

[fnordfish]

Right. I guess, I'm just trying to understand what could go wrong if I'd just do exactly that (Faktory::Client.new.push(...)) and not go through client_middleware.

[mperham]

Yep, I see what you mean. There are several middleware which would be skipped. We could move the middleware execution point into the Faktory::Client#push method so it is not skipped. WDYT?

[fnordfish]

Sorry for the delay ... time zones.

Honestly, I like how simple Faktory::Client is :)

I've done some quick hacking and come up with this (naming might need some work)

module Faktory
  module Handler
    def perform_async(jobtype, args, faktory_options = Faktory.default_job_options)
      client_push({'jobtype'.freeze => jobtype, 'args'.freeze => args}, faktory_options)
    end

    def perform_in(interval, jobtype, args, faktory_options = Faktory.default_job_options)
      int = interval.to_f
      now = Time.now.to_f
      ts = (int < 1_000_000_000 ? now + int : int)
      item = {'jobtype'.freeze => jobtype, 'args'.freeze => args}
      item['at'.freeze] = Time.at(ts).utc.to_datetime.rfc3339(9) if ts > now

      client_push(item, faktory_options)
    end
    alias_method :perform_at, :perform_in

    def client_push(item, faktory_options = Faktory.default_job_options) # :nodoc:
      pool = Thread.current[:faktory_via_pool] || faktory_options['pool'.freeze] || Faktory.server_pool
      item = faktory_options.merge(item)
      # stringify
      item.keys.each do |key|
        item[key.to_s] = item.delete(key)
      end
      item["jid"] ||= SecureRandom.hex(12)
      item["queue"] ||= "default"

      Faktory.client_middleware.invoke(item, pool) do
        pool.with do |c|
          c.push(item)
        end
      end
    end

    module_function :perform_at, :perform_in, :perform_async, :client_push
  end
end

... and letting the Jobs perform_async, perform_in and client_push just delegate to Handler.

Now, that polyglot use case would just look like this

Faktory::Handler.perform_async('rs_func', [1,2,3], queue: 'rs')

What's your take on this approach?

[mperham]

Hmm, I'm fine overall with that approach. I think I'd prefer it on Faktory::Job and would rename client_push to just push. Want to open a new issue or send a PR?

[fnordfish]

Sounds like a good idea. I’ve god a long train ride coming up :)

From: Mike Perham notifications@github.commailto:notifications@github.com Reply: contribsys/faktory_worker_ruby reply@reply.github.commailto:reply@reply.github.com Date: 20. December 2019 at 18:13:20 To: contribsys/faktory_worker_ruby faktory_worker_ruby@noreply.github.commailto:faktory_worker_ruby@noreply.github.com CC: Robert Schulze robert@dotless.demailto:robert@dotless.de, Comment comment@noreply.github.commailto:comment@noreply.github.com Subject: Re: [contribsys/faktory_worker_ruby] jobtype in polyglot (#30)

Hmm, I'm fine overall with that approach. I think I'd prefer it on Faktory::Job and would rename client_push to just push. Want to open a new issue or send a PR?

— You are receiving this because you commented. Reply to this email directly, view it on GitHubhttps://github.com/contribsys/faktory_worker_ruby/issues/30?email_source=notifications&email_token=AAADEJ6DOPYDNVHECYXQVILQZT4LBA5CNFSM4GL33EIKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEHNRFIA#issuecomment-568005280, or unsubscribehttps://github.com/notifications/unsubscribe-auth/AAADEJY36RIL5LA2CD7GVFTQZT4LBANCNFSM4GL33EIA.