Add execution plan chaining

examples/execution_plan_chaining.rb

@@ -0,0 +1,56 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require_relative 'example_helper'
+
+class DelayedAction < Dynflow::Action
+  def plan(should_fail = false)
+    plan_self :should_fail => should_fail
+  end
+
+  def run
+    sleep 5
+    raise "Controlled failure" if input[:should_fail]
+  end
+
+  def rescue_strategy
+    Dynflow::Action::Rescue::Fail
+  end
+end
+
+if $PROGRAM_NAME == __FILE__
+  world = ExampleHelper.create_world do |config|
+    config.auto_rescue = true
+  end
+  world.action_logger.level = 1
+  world.logger.level = 0
+
+  plan1 = world.trigger(DelayedAction)
+  plan2 = world.chain(plan1.execution_plan_id, DelayedAction)
+  plan3 = world.chain(plan2.execution_plan_id, DelayedAction)
+  plan4 = world.chain(plan2.execution_plan_id, DelayedAction)
+
+  plan5 = world.trigger(DelayedAction, true)
+  plan6 = world.chain(plan5.execution_plan_id, DelayedAction)
+
+  puts <<-MSG.gsub(/^.*\|/, '')
+    |
+    |  Execution Plan Chaining example
+    |  ========================
+    |
+    |  This example shows the execution plan chaining functionality of Dynflow, which allows execution plans to wait until another execution plan finishes.
+    |
+    |  Execution plans:
+    |    #{plan1.id} runs immediately and should run successfully.
+    |    #{plan2.id} is delayed and should run once #{plan1.id} finishes.
+    |    #{plan3.id} and #{plan4.id} are delayed and should run once #{plan2.id} finishes.
+    |
+    |    #{plan5.id} runs immediately and is expected to fail.
+    |    #{plan6.id} should not run at all as its prerequisite failed.
+    |
+    |  Visit #{ExampleHelper::DYNFLOW_URL} to see their status.
+    |
+  MSG
+
+  ExampleHelper.run_web_console(world)
+end

lib/dynflow/debug/telemetry/persistence.rb

@@ -19,7 +19,7 @@ module Persistence
           :load_execution_plan,
           :save_execution_plan,
           :find_old_execution_plans,
-          :find_past_delayed_plans,
+          :find_ready_delayed_plans,
           :delete_delayed_plans,
           :save_delayed_plan,
           :set_delayed_plan_frozen,

lib/dynflow/delayed_executors/abstract_core.rb

@@ -32,7 +32,7 @@ def time
 
       def delayed_execution_plans(time)
         with_error_handling([]) do
-          world.persistence.find_past_delayed_plans(time)
+          world.persistence.find_ready_delayed_plans(time)
         end
       end
 

lib/dynflow/delayed_plan.rb

@@ -31,6 +31,12 @@ def timeout
       error("Execution plan could not be started before set time (#{@start_before})", 'timeout')
     end
 
+    def failed_dependencies(uuids)
+      bullets = uuids.map { |u| "- #{u}" }.join("\n")
+      msg = "Execution plan could not be started because some of its prerequisite execution plans failed:\n#{bullets}"
+      error(msg, 'failed-dependency')
+    end
+
     def error(message, history_entry = nil)
       execution_plan.root_plan_step.state = :error
       execution_plan.root_plan_step.error = ::Dynflow::ExecutionPlan::Steps::Error.new(message)

lib/dynflow/director.rb

@@ -114,7 +114,15 @@ def execute
         plan = world.persistence.load_delayed_plan(execution_plan_id)
         return if plan.nil? || plan.execution_plan.state != :scheduled
 
-        if !plan.start_before.nil? && plan.start_before < Time.now.utc()
+        if plan.start_before.nil?
+          blocker_ids = world.persistence.find_execution_plan_dependencies(execution_plan_id)
+          statuses = world.persistence.find_execution_plan_statuses({ filters: { uuid: blocker_ids } })
+          failed = statuses.select { |_uuid, status| status[:state] == 'stopped' && status[:result] == 'error' }
+          if failed.any?
+            plan.failed_dependencies(failed.keys)
+            return
+          end
+        elsif plan.start_before < Time.now.utc()
           plan.timeout
           return
         end

lib/dynflow/persistence.rb

@@ -101,8 +101,16 @@ def find_old_execution_plans(age)
       end
     end
 
-    def find_past_delayed_plans(time)
-      adapter.find_past_delayed_plans(time).map do |plan|
+    def find_execution_plan_dependencies(execution_plan_id)
+      adapter.find_execution_plan_dependencies(execution_plan_id)
+    end
+
+    def find_blocked_execution_plans(execution_plan_id)
+      adapter.find_blocked_execution_plans(execution_plan_id)
+    end
+
+    def find_ready_delayed_plans(time)
+      adapter.find_ready_delayed_plans(time).map do |plan|
         DelayedPlan.new_from_hash(@world, plan)
       end
     end
@@ -163,5 +171,9 @@ def prune_envelopes(receiver_ids)
     def prune_undeliverable_envelopes
       adapter.prune_undeliverable_envelopes
     end
+
+    def chain_execution_plan(first, second)
+      adapter.chain_execution_plan(first, second)
+    end
   end
 end

lib/dynflow/persistence_adapters/abstract.rb

