Make SSHKit::Runner::Parallel fail slow

Using a different SSHKit runner doesn't work well, because the group
runner uses the Parallel runner internally. So instead we'll patch its
behaviour to fail slow.

We'll also get it to return all the errors so we can report on all the
hosts that failed.

Author: djmb

bin/kamal

@@ -7,6 +7,15 @@ require "kamal"
 
 begin
   Kamal::Cli::Main.start(ARGV)
+rescue SSHKit::Runner::MultipleExecuteError => e
+  e.execute_errors.each do |execute_error|
+    puts "  \e[31mERROR (#{execute_error.cause.class}): #{execute_error.message}\e[0m"
+  end
+  if ENV["VERBOSE"]
+    puts "Backtrace for the first error:"
+    puts e.execute_errors.first.cause.backtrace
+  end
+  exit 1
 rescue SSHKit::Runner::ExecuteError => e
   puts "  \e[31mERROR (#{e.cause.class}): #{e.message}\e[0m"
   puts e.cause.backtrace if ENV["VERBOSE"]

lib/kamal/commander.rb

@@ -150,7 +150,6 @@ def configure_sshkit_with(config)
         sshkit.max_concurrent_starts = config.sshkit.max_concurrent_starts
         sshkit.ssh_options = config.ssh.options
       end
-      SSHKit.config.default_runner = SSHKit::Runner::ParallelCompleteAll
       SSHKit.config.command_map[:docker] = "docker" # No need to use /usr/bin/env, just clogs up the logs
       SSHKit.config.output_verbosity = verbosity
     end

lib/kamal/sshkit_with_ext.rb

@@ -104,33 +104,46 @@ def start_with_concurrency_limit(*args)
   prepend LimitConcurrentStartsInstance
 end
 
-require "thread"
-
-module SSHKit
-  module Runner
-    class ParallelCompleteAll < Abstract
-      def execute
-        threads = hosts.map do |host|
-          Thread.new(host) do |h|
-            begin
-              backend(h, &block).run
-            rescue ::StandardError => e
-              e2 = SSHKit::Runner::ExecuteError.new e
-              raise e2, "Exception while executing #{host.user ? "as #{host.user}@" : "on host "}#{host}: #{e.message}"
-            end
-          end
+class SSHKit::Runner::MultipleExecuteError < SSHKit::StandardError
+  attr_reader :execute_errors
+
+  def initialize(execute_errors)
+    @execute_errors = execute_errors
+  end
+end
+
+class SSHKit::Runner::Parallel
+  # SSHKit joins the threads in sequence and fails on the first error it encounters, which means that we wait threads
+  # before the first failure to complete but not for ones after.
+  #
+  # We'll patch it to wait for them all to complete, and to record all the threads that errored so we can see when a
+  # problem occurs on multiple hosts.
+  module CompleteAll
+    def execute
+      threads = hosts.map do |host|
+        Thread.new(host) do |h|
+          backend(h, &block).run
+        rescue ::StandardError => e
+          e2 = SSHKit::Runner::ExecuteError.new e
+          raise e2, "Exception while executing #{host.user ? "as #{host.user}@" : "on host "}#{host}: #{e.message}"
         end
+      end
 
-        exception = nil
-        threads.each do |t|
-          begin
-            t.join
-          rescue SSHKit::Runner::ExecuteError => e
-            exception ||= e
-          end
+      exceptions = []
+      threads.each do |t|
+        begin
+          t.join
+        rescue SSHKit::Runner::ExecuteError => e
+          exceptions << e
         end
-        raise exception if exception
+      end
+      if exceptions.one?
+        raise exceptions.first
+      elsif exceptions.many?
+        raise SSHKit::Runner::MultipleExecuteError.new(exceptions)
       end
     end
   end
+
+  prepend CompleteAll
 end