module Dynflow
  module Executors
    class Parallel < Abstract
      class Worker < Actor
        def initialize(pool, transaction_adapter, telemetry_options = {})
          @pool                = Type! pool, Concurrent::Actor::Reference
          @transaction_adapter = Type! transaction_adapter, TransactionAdapters::Abstract
          @telemetry_options   = telemetry_options
        end

        def on_message(work_item)
          ok = false
          Executors.run_user_code do
            work_item.execute
            ok = true
          end
        rescue Errors::PersistenceError => e
          @pool.tell([:handle_persistence_error, reference, e, work_item])
          ok = false
        ensure
          Dynflow::Telemetry.with_instance { |t| t.increment_counter(:dynflow_worker_events, 1, @telemetry_options) }
          @pool.tell([:worker_done, reference, work_item]) if ok
        end
      end
    end
  end
end

module Dynflow
  module Executors
    class Parallel < Abstract
      class Worker < Actor
        def initialize(pool, transaction_adapter, telemetry_options = {})
          @pool                = Type! pool, Concurrent::Actor::Reference
          @transaction_adapter = Type! transaction_adapter, TransactionAdapters::Abstract
          @telemetry_options   = telemetry_options
        end

        def on_message(work_item)
          already_responded = false
          Executors.run_user_code do
            work_item.execute
          end
        rescue Errors::PersistenceError => e
          @pool.tell([:handle_persistence_error, reference, e, work_item])
          already_responded = true
        ensure
          Dynflow::Telemetry.with_instance { |t| t.increment_counter(:dynflow_worker_events, 1, @telemetry_options) }
          if !already_responded && Concurrent.global_io_executor.running?
            @pool.tell([:worker_done, reference, work_item])
          end
        end
      end
    end
  end
end

Enhance fix for termination with running action
Enhance fix for termination with running action

Previously, we would not respond to the pool in case of any exception
coming from the `work_item.execute`. Even though I don't think we expect
any other exceptions other than PeristenceError, it didn't feel right not to
respond on any exception that might occur.

When I looked at API of concurrent ruby, it seems like we can use check on
`global_io_executor` to check it's still operational.

See https://projects.theforeman.org/issues/25593 for reasons on why we
need this check