@@ -72,7 +72,15 @@ def save_execution_plan(execution_plan_id, value)
         raise NotImplementedError
       end
 
-      def find_past_delayed_plans(options = {})
+      def find_execution_plan_dependencies(execution_plan_id)
+        raise NotImplementedError
+      end
+
+      def find_blocked_execution_plans(execution_plan_id)
+        raise NotImplementedError
+      end
+
+      def find_ready_delayed_plans(options = {})
         raise NotImplementedError
       end
 

lib/dynflow/persistence_adapters/sequel.rb

@@ -39,7 +39,8 @@ class action_class execution_plan_uuid queue),
                     envelope:            %w(receiver_id),
                     coordinator_record:  %w(id owner_id class),
                     delayed:             %w(execution_plan_uuid start_at start_before args_serializer frozen),
-                    output_chunk:        %w(execution_plan_uuid action_id kind timestamp) }
+                    output_chunk:        %w(execution_plan_uuid action_id kind timestamp),
+                    execution_plan_dependency: %w(execution_plan_uuid blocked_by_uuid) }
 
       SERIALIZABLE_COLUMNS = { action:  %w(input output),
                                delayed: %w(serialized_args),
@@ -153,12 +154,31 @@ def find_old_execution_plans(age)
         records.map { |plan| execution_plan_column_map(load_data plan, table_name) }
       end
 
-      def find_past_delayed_plans(time)
+      def find_execution_plan_dependencies(execution_plan_id)
+        table(:execution_plan_dependency)
+          .where(execution_plan_uuid: execution_plan_id)
+          .select_map(:blocked_by_uuid)
+      end
+
+      def find_blocked_execution_plans(execution_plan_id)
+        table(:execution_plan_dependency)
+          .where(blocked_by_uuid: execution_plan_id)
+          .select_map(:execution_plan_uuid)
+      end
+
+      def find_ready_delayed_plans(time)
         table_name = :delayed
+        # Subquery to find delayed plans that have at least one non-stopped dependency
+        plans_with_unfinished_deps = table(:execution_plan_dependency)
+                                     .join(TABLES[:execution_plan], uuid: :blocked_by_uuid)
+                                     .where(::Sequel.~(state: 'stopped'))
+                                     .select(:execution_plan_uuid)
+
         records = with_retry do
           table(table_name)
-            .where(::Sequel.lit('start_at <= ? OR (start_before IS NOT NULL AND start_before <= ?)', time, time))
+            .where(::Sequel.lit('start_at IS NULL OR (start_at <= ? OR (start_before IS NOT NULL AND start_before <= ?))', time, time))
             .where(:frozen => false)
+            .exclude(execution_plan_uuid: plans_with_unfinished_deps)
             .order_by(:start_at)
             .all
         end
@@ -175,6 +195,10 @@ def save_delayed_plan(execution_plan_id, value)
         save :delayed, { execution_plan_uuid: execution_plan_id }, value, with_data: false
       end
 
+      def chain_execution_plan(first, second)
+        save :execution_plan_dependency, {}, { execution_plan_uuid: second, blocked_by_uuid: first }, with_data: false
+      end
+
       def load_step(execution_plan_id, step_id)
         load :step, execution_plan_uuid: execution_plan_id, id: step_id
       end
@@ -319,7 +343,8 @@ def abort_if_pending_migrations!
                  envelope:            :dynflow_envelopes,
                  coordinator_record:  :dynflow_coordinator_records,
                  delayed:             :dynflow_delayed_plans,
-                 output_chunk:        :dynflow_output_chunks }
+                 output_chunk:        :dynflow_output_chunks,
+                 execution_plan_dependency: :dynflow_execution_plan_dependencies }
 
       def table(which)
         db[TABLES.fetch(which)]

lib/dynflow/persistence_adapters/sequel_migrations/025_create_execution_plan_dependencies.rb

@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+Sequel.migration do
+  up do
+    type = database_type
+    create_table(:dynflow_execution_plan_dependencies) do
+      column_properties = if type.to_s.include?('postgres')
+                            { type: :uuid }
+                          else
+                            { type: String, size: 36, fixed: true, null: false }
+                          end
+      foreign_key :execution_plan_uuid, :dynflow_execution_plans, on_delete: :cascade, **column_properties
+      foreign_key :blocked_by_uuid, :dynflow_execution_plans, on_delete: :cascade, **column_properties
+      index :blocked_by_uuid
+      index :execution_plan_uuid
+    end
+  end
+
+  down do
+    drop_table(:dynflow_execution_plan_dependencies)
+  end
+end

lib/dynflow/world.rb

@@ -202,6 +202,16 @@ def delay_with_options(action_class:, args:, delay_options:, id: nil, caller_act
       Scheduled[execution_plan.id]
     end
 
+    def chain(plan_uuids, action_class, *args)
+      plan_uuids = [plan_uuids] unless plan_uuids.is_a? Array
+      result = delay_with_options(action_class: action_class, args: args, delay_options: { frozen: true })
+      plan_uuids.each do |plan_uuid|
+        persistence.chain_execution_plan(plan_uuid, result.execution_plan_id)
+      end
+      persistence.set_delayed_plan_frozen(result.execution_plan_id, false)
+      result
+    end
+
     def plan_elsewhere(action_class, *args)
       execution_plan = ExecutionPlan.new(self, nil)
       execution_plan.delay(nil, action_class, {}, *args)