diff --git a/Gemfile.lock b/Gemfile.lock index 17e5800..9b53a40 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - entitlements-app (1.2.4) + entitlements-app (1.2.5) concurrent-ruby (~> 1.3, >= 1.3.1) dogstatsd-ruby (~> 5.7) faraday (~> 2.0) diff --git a/lib/entitlements.rb b/lib/entitlements.rb index 21dc118..2be391f 100644 --- a/lib/entitlements.rb +++ b/lib/entitlements.rb @@ -22,6 +22,7 @@ require "ostruct" require "resolv" require "stringio" +require "thread" require "uri" require "yaml" @@ -457,19 +458,33 @@ def self.calculate_actions # Calculate old and new membership in each group. thread_pool = Concurrent::FixedThreadPool.new(max_parallelism) logger.debug("Begin prefetch and validate for all groups") + prep_start = Time.now - futures = Entitlements.child_classes.map do |group_name, obj| - Concurrent::Future.execute({ executor: thread_pool }) do - group_start = Time.now - logger.debug("Begin prefetch and validate for #{group_name}") - provider = Entitlements.config["groups"].fetch(group_name).fetch("type") - timed_operation(phase: "prefetch", provider: provider, target: group_name, concurrent: true) { obj.prefetch } - timed_operation(phase: "validate", provider: provider, target: group_name, concurrent: true) { obj.validate } - logger.debug("Finished prefetch and validate for #{group_name} in #{Time.now - group_start}") + jobs = Entitlements.child_classes + completions = Queue.new + + begin + jobs.each do |group_name, obj| + thread_pool.post do + group_start = Time.now + logger.debug("Begin prefetch and validate for #{group_name}") + provider = Entitlements.config["groups"].fetch(group_name).fetch("type") + timed_operation(phase: "prefetch", provider: provider, target: group_name, concurrent: true) { obj.prefetch } + timed_operation(phase: "validate", provider: provider, target: group_name, concurrent: true) { obj.validate } + logger.debug("Finished prefetch and validate for #{group_name} in #{Time.now - group_start}") + completions << nil + rescue => e + completions << e + end end - end - futures.each(&:value!) + jobs.size.times do + exception = completions.pop + raise exception if exception + end + ensure + thread_pool.kill + end logger.debug("Finished all prefetch and validate in #{Time.now - prep_start}") logger.debug("Begin all calculations") diff --git a/lib/version.rb b/lib/version.rb index 6908551..0dd6539 100644 --- a/lib/version.rb +++ b/lib/version.rb @@ -2,6 +2,6 @@ module Entitlements module Version - VERSION = "1.2.4" + VERSION = "1.2.5" end end diff --git a/spec/unit/entitlements_spec.rb b/spec/unit/entitlements_spec.rb index cad08fb..9d54a1b 100644 --- a/spec/unit/entitlements_spec.rb +++ b/spec/unit/entitlements_spec.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require_relative "spec_helper" +require "timeout" describe Entitlements do let(:subject) { Entitlements } @@ -346,6 +347,26 @@ expect(cache[:change_count]).to eq(3) end + + it "raises a worker failure without waiting for an earlier job" do + Entitlements.config["max_parallelism"] = 2 + started = Concurrent::Event.new + blocker = Concurrent::Event.new + allow(Entitlements).to receive(:child_classes) + .and_return("ldap-dir" => ldap_controller, "other-ldap-dir" => other_controller) + allow(ldap_controller).to receive(:prefetch) do + started.set + blocker.wait + end + allow(other_controller).to receive(:prefetch) do + started.wait + raise "Boom" + end + + expect do + Timeout.timeout(1) { described_class.calculate } + end.to raise_error(RuntimeError, "Boom") + end end describe "#execute" do