Make halting work for execution plans in almost any state

It is the caller's responsibility to not try to halt an already stopped
execution plan.

Author: adamruzicka

lib/dynflow/coordinator.rb

@@ -266,6 +266,19 @@ def self.valid_classes
       end
     end
 
+    class ExecutionInhibitionLock < Lock
+      def initialize(execution_plan_id)
+        super
+        @data[:owner_id] = "execution-plan:#{execution_plan_id}"
+        @data[:execution_plan_id] = execution_plan_id
+        @data[:id] = self.class.lock_id(execution_plan_id)
+      end
+
+      def self.lock_id(execution_plan_id)
+        "execution-plan:#{execution_plan_id}"
+      end
+    end
+
     class ExecutionLock < LockByWorld
       def initialize(world, execution_plan_id, client_world_id, request_id)
         super(world)

lib/dynflow/director.rb

@@ -244,15 +244,26 @@ def terminate
     end
 
     def halt(event)
-      manager = @execution_plan_managers[event.execution_plan_id]
-      return unless manager
+      halt_execution(event.execution_plan_id)
+    end
+
+    private
+
+    def halt_execution(execution_plan_id)
+      manager = @execution_plan_managers[execution_plan_id]
+      @logger.warn "Halting execution plan #{execution_plan_id}"
+      return halt_inactive(execution_plan_id) unless manager
 
-      @logger.warn "Halting execution plan #{event.execution_plan_id}"
       manager.halt
       finish_manager manager
     end
 
-    private
+    def halt_inactive(execution_plan_id)
+      plan = @world.persistence.load_execution_plan(execution_plan_id)
+      plan.update_state(:stopped)
+    rescue => e
+      @logger.error e
+    end
 
     def unless_done(manager, work_items)
       return [] unless manager
@@ -316,6 +327,14 @@ def track_execution_plan(execution_plan_id, finished)
               "cannot execute execution_plan_id:#{execution_plan_id} it's stopped"
       end
 
+      lock_class = Coordinator::ExecutionInhibitionLock
+      filters = { class: lock_class.to_s, owner_id: lock_class.lock_id(execution_plan_id) }
+      if @world.coordinator.find_records(filters).any?
+        halt_execution(execution_plan_id)
+        raise Dynflow::Error,
+              "cannot execute execution_plan_id:#{execution_plan_id} it's execution is inhibited"
+      end
+
       @execution_plan_managers[execution_plan_id] =
           ExecutionPlanManager.new(@world, execution_plan, finished)
     rescue Dynflow::Error => e

lib/dynflow/dispatcher/client_dispatcher.rb

@@ -137,10 +137,14 @@ def dispatch_request(request, client_world_id, request_id)
                             (on ~Execution | ~Planning do |execution|
                                AnyExecutor
                              end),
-                            (on ~Event | ~Halt do |event|
+                            (on ~Event do |event|
                                ignore_unknown = event.optional
                                find_executor(event.execution_plan_id)
                              end),
+                            (on ~Halt do |event|
+                               executor = find_executor(event.execution_plan_id)
+                               executor == Dispatcher::UnknownWorld ? AnyExecutor : executor
+                             end),
                             (on Ping.(~any, ~any) | Status.(~any, ~any) do |receiver_id, _|
                                receiver_id
                              end)

lib/dynflow/execution_plan.rb

@@ -133,7 +133,9 @@ def update_state(state, history_notice: :auto)
                               telemetry_common_options.merge(:result => key.to_s))
         end
         hooks_to_run << key
+        world.persistence.delete_delayed_plans(:execution_plan_uuid => id) if delay_record && original == :scheduled
         unlock_all_singleton_locks!
+        unlock_execution_inhibition_lock!
       when :paused
         unlock_all_singleton_locks!
       else
@@ -566,6 +568,14 @@ def unlock_all_singleton_locks!
       end
     end
 
+    def unlock_execution_inhibition_lock!
+      filter = { :owner_id => 'execution-plan:' + self.id,
+                 :class => Dynflow::Coordinator::ExecutionInhibitionLock.to_s }
+      world.coordinator.find_locks(filter).each do |lock|
+        world.coordinator.release(lock)
+      end
+    end
+
     def toggle_telemetry_state(original, new)
       return if original == new
       @label = root_plan_step.action_class if @label.nil?

lib/dynflow/world.rb

@@ -252,6 +252,7 @@ def get_execution_status(world_id, execution_plan_id, timeout, done = Concurrent
     end
 
     def halt(execution_plan_id, accepted = Concurrent::Promises.resolvable_future)
+      coordinator.acquire(Coordinator::ExecutionInhibitionLock.new(execution_plan_id))
       publish_request(Dispatcher::Halt[execution_plan_id], accepted, false)
     end
 

lib/dynflow/world/invalidation.rb

@@ -28,12 +28,27 @@ def invalidate(world)
             end
           end
 
+          prune_execution_inhibition_locks!
+
           pruned = persistence.prune_envelopes(world.id)
           logger.error("Pruned #{pruned} envelopes for invalidated world #{world.id}") unless pruned.zero?
           coordinator.delete_world(world)
         end
       end
 
+      # Prunes execution inhibition locks which got somehow left behind.
+      # Any execution inhibition locks, which have their corresponding execution
+      # plan in stopped state, will be removed.
+      def prune_execution_inhibition_locks!
+        locks = coordinator.find_locks(class: Coordinator::ExecutionInhibitionLock.name)
+        uuids = locks.map { |lock| lock.data[:execution_plan_id] }
+        plan_uuids = persistence.find_execution_plans(filters: { uuid: uuids, state: 'stopped' }).map(&:id)
+
+        locks.select { |lock| plan_uuids.include? lock.data[:execution_plan_id] }.each do |lock|
+          coordinator.release(lock)
+        end
+      end
+
       def invalidate_planning_lock(planning_lock)
         with_valid_execution_plan_for_lock(planning_lock) do |plan|
           plan.steps.values.each { |step| invalidate_step step }

test/executor_test.rb

@@ -735,6 +735,99 @@ def assert_next_steps(expected_next_step_ids, finished_step_id = nil, success =
           assert [world.terminate, world.terminate].map(&:value).all?
         end
       end
+
+      describe 'halting' do
+        include TestHelpers
+        let(:world) { WorldFactory.create_world }
+
+        it 'halts an execution plan with a suspended step' do
+          triggered = world.trigger(Support::DummyExample::PlanEventsAction, ping_time: 1)
+          plan = world.persistence.load_execution_plan(triggered.id)
+          wait_for do
+            plan = world.persistence.load_execution_plan(triggered.id)
+            plan.state == :running
+          end
+          world.halt(triggered.id)
+          wait_for('the execution plan to halt') do
+            plan = world.persistence.load_execution_plan(triggered.id)
+            plan.state == :stopped
+          end
+          _(plan.steps[2].state).must_equal :suspended
+        end
+
+        it 'halts a paused execution plan' do
+          triggered = world.trigger(Support::DummyExample::FailingDummy)
+          plan = world.persistence.load_execution_plan(triggered.id)
+          wait_for do
+            plan = world.persistence.load_execution_plan(triggered.id)
+            plan.state == :paused
+          end
+          world.halt(plan.id)
+          wait_for('the execution plan to halt') do
+            plan = world.persistence.load_execution_plan(triggered.id)
+            plan.state == :stopped
+          end
+          _(plan.steps[2].state).must_equal :error
+        end
+
+        it 'halts a planned execution plan' do
+          plan = world.plan(Support::DummyExample::Dummy)
+          wait_for do
+            plan = world.persistence.load_execution_plan(plan.id)
+            plan.state == :planned
+          end
+          world.halt(plan.id)
+          wait_for('the execution plan to halt') do
+            plan = world.persistence.load_execution_plan(plan.id)
+            plan.state == :stopped
+          end
+          _(plan.steps[2].state).must_equal :pending
+        end
+
+        it 'halts a scheduled execution plan' do
+          plan = world.delay(Support::DummyExample::Dummy, {start_at: Time.now + 120})
+          wait_for do
+            plan = world.persistence.load_execution_plan(plan.id)
+            plan.state == :scheduled
+          end
+          world.halt(plan.id)
+          wait_for('the execution plan to halt') do
+            plan = world.persistence.load_execution_plan(plan.id)
+            plan.state == :stopped
+          end
+          _(plan.delay_record).must_be :nil?
+          _(plan.steps[1].state).must_equal :pending
+        end
+
+        it 'halts a pending execution plan' do
+          plan = ExecutionPlan.new(world, nil)
+          plan.save
+          world.halt(plan.id)
+          wait_for('the execution plan to halt') do
+            plan = world.persistence.load_execution_plan(plan.id)
+            plan.state == :stopped
+          end
+        end
+      end
+
+      describe 'execution inhibition locks' do
+        include TestHelpers
+        let(:world) { WorldFactory.create_world }
+
+        it 'inhibits execution' do
+          plan = world.plan(Support::DummyExample::Dummy)
+          world.coordinator.acquire(Coordinator::ExecutionInhibitionLock.new(plan.id))
+          triggered = world.execute(plan.id)
+          triggered.wait
+          _(triggered).must_be :rejected?
+
+          plan = world.persistence.load_execution_plan(plan.id)
+          _(plan.state).must_equal :stopped
+
+          locks = world.coordinator.find_locks({ class: Coordinator::ExecutionInhibitionLock.to_s, owner_id: "execution-plan:#{plan.id}" })
+          _(locks).must_be :empty?
+        end
+      end
     end
   end
 end