fractaledmind / acidic_job Goto Github PK

In the ExceptionSerializer, we serialize and deserialize exceptions stored in Run records. Some exceptions (like the ActionView::Template::Error) expect the global sigil $! to be defined, as if the exception was raised naturally. By having this $! sigil undefined, we get errors when trying to deserialize exceptions.

Thank you for you work on this gem @fractaledmind 💛 Not knowing about this gem I created https://github.com/adamniedzielski/sidekiq-staged_push which focuses on Transactionally Staged Jobs as described in https://brandur.org/job-drain and Sidekiq specifically 😅

It has a much more narrow scope and I managed to put it up in the Christmas break. I'll go through your code to compare the approaches. Feel free to close this issue.

I couldn't find any documentation on deployment and runtime related issues such as: how to start AJ, how it manages queues, monitoring, etc. Is it possible I missed the details somewhere?

If not, I think the lack of this information would stop people from even trying AJ, which is a shame as it's a very, very interesting project that I would have loved to use in my application.

The default GitHub Action setup doesn't run the tests properly or successfully

Anybody know of a Rails gem that uses HTTP 202 Accepted as a controller response for long-running jobs, queues the job, returns 202’s to requests until it processes, and when done returns a 200 OK with a response?

https://x.com/bradgessler/status/1697017603431215351?s=46&t=cXXA-9kGHUFcaLI78uAbwg

Imagine a scenario like this. We have a workflow where we need the author of a Slack message. Maybe a webhook has already processed and created our internal database representation, maybe it hasn't. So, in our job, we need to

1. try and find the internal record
2. if we don't have it, make an API call to fetch the data
3. if the API read succeeds, create the internal record
4. find the newly created internal record and set the variable

for example:

def find_author
  self.slack_author = event.installation.profiles.find_by(uid: @user_uid)
end

def fetch_unknown_author_from_slack
  return [] if @slack_author

  api_response = event.installation.client.user(uid: @user_uid)
  if api_response[:ok]
    [Slack::ProcessProfileSlackJob.with(event.installation, api_response[:user])]
  else
    raise DoNotRetryJob
  end
end

def ensure_author_known
   self.slack_author ||= event.installation.profiles.find_by(uid: @user_uid)
  raise DoNotRetryJob and return if @slack_author.nil?
end

Should this be something that the DSL makes more straight-forward?

Using a Custom Idempotent Key and an iterable step (using a for_each:), results in the job queue not running any jobs after the first job is completed. The AcidicJob will only run the first job, including all its steps, but all others jobs enqueued after this are only initialized but never run. In effect, this Custom Indempotent queue is stuck.

Sample code :

class MyJob < ApplicationJob
  # custom idempotent key so that all transactions with the distant server are sequential
  acidic_by do
    distant_server_config = arguments.first   # First parameter to the .perform method
    "DistantServer-" + distant_server_config.api_key
  end

  def perform(distant_server_config)
    @distant_server_config = distant_server_config

    with_acidic_workflow persisting: { collection: [1, 2, 3, 4, 5] } do |workflow|
      workflow.step :step1, for_each: :collection
      workflow.step :step2
    end

  def step1(num)
    p "in step1 with " + num.to_s
    true
  end

  def step2
    p "in step2"
    true
  end
end

Then do
dist_server.api_key = "abcdef"
MyJob.perform_later(dist_server)
sleep 2
MyJob.perform_later(dist_server)

Result
step1 is displayed 5 times, followed by step2
(the second job is never executed)

Expected result
step1 is displayed 5 times, followed by step2
step1 is displayed 5 times, followed by step2

Hi,
First day with AcidicJob.

I've created a Custom Idempotency Key and then tried to add a collection to one of the steps. I may have had a syntax error or something in the step or workflow, I'm not sure. But now it wont run any steps in this job, nor any more job (on this idempotency key). How can I reset the job or the whole gem?

I tried to AcidicJobRuns.destroy_all in the console but it fails with uninitialized constant AcidicJobRuns (NameError) Actually AcidicJobRuns.all fails with the same error.

Stuck :/

Serializable transactions are prone to failure: https://stackoverflow.com/a/21715207/2884386

We need to build automatic retries into our code, as this failure will naturally resolve.

It is useful to know if a Run is in a retry or on the first run. For example, imagine a step method that makes an API call to create an external resource. That step method could fail during the "execution" stage such that the API call was successful, but the run errors while trying to close the transaction or update the Run record. In this case, the job will be retried and re-run this step method.

We can add another API call as a guard to check if the resource already exists in the external system, but on the first run of the job, this is guaranteed not to be the case (presuming the job is setup properly to be idempotent). We could save the API call on the first run and only check in subsequent runs, if we knew whether the run was a first run or not.

Imagine a workflow step method that makes an API call to create a record in an external system:

def create_external_record
  self.external_record = ExternalClient.create(...)
end

Now, imagine that we get a timeout error when AcidicJob is trying to write the value to the Run#attr_accessors field.

In this case, the record is created in the external system, but we don't have a persisted copy of that record on our end.

After exploring the problem space more deeply in my RubyConf talk, I prefer the semantics of AcidicJob::Run

Currently for_each can only provide the key name for a key present in the persisting hash. This means that steps can only iterate over data present before the workflow begins. We need to be able to allow a workflow to iterate over data generated within the workflow itself. Consider, for example, a workflow job that traverses and processes the paginated responses from an API endpoint. We can have a step method to process each page, but we would need an earlier step method to make the pagination API requests to feed the iteration.

I think that everything can be handled by one AcidicJob::Run model, but I need to investigate further

Imagine having a job like this:

with_acidic_workflow persisting: { } do |workflow|
  workflow.step :no_op, awaits: [PreprocessingJob.new]
  workflow.step :do_thing
  workflow.step :no_op, awaits: :post_process_if_necessary
end

Currently, workflow steps with the same name will conflict and only one will be set in the Run#workflow hash. This is a foot-gun. We don't like foot-guns. We should resolve

Currently, we store any error object in the error_object field. This doesn't provide any context on which step caused the error, or which execution the error occurred in. We should track and store both bits of context.

When working locally (or even in production), it is possible to run a job (r1), have it error and stop on some step, update the code of the job changing the shape of the workflow, and then retry the job (r2). There are a number of pain-states that can occur in this situation:

• if persisted attributes were added, and the code to persist them was added to steps that occur before the run's recovery_point, that attribute will never get persisted

After following the prescribed setup, I keep running into the issue of Uninitialized constant AcidicJob::ActiveKiq when dealing with pure Sidekiq jobs.

I have created a small rails app for an example:
You can clone it: https://github.com/RomanTurner/sidekiq-practice
After cloning, you can run rails db:setup. This will create/migrate/seed for the database.

To find the error you can either travel to the route at localhost:3000/students/:id/example
or in the rails console run: Students::PureSidekiqJob.perform_async(:student_id)

class Students::PureSidekiqJob < AcidicJob::ActiveKiq
  include Sidekiq::Job

  def perform(student_id)
    @student = Student.find(student_id)

    with_acidic_workflow do |workflow|
      workflow.step :student_says_hello
    end
  end

  def student_says_hello
    p "Student #{@student.first_name} #{@student.last_name} says 'Hello!'"
  end
end

We need to manually check (and hopefully then write an automated test) to ensure that if a job called as an await job for a workflow step fails and ends up in the Dead Set, and is then manually re-enqueued and succeeds, that the workflow callback successfully fires and the next step in the workflow is kicked off