WIP Workflows

.gitignore

@@ -1,6 +1,7 @@
 /.bundle/
 /doc/
 /log/*.log
+/log/*.log.*
 /pkg/
 /tmp/
 /test/dummy/db/*.sqlite3

Gemfile

@@ -3,3 +3,17 @@ git_source(:github) { |repo| "https://github.com/#{repo}.git" }
 
 # Specify your gem's dependencies in solid_queue.gemspec.
 gemspec
+
+# https://github.com/brianmario/mysql2/issues/1345#issuecomment-1963508372
+gem "mysql2"
+gem "pg"
+gem "sqlite3"
+gem "pry-byebug"
+
+gem "awesome_print"
+gem "good_job"
+# gem "sidekiq", "~> 6.5"
+gem "sidekiq", "~> 7.2"
+gem "sidekiq-pro", "~> 7.2", source: "https://enterprise.contribsys.com/"
+# gem "sidekiq-batch"
+gem "after_commit_everywhere"

Gemfile.lock

@@ -9,6 +9,12 @@ PATH
       railties (>= 7.1)
       thor (~> 1.3.1)
 
+GEM
+  remote: https://enterprise.contribsys.com/
+  specs:
+    sidekiq-pro (7.2.0)
+      sidekiq (>= 7.2.0, < 8)
+
 GEM
   remote: https://rubygems.org/
   specs:
@@ -47,10 +53,19 @@ GEM
       minitest (>= 5.1)
       mutex_m
       tzinfo (~> 2.0)
+    after_commit_everywhere (1.3.1)
+      activerecord (>= 4.2)
+      activesupport
     ast (2.4.2)
+    awesome_print (1.9.2)
+    base64 (0.1.1)
     base64 (0.2.0)
+    bigdecimal (3.1.4)
     bigdecimal (3.1.8)
+    builder (3.2.4)
     builder (3.3.0)
+    byebug (11.1.3)
+    coderay (1.1.3)
     concurrent-ruby (1.3.4)
     connection_pool (2.4.1)
     crass (1.0.6)
@@ -66,8 +81,16 @@ GEM
       raabro (~> 1.4)
     globalid (1.2.1)
       activesupport (>= 6.1)
-    i18n (1.14.5)
+    good_job (3.24.0)
+      activejob (>= 6.0.0)
+      activerecord (>= 6.0.0)
+      concurrent-ruby (>= 1.0.2)
+      fugit (>= 1.1)
+      railties (>= 6.0.0)
+      thor (>= 0.14.1)
+    i18n (1.14.1)
       concurrent-ruby (~> 1.0)
+    i18n (1.14.5)
     io-console (0.6.0)
     irb (1.6.2)
       reline (>= 0.3.0)
@@ -76,12 +99,15 @@ GEM
     loofah (2.22.0)
       crass (~> 1.0.2)
       nokogiri (>= 1.12.0)
+    method_source (1.0.0)
     mini_portile2 (2.8.1)
     minitest (5.24.0)
     mocha (2.1.0)
       ruby2_keywords (>= 0.0.5)
+    mutex_m (0.1.2)
     mutex_m (0.2.0)
     mysql2 (0.5.4)
+    mysql2 (0.5.6)
     nio4r (2.7.0)
     nokogiri (1.16.6-arm64-darwin)
       racc (~> 1.4)
@@ -94,6 +120,12 @@ GEM
       ast (~> 2.4.1)
       racc
     pg (1.5.4)
+    pry (0.14.2)
+      coderay (~> 1.1)
+      method_source (~> 1.0)
+    pry-byebug (3.10.1)
+      byebug (~> 11.0)
+      pry (>= 0.13, < 0.15)
     puma (6.4.2)
       nio4r (~> 2.0)
     raabro (1.4.0)
@@ -123,6 +155,8 @@ GEM
       zeitwerk (~> 2.6)
     rainbow (3.1.1)
     rake (13.0.6)
+    redis-client (0.20.0)
+      connection_pool
     regexp_parser (2.9.0)
     reline (0.3.2)
       io-console (~> 0.5)
@@ -159,6 +193,11 @@ GEM
       rubocop-rails
     ruby-progressbar (1.13.0)
     ruby2_keywords (0.0.5)
+    sidekiq (7.2.1)
+      concurrent-ruby (< 2)
+      connection_pool (>= 2.3.0)
+      rack (>= 2.2.4)
+      redis-client (>= 0.19.0)
     sqlite3 (1.5.4)
       mini_portile2 (~> 2.8.0)
     strscan (3.1.0)
@@ -178,12 +217,18 @@ PLATFORMS
   x86_64-linux
 
 DEPENDENCIES
+  after_commit_everywhere
+  awesome_print
   debug
+  good_job
   mocha
   mysql2
   pg
+  pry-byebug
   puma
   rubocop-rails-omakase
+  sidekiq (~> 7.2)
+  sidekiq-pro (~> 7.2)!
   solid_queue!
   sqlite3
 

README.md

@@ -1,3 +1,6 @@
+SolidQueue::Supervisor.start(mode: :work)
+SolidQueue::Supervisor.start(mode: :dispatch)
+
 # Solid Queue
 
 Solid Queue is a DB-based queuing backend for [Active Job](https://edgeguides.rubyonrails.org/active_job_basics.html), designed with simplicity and performance in mind.
@@ -335,6 +338,17 @@ class ApplicationMailer < ActionMailer::Base
     Rails.error.report(exception)
     raise exception
   end
+```
+
+## Batch jobs
+
+```rb
+SolidQueue::JobBatch.enqueue(on_finish: BatchCompletionJob) do
+  5.times.map { |i| SleepyJob.perform_later(i) }
+end
+
+SolidQueue::JobBatch.enqueue(on_success: BatchCompletionJob) do
+  5.times.map { |i| SleepyJob.perform_later(i) }
 end
 ```
 

app/models/solid_queue/claimed_execution.rb

@@ -65,6 +65,8 @@ def perform
     else
       failed_with(result.error)
     end
+
+    job.job_batch.touch(:changed_at, :last_changed_at) if job.batch_id.present?
   ensure
     job.unblock_next_blocked_job
   end

app/models/solid_queue/client_middleware.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class ClientMiddleware
+    def call(_worker_class, job, _queue, _redis_pool)
+      # {
+      #   "retry"=>true,
+      #   "queue"=>"default",
+      #   "class"=>"ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper",
+      #   "wrapped"=>SolidQueue::WorkflowTest::SidekiqJob,
+      #   "args"=>
+      #     [{"job_class"=>"SolidQueue::WorkflowTest::SidekiqJob",
+      #       "job_id"=>"e5db262e-2c49-4556-a2e4-78c009105e5d",
+      #       "provider_job_id"=>nil,
+      #       "queue_name"=>"default",
+      #       "priority"=>nil,
+      #       "arguments"=>["world"],
+      #       "executions"=>0,
+      #       "exception_executions"=>{},
+      #       "locale"=>"en",
+      #       "timezone"=>"UTC",
+      #       "enqueued_at"=>"2024-02-14T16:29:36.752639000Z",
+      #       "scheduled_at"=>nil,
+      #       "batch_id"=>nil}],
+      #   "jid"=>"185f9fb47e6610a31a576f72",
+      #   "created_at"=>1707928176.753821
+      # }
+      # job["args"].first["batch_id"] = JobBatch.current_batch_id
+      # process_queue_job = job["args"].first
+      # job["process_queue_id"] = process_queue_job.id
+      # job["args"] = process_queue_job.args
+      # update_jid(process_queue_job)
+      if SolidQueue::Workflow.current_workflow.present?
+        job["workflow_id"] = SolidQueue::Workflow.current_workflow.id
+      end
+
+      yield
+    end
+  end
+end

app/models/solid_queue/job.rb

@@ -8,6 +8,8 @@ class EnqueueError < StandardError; end
 
     serialize :arguments, coder: JSON
 
+    belongs_to :job_batch, foreign_key: :batch_id, optional: true
+
     class << self
       def enqueue_all(active_jobs)
         active_jobs_by_job_id = active_jobs.index_by(&:job_id)
@@ -53,14 +55,16 @@ def create_all_from_active_jobs(active_jobs)
         end
 
         def attributes_from_active_job(active_job)
+          active_job.batch_id = JobBatch.current_batch_id || active_job.batch_id
           {
             queue_name: active_job.queue_name || DEFAULT_QUEUE_NAME,
             active_job_id: active_job.job_id,
             priority: active_job.priority || DEFAULT_PRIORITY,
             scheduled_at: active_job.scheduled_at,
             class_name: active_job.class.name,
             arguments: active_job.serialize,
-            concurrency_key: active_job.concurrency_key
+            concurrency_key: active_job.concurrency_key,
+            batch_id: active_job.batch_id
           }
         end
     end

app/models/solid_queue/job/executable.rb

@@ -76,7 +76,7 @@ def dispatch_bypassing_concurrency_limits
       end
 
       def finished!
-        if SolidQueue.preserve_finished_jobs?
+        if SolidQueue.preserve_finished_jobs? || batch_id.present?
           touch(:finished_at)
         else
           destroy!

app/models/solid_queue/job/workflow_callback_job.rb

@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowCallbackJob < ActiveJob::Base
+      def perform(batch, workflow, workflow_level_id)
+        # puts "checking batch for workflow_level_id #{workflow_level_id}"
+        # apply job finished to nodes
+        jobs = batch.jobs
+        jobs_by_job_id = jobs.index_by(&:active_job_id)
+
+        ActiveRecord::Base.transaction do
+          workflow.workflow_nodes.where(active_job_id: jobs.pluck(:active_job_id)).each do |node|
+            # job = jobs.find { |j| j.active_job_id == node.active_job_id }
+            jobs = Array.wrap(jobs_by_job_id[node.active_job_id])
+            if jobs.any?(&:finished?)
+              # node.update!(finished_at: job.finished_at)
+              node.update!(finished_at: Time.current)
+            end
+          end
+        end
+
+        # finish workflow if all nodes are done
+        workflow.with_lock do
+          break if workflow.batch_id?
+          break if workflow.workflow_nodes.incomplete.any?
+          puts "and here we are..."
+
+          active_job = SolidQueue::Job::WorkflowMonitorJob.perform_later(workflow)
+          workflow.update!(batch_id: active_job.job_id)
+          puts "#{workflow.id}: and here we are... #{workflow.batch_id} #{workflow_level_id} #{workflow.workflow_nodes.incomplete.count}"
+        rescue => e
+          binding.pry
+          raise
+        end
+
+        return if workflow.batch_id?
+
+        workflow.workflow_nodes.incomplete.where(parent_workflow_level_id: workflow_level_id).each do |next_node|
+          jobs_to_insert = Workflow.prep_level(workflow, next_node.workflow_level_id)
+          Workflow.enqueue_solid_queue(workflow, next_node.workflow_level_id, jobs_to_insert)
+        end
+      rescue => e
+        puts "Error: #{e.message}"
+        puts e.backtrace
+        raise
+      end
+    end
+  end
+end

app/models/solid_queue/job/workflow_good_job_callback_job.rb

@@ -0,0 +1,86 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowGoodJobCallbackJob < ActiveJob::Base
+      def perform(batch, params)
+        workflow = batch.properties[:workflow]
+        workflow_level_id = batch.properties[:workflow_level_id]
+
+        ActiveRecord::Base.transaction do
+          # Mark nodes a finished or failed
+          any_discarded = mark_node_completion(workflow, batch)
+
+          # If any jobs in the level were discarded, discard the level and all its descendents
+          discard_descendents(workflow, workflow_level_id) if any_discarded
+        end
+
+        attempt_finish(workflow, workflow_level_id)
+
+        return if workflow.batch_id?
+
+        # enqueue next level of jobs
+        enqueue_next_levels(workflow, workflow_level_id)
+      rescue StandardError => e
+        puts "Error: #{e.message}"
+        puts e.backtrace
+        raise
+      end
+
+      def attempt_finish(workflow, workflow_level_id)
+        # Finish workflow if all nodes are done
+        workflow.with_lock do
+          return if workflow.batch_id?
+          return if workflow.workflow_nodes.incomplete.any?
+
+          active_job = SolidQueue::Job::WorkflowMonitorJob.perform_later(workflow)
+          workflow.update!(batch_id: active_job.job_id)
+          puts "#{workflow.id}: and here we are... #{workflow.batch_id} #{workflow_level_id} #{workflow.workflow_nodes.incomplete.count}"
+        rescue StandardError => e
+          puts "Error: #{e.message}"
+          puts e.backtrace
+          raise
+        end
+      end
+
+      def mark_node_completion(workflow, batch)
+        active_jobs = batch.active_jobs
+        good_jobs = GoodJob::Job.where(active_job_id: active_jobs.map(&:job_id)) # only get the last run
+        jobs_by_job_id = good_jobs.index_by(&:active_job_id)
+        any_discarded = false
+
+        workflow.workflow_nodes.where(active_job_id: jobs_by_job_id.keys).each do |node|
+          next unless node.incomplete?
+
+          jobs = Array.wrap(jobs_by_job_id[node.active_job_id])
+          if jobs.any?(&:discarded?)
+            any_discarded = true
+            # node.update!(discarded_at: job.discarded_at)
+            node.update!(discarded_at: Time.current)
+          elsif jobs.any?(&:finished?)
+            # node.update!(finished_at: job.finished_at)
+            node.update!(finished_at: Time.current)
+          end
+        end
+
+        any_discarded
+      end
+
+      def discard_descendents(workflow, workflow_level_id)
+        # get descendents and discard them
+        child_nodes = WorkflowNode.descendants(workflow, workflow_level_id)
+        SolidQueue::WorkflowNode
+          .where(workflow: workflow, workflow_level_id: child_nodes.pluck(:workflow_level_id))
+          .update_all(discarded_at: Time.current)
+      end
+
+      def enqueue_next_levels(workflow, workflow_level_id)
+        # ActiveRecord::Base.transaction - causes a deadlock in #prep_node
+        workflow.workflow_nodes.incomplete.where(parent_workflow_level_id: workflow_level_id).each do |next_node|
+          jobs_to_insert = Workflow.prep_level(workflow, next_node.workflow_level_id)
+          Workflow.enqueue_good_job(workflow, next_node.workflow_level_id, jobs_to_insert)
+        end
+      end
+    end
+  end
+end