Choria transport fixes and cleanup

developer-docs/choria/choria-transport-dev.md

@@ -276,8 +276,9 @@ The shell polling loop is the heart of asynchronous command execution:
 **`shell_list`** does a single round:
 1. Sends one batched `shell.list` RPC call to all remaining targets
 2. For each target, matches its handle against the returned job list
-3. Returns `[done_hash, rpc_failed_boolean]` where done contains completed
-   targets (status in `SHELL_DONE_STATUSES`: `stopped` or `failed`)
+3. Returns `[finished, failed, rpc_failed]`, where finished contains the
+   targets whose process ended (status in `SHELL_DONE_STATUSES`: `stopped` or
+   `failed`) and failed contains an error hash for each target whose poll failed
 
 **`shell_statuses`** retrieves results for completed targets:
 1. Sends one batched `shell.statuses` RPC call with all completed handles
@@ -295,11 +296,15 @@ behaviors:
 Model for why this is necessary.
 
 **Error classification**: Splits results into three categories:
-- **Success** (statuscode 0 or 1): The agent responded. Statuscode 1 means
-  "application error but action completed" (the task ran but had issues).
-- **Agent error** (statuscode >= 2): The agent itself had a problem. Returned
-  as failures with `bolt/choria-agent-error`.
-- **No response**: Target didn't reply. Returned as failures with
+- **Success** (statuscode 0, or 1 when the caller passes `accept_aborted:
+  true`): The agent responded. The `bolt_tasks` calls accept statuscode 1,
+  which that agent answers for a task that ran and failed, and so does the
+  ping behind `batch_connected?`. The status message of such a reply is
+  returned under `:aborted`.
+- **Agent error** (statuscode >= 2, or 1 for every other call): The action
+  did not run or failed, as when an action policy denies the caller. Returned
+  as failures with `bolt/choria-rpc-error`.
+- **No response**: Target did not reply. Returned as failures with
   `bolt/choria-no-response`.
 
 **Absorption of StandardError**: If the RPC call raises a `StandardError`
@@ -426,14 +431,18 @@ handles its own concurrency: `batches()` groups targets by collective
 - **Agent discovery for offline nodes:** Non-responding targets are not cached
   (to allow retry), so each operation re-queries offline nodes. At 10,000
   targets with many offline, this adds discovery overhead.
-- **Per-target options in batches:** Batch methods use `targets.first.options`
-  for shared settings (timeout, tmpdir, cleanup). Per-target option differences
-  within a batch are silently ignored. This is inherent to the batch execution
-  model.
-- **Base64 upload size:** The `upload_file_content` method sends files
-  in a single NATS message. After base64 encoding (~33% expansion) and RPC
-  overhead, the max raw file size is roughly 700-750KB with the default 1MB
-  message limit. This will be addressed in Phase 4 with chunked file transfer.
+- **Per-target options in batches:** Targets in one collective share the
+  request-level options: the timeouts, the task agent, the interpreters, the
+  Puppet environment, the shell tmpdir base and `cleanup`. A batch reads them
+  from its first target, so targets whose values differ get the first target's
+  values. Batching by option values as well would serialize one batch per
+  distinct set of values, which is far slower when the values vary per target.
+- **Base64 upload size:** The `upload_file_content` method sends the whole
+  file base64-encoded inside one `shell.run` command, and the shell agent's
+  DDL caps a command at 10,240 characters, so a file can be about 7.5 KB on
+  POSIX and under 3 KB on Windows, where the PowerShell wrapper is encoded as
+  UTF-16LE base64 on top of that. This will be addressed in Phase 4 with
+  chunked file transfer.
 
 ## Testing
 

lib/bolt/transport/choria.rb

@@ -45,10 +45,9 @@ def initialize
         @default_collective = nil
       end
 
-      # Advertise both shell and powershell so tasks with either requirement
-      # can be selected. The per-target selection happens in
-      # select_implementation below, which picks the right feature set based
-      # on the target's detected OS.
+      # Task selection does not use this list, since select_implementation
+      # below overrides the Base method that reads it. The list is kept for
+      # apply_prep, which reads a transport's provided features.
       def provided_features
         %w[shell powershell]
       end
@@ -107,7 +106,6 @@ def batch_task(targets, task, arguments, _options = {}, position = [], &callback
         logger.debug { "Task #{task.name} routing: agent: #{chosen_agent}, #{capable.size} capable / #{targets.size - capable.size} incapable" }
 
         unless capable.empty?
-          capable.each { |target| callback&.call(type: :node_start, target: target) }
           arguments = unwrap_sensitive_args(arguments)
 
           results += case chosen_agent
@@ -148,12 +146,18 @@ def batch_task(targets, task, arguments, _options = {}, position = [], &callback
       # @param callback [Proc] Called with :node_start and :node_result events
       # @return [Array<Bolt::Result>] Results for all targets
       def batch_task_with(targets, task, target_mapping, options = {}, position = [], &callback)
-        # Pre-warm the agent cache so individual batch_task calls are cache hits
+        # Pre-warm the agent cache so individual batch_task calls are cache hits.
+        # A target that did not answer has no cache entry, and batch_task would
+        # discover it again and wait out the timeout again, so it gets its
+        # error here instead.
         configure_client(targets.first)
         discover_agents(targets)
+        discovered, silent = targets.partition { |target| discovered?(target) }
 
-        results = []
-        targets.each do |target|
+        silent_errors = silent.to_h { |target| [target, undiscovered_error(target)] }
+        silent.each { |target| callback&.call(type: :node_start, target: target) }
+        results = emit_results(silent_errors, action: 'task', name: task.name, position: position, &callback)
+        discovered.each do |target|
           results += batch_task([target], task, target_mapping[target], options, position, &callback)
         end
         results
@@ -169,15 +173,11 @@ def batch_connected?(targets)
         first_target = targets.first
         configure_client(first_target)
 
-        response = rpc_request('rpcutil', targets, 'rpcutil.ping') do |client|
+        # A target that denies the caller still answers, so it counts as connected.
+        response = rpc_request('rpcutil', targets, 'rpcutil.ping', accept_aborted: true) do |client|
           client.ping
         end
         response[:responded].length == targets.length
-      rescue StandardError => e
-        raise if e.is_a?(Bolt::Error)
-
-        logger.warn { "Batch connectivity check failed: #{e.class}: #{e.message}" }
-        false
       end
 
       def upload(_target, _source, _destination, _options = {}, _position = [])

lib/bolt/transport/choria/agent_discovery.rb

@@ -15,17 +15,16 @@ class Choria
       #
       # @param targets [Array<Bolt::Target>] Targets to discover agents on
       def discover_agents(targets)
-        uncached = targets.reject { |target| @agent_cache.key?(choria_identity(target)) }
+        uncached = targets.reject { |target| discovered?(target) }
         return if uncached.empty?
 
         logger.debug { "Discovering agents on #{target_count(uncached)}" }
         discover_agent_list(uncached)
         discover_os_family(uncached)
+      end
 
-        uncached.each do |target|
-          identity = choria_identity(target)
-          logger.warn { "No response from #{identity} during agent discovery" } unless @agent_cache.key?(identity)
-        end
+      def discovered?(target)
+        @agent_cache.key?(choria_identity(target))
       end
 
       def has_agent?(target, agent_name)
@@ -44,7 +43,7 @@ def discover_agent_list(targets)
         response = rpc_request('rpcutil', targets, 'rpcutil.agent_inventory') do |client|
           client.agent_inventory
         end
-        response[:errors].each { |target, output| logger.debug { "agent_inventory failed for #{target.safe_name}: #{output[:error]}" } }
+        response[:errors].each { |target, output| logger.warn { "agent_inventory failed for #{target.safe_name}: #{output[:error]}" } }
 
         response[:responded].each do |target, data|
           sender = choria_identity(target)
@@ -68,7 +67,7 @@ def discover_agent_list(targets)
       # @param targets [Array<Bolt::Target>] Targets to detect OS on
       def discover_os_family(targets)
         # Only fetch OS for targets that responded to agent_inventory
-        responded = targets.select { |target| @agent_cache.key?(choria_identity(target)) }
+        responded = targets.select { |target| discovered?(target) }
         return if responded.empty?
 
         response = rpc_request('rpcutil', responded, 'rpcutil.get_fact') do |client|

lib/bolt/transport/choria/bolt_tasks.rb

@@ -51,35 +51,34 @@ def run_task_via_bolt_tasks(targets, task, arguments, result_opts, &callback)
       # @return [Hash] with keys:
       #   - :failed_results [Array<Bolt::Result>] Error results from setup phase
       #   - :targets [Array<Bolt::Target>] Targets that successfully started
-      #   - :task_id [String, nil] Shared task ID for polling, nil if nothing started
+      #   - :task_id [String] Shared task ID for polling
+      #   When nothing started, only :failed_results is set.
       def download_and_start_task(targets, task, implementation, arguments, result_opts, &callback)
         environment = targets.first.options['puppet-environment']
         input_method = implementation['input_method']
         impl_files = [{ 'name' => File.basename(implementation['name']), 'path' => implementation['path'] }] +
-                     (implementation['files'] || [])
+                     implementation['files']
         file_specs_json = impl_files.map { |file| task_file_spec(file, task.module_name, environment) }.to_json
 
         # The failed_results reference will get updated and if we ever end up without
         # any targets left to act on, we can return it immediately.
         failed_results = []
-        none_started_result = { failed_results: failed_results, targets: [], task_id: nil }
+        none_started_result = { failed_results: failed_results }
 
         # Download task files
         logger.debug { "Downloading task #{task.name} files via bolt_tasks to #{target_count(targets)}" }
-        response = rpc_request('bolt_tasks', targets, 'bolt_tasks.download') do |client|
+        response = rpc_request('bolt_tasks', targets, 'bolt_tasks.download', accept_aborted: true) do |client|
           client.download(task: task.name, files: file_specs_json, environment: environment)
         end
         # The bolt_tasks agent uses reply.fail! with statuscode 1 for download
-        # failures, which rpc_request routes to :responded since statuscode 0-1
-        # means the action completed. Check rpc_statuscodes to catch these and
+        # failures, which rpc_request keeps under :responded here since the
+        # call accepts aborted replies. Check :aborted to catch these and
         # report the download failure clearly instead of letting run_no_wait
         # fail with a confusing "task not available" error.
         dl_errors = response[:errors]
-        response[:rpc_statuscodes].each do |target, code|
-          next if code.zero? || dl_errors.key?(target)
-
+        response[:aborted].each do |target, statusmsg|
           dl_errors[target] = error_output(
-            "bolt_tasks.download on #{target.safe_name} failed to download task files",
+            "bolt_tasks.download on #{target.safe_name} failed to download task files: #{statusmsg}",
             'bolt/choria-download-failed'
           )
         end
@@ -90,18 +89,29 @@ def download_and_start_task(targets, task, implementation, arguments, result_opt
 
         # Start task execution
         logger.debug { "Starting task #{task.name} on #{target_count(remaining)}" }
-        response = rpc_request('bolt_tasks', remaining, 'bolt_tasks.run_no_wait') do |client|
+        response = rpc_request('bolt_tasks', remaining, 'bolt_tasks.run_no_wait', accept_aborted: true) do |client|
           client.run_no_wait(task: task.name, input_method: input_method,
                              files: file_specs_json, input: arguments.to_json)
         end
-        failed_results.concat(emit_results(response[:errors], **result_opts, &callback))
-        return none_started_result if response[:responded].empty?
+        # A reply with statuscode 1 lands in :responded like a download
+        # failure does, so it is moved to the errors the same way. Otherwise
+        # the target would be polled for a task that never ran, or its reply
+        # without a task_id would stand in for the whole group.
+        start_errors = response[:errors]
+        response[:aborted].each do |target, statusmsg|
+          start_errors[target] = error_output(
+            "bolt_tasks.run_no_wait on #{target.safe_name} failed to start the task: #{statusmsg}",
+            'bolt/choria-task-start-failed'
+          )
+        end
+        failed_results.concat(emit_results(start_errors, **result_opts, &callback))
+        started = response[:responded].except(*start_errors.keys)
 
         # Extract the shared task_id (all targets get the same one from
         # the single run_no_wait call that fanned out to all of them)
-        task_id = response[:responded].values.first&.dig(:task_id)
+        task_id = started.values.first&.dig(:task_id)
         unless task_id
-          no_id_errors = response[:responded].each_with_object({}) do |(target, _), errors|
+          no_id_errors = started.each_with_object({}) do |(target, _), errors|
             errors[target] = error_output(
               "bolt_tasks.run_no_wait on #{target.safe_name} succeeded but returned no task_id",
               'bolt/choria-missing-task-id'
@@ -111,8 +121,8 @@ def download_and_start_task(targets, task, implementation, arguments, result_opt
           return none_started_result
         end
 
-        logger.debug { "Started task #{task.name} on #{target_count(response[:responded])}, task_id: #{task_id}" }
-        { failed_results: failed_results, targets: response[:responded].keys, task_id: task_id }
+        logger.debug { "Started task #{task.name} on #{target_count(started)}, task_id: #{task_id}" }
+        { failed_results: failed_results, targets: started.keys, task_id: task_id }
       end
 
       # Poll bolt_tasks.task_status until all targets complete or timeout.
@@ -125,12 +135,14 @@ def poll_task_status(targets, task_id, task)
         timeout = targets.first.options['task-timeout']
 
         poll_result = poll_with_retries(targets, timeout, 'bolt_tasks.task_status') do |remaining|
-          response = rpc_request('bolt_tasks', remaining, 'bolt_tasks.task_status') do |client|
+          # The agent answers statuscode 1 for a task that ran and failed,
+          # with the exit code and output in the data.
+          response = rpc_request('bolt_tasks', remaining, 'bolt_tasks.task_status', accept_aborted: true) do |client|
             client.task_status(task_id: task_id)
           end
-          next { rpc_failed: true, done: {} } if response[:rpc_failed]
+          next { rpc_failed: true } if response[:rpc_failed]
 
-          done = response[:errors].dup
+          done = response[:errors]
 
           response[:responded].each do |target, data|
             if data.nil?
@@ -149,14 +161,8 @@ def poll_task_status(targets, task_id, task)
         end
 
         remaining_errors = poll_result[:remaining].each_with_object({}) do |target, errors|
-          errors[target] =
-            if poll_result[:rpc_persistent_failure]
-              error_output("RPC requests to poll task status on #{target.safe_name} failed persistently",
-                           'bolt/choria-poll-failed')
-            else
-              error_output("Task #{task.name} timed out after #{timeout} seconds on #{target.safe_name}",
-                           'bolt/choria-task-timeout')
-            end
+          errors[target] = error_output("Task #{task.name} timed out after #{timeout} seconds on #{target.safe_name}",
+                                        'bolt/choria-task-timeout')
         end
 
         poll_result[:completed].merge(remaining_errors)
@@ -168,7 +174,7 @@ def poll_task_status(targets, task_id, task)
       # @param target [Bolt::Target] Target for logging and stdout unwrapping context
       # @return [Hash] Output hash from output() or error_output()
       def extract_task_output(data, target)
-        exitcode = exitcode_from(data, target, 'task')
+        exitcode = exitcode_from(data[:exitcode], target, 'task')
         output(stdout: unwrap_bolt_tasks_stdout(data[:stdout]),
                stderr: data[:stderr], exitcode: exitcode)
       end