Race condition when a task is interrupted leads to exceptions

[etiennebarrie]

First Job	Second Job	Run status
Task is created, job enqueued in Runner.run		enqueued
TaskJob#before_enqueue		running
Job is re-enqueued in JobIteration::Iteration#reenqueue_iteration_job
~	TaskJob#before_enqueue	running
TaskJob#after_perform		interrupted
~	Task is processing	interrupted

At this point you have a running Task which a job is actively processing, so in the UI it's a bit weird because you can see progress being made to a seemingly interrupted Task.
An exception is then raised when the Task succeeds, because we try to move the status from interrupted to succeeded.

Before 1.1.0, this exception is caught and another one is raised when we try to move from interrupted to errored.

What happened next with 1.0.0 and sidekiq is that the job was retried because there was an unhandled exception (with the previous cursor as explained in the README). That finally led the task to finish and succeed. Re-using the last cursor meant that elements were processed multiple times, luckily the process was idempotent.

With 1.1.0, the exception is caught, and the task is moved to errored thanks to the second commit of #293, which was meant for deleted tasks but applies here as well: dbe3234.
The only solution at this point is to run the task again, which may skip the already processed items with the collection method, or last line of defense hopefully the process method is idempotent.

[adrianna-chang-shopify]

Okay, so if I understand correctly, the current situation with 1.1 is this:

• A Task can get interrupted and reenqueued so fast that you have an "interrupted" Task that is actually running, because the status gets set to running when it is starts performing again, and then is set to interrupted afterwards when the initial job goes through it's shutdown callback
• When the Task finishes successfully, it will get set to "succeeded", which is invalid
• This triggers a transition from "interrupted" -> "errored", which is now valid in 1.1

So, with 1.1 we won't run into the issue of Sidekiq retrying the job because of an unhandled exception, because we're rescuing appropriately now. But, we may have missed some iterations because the cursor was not correct when the job was reenqueued (since this happened before the shutdown callback).

Can you clarify

The only solution at this point is to run the task again, which may skip the already processed items with the collection method

Wouldn't process need to be idempotent? Running the Task again will re-process every element in the collection, so unless it is completely idempotent, I think we'd run into trouble here? Either way, I guess the fact that we re-use an old cursor any time we get into this race condition is going to require process to be idempotent, unless we fix it 😅

Could we solve this by simply waiting a bit to retry the job after it gets interrupted so that the shutdown has time to complete?

[etiennebarrie]

So, with 1.1 we won't run into the issue of Sidekiq retrying the job because of an unhandled exception, because we're rescuing appropriately now. But, we may have missed some iterations because the cursor was not correct when the job was reenqueued (since this happened before the shutdown callback).

Oh sorry in 1.1 the job is errored and stops, and we don't miss any iterations because it was finished. It's a successful job that appears errored. And so there's no need to run the task again, I got that wrong.

But, we may have missed some iterations because the cursor was not correct when the job was reenqueued (since this happened before the shutdown callback).

I think this part may be safe because we use the cursor from the re-enqueued job rather than the one from the Run if it's available (in TaskJob#build_enumerator passed by JobIteration). We only use the one from the Run when the Run was paused.

Could we solve this by simply waiting a bit to retry the job after it gets interrupted so that the shutdown has time to complete?

Yeah, though I wonder if we shouldn't fix this in job-iteration instead, something similar can probably happen for any user of that gem, with an after_ or around_perform. Also not sure how we can make sure that callback would run last.
I'd rather enqueue at the right time (after persisting) than using delays though if we can.

[adrianna-chang-shopify]

I think this part may be safe because we use the cursor from the re-enqueued job rather than the one from the Run if it's available

Ah, yes right 🤦‍♀️ OK so yeah we should be good there, we should be able to rely on the cursor being correct, it's really just a race in terms of updating the Run.

Yeah, though I wonder if we shouldn't fix this in job-iteration instead, something similar can probably happen for any user of that gem, with an after_ or around_perform.

Yeah, it seems any of the after_perform, around_perform and on_shutdown hooks are at risk of not finishing prior to the next job starting up. But you're right, it's a bit brittle to just use delays.

I'd rather enqueue at the right time (after persisting) than using delays though if we can.

Another idea, what if we still used delays, but basically repeatedly retried using delays until the Run is in a status we want?

module MaintenanceTasks
  class TaskJob < ActiveJob::Base
    include JobIteration::Iteration

    def interruptible_perform(*arguments)
      unless @run.previous_changes[:status]
        self.class.set(wait: 10.seconds).perform_later(run)
        return
      end

      super(*arguments)
    end

So here, if the @run has gone from running -> running, @run.previous_changes[:status] will be nil. If that's the case, we reenqueue the job to be performed in 10 seconds, and then check again to see if a status change has happened.

So the flow would look something like:

First Job	Second Job	Run status
Task is created, job enqueued in Runner.run		enqueued
TaskJob#before_enqueue		running
Job is re-enqueued in JobIteration::Iteration#reenqueue_iteration_job
~	TaskJob#before_enqueue	running
~	TaskJob#interruptible_perform => Delays running the job by 10 seconds bc no status change	running
TaskJob#on_shutdown		interrupted
~	TaskJob#before_enqueue	running
~	TaskJob#interruptible_perform => Status condition change is now ["interrupted", "running"], so Task begins processing	running

This might be completely crazy, LMK what you think 😁

[etiennebarrie]

I think rather than deal with the job being enqueued too early, we should enqueue it when we're ready for it to be picked up by another worker. I think we could instead override reenqueue_iteration_job, preventing job-iteration from retrying the job too early, then in after_perform, once we persisted the job, we could enqueue at this point.

[adrianna-chang-shopify]

This is closed by #345 , although we may revisit patching this in JobIteration eventually.