Enqueue awaited job

lib/acidic_job.rb

@@ -50,13 +50,9 @@ def self.wire_everything_up(klass)
     # TODO: write test for a staged job that uses awaits
     klass.set_callback :perform, :after, :reenqueue_awaited_by_job,
                        if: -> { was_awaited_job? && !was_workflow_job? }
-    klass.set_callback :perform, :after, :delete_staged_job_record,
-                       if: -> { was_staged_job? && !was_awaited_job? }
     klass.define_callbacks :finish
     klass.set_callback :finish, :after, :reenqueue_awaited_by_job,
                        if: -> { was_workflow_job? && was_awaited_job? }
-    klass.set_callback :finish, :after, :delete_staged_job_record,
-                       if: -> { was_workflow_job? && was_staged_job? && !was_awaited_job? }
 
     klass.instance_variable_set(:@acidic_identifier, :job_id)
     klass.define_singleton_method(:acidic_by_job_id) { @acidic_identifier = :job_id }

lib/acidic_job/awaiting.rb

@@ -32,11 +32,10 @@ def reenqueue_awaited_by_job
 
       return if run.finished?
 
-      # job = job_class.constantize.deserialize(serialized_staged_job)
-      # job.enqueue
-
       # when a batch of jobs for a step succeeds, we begin processing the `AcidicJob::Run` record again
-      process_run(run)
+      # process_run(run)
+      run.update_column(:locked_at, nil)
+      job.enqueue
     end
 
     def enqueue_step_parallel_jobs(jobs_or_jobs_getter, run, step_result)

lib/acidic_job/extensions/sidekiq.rb

@@ -58,7 +58,7 @@ def enqueue
           ::Sidekiq::Client.push(
             "class" => self.class,
             "args" => @args,
-            "jid" => @jid
+            "jid" => @acidic_job_run&.staged_job_id || @jid
           )
         end
       end

lib/acidic_job/run.rb

@@ -64,16 +64,19 @@ def failed?
       error_object.present?
     end
 
-    private
-
-    def enqueue_staged_job
-      return unless staged?
-
+    def staged_job_id
       # encode the identifier for this record in the job ID
       # base64 encoding for minimal security
       global_id = to_global_id.to_s.remove("gid://")
       encoded_global_id = Base64.encode64(global_id).strip
-      staged_job_id = "STG__#{idempotency_key}__#{encoded_global_id}"
+
+      "STG__#{idempotency_key}__#{encoded_global_id}"
+    end
+
+    private
+
+    def enqueue_staged_job
+      return unless staged?
 
       serialized_staged_job = if serialized_job.key?("jid")
                                 serialized_job.merge("jid" => staged_job_id)

lib/acidic_job/staging.rb

@@ -9,15 +9,6 @@ module Staging
 
     private
 
-    def delete_staged_job_record
-      return unless was_staged_job?
-
-      staged_job_run.delete
-      true
-    rescue ActiveRecord::RecordNotFound
-      true
-    end
-
     def was_staged_job?
       identifier.start_with? "STG__"
     end

test/acidic_job_test.rb

@@ -79,7 +79,7 @@ def test_passes_for_a_new_key
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_returns_a_stored_result
@@ -111,7 +111,7 @@ def test_passes_for_keys_that_are_unlocked
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_passes_for_keys_with_a_stale_locked_at
@@ -128,7 +128,7 @@ def test_passes_for_keys_with_a_stale_locked_at
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_stores_results_for_a_permanent_failure
@@ -185,7 +185,7 @@ def test_continues_from_recovery_point_create_ride_and_audit_record
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
       assert_equal key.attr_accessors,
                    { "user_id" => @valid_user.id, "params" => @valid_params, "ride" => Ride.first }
     end
@@ -207,7 +207,7 @@ def test_continues_from_recovery_point_create_stripe_charge
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 0, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
       assert_equal key.attr_accessors,
                    { "user_id" => @valid_user.id, "params" => @valid_params, "ride" => Ride.first }
     end
@@ -226,7 +226,7 @@ def test_continues_from_recovery_point_send_receipt
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 0, Ride.count
       assert_equal 0, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
       assert_equal key.attr_accessors,
                    { "user_id" => @valid_user.id, "params" => @valid_params, "ride" => nil }
     end
@@ -245,7 +245,7 @@ def test_halts_execution_of_steps_when_safely_finish_acidic_job_returned
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 0, Ride.count
       assert_equal 0, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
       assert_equal key.attr_accessors,
                    { "user_id" => @valid_user.id, "params" => @valid_params, "ride" => nil }
     end

test/acidic_worker_test.rb

@@ -75,7 +75,7 @@ def test_passes_for_a_new_key
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_returns_a_stored_result
@@ -107,7 +107,7 @@ def test_passes_for_keys_that_are_unlocked
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_passes_for_keys_with_a_stale_locked_at
@@ -124,7 +124,7 @@ def test_passes_for_keys_with_a_stale_locked_at
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_stores_results_for_a_permanent_failure
@@ -160,7 +160,7 @@ def test_continues_from_recovery_point_create_ride_and_audit_record
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 1, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_continues_from_recovery_point_create_stripe_charge
@@ -180,7 +180,7 @@ def test_continues_from_recovery_point_create_stripe_charge
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 1, Ride.count
       assert_equal 0, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
 
     def test_continues_from_recovery_point_send_receipt
@@ -197,7 +197,7 @@ def test_continues_from_recovery_point_send_receipt
       assert_equal 1, AcidicJob::Run.unstaged.count
       assert_equal 0, Ride.count
       assert_equal 0, Audit.count
-      assert_equal 0, AcidicJob::Run.staged.count
+      assert_equal 1, AcidicJob::Run.staged.count
     end
   end
 

test/workflow_test.rb

@@ -345,4 +345,36 @@ def dynamic_awaiting
     assert_equal 1, retry_set.size
     assert_equal ["ErroringDynamicAwaitFromStringWorker"], retry_set.map { _1.item["class"] }
   end
+
+  def test_step_with_awaits_followed_by_another_step_is_run_properly
+    dynamic_class = Class.new(ApplicationWorker) do
+      dynamic_step_job = Class.new(ApplicationWorker) do
+        def perform; end
+      end
+      Object.const_set("SimpleAwaitedWorker", dynamic_step_job)
+
+      def perform
+        with_acidity providing: {} do
+          step :await_step, awaits: [SimpleAwaitedWorker]
+          step :do_something
+        end
+      end
+
+      def do_something; end
+    end
+    Object.const_set("WorkerWithAwaitStepFollowedByAnotherStep", dynamic_class)
+
+    WorkerWithAwaitStepFollowedByAnotherStep.new.perform
+    Sidekiq::Worker.drain_all
+
+    assert_equal 2, AcidicJob::Run.count
+
+    parent_run = AcidicJob::Run.find_by(job_class: "WorkerWithAwaitStepFollowedByAnotherStep")
+    assert_equal "FINISHED", parent_run.recovery_point
+
+    child_run = AcidicJob::Run.find_by(job_class: "SimpleAwaitedWorker")
+    assert_nil child_run.recovery_point
+
+    assert_equal 0, Sidekiq::RetrySet.new.size
+  end
 end