Correct outbound validation.

.docker/production/Dockerfile.gha

@@ -142,6 +142,7 @@ COPY --chown=$USERNAME:$USERNAME ./components/mitc_service/lib/mitc_service/vers
 
 COPY --chown=$USERNAME:$USERNAME ./Gemfile $HOME/Gemfile
 COPY --chown=$USERNAME:$USERNAME ./Gemfile.lock $HOME/Gemfile.lock
+COPY --chown=$USERNAME:$USERNAME ./project_gems $HOME/project_gems
 
 RUN apt-get update -qq \
   && apt-get install -yq --no-install-recommends \

.rubocop.yml

@@ -10,6 +10,9 @@ AllCops:
     - "config/deploy.rb"
     - "config/unicorn.rb"
     - "config/deploy/**/*"
+    # This gem is included from an external repository.  Style failures should
+    # be fixed upstream.
+    - "project_gems/rex_port/**/*"
 
 Layout/EmptyLineAfterGuardClause:
   Enabled: false

Gemfile

@@ -55,6 +55,8 @@ gem 'mime-types'
 gem 'httparty',            '~> 0.16'
 gem 'nokogiri', '>= 1.10.8', platforms: [:ruby, :mri]
 gem 'pundit', '~> 2.2.0'
+gem 'pond'
+gem 'rex_port', path: './project_gems/rex_port'
 
 group :development, :test do
   # Call 'byebug' anywhere in the code to stop execution and get a debugger console

Gemfile.lock

@@ -81,6 +81,11 @@ PATH
       mongoid (~> 7.2.1)
       rails (~> 6.1.3)
 
+PATH
+  remote: project_gems/rex_port
+  specs:
+    rex_port (0.1.0)
+
 GEM
   remote: https://rubygems.org/
   specs:
@@ -341,6 +346,7 @@ GEM
     parallel (1.21.0)
     parser (3.0.2.0)
       ast (~> 2.4.1)
+    pond (0.5.0)
     popper_js (2.9.3)
     pry (0.13.1)
       coderay (~> 1.1)
@@ -564,6 +570,7 @@ DEPENDENCIES
   mitc_service!
   mongoid (~> 7.2.1)
   nokogiri (>= 1.10.8)
+  pond
   pry-byebug
   puma (~> 5.0)
   pundit (~> 2.2.0)
@@ -572,6 +579,7 @@ DEPENDENCIES
   redis (~> 4.0)
   redis-session-store
   resource_registry!
+  rex_port!
   rspec-rails (~> 5.0)
   rubocop (~> 1.13.0)
   sass-rails (>= 6)

app/models/atp_business_rules_validation_proxy.rb

@@ -7,99 +7,21 @@
 # The jar itself lives in lib/atp_validator-0.1.0-jar-with-dependencies.jar.
 # The repository for the code is: https://github.com/ideacrew/atp_validator
 class AtpBusinessRulesValidationProxy
-  include Singleton
-
   WORKING_DIRECTORY = File.expand_path(Rails.root).freeze
 
-  def self.boot!
-    instance.boot_port_process
-  end
-
-  def self.reconnect!
-    instance.reconnect!
-  end
-
-  def self.run_validation(data)
-    instance.run_validation(data)
-  end
-
-  def boot_port_process
-    @error_reader, @errors_from_validator = IO.pipe(binmode: true)
-    @reader, @write_from_validator = IO.pipe(binmode: true)
-    @read_into_validator, @writer = IO.pipe(binmode: true)
-    @pid = Process.spawn(
-      "java -jar lib/atp_validator-0.1.0-jar-with-dependencies.jar",
-      chdir: WORKING_DIRECTORY,
-      :in => @read_into_validator,
-      :out => @write_from_validator,
-      :err => @errors_from_validator,
-      @error_reader.fileno => :close,
-      @reader.fileno => :close,
-      @writer.fileno => :close
+  def self.boot!(count = 5)
+    config = RexPort::ChildConfig.new(
+      "RUBYOPT=\"-W0\" java -jar lib/atp_validator-0.2.0-jar-with-dependencies.jar",
+      WORKING_DIRECTORY
     )
-    Process.detach(@pid)
-    @write_from_validator.close
-    @read_into_validator.close
-    @errors_from_validator.close
-    check_process_death
-    @pid
-  end
-
-  def check_process_death(stage = "boot")
-    pid_status = nil
-    begin
-      pid_status = Process.waitpid(@pid, Process::WNOHANG)
-      return unless pid_status
-    rescue Errno::ECHILD
-      pid_status = -1
-    end
-    read_death_data = @error_reader.read_nonblock(2**16)
-    Rails.logger.error do
-      "Process died during #{stage}: #{@pid}\n#{read_death_data}"
-    end
-    pid_status
-  end
-
-  def run_validation(data)
-    packet_size = [data.bytesize].pack("l>*")
-    reconnect! unless @writer
-    @writer.write(packet_size)
-    @writer.write(data)
-    @writer.flush
-    readable = IO.select([@reader, @error_reader], [], [@error_reader], 10)
-    reconnect! unless readable
-    first_readable_array = readable.detect { |item| !item.empty? }
-    if first_readable_array.first.fileno == @error_reader.fileno
-      read = first_readable_array.first.read_nonblock(2**16)
-      Rails.logger.error { "Validator Crashed:\n#{read}" }
-      reconnect!
-      raise StandardError, read
-    end
-
-    packet_response_size = @reader.read(4)
-    read_size = packet_response_size.unpack1("L>*")
-    read_buff = @reader.read(read_size)
-    pid_status = check_process_death("run")
-    unless pid_status.nil?
-      reconnect!
-      raise StandardError, "process crashed!"
+    @pool = Pond.new(:maximum_size => count, :timeout => 3, :eager => true) do
+      RexPort::Child.new(config, true)
     end
-    read_buff
   end
 
-  # rubocop:disable Lint/SuppressedException
-  def reconnect!
-    unless @pid.blank?
-      @reader.close
-      @writer.close
-      @error_reader.close
-      begin
-        Process.kill(9, @pid)
-        Process.waitpid(@pid)
-      rescue Errno::ECHILD, Errno::ESRCH
-      end
+  def self.run_validation(data)
+    @pool.checkout do |port|
+      port.request(data)
     end
-    boot_port_process
   end
-  # rubocop:enable Lint/SuppressedException
 end

app/operations/transfers/execute_business_xml_validations.rb

@@ -28,7 +28,6 @@ def execute_validator(payload)
       Success(result)
     rescue StandardError => _e
       # try to reconnect the proxy and run validation again if initial attempt crashed
-      AtpBusinessRulesValidationProxy.reconnect!
       attempt = Try do
         AtpBusinessRulesValidationProxy.run_validation(payload)
       end

app/operations/transfers/to_service.rb

@@ -18,8 +18,8 @@ def call(params, transfer_id = "")
       flagged_params = yield add_param_flags(params, transfer_id)
       xml = yield generate_xml(flagged_params, transfer_id)
       validated = yield schema_validation(xml, transfer_id)
-      # validated  = yield business_validation(validated)
-      transfer_response = yield initiate_transfer(validated, transfer_id)
+      business_validated = yield business_validation(validated)
+      transfer_response = yield initiate_transfer(business_validated, transfer_id)
       update_transfer(transfer_response)
     end
 

config/puma.rb

@@ -43,8 +43,3 @@
 
 # Allow puma to be restarted by `rails restart` command.
 plugin :tmp_restart
-
-on_worker_boot do
-  Rails.logger.info { "Rebooting Validation Proxy for worker" }
-  AtpBusinessRulesValidationProxy.reconnect!
-end

project_gems/rex_port/lib/rex_port.rb

@@ -0,0 +1,6 @@
+require_relative "rex_port/errors"
+require_relative "rex_port/child_config"
+require_relative "rex_port/child"
+
+module RexPort
+end

project_gems/rex_port/lib/rex_port/child.rb

@@ -0,0 +1,107 @@
+module RexPort
+  class Child
+    attr_reader :pid, :reader, :writer, :error_reader, :child_config
+
+    def initialize(child_config, boot = false)
+      @child_config = child_config
+      boot! if boot
+    end
+
+    def request(message)
+      tell(message)
+      listen
+    end
+
+    def reboot!
+      kill!
+      boot!
+    end
+
+    def boot!
+      @pid, @reader, @writer, @error_reader = @child_config.boot!
+      check_status = check_process_death
+      unless check_status.empty?
+        raise Errors::StartupError.new(check_status.last)
+      end
+    end
+
+    def kill!
+      @writer.close
+      @reader.close
+      @error_reader.close
+      begin
+        Process.kill(9, @pid)
+        Process.waitpid(@pid)
+      rescue Errno::ECHILD, Errno::ESRCH
+      end
+    end
+
+    protected
+
+    def tell(message)
+      check_status = check_process_death
+      unless check_status.empty?
+        raise Errors::DiedBeforeListeningError.new(check_status.last)
+      end
+      packet_size = [message.bytesize].pack("l>*")
+      @writer.write(packet_size)
+      @writer.write(message)
+      @writer.flush
+    end
+
+    def listen
+      readable = nil
+      if @child_config.timeout > 0
+        readable = IO.select([@reader, @error_reader], [], [@error_reader], @child_config.timeout)
+      else
+        readable = IO.select([@reader, @error_reader], [], [@error_reader])
+      end
+      unless readable
+        reboot!
+        raise Errors::ResponseTimeoutError, "process timeout!"
+      end
+
+      first_readable_array = readable.detect { |item| !item.empty? }
+      if first_readable_array.first.fileno == @error_reader.fileno
+        read = try_read_nonblock(first_readable_array.first)
+        reboot!
+        raise Errors::ResponseReadError, read
+      end
+
+      packet_response_size = @reader.read(4)
+      if packet_response_size.nil? || packet_response_size.bytesize < 4
+        check_result = check_process_death
+        reboot!
+        raise Errors::ResponseReadError, "process crashed:\n#{check_result.last}"
+      end
+      read_size = packet_response_size.unpack("L>*")
+      read_buff = @reader.read(read_size.first)
+      check_result = check_process_death
+      unless check_result.empty?
+        reboot!
+        raise Errors::ResponseReadError, "process crashed:\n#{check_result.last}"
+      end
+      read_buff
+    end
+
+    def check_process_death
+      pid_status = nil
+      begin
+        pid_status = Process.waitpid(@pid, Process::WNOHANG)
+        return [] unless pid_status
+      rescue Errno::ECHILD
+        pid_status = -1
+      end
+      read_death_data = try_read_nonblock(@error_reader)
+      [pid_status, read_death_data]
+    end
+
+    def try_read_nonblock(io)
+      begin
+        io.read_nonblock(2**16)
+      rescue EOFError
+        ""
+      end
+    end
+  end
+end

project_gems/rex_port/lib/rex_port/child_config.rb

@@ -0,0 +1,37 @@
+module RexPort
+  class ChildConfig
+    attr_reader :chdir, :timeout, :command
+
+    def initialize(command, working_dir, timeout = -1)
+      @chdir = working_dir
+      @command = command
+      @timeout = timeout
+    end
+
+    def boot!
+      error_reader, errors_from_validator = IO.pipe(binmode: true)
+      reader, write_from_validator = IO.pipe(binmode: true)
+      read_into_validator, writer = IO.pipe(binmode: true)
+      pid = nil
+      begin
+        pid = Process.spawn(
+          @command,
+          chdir: @chdir,
+          :in => read_into_validator,
+          :out => write_from_validator,
+          :err => errors_from_validator,
+          error_reader.fileno => :close,
+          reader.fileno => :close,
+          writer.fileno => :close
+        )
+      rescue StandardError => e
+        raise Errors::StartupError.new(e)
+      end
+      Process.detach(pid)
+      write_from_validator.close
+      read_into_validator.close
+      errors_from_validator.close
+      [pid, reader, writer, error_reader]
+    end
+  end
+end

project_gems/rex_port/lib/rex_port/errors.rb

@@ -0,0 +1,11 @@
+module RexPort
+  module Errors
+    class StartupError < RuntimeError; end
+
+    class DiedBeforeListeningError < RuntimeError; end
+
+    class ResponseTimeoutError < RuntimeError; end
+
+    class ResponseReadError < RuntimeError; end
+  end
+end

project_gems/rex_port/rex_port.gemspec

@@ -0,0 +1,10 @@
+Gem::Specification.new do |s|
+  s.name        = 'rex_port'
+  s.version     = '0.1.0'
+  s.summary     = "Communicate with external programs via ports."
+  s.description = "Use the port protocol to communicate with external programs via child processes."
+  s.authors     = ["Trey Evans"]
+  s.require_paths = ["lib"]
+  s.email       = 'lewis.r.evans@gmail.com'
+  s.license       = 'MIT'
+end

project_gems/rex_port/spec/example_programs/instant_death.rb

@@ -0,0 +1 @@
+exit(0)

project_gems/rex_port/spec/example_programs/pingpong.rb

@@ -0,0 +1,12 @@
+while (true) do
+  msg_length_bytes = STDIN.read(4)
+  msg_length = msg_length_bytes.unpack("L>*")
+  msg = STDIN.read(msg_length.first)
+
+  response = msg + " PONG!"
+  response_size_bytes = [response.bytesize].pack("L>*")
+  STDOUT.write(response_size_bytes)
+  STDOUT.flush
+  STDOUT.write(response)
+  STDOUT.flush
+end